"""Version and checkpoint the tamper-evident audit chain. Revision ID: 20260830_0024 Revises: 20260830_0023 """ from __future__ import annotations import hashlib import json import re import uuid from datetime import UTC, datetime from typing import Any import sqlalchemy as sa from sqlalchemy.engine import Connection from alembic import op revision = "20260830_0024" down_revision = "20260830_0023" branch_labels = None depends_on = None _LEGACY_PREFIX_DOMAIN = b"modelforge:audit:legacy-prefix:v1\n" _SHA256 = re.compile(r"^[0-9a-f]{64}$") _AUDIT_CHAIN_LOCK_KEY = int.from_bytes(b"MF_AUDIT", byteorder="big", signed=False) _AUDIT_OWNER_ROLE = "modelforge" _AUDIT_RUNTIME_ROLE = "modelforge_runtime" _POSTGRES_AUDIT_BOUNDARY_SQL = r""" create schema if not exists modelforge_audit authorization modelforge; alter schema modelforge_audit owner to modelforge; revoke all on schema modelforge_audit from public; revoke create on schema public from modelforge_runtime; grant usage on schema public, modelforge_audit to modelforge_runtime; create or replace function modelforge_audit.enforce_owner_mutation() returns trigger language plpgsql security invoker set search_path = pg_catalog as $guard$ begin if current_user <> 'modelforge' then raise exception 'audit tables are writable only through the canonical append function' using errcode = '42501'; end if; if tg_op = 'DELETE' then return old; elsif tg_op = 'TRUNCATE' then return null; end if; return new; end $guard$; alter function modelforge_audit.enforce_owner_mutation() owner to modelforge; revoke all on function modelforge_audit.enforce_owner_mutation() from public; revoke all on function modelforge_audit.enforce_owner_mutation() from modelforge_runtime; drop trigger if exists trg_modelforge_audit_events_owner on public.audit_events; create trigger trg_modelforge_audit_events_owner before insert or update or delete on public.audit_events for each row execute function modelforge_audit.enforce_owner_mutation(); drop trigger if exists trg_modelforge_audit_events_truncate_owner on public.audit_events; create trigger trg_modelforge_audit_events_truncate_owner before truncate on public.audit_events for each statement execute function modelforge_audit.enforce_owner_mutation(); drop trigger if exists trg_modelforge_audit_head_owner on public.audit_chain_heads; create trigger trg_modelforge_audit_head_owner before insert or update or delete on public.audit_chain_heads for each row execute function modelforge_audit.enforce_owner_mutation(); drop trigger if exists trg_modelforge_audit_head_truncate_owner on public.audit_chain_heads; create trigger trg_modelforge_audit_head_truncate_owner before truncate on public.audit_chain_heads for each statement execute function modelforge_audit.enforce_owner_mutation(); create or replace function modelforge_audit.append_event_v2( p_event_id uuid, p_occurred_at timestamptz, p_correlation_id text, p_actor_type text, p_actor_id text, p_action text, p_resource_type text, p_resource_id text, p_outcome text, p_details jsonb, p_expected_event_count bigint, p_expected_last_sequence bigint, p_expected_last_event_hash text, p_expected_hash_format text, p_expected_v2_start_sequence bigint, p_expected_legacy_prefix_count bigint, p_expected_legacy_prefix_seal text ) returns table(event_id uuid, sequence bigint, event_hash text, occurred_at timestamptz) language plpgsql security definer set search_path = pg_catalog as $append$ declare v_head public.audit_chain_heads%rowtype; v_tail public.audit_events%rowtype; v_max_sequence bigint; v_predecessor_hash text; v_sequence bigint; v_payload text; v_hash text; v_updated bigint; begin perform pg_catalog.pg_advisory_xact_lock(5568242723498248532); if p_event_id is null or p_occurred_at is null or p_correlation_id is null or p_actor_type is null or p_actor_id is null or p_action is null or p_resource_type is null or p_outcome is null or p_details is null then raise exception 'canonical audit append arguments must not be null' using errcode = '23502'; end if; if not pg_catalog.isfinite(p_occurred_at) then raise exception 'canonical audit occurred_at must be finite' using errcode = '22008'; end if; if pg_catalog.jsonb_typeof(p_details) <> 'object' then raise exception 'canonical audit details must be a JSON object' using errcode = '22023'; end if; select head.* into v_head from public.audit_chain_heads as head where head.singleton_id = 1 for update; if not found then raise exception 'audit checkpoint is missing' using errcode = '23514'; end if; if v_head.event_count is distinct from p_expected_event_count or v_head.last_sequence is distinct from p_expected_last_sequence or v_head.last_event_hash is distinct from p_expected_last_event_hash or v_head.hash_format is distinct from p_expected_hash_format or v_head.v2_start_sequence is distinct from p_expected_v2_start_sequence or v_head.legacy_prefix_count is distinct from p_expected_legacy_prefix_count or v_head.legacy_prefix_seal is distinct from p_expected_legacy_prefix_seal then raise exception 'audit checkpoint changed before canonical append' using errcode = '40001'; end if; if v_head.hash_format <> 'v2' or v_head.event_count <> v_head.last_sequence or v_head.event_count < 0 or v_head.v2_start_sequence < 1 or v_head.legacy_prefix_count <> v_head.v2_start_sequence - 1 or v_head.legacy_prefix_count > v_head.event_count or v_head.legacy_prefix_seal !~ '^[0-9a-f]{64}$' then raise exception 'audit checkpoint invariants are invalid' using errcode = '23514'; end if; select events.sequence into v_max_sequence from public.audit_events as events order by events.sequence desc, events.id desc limit 1; if v_head.event_count = 0 then if v_max_sequence is not null or v_head.last_event_hash is not null then raise exception 'empty audit checkpoint has retained events' using errcode = '23514'; end if; else if v_max_sequence is distinct from v_head.last_sequence or v_head.last_event_hash is null then raise exception 'audit checkpoint does not identify the retained tail' using errcode = '23514'; end if; select events.* into v_tail from public.audit_events as events where events.sequence = v_head.last_sequence; if not found or v_tail.event_hash is distinct from v_head.last_event_hash or v_tail.event_hash !~ '^[0-9a-f]{64}$' then raise exception 'audit retained tail is missing or does not match the checkpoint' using errcode = '23514'; end if; if v_tail.hash_format = 'v2' then if v_tail.canonical_payload is null or pg_catalog.encode( pg_catalog.sha256(pg_catalog.convert_to(v_tail.canonical_payload, 'UTF8')), 'hex' ) <> v_tail.event_hash or v_tail.canonical_payload::jsonb is distinct from pg_catalog.jsonb_build_object( 'hash_format', 'v2', 'id', v_tail.id::text, 'occurred_at', pg_catalog.to_char( v_tail.occurred_at at time zone 'UTC', 'YYYY-MM-DD"T"HH24:MI:SS.US"Z"' ), 'correlation_id', v_tail.correlation_id, 'actor_type', v_tail.actor_type, 'actor_id', v_tail.actor_id, 'action', v_tail.action, 'resource_type', v_tail.resource_type, 'resource_id', v_tail.resource_id, 'outcome', v_tail.outcome, 'details', v_tail.details::jsonb, 'previous_event_hash', v_tail.previous_event_hash ) then raise exception 'v2 audit retained tail payload is invalid' using errcode = '23514'; end if; elsif v_tail.hash_format <> 'v1' or v_tail.sequence <> v_head.v2_start_sequence - 1 then raise exception 'audit retained tail hash format is invalid' using errcode = '23514'; end if; if v_tail.sequence = 1 then if v_tail.previous_event_hash is not null then raise exception 'first audit event has a previous hash' using errcode = '23514'; end if; else select events.event_hash into v_predecessor_hash from public.audit_events as events where events.sequence = v_tail.sequence - 1; if not found or v_tail.previous_event_hash is distinct from v_predecessor_hash then raise exception 'audit retained tail link is invalid' using errcode = '23514'; end if; end if; end if; v_sequence := v_head.last_sequence + 1; v_payload := pg_catalog.jsonb_build_object( 'hash_format', 'v2', 'id', p_event_id::text, 'occurred_at', pg_catalog.to_char( p_occurred_at at time zone 'UTC', 'YYYY-MM-DD"T"HH24:MI:SS.US"Z"' ), 'correlation_id', p_correlation_id, 'actor_type', p_actor_type, 'actor_id', p_actor_id, 'action', p_action, 'resource_type', p_resource_type, 'resource_id', p_resource_id, 'outcome', p_outcome, 'details', p_details, 'previous_event_hash', v_head.last_event_hash )::text; v_hash := pg_catalog.encode( pg_catalog.sha256(pg_catalog.convert_to(v_payload, 'UTF8')), 'hex' ); insert into public.audit_events ( id, sequence, occurred_at, correlation_id, actor_type, actor_id, action, resource_type, resource_id, outcome, details, previous_event_hash, event_hash, hash_format, canonical_payload ) values ( p_event_id, v_sequence, p_occurred_at, p_correlation_id, p_actor_type, p_actor_id, p_action, p_resource_type, p_resource_id, p_outcome, p_details, v_head.last_event_hash, v_hash, 'v2', v_payload ); update public.audit_chain_heads as head set event_count = v_head.event_count + 1, last_sequence = v_sequence, last_event_hash = v_hash, updated_at = p_occurred_at where head.singleton_id = 1 and head.event_count = v_head.event_count and head.last_sequence = v_head.last_sequence and head.last_event_hash is not distinct from v_head.last_event_hash and head.hash_format = v_head.hash_format and head.v2_start_sequence = v_head.v2_start_sequence and head.legacy_prefix_count = v_head.legacy_prefix_count and head.legacy_prefix_seal = v_head.legacy_prefix_seal; get diagnostics v_updated = row_count; if v_updated <> 1 then raise exception 'audit checkpoint compare-and-set failed' using errcode = '40001'; end if; return query select p_event_id, v_sequence, v_hash, p_occurred_at; end $append$; alter function modelforge_audit.append_event_v2( uuid, timestamptz, text, text, text, text, text, text, text, jsonb, bigint, bigint, text, text, bigint, bigint, text ) owner to modelforge; revoke all on function modelforge_audit.append_event_v2( uuid, timestamptz, text, text, text, text, text, text, text, jsonb, bigint, bigint, text, text, bigint, bigint, text ) from public; grant execute on function modelforge_audit.append_event_v2( uuid, timestamptz, text, text, text, text, text, text, text, jsonb, bigint, bigint, text, text, bigint, bigint, text ) to modelforge_runtime; grant select, insert, update, delete on all tables in schema public to modelforge_runtime; grant usage, select on all sequences in schema public to modelforge_runtime; revoke execute on all functions in schema public from public, modelforge_runtime; revoke insert, update, delete, truncate, references, trigger on public.audit_events, public.audit_chain_heads from modelforge_runtime; grant select on public.audit_events, public.audit_chain_heads to modelforge_runtime; alter default privileges for role modelforge in schema public grant select, insert, update, delete on tables to modelforge_runtime; alter default privileges for role modelforge in schema public grant usage, select on sequences to modelforge_runtime; alter default privileges for role modelforge in schema public revoke execute on functions from public; do $database_privileges$ begin execute pg_catalog.format( 'revoke create, temporary on database %I from modelforge_runtime', pg_catalog.current_database() ); end $database_privileges$; """ def _normalise_timestamp(value: datetime | str) -> str: if isinstance(value, datetime): moment = value elif isinstance(value, str): candidate = value.strip() if candidate.endswith("Z"): candidate = candidate[:-1] + "+00:00" try: moment = datetime.fromisoformat(candidate) except ValueError as error: raise RuntimeError("legacy audit event has an invalid occurred_at") from error else: raise RuntimeError("legacy audit event has an invalid occurred_at") if moment.tzinfo is None: moment = moment.replace(tzinfo=UTC) return moment.astimezone(UTC).isoformat(timespec="microseconds").replace("+00:00", "Z") def _details(value: Any) -> dict[str, Any]: if isinstance(value, str): try: value = json.loads(value) except json.JSONDecodeError as error: raise RuntimeError("legacy audit event details are not valid JSON") from error if not isinstance(value, dict): raise RuntimeError("legacy audit event details must be a JSON object") return value def _legacy_hash(row: sa.RowMapping) -> str: payload = { "correlation_id": row["correlation_id"], "actor_type": row["actor_type"], "actor_id": row["actor_id"], "action": row["action"], "resource_type": row["resource_type"], "resource_id": row["resource_id"], "outcome": row["outcome"], "details": _details(row["details"]), "previous_event_hash": row["previous_event_hash"], } encoded = json.dumps(payload, sort_keys=True, separators=(",", ":")) return hashlib.sha256(encoded.encode("utf-8")).hexdigest() def _prefix_entry(row: sa.RowMapping) -> bytes: try: event_id = str(uuid.UUID(str(row["id"]))) except (AttributeError, TypeError, ValueError) as error: raise RuntimeError("legacy audit event id is not a UUID") from error payload = { "sequence": int(row["sequence"]), "id": event_id, "occurred_at": _normalise_timestamp(row["occurred_at"]), "event_hash": str(row["event_hash"]), } return json.dumps(payload, sort_keys=True, separators=(",", ":")).encode("utf-8") + b"\n" def _validate_legacy_rows(rows: list[sa.RowMapping]) -> dict[str, Any]: """Validate the production-shaped v1 chain and return its immutable cutover state.""" previous_hash: str | None = None prefix = hashlib.sha256() prefix.update(_LEGACY_PREFIX_DOMAIN) for expected_sequence, row in enumerate(rows, start=1): try: sequence = int(row["sequence"]) except (TypeError, ValueError) as error: raise RuntimeError("legacy audit event sequence is not an integer") from error if sequence != expected_sequence: raise RuntimeError( f"legacy audit chain has sequence {sequence}; expected {expected_sequence}" ) event_hash = str(row["event_hash"]) if _SHA256.fullmatch(event_hash) is None: raise RuntimeError(f"legacy audit event {sequence} has a malformed event hash") if row["previous_event_hash"] != previous_hash: raise RuntimeError(f"legacy audit event {sequence} has an invalid previous hash") if _legacy_hash(row) != event_hash: raise RuntimeError(f"legacy audit event {sequence} content hash is invalid") prefix.update(_prefix_entry(row)) previous_hash = event_hash count = len(rows) return { "event_count": count, "last_sequence": count, "last_event_hash": previous_hash, "v2_start_sequence": count + 1, "legacy_prefix_count": count, "legacy_prefix_seal": prefix.hexdigest(), } def _read_and_validate_legacy_chain(connection: Connection) -> dict[str, Any]: rows = list( connection.execute( sa.text( "select id, sequence, occurred_at, correlation_id, actor_type, actor_id, " "action, resource_type, resource_id, outcome, details, previous_event_hash, " "event_hash from audit_events order by sequence, id" ) ).mappings() ) return _validate_legacy_rows(rows) def _lock_legacy_audit_chain(connection: Connection) -> None: """Serialize validation and checkpoint seed against every legacy writer. The shared advisory key coordinates with 0024-aware writers. ``ACCESS EXCLUSIVE`` also blocks pre-0024 applications, which do not know that key, until this migration transaction commits. SQLite is test-only; a no-op write upgrades its deferred transaction to a writer before the validation read so another test connection cannot append into the validation/seed window. """ dialect = connection.dialect.name if dialect == "postgresql": connection.execute( sa.text("select pg_advisory_xact_lock(:lock_key)"), {"lock_key": _AUDIT_CHAIN_LOCK_KEY}, ) connection.execute(sa.text("lock table audit_events in access exclusive mode")) return if dialect == "sqlite": connection.execute(sa.text("update audit_events set event_hash = event_hash where 1 = 0")) return raise RuntimeError(f"audit-chain migration does not support the {dialect!r} dialect") def _validate_postgres_role_preflight(connection: Connection) -> None: """Require the separately provisioned non-superuser owner/runtime roles before DDL. Existing 1.2.1 installations commonly made ``modelforge`` the cluster bootstrap superuser. That credential cannot be converted into the API boundary implicitly by an application migration. Operators must first run the documented admin-owned provisioning step; failure is deliberately before this migration changes a column or seeds a checkpoint. """ if connection.dialect.name != "postgresql": return roles = list( connection.execute( sa.text( "select rolname, rolsuper, rolinherit, rolcreaterole, rolcreatedb, " "rolcanlogin, rolreplication, rolbypassrls from pg_catalog.pg_roles " "where rolname in (:owner_role, :runtime_role) order by rolname" ), {"owner_role": _AUDIT_OWNER_ROLE, "runtime_role": _AUDIT_RUNTIME_ROLE}, ).mappings() ) by_name = {str(row["rolname"]): row for row in roles} if set(by_name) != {_AUDIT_OWNER_ROLE, _AUDIT_RUNTIME_ROLE}: raise RuntimeError( "audit migration preflight requires separately provisioned modelforge owner and " "modelforge_runtime roles; run the v1.2.1-to-schema-0024 role provisioning step" ) current_role = str(connection.scalar(sa.text("select current_user"))) if current_role != _AUDIT_OWNER_ROLE: raise RuntimeError( "audit migration must run with the non-superuser modelforge owner credential" ) session_role = str(connection.scalar(sa.text("select session_user"))) if session_role != _AUDIT_OWNER_ROLE: raise RuntimeError( "audit migration must authenticate directly as modelforge, not SET ROLE from admin" ) for role_name, require_noinherit in ( (_AUDIT_OWNER_ROLE, False), (_AUDIT_RUNTIME_ROLE, True), ): role = by_name[role_name] forbidden = any( bool(role[field]) for field in ( "rolsuper", "rolcreaterole", "rolcreatedb", "rolreplication", "rolbypassrls", ) ) if forbidden or not bool(role["rolcanlogin"]): raise RuntimeError(f"database role {role_name} has forbidden administrative powers") if require_noinherit and bool(role["rolinherit"]): raise RuntimeError("modelforge_runtime must be provisioned NOINHERIT") app_role_membership_count = int( connection.scalar( sa.text( "select count(*) from pg_catalog.pg_auth_members as membership " "join pg_catalog.pg_roles as member on member.oid = membership.member " "where member.rolname in (:runtime_role, :owner_role)" ), {"runtime_role": _AUDIT_RUNTIME_ROLE, "owner_role": _AUDIT_OWNER_ROLE}, ) or 0 ) if app_role_membership_count: raise RuntimeError( "modelforge and modelforge_runtime must have no SET ROLE-capable memberships" ) def _install_postgres_audit_boundary(connection: Connection) -> None: if connection.dialect.name == "postgresql": for statement in _postgres_sql_statements(_POSTGRES_AUDIT_BOUNDARY_SQL): _exec_postgres_sql(connection, statement) def _exec_postgres_sql(connection: Connection, statement: str) -> None: """Execute trusted static SQL without exposing PostgreSQL percent syntax to DBAPI parsing.""" paramstyle = getattr(connection.dialect, "paramstyle", None) driver_statement = ( statement.replace("%", "%%") if paramstyle in {"format", "pyformat"} else statement ) connection.exec_driver_sql(driver_statement) def _postgres_sql_statements(script: str) -> list[str]: """Split this migration's trusted static SQL without splitting function bodies.""" statements: list[str] = [] start = 0 index = 0 quote: str | None = None while index < len(script): if quote is not None: if quote == "'" and script.startswith("''", index): index += 2 continue if script.startswith(quote, index): index += len(quote) quote = None continue index += 1 continue character = script[index] if character == "'": quote = "'" index += 1 continue if character == "$": delimiter = re.match(r"\$[A-Za-z_][A-Za-z0-9_]*\$|\$\$", script[index:]) if delimiter is not None: quote = delimiter.group(0) index += len(quote) continue if character == ";": statement = script[start:index].strip() if statement: statements.append(statement) start = index + 1 index += 1 trailing = script[start:].strip() if quote is not None: raise RuntimeError("generated PostgreSQL audit boundary SQL has an unterminated literal") if trailing: statements.append(trailing) return statements def _remove_postgres_audit_boundary(connection: Connection) -> None: if connection.dialect.name != "postgresql": return script = ( "drop trigger if exists trg_modelforge_audit_events_owner on public.audit_events; " "drop trigger if exists trg_modelforge_audit_events_truncate_owner " "on public.audit_events; " "drop trigger if exists trg_modelforge_audit_head_owner on public.audit_chain_heads; " "drop trigger if exists trg_modelforge_audit_head_truncate_owner " "on public.audit_chain_heads; " "drop function if exists modelforge_audit.append_event_v2(" "uuid, timestamptz, text, text, text, text, text, text, text, jsonb, " "bigint, bigint, text, text, bigint, bigint, text); " "drop function if exists modelforge_audit.enforce_owner_mutation(); " "drop schema if exists modelforge_audit; " "grant select, insert on public.audit_events to modelforge_runtime" ) for statement in _postgres_sql_statements(script): _exec_postgres_sql(connection, statement) def upgrade() -> None: connection = op.get_bind() _validate_postgres_role_preflight(connection) _lock_legacy_audit_chain(connection) # Validation deliberately precedes every schema mutation. In particular, legacy recovery # markers written with a random hash stop the migration instead of being blessed by a seal. legacy = _read_and_validate_legacy_chain(connection) op.add_column( "audit_events", sa.Column("hash_format", sa.String(length=16), nullable=True), ) op.add_column( "audit_events", sa.Column("canonical_payload", sa.Text(), nullable=True), ) connection.execute(sa.text("update audit_events set hash_format = 'v1'")) with op.batch_alter_table("audit_events") as batch: batch.alter_column( "hash_format", existing_type=sa.String(length=16), nullable=False, ) batch.create_check_constraint( "ck_audit_event_hash_format", "hash_format IN ('v1', 'v2')" ) batch.create_check_constraint( "ck_audit_event_canonical_payload", "((hash_format = 'v1' AND canonical_payload IS NULL) OR " "(hash_format = 'v2' AND canonical_payload IS NOT NULL))", ) op.create_table( "audit_chain_heads", sa.Column("singleton_id", sa.Integer(), nullable=False), sa.Column("event_count", sa.BigInteger(), nullable=False), sa.Column("last_sequence", sa.BigInteger(), nullable=False), sa.Column("last_event_hash", sa.String(length=64), nullable=True), sa.Column("hash_format", sa.String(length=16), nullable=False), sa.Column("v2_start_sequence", sa.BigInteger(), nullable=False), sa.Column("legacy_prefix_count", sa.BigInteger(), nullable=False), sa.Column("legacy_prefix_seal", sa.String(length=64), nullable=False), sa.Column( "updated_at", sa.DateTime(timezone=True), server_default=sa.text("CURRENT_TIMESTAMP"), nullable=False, ), sa.CheckConstraint("singleton_id = 1", name="ck_audit_chain_head_singleton"), sa.CheckConstraint("event_count >= 0", name="ck_audit_chain_head_count"), sa.CheckConstraint("last_sequence >= 0", name="ck_audit_chain_head_sequence"), sa.CheckConstraint( "event_count = last_sequence", name="ck_audit_chain_head_count_sequence", ), sa.CheckConstraint("v2_start_sequence >= 1", name="ck_audit_chain_head_cutover"), sa.CheckConstraint( "legacy_prefix_count = v2_start_sequence - 1", name="ck_audit_chain_head_prefix_count", ), sa.CheckConstraint( "legacy_prefix_count <= event_count", name="ck_audit_chain_head_prefix_within_chain", ), sa.CheckConstraint("hash_format = 'v2'", name="ck_audit_chain_head_hash_format"), sa.CheckConstraint( "length(legacy_prefix_seal) = 64", name="ck_audit_chain_head_prefix_seal", ), sa.CheckConstraint( "((event_count = 0 AND last_sequence = 0 AND last_event_hash IS NULL) OR " "(event_count > 0 AND last_sequence > 0 AND last_event_hash IS NOT NULL))", name="ck_audit_chain_head_shape", ), sa.PrimaryKeyConstraint("singleton_id"), ) connection.execute( sa.text( "insert into audit_chain_heads (singleton_id, event_count, last_sequence, " "last_event_hash, hash_format, v2_start_sequence, legacy_prefix_count, " "legacy_prefix_seal) values (1, :event_count, :last_sequence, :last_event_hash, " "'v2', :v2_start_sequence, :legacy_prefix_count, :legacy_prefix_seal)" ), legacy, ) _install_postgres_audit_boundary(connection) def downgrade() -> None: connection = op.get_bind() _validate_postgres_role_preflight(connection) _lock_legacy_audit_chain(connection) non_legacy = int( connection.scalar( sa.text("select count(*) from audit_events where hash_format <> 'v1'") ) or 0 ) if non_legacy: raise RuntimeError( "cannot downgrade audit hash format after v2 events exist without rewriting history" ) legacy = _read_and_validate_legacy_chain(connection) head = connection.execute( sa.text( "select event_count, last_sequence, last_event_hash, hash_format, " "v2_start_sequence, legacy_prefix_count, legacy_prefix_seal " "from audit_chain_heads where singleton_id = 1" ) ).mappings().one_or_none() if head is None or head["hash_format"] != "v2": raise RuntimeError("cannot downgrade a missing or malformed audit checkpoint") for key, expected in legacy.items(): if head[key] != expected: raise RuntimeError(f"cannot downgrade: audit checkpoint {key} is inconsistent") _remove_postgres_audit_boundary(connection) op.drop_table("audit_chain_heads") with op.batch_alter_table("audit_events") as batch: batch.drop_constraint("ck_audit_event_canonical_payload", type_="check") batch.drop_constraint("ck_audit_event_hash_format", type_="check") batch.drop_column("canonical_payload") batch.drop_column("hash_format")