diff --git a/alembic/versions/20261028_08c7cf36ddbe_v141_traceability_events_append_only.py b/alembic/versions/20261028_08c7cf36ddbe_v141_traceability_events_append_only.py new file mode 100644 index 000000000..ea4b236c4 --- /dev/null +++ b/alembic/versions/20261028_08c7cf36ddbe_v141_traceability_events_append_only.py @@ -0,0 +1,198 @@ +"""v141 — fsma.traceability_events mutation policy (column allowlist trigger). + +``fsma.traceability_events`` is the canonical FSMA 204 evidence table and +the chain hash is computed over its payload columns, yet until now nothing +in the database stopped an UPDATE or DELETE on it. ``fsma.cte_events`` has +had an append-only trigger since v107; the canonical table had none, so a +``UPDATE fsma.traceability_events SET quantity = ...`` from any role with +UPDATE privilege silently broke the evidence the FDA package is built on. + +This revision installs a BEFORE UPDATE OR DELETE row trigger plus a BEFORE +TRUNCATE statement trigger that together enforce the sanctioned mutation +surface, which is exactly what the three production writers do today: + +* ``services/shared/canonical_persistence/writer.py`` marks the row an + amendment supersedes: ``status = 'superseded', amended_at = NOW()``. +* ``services/shared/rules/engine.py`` records the event-level verdict: + ``evaluation_verdict``, ``no_verdict_reason``, ``evaluated_at``, and the + evaluation error marker: ``evaluation_error``, ``evaluated_at``. + +Allowed on UPDATE: changes confined to the allowlisted columns +(``status``, ``amended_at``, ``evaluation_verdict``, ``no_verdict_reason``, +``evaluation_error``, ``evaluated_at``), and ``status`` may only move from +``active`` to ``superseded``. Any other column change, any other status +transition, every DELETE and every TRUNCATE raise ``check_violation``. + +Break-glass: ``SET LOCAL fsma.allow_mutation = 'true'`` lets the statement +through with a WARNING, the same GUC v095 and v107 use. The runtime resets +it per transaction (``services/shared/tenant_context.py``) and per pooled +connection (``services/admin/app/database.py``); the retention purge is +its only production user. + +The comparison is ``to_jsonb(NEW) - allowlist = to_jsonb(OLD) - allowlist`` +rather than a list of frozen column names, so the trigger is correct on +every catalog variant of the table (the Supabase twin types the tenant +column as text and has no ``ingested_at``) and keeps protecting columns +added later. + +Idempotent: ``CREATE OR REPLACE`` functions, ``DROP TRIGGER IF EXISTS`` +before ``CREATE TRIGGER``, ``to_regclass`` guard. No grant changes (v111 +still grants UPDATE to the runtime role; the trigger is the enforcement). + +Revision ID: 08c7cf36ddbe +Revises: b8c1e4f2a901 +Create Date: 2026-10-28 +""" + +from typing import Sequence, Union + +from alembic import op + +revision: str = "08c7cf36ddbe" # pragma: allowlist secret +down_revision: Union[str, Sequence[str], None] = "b8c1e4f2a901" # pragma: allowlist secret +branch_labels = None +depends_on = None + +TABLE = "fsma.traceability_events" +ROW_TRIGGER = "traceability_events_mutation_policy" +TRUNCATE_TRIGGER = "traceability_events_no_truncate" +ROW_FUNCTION = "fsma.prevent_traceability_event_mutation" +TRUNCATE_FUNCTION = "fsma.prevent_traceability_event_truncate" + +# Columns the sanctioned writers may change. Keep in step with the module +# docstring and docs/engineering/SCHEMA_CHANGE_POLICY.md §5. +ALLOWLIST: tuple[str, ...] = ( + "status", + "amended_at", + "evaluation_verdict", + "no_verdict_reason", + "evaluation_error", + "evaluated_at", +) + +_ALLOWLIST_SQL = ", ".join(f"'{c}'" for c in ALLOWLIST) + +_ROW_FUNCTION_SQL = f""" +CREATE OR REPLACE FUNCTION {ROW_FUNCTION}() +RETURNS TRIGGER LANGUAGE plpgsql AS $$ +DECLARE + allowlist text[] := ARRAY[{_ALLOWLIST_SQL}]; + changed text[]; +BEGIN + IF current_setting('fsma.allow_mutation', true) = 'true' THEN + RAISE WARNING + 'traceability_events modification via break-glass. op=% user=% event_id=%', + TG_OP, current_user, COALESCE(NEW.event_id::text, OLD.event_id::text); + IF TG_OP = 'UPDATE' THEN + RETURN NEW; + END IF; + RETURN OLD; + END IF; + + IF TG_OP = 'DELETE' THEN + RAISE EXCEPTION + 'traceability_events is append-only (FDA 21 CFR 1.1455). DELETE rejected for event_id=%. Set fsma.allow_mutation=true for authorized break-glass.', + OLD.event_id + USING ERRCODE = 'check_violation'; + END IF; + + -- UPDATE: every changed column must be on the allowlist. + SELECT COALESCE(array_agg(key ORDER BY key), ARRAY[]::text[]) + INTO changed + FROM ( + SELECT key FROM jsonb_each(to_jsonb(NEW) - allowlist) + EXCEPT + SELECT key FROM jsonb_each(to_jsonb(OLD) - allowlist) + UNION + SELECT n.key + FROM jsonb_each(to_jsonb(NEW) - allowlist) n + JOIN jsonb_each(to_jsonb(OLD) - allowlist) o ON o.key = n.key + WHERE n.value IS DISTINCT FROM o.value + ) diff; + IF array_length(changed, 1) IS NOT NULL THEN + RAISE EXCEPTION + 'traceability_events is append-only (FDA 21 CFR 1.1455). UPDATE rejected for event_id=%: column(s) % are not amendable. Corrections are new events that supersede the old one. Set fsma.allow_mutation=true for authorized break-glass.', + OLD.event_id, array_to_string(changed, ', ') + USING ERRCODE = 'check_violation'; + END IF; + + -- status may only move active -> superseded (an amendment landed). + -- IS NOT TRUE, not NOT (...): the Supabase twin leaves status nullable, + -- and a NULL on either side would turn NOT (...) into NULL and let the + -- change through. + IF NEW.status IS DISTINCT FROM OLD.status + AND (OLD.status = 'active' AND NEW.status = 'superseded') IS NOT TRUE THEN + RAISE EXCEPTION + 'traceability_events status may only move from active to superseded; % -> % rejected for event_id=%. Set fsma.allow_mutation=true for authorized break-glass.', + OLD.status, NEW.status, OLD.event_id + USING ERRCODE = 'check_violation'; + END IF; + + RETURN NEW; +END; +$$; +""" + +_TRUNCATE_FUNCTION_SQL = f""" +CREATE OR REPLACE FUNCTION {TRUNCATE_FUNCTION}() +RETURNS TRIGGER LANGUAGE plpgsql AS $$ +BEGIN + IF current_setting('fsma.allow_mutation', true) = 'true' THEN + RAISE WARNING 'traceability_events TRUNCATE via break-glass. user=%', current_user; + RETURN NULL; + END IF; + RAISE EXCEPTION + 'traceability_events is append-only (FDA 21 CFR 1.1455). TRUNCATE rejected. Set fsma.allow_mutation=true for authorized break-glass.' + USING ERRCODE = 'check_violation'; +END; +$$; +""" + +_TRIGGERS_SQL = f""" +DO $$ +BEGIN + IF to_regclass('{TABLE}') IS NULL THEN + RETURN; + END IF; + DROP TRIGGER IF EXISTS {ROW_TRIGGER} ON {TABLE}; + CREATE TRIGGER {ROW_TRIGGER} + BEFORE UPDATE OR DELETE ON {TABLE} + FOR EACH ROW EXECUTE FUNCTION {ROW_FUNCTION}(); + DROP TRIGGER IF EXISTS {TRUNCATE_TRIGGER} ON {TABLE}; + CREATE TRIGGER {TRUNCATE_TRIGGER} + BEFORE TRUNCATE ON {TABLE} + FOR EACH STATEMENT EXECUTE FUNCTION {TRUNCATE_FUNCTION}(); +END $$; +""" + +_DOWNGRADE_SQL = f""" +DO $$ +BEGIN + IF to_regclass('{TABLE}') IS NOT NULL THEN + DROP TRIGGER IF EXISTS {ROW_TRIGGER} ON {TABLE}; + DROP TRIGGER IF EXISTS {TRUNCATE_TRIGGER} ON {TABLE}; + END IF; +END $$; +DROP FUNCTION IF EXISTS {ROW_FUNCTION}(); +DROP FUNCTION IF EXISTS {TRUNCATE_FUNCTION}(); +""" + + +# CREATE/DROP TRIGGER takes SHARE ROW EXCLUSIVE on the table and would queue +# behind a long ingest or export transaction, holding every later writer +# behind it. Bound the wait (same 10s as v134) so the migration fails fast +# and can be retried instead of stalling production. Alembic runs each +# path in one transaction, so SET LOCAL covers every statement below. +_LOCK_TIMEOUT_SQL = "SET LOCAL lock_timeout = '10s'" + + +def upgrade() -> None: + op.execute(_LOCK_TIMEOUT_SQL) + op.execute(_ROW_FUNCTION_SQL) + op.execute(_TRUNCATE_FUNCTION_SQL) + op.execute(_TRIGGERS_SQL) + + +def downgrade() -> None: + op.execute(_LOCK_TIMEOUT_SQL) + op.execute(_DOWNGRADE_SQL) diff --git a/docs/engineering/SCHEMA_CHANGE_POLICY.md b/docs/engineering/SCHEMA_CHANGE_POLICY.md index abe8166da..ae02ab166 100644 --- a/docs/engineering/SCHEMA_CHANGE_POLICY.md +++ b/docs/engineering/SCHEMA_CHANGE_POLICY.md @@ -89,6 +89,37 @@ Engineering must implement. Reviewers must enforce. | Every verdict records `rule_version_id`, `fact_version_ids`, `authority_pointer_ids` | NOT NULL + FK constraints | | Missing data → `INDETERMINATE` | Application-level (never NULL verdicts) | | Corrections are new versions | No UPDATE on verdict tables (trigger-enforced) | +| `fsma.traceability_events` is append-only | v141 trigger (`traceability_events_mutation_policy`, `traceability_events_no_truncate`); see 5.1 | +| Legacy CTE tables are read-only retention | No new writers, columns or indexes on `fsma.cte_events` and `fsma.critical_tracking_events`; see 5.2 | + +### 5.1 Sanctioned mutations of `fsma.traceability_events` + +The canonical evidence table rejects every UPDATE, DELETE and TRUNCATE at the database (migration +v141, `08c7cf36ddbe`) except the two writers the application already relies on: + +| Writer | Columns it may change | Constraint | +|--------|-----------------------|------------| +| Amendment (`services/shared/canonical_persistence/writer.py`) | `status`, `amended_at` | `status` may only move `active -> superseded` | +| Rules engine (`services/shared/rules/engine.py`) | `evaluation_verdict`, `no_verdict_reason`, `evaluation_error`, `evaluated_at` | none | + +Any statement that changes another column, moves `status` anywhere else, deletes or truncates +raises `check_violation`. A correction is a new event that supersedes the old one, never an edit. +The break-glass is `SET LOCAL fsma.allow_mutation = 'true'` inside one transaction (the same GUC +as `fsma.cte_events` and `fsma.correction_patches`); the runtime resets it per transaction and per +pooled connection, and the retention purge is its only production user. Adding a column to the +allowlist is a schema change under this policy: update the trigger in a new migration, this table, +and `tests/migrations/test_v141_traceability_events_append_only.py` together. + +### 5.2 Read freeze on the legacy CTE tables (SEL-589) + +`fsma.traceability_events` is the committed canonical contract for all new FSMA evidence (product +decision 2026-10-27, `docs/product/PRODUCT_PRD_2026-10.md`). `fsma.cte_events` and +`fsma.critical_tracking_events` are read-only retention: no new writers, columns or indexes. Their +reads are not frozen yet: the v1 FDA export and recall readers still default to `fsma.cte_events` +behind `FDA_EXPORT_V1_READ_CANONICAL` (default off, `services/ingestion/app/fda_export/queries.py`). +Each export, recall, trace and verification reader is cut over one at a time, only after its +historical-row and canonical coverage is documented. A migration that adds to either legacy table +is rejected under section 7. --- @@ -115,6 +146,6 @@ Engineering must implement. Reviewers must enforce. ## Appendix: Current Schema Version -- **Version:** 1.1.2 -- **Last Updated:** 2026-01-22 +- **Version:** 1.2.0 +- **Last Updated:** 2026-10-28 (v141 mutation policy on `fsma.traceability_events`; SEL-589 read freeze) - **Git SHA:** (auto-populated by migration) diff --git a/services/ingestion/app/exception_router.py b/services/ingestion/app/exception_router.py index e2288fd6a..a328b4448 100644 --- a/services/ingestion/app/exception_router.py +++ b/services/ingestion/app/exception_router.py @@ -88,8 +88,13 @@ class ResolveRequest(BaseModel): class WaiveRequest(BaseModel): + """Waiver body. The approver is the authenticated principal, never a body field. + + A client-supplied ``waiver_approved_by`` is ignored (pydantic drops + unknown fields) so a caller cannot sign a waiver in someone else's name. + """ + waiver_reason: str - waiver_approved_by: str class AddCommentRequest(BaseModel): @@ -237,9 +242,23 @@ async def resolve_exception( return {"case_id": case_id, "status": "resolved"} +def _actor_from_principal(principal: IngestionPrincipal) -> str: + """Identity string recorded as the actor of a signoff. + + Mirrors ``_audit_identity`` in the FDA export router: identity fields + resolve from the authenticated principal, never from the request body. + ``key_id`` is ``session:`` for browser sessions and the API key + id for scoped keys. An empty principal cannot approve anything. + """ + actor = (getattr(principal, "key_id", None) or "").strip() + if not actor: + raise HTTPException(status_code=401, detail="Authenticated principal required to approve a waiver") + return actor + + @router.patch( "/{case_id}/waive", - summary="Waive exception (requires reason + approver)", + summary="Waive exception (requires reason; approver is the authenticated principal)", ) async def waive_exception( case_id: str, @@ -250,7 +269,7 @@ async def waive_exception( ): tid = resolve_tenant(tenant_id, principal) svc = _get_service(db_session) - svc.waive_exception(tid, case_id, body.waiver_reason, body.waiver_approved_by) + svc.waive_exception(tid, case_id, body.waiver_reason, _actor_from_principal(principal)) return {"case_id": case_id, "status": "waived"} diff --git a/services/ingestion/tests/test_auditor_pagination_ties_pg.py b/services/ingestion/tests/test_auditor_pagination_ties_pg.py index 78ab66871..b255efa62 100644 --- a/services/ingestion/tests/test_auditor_pagination_ties_pg.py +++ b/services/ingestion/tests/test_auditor_pagination_ties_pg.py @@ -139,6 +139,7 @@ def seeded_tenant(): yield engine, tenant_id, set(lot_codes) finally: with engine.begin() as conn: + conn.execute(text("SET LOCAL fsma.allow_mutation = 'true'")) # v141 append-only conn.execute( text("DELETE FROM fsma.traceability_events WHERE tenant_id = :tid"), {"tid": tenant_id}, diff --git a/services/ingestion/tests/test_disaster_recovery_data_volume_pg.py b/services/ingestion/tests/test_disaster_recovery_data_volume_pg.py index 9b57af55b..6496aef99 100644 --- a/services/ingestion/tests/test_disaster_recovery_data_volume_pg.py +++ b/services/ingestion/tests/test_disaster_recovery_data_volume_pg.py @@ -98,6 +98,7 @@ def seeded_tenants(): yield engine, tenant_id finally: with engine.begin() as conn: + conn.execute(text("SET LOCAL fsma.allow_mutation = 'true'")) # v141 append-only conn.execute( text("DELETE FROM fsma.traceability_events WHERE tenant_id IN (:tid, :other)"), {"tid": tenant_id, "other": other_tenant_id}, diff --git a/services/ingestion/tests/test_exception_router.py b/services/ingestion/tests/test_exception_router.py index 673e49484..0c2fa9c0d 100644 --- a/services/ingestion/tests/test_exception_router.py +++ b/services/ingestion/tests/test_exception_router.py @@ -236,13 +236,11 @@ def test_resolve_exception(client, mock_svc): # --------------------------------------------------------------------------- -def test_waive_exception(client, mock_svc): +def test_waive_exception_records_principal_as_approver(client, mock_svc): + """S5: the approver is the authenticated principal's key_id, not a body field.""" resp = client.patch( "/api/v1/exceptions/exc-1/waive", - json={ - "waiver_reason": "Low-risk supplier", - "waiver_approved_by": "admin-1", - }, + json={"waiver_reason": "Low-risk supplier"}, ) assert resp.status_code == 200 assert resp.json()["status"] == "waived" @@ -250,8 +248,40 @@ def test_waive_exception(client, mock_svc): TENANT, "exc-1", "Low-risk supplier", - "admin-1", + "test-key", + ) + + +def test_waive_exception_ignores_body_supplied_approver(client, mock_svc): + """A caller cannot sign a waiver in someone else's name.""" + resp = client.patch( + "/api/v1/exceptions/exc-1/waive", + json={ + "waiver_reason": "Low-risk supplier", + "waiver_approved_by": "someone-else", + }, + ) + assert resp.status_code == 200 + assert mock_svc.waive_exception.call_args.args[3] == "test-key" + + +def test_waive_exception_requires_reason(client, mock_svc): + resp = client.patch("/api/v1/exceptions/exc-1/waive", json={}) + assert resp.status_code == 422 + mock_svc.waive_exception.assert_not_called() + + +def test_waive_exception_rejects_empty_principal_identity(): + principal = _make_principal() + principal = principal.model_copy(update={"key_id": ""}) + svc = MagicMock() + client = _build_client(principal, svc) + resp = client.patch( + "/api/v1/exceptions/exc-1/waive", + json={"waiver_reason": "Low-risk supplier"}, ) + assert resp.status_code == 401 + svc.waive_exception.assert_not_called() # --------------------------------------------------------------------------- diff --git a/services/ingestion/tests/test_product_catalog_derive_events_pg.py b/services/ingestion/tests/test_product_catalog_derive_events_pg.py index d359f4382..395535e68 100644 --- a/services/ingestion/tests/test_product_catalog_derive_events_pg.py +++ b/services/ingestion/tests/test_product_catalog_derive_events_pg.py @@ -98,6 +98,7 @@ def seeded_tenant(): yield engine, tenant_id finally: with engine.begin() as conn: + conn.execute(text("SET LOCAL fsma.allow_mutation = 'true'")) # v141 append-only conn.execute( text("DELETE FROM fsma.traceability_events WHERE tenant_id = :tid"), {"tid": tenant_id}, diff --git a/tests/invariants/test_backfill_canonical.py b/tests/invariants/test_backfill_canonical.py index f83f96c0d..431a66bf4 100644 --- a/tests/invariants/test_backfill_canonical.py +++ b/tests/invariants/test_backfill_canonical.py @@ -252,6 +252,8 @@ def _persist_dual(session, event): def _orphan_the_canonical_row(session, tenant_id, event_id): """Simulate the #1277 failure: legacy row survives, canonical row is lost.""" + # v141 append-only trigger: the simulated loss needs the break-glass GUC. + session.execute(text("SET LOCAL fsma.allow_mutation = 'true'")) session.execute( text( "DELETE FROM fsma.traceability_events " diff --git a/tests/invariants/test_verify_canonical_parity_live.py b/tests/invariants/test_verify_canonical_parity_live.py index 4936fa8e0..ccdfb66aa 100644 --- a/tests/invariants/test_verify_canonical_parity_live.py +++ b/tests/invariants/test_verify_canonical_parity_live.py @@ -758,6 +758,8 @@ def test_canonical_drift_on_recall_predicate_columns_is_detected(self, seeded_te database_url, engine, tenant = seeded_tenant with engine.begin() as conn: conn.execute(text("SELECT set_config('app.tenant_id', :tid, true)"), {"tid": tenant}) + # v141 append-only trigger: simulated drift needs the break-glass GUC. + conn.execute(text("SET LOCAL fsma.allow_mutation = 'true'")) conn.execute( text( "UPDATE fsma.traceability_events SET quantity = 999.0, " diff --git a/tests/migrations/test_v141_traceability_events_append_only.py b/tests/migrations/test_v141_traceability_events_append_only.py new file mode 100644 index 000000000..15a9144fa --- /dev/null +++ b/tests/migrations/test_v141_traceability_events_append_only.py @@ -0,0 +1,473 @@ +"""v141 — ``fsma.traceability_events`` mutation policy (S4). + +Static checks run everywhere. Behavioural checks run against a real +PostgreSQL at alembic head (``REGENGINE_MIGRATION_TEST_DB`` or a +testcontainers Postgres, same contract as ``test_client_role_lockdown_v092``) +and verify the sanctioned mutation surface end to end: + +1. Allowlisted column UPDATEs (verdict markers, amended_at) succeed. +2. ``status`` may move ``active -> superseded`` and nothing else. +3. A payload column UPDATE, a DELETE and a TRUNCATE raise ``check_violation``. +4. ``SET LOCAL fsma.allow_mutation = 'true'`` lets UPDATE and DELETE through. +5. Re-applying the upgrade SQL is idempotent; the downgrade removes both + triggers and both functions; re-upgrading restores them. +""" + +from __future__ import annotations + +import ast +import importlib.util +import subprocess +import sys +import uuid +from pathlib import Path +from typing import Iterator + +import pytest +from sqlalchemy import text +from sqlalchemy.engine import Engine +from sqlalchemy.exc import DBAPIError + +from tests.migrations.rls_db import rls_postgres_or_skip, skip_or_fail_db + +REPO_ROOT = Path(__file__).resolve().parents[2] +MIGRATION = REPO_ROOT / "alembic" / "versions" / "20261028_08c7cf36ddbe_v141_traceability_events_append_only.py" + +TENANT = "00000000-0000-4000-8000-0000000138aa" + + +def _load_migration(): + spec = importlib.util.spec_from_file_location("v141_traceability_events_append_only", MIGRATION) + assert spec is not None and spec.loader is not None + mod = importlib.util.module_from_spec(spec) + sys.modules[spec.name] = mod + spec.loader.exec_module(mod) + return mod + + +# --------------------------------------------------------------------------- +# Static +# --------------------------------------------------------------------------- + + +def test_revision_chain_and_shape() -> None: + mod = _load_migration() + assert mod.revision == "08c7cf36ddbe" # pragma: allowlist secret + assert mod.down_revision == "b8c1e4f2a901" # pragma: allowlist secret + tree = ast.parse(MIGRATION.read_text()) + names = {n.name for n in tree.body if isinstance(n, ast.FunctionDef)} + assert {"upgrade", "downgrade"} <= names + + +def test_trigger_sql_covers_update_delete_and_truncate() -> None: + mod = _load_migration() + assert "BEFORE UPDATE OR DELETE ON fsma.traceability_events" in mod._TRIGGERS_SQL + assert "BEFORE TRUNCATE ON fsma.traceability_events" in mod._TRIGGERS_SQL + assert "FOR EACH STATEMENT" in mod._TRIGGERS_SQL + assert "to_regclass('fsma.traceability_events')" in mod._TRIGGERS_SQL + assert "DROP TRIGGER IF EXISTS" in mod._TRIGGERS_SQL + assert "ERRCODE = 'check_violation'" in mod._ROW_FUNCTION_SQL + assert "current_setting('fsma.allow_mutation', true)" in mod._ROW_FUNCTION_SQL + assert "current_setting('fsma.allow_mutation', true)" in mod._TRUNCATE_FUNCTION_SQL + # Null-safe transition guard (the Supabase twin leaves status nullable). + assert "(OLD.status = 'active' AND NEW.status = 'superseded') IS NOT TRUE" in mod._ROW_FUNCTION_SQL + + +def test_allowlist_matches_the_production_writers() -> None: + """The three sanctioned writers only touch these columns.""" + mod = _load_migration() + assert set(mod.ALLOWLIST) == { + "status", + "amended_at", + "evaluation_verdict", + "no_verdict_reason", + "evaluation_error", + "evaluated_at", + } + writer = (REPO_ROOT / "services/shared/canonical_persistence/writer.py").read_text() + assert "SET status = 'superseded', amended_at = NOW()" in writer + engine = (REPO_ROOT / "services/shared/rules/engine.py").read_text() + assert "SET evaluation_verdict = :verdict" in engine + assert "SET evaluation_error = :error_message" in engine + + +def test_downgrade_removes_everything_it_created() -> None: + mod = _load_migration() + for name in (mod.ROW_TRIGGER, mod.TRUNCATE_TRIGGER): + assert f"DROP TRIGGER IF EXISTS {name}" in mod._DOWNGRADE_SQL + for fn in (mod.ROW_FUNCTION, mod.TRUNCATE_FUNCTION): + assert f"DROP FUNCTION IF EXISTS {fn}()" in mod._DOWNGRADE_SQL + + +# --------------------------------------------------------------------------- +# Behavioural (live PostgreSQL) +# --------------------------------------------------------------------------- + + +@pytest.fixture(scope="module") +def upgraded_engine() -> Iterator[Engine]: + """Real Postgres at alembic head (same contract as test_client_role_lockdown_v092).""" + with rls_postgres_or_skip() as (plain_url, engine): + env = { + "MIGRATIONS_DATABASE_URL": plain_url, + "DATABASE_URL": plain_url, + "PATH": subprocess.os.environ["PATH"], + } + for k in ("REGENGINE_ENV", "REGENGINE_ALLOW_BASELINE_DOWNGRADE"): + v = subprocess.os.environ.get(k) + if v is not None: + env[k] = v + result = subprocess.run( + [sys.executable, "-m", "alembic", "upgrade", "head"], + cwd=REPO_ROOT, + env=env, + capture_output=True, + text=True, + timeout=300, + ) + if result.returncode != 0: + skip_or_fail_db( + f"alembic upgrade head failed (returncode={result.returncode}):" + f"\nstdout: {result.stdout}\nstderr: {result.stderr}" + ) + with engine.connect() as conn: + if conn.execute(text("SELECT to_regclass('fsma.traceability_events')")).scalar() is None: + skip_or_fail_db("fsma.traceability_events missing at alembic head") + yield engine + + +def _insert_event(conn, *, status: str = "active") -> str: + event_id = str(uuid.uuid4()) + conn.execute( + text( + """ + INSERT INTO fsma.traceability_events + (event_id, tenant_id, source_system, event_type, event_timestamp, + traceability_lot_code, quantity, unit_of_measure, status) + VALUES + (CAST(:eid AS uuid), CAST(:tid AS uuid), 'test', 'shipping', NOW(), + 'TLC-V138', 10.0, 'case', :status) + """ + ), + {"eid": event_id, "tid": TENANT, "status": status}, + ) + return event_id + + +@pytest.fixture +def seeded(upgraded_engine: Engine) -> Iterator[tuple[Engine, str]]: + with upgraded_engine.begin() as conn: + event_id = _insert_event(conn) + try: + yield upgraded_engine, event_id + finally: + with upgraded_engine.begin() as conn: + conn.execute(text("SET LOCAL fsma.allow_mutation = 'true'")) + conn.execute( + text("DELETE FROM fsma.traceability_events WHERE tenant_id = CAST(:tid AS uuid)"), + {"tid": TENANT}, + ) + + +def _expect_rejected(engine: Engine, sql: str, params: dict) -> None: + with pytest.raises(DBAPIError) as exc: + with engine.begin() as conn: + conn.execute(text(sql), params) + message = str(exc.value) + assert "append-only" in message or "status may only move" in message, message + # psycopg 3 (rls_db forces it) exposes the server SQLSTATE directly; + # no fallback, so a driver that lacks it fails loudly. + assert exc.value.orig.sqlstate == "23514", message # check_violation + + +def test_triggers_installed_at_head(upgraded_engine: Engine) -> None: + with upgraded_engine.connect() as conn: + rows = conn.execute( + text( + """ + SELECT tgname FROM pg_trigger + WHERE tgrelid = 'fsma.traceability_events'::regclass AND NOT tgisinternal + """ + ) + ).scalars() + names = set(rows) + assert {"traceability_events_mutation_policy", "traceability_events_no_truncate"} <= names + + +def test_verdict_marker_update_is_allowed(seeded) -> None: + engine, event_id = seeded + with engine.begin() as conn: + conn.execute( + text( + "UPDATE fsma.traceability_events" + " SET evaluation_verdict = 'pass', no_verdict_reason = NULL, evaluated_at = NOW()" + " WHERE tenant_id = CAST(:tid AS uuid) AND event_id = CAST(:eid AS uuid)" + ), + {"tid": TENANT, "eid": event_id}, + ) + conn.execute( + text( + "UPDATE fsma.traceability_events SET evaluation_error = 'boom', evaluated_at = NOW()" + " WHERE tenant_id = CAST(:tid AS uuid) AND event_id = CAST(:eid AS uuid)" + ), + {"tid": TENANT, "eid": event_id}, + ) + with engine.connect() as conn: + row = conn.execute( + text("SELECT evaluation_verdict, evaluation_error FROM fsma.traceability_events WHERE event_id = CAST(:eid AS uuid)"), + {"eid": event_id}, + ).one() + assert tuple(row) == ("pass", "boom") + + +def test_active_to_superseded_is_allowed(seeded) -> None: + engine, event_id = seeded + with engine.begin() as conn: + result = conn.execute( + text( + "UPDATE fsma.traceability_events SET status = 'superseded', amended_at = NOW()" + " WHERE event_id = CAST(:eid AS uuid) AND tenant_id = CAST(:tid AS uuid) AND status = 'active'" + " RETURNING event_id" + ), + {"tid": TENANT, "eid": event_id}, + ) + assert result.fetchone() is not None + + +def test_superseded_cannot_go_back_to_active(seeded) -> None: + engine, event_id = seeded + with engine.begin() as conn: + conn.execute( + text("UPDATE fsma.traceability_events SET status = 'superseded' WHERE event_id = CAST(:eid AS uuid)"), + {"eid": event_id}, + ) + _expect_rejected( + engine, + "UPDATE fsma.traceability_events SET status = 'active' WHERE event_id = CAST(:eid AS uuid)", + {"eid": event_id}, + ) + + +@pytest.mark.parametrize("target", ["rejected", "draft"]) +def test_other_status_transitions_rejected(seeded, target) -> None: + engine, event_id = seeded + _expect_rejected( + engine, + f"UPDATE fsma.traceability_events SET status = '{target}' WHERE event_id = CAST(:eid AS uuid)", + {"eid": event_id}, + ) + + +def test_null_status_rejected_on_nullable_catalogs(seeded) -> None: + """A NULL ``status`` must not slip past the transition guard. + + The alembic catalog declares ``status NOT NULL``, but the Supabase twin + (``supabase/migrations/20260403234813_*.sql``) leaves it nullable, and + ``NOT (OLD.status = 'active' AND NEW.status = 'superseded')`` is NULL, + not TRUE, when either side is NULL. Both transactions below drop the + NOT NULL constraint only inside a transaction that the rejection rolls + back, so the catalog is unchanged afterwards. + """ + engine, event_id = seeded + drop_not_null = text("ALTER TABLE fsma.traceability_events ALTER COLUMN status DROP NOT NULL") + set_status = text("UPDATE fsma.traceability_events SET status = :status WHERE event_id = CAST(:eid AS uuid)") + + # active -> NULL + with pytest.raises(DBAPIError) as exc: + with engine.begin() as conn: + conn.execute(drop_not_null) + conn.execute(set_status, {"status": None, "eid": event_id}) + assert "status may only move" in str(exc.value), str(exc.value) + assert exc.value.orig.sqlstate == "23514" + + # NULL -> active (the row got its NULL through break-glass) + with pytest.raises(DBAPIError) as exc: + with engine.begin() as conn: + conn.execute(drop_not_null) + conn.execute(text("SET LOCAL fsma.allow_mutation = 'true'")) + conn.execute(set_status, {"status": None, "eid": event_id}) + conn.execute(text("SET LOCAL fsma.allow_mutation = 'false'")) + conn.execute(set_status, {"status": "active", "eid": event_id}) + assert "status may only move" in str(exc.value), str(exc.value) + assert exc.value.orig.sqlstate == "23514" + + with engine.connect() as conn: + nullable = conn.execute( + text( + "SELECT is_nullable FROM information_schema.columns" + " WHERE table_schema = 'fsma' AND table_name = 'traceability_events' AND column_name = 'status'" + ) + ).scalar() + assert nullable == "NO" + + +@pytest.mark.parametrize( + "assignment", + [ + "quantity = quantity + 0.25", + "event_timestamp = event_timestamp + interval '1 minute'", + "traceability_lot_code = 'TLC-MUTATED'", + "kdes = '{\"x\": 1}'::jsonb", + "sha256_hash = repeat('a', 64)", + "tenant_id = '00000000-0000-4000-8000-0000000138bb'", + ], +) +def test_payload_column_update_rejected(seeded, assignment) -> None: + engine, event_id = seeded + _expect_rejected( + engine, + f"UPDATE fsma.traceability_events SET {assignment} WHERE event_id = CAST(:eid AS uuid)", + {"eid": event_id}, + ) + + +@pytest.mark.parametrize( + "assignment", + [ + "sha256_hash = NULL", + "chain_hash = NULL, status = 'superseded', amended_at = NOW()", + "product_reference = NULL, evaluation_verdict = 'pass', evaluated_at = NOW()", + "from_facility_reference = NULL", + ], +) +def test_clearing_populated_column_to_null_rejected(seeded, assignment) -> None: + """Clearing a protected column to NULL is a change the diff must see. + + ``to_jsonb(NEW)`` keeps a NULL column as a ``null``-valued key rather + than dropping it, so the key-join branch of the diff compares + ``null`` against the old value and reports the column. This pins that, + including when the clear hides behind a sanctioned status or verdict + update. The seeded row leaves these columns NULL, so they are populated + through break-glass first; a NULL -> NULL assignment is a no-op and is + not what this test is about. + """ + engine, event_id = seeded + with engine.begin() as conn: + conn.execute(text("SET LOCAL fsma.allow_mutation = 'true'")) + conn.execute( + text( + "UPDATE fsma.traceability_events" + " SET sha256_hash = repeat('a', 64), chain_hash = repeat('b', 64)," + " product_reference = 'PROD-1', from_facility_reference = 'FAC-1'" + " WHERE event_id = CAST(:eid AS uuid)" + ), + {"eid": event_id}, + ) + _expect_rejected( + engine, + f"UPDATE fsma.traceability_events SET {assignment} WHERE event_id = CAST(:eid AS uuid)", + {"eid": event_id}, + ) + with engine.connect() as conn: + row = conn.execute( + text( + "SELECT sha256_hash, chain_hash, product_reference, from_facility_reference" + " FROM fsma.traceability_events WHERE event_id = CAST(:eid AS uuid)" + ), + {"eid": event_id}, + ).one() + assert None not in tuple(row) + + +def test_allowlisted_and_payload_change_in_one_statement_rejected(seeded) -> None: + """Hiding a payload change behind a sanctioned marker update does not work.""" + engine, event_id = seeded + _expect_rejected( + engine, + "UPDATE fsma.traceability_events SET evaluated_at = NOW(), quantity = 1.0 WHERE event_id = CAST(:eid AS uuid)", + {"eid": event_id}, + ) + + +def test_delete_rejected(seeded) -> None: + engine, event_id = seeded + _expect_rejected( + engine, + "DELETE FROM fsma.traceability_events WHERE event_id = CAST(:eid AS uuid)", + {"eid": event_id}, + ) + + +def test_truncate_rejected(upgraded_engine: Engine) -> None: + # CASCADE so PostgreSQL gets past its own FK check (evidence_attachments + # references this table) and reaches the BEFORE TRUNCATE trigger, which + # raises before any table loses a row. The transaction rolls back. + with pytest.raises(DBAPIError) as exc: + with upgraded_engine.begin() as conn: + conn.execute(text("TRUNCATE fsma.traceability_events CASCADE")) + assert "TRUNCATE rejected" in str(exc.value) + + +def test_break_glass_allows_update_and_delete(seeded) -> None: + engine, event_id = seeded + with engine.begin() as conn: + conn.execute(text("SET LOCAL fsma.allow_mutation = 'true'")) + conn.execute( + text("UPDATE fsma.traceability_events SET quantity = 99.0 WHERE event_id = CAST(:eid AS uuid)"), + {"eid": event_id}, + ) + with engine.connect() as conn: + assert conn.execute( + text("SELECT quantity FROM fsma.traceability_events WHERE event_id = CAST(:eid AS uuid)"), + {"eid": event_id}, + ).scalar() == 99.0 + with engine.begin() as conn: + conn.execute(text("SET LOCAL fsma.allow_mutation = 'true'")) + conn.execute( + text("DELETE FROM fsma.traceability_events WHERE event_id = CAST(:eid AS uuid)"), + {"eid": event_id}, + ) + with engine.connect() as conn: + assert conn.execute( + text("SELECT COUNT(*) FROM fsma.traceability_events WHERE event_id = CAST(:eid AS uuid)"), + {"eid": event_id}, + ).scalar() == 0 + + +def test_break_glass_is_transaction_local(seeded) -> None: + """SET LOCAL does not leak: the next transaction is protected again.""" + engine, event_id = seeded + with engine.begin() as conn: + conn.execute(text("SET LOCAL fsma.allow_mutation = 'true'")) + conn.execute(text("SELECT 1")) + _expect_rejected( + engine, + "UPDATE fsma.traceability_events SET quantity = 1.0 WHERE event_id = CAST(:eid AS uuid)", + {"eid": event_id}, + ) + + +def test_upgrade_is_idempotent_and_downgrade_removes_triggers(upgraded_engine: Engine) -> None: + mod = _load_migration() + with upgraded_engine.begin() as conn: + for stmt in (mod._ROW_FUNCTION_SQL, mod._TRUNCATE_FUNCTION_SQL, mod._TRIGGERS_SQL): + conn.execute(text(stmt)) + query = text( + """ + SELECT tgname FROM pg_trigger + WHERE tgrelid = 'fsma.traceability_events'::regclass AND NOT tgisinternal + AND tgname IN ('traceability_events_mutation_policy', 'traceability_events_no_truncate') + """ + ) + with upgraded_engine.connect() as conn: + assert len(set(conn.execute(query).scalars())) == 2 + try: + with upgraded_engine.begin() as conn: + conn.exec_driver_sql(mod._DOWNGRADE_SQL) + with upgraded_engine.connect() as conn: + assert set(conn.execute(query).scalars()) == set() + functions = set( + conn.execute( + text( + "SELECT p.proname FROM pg_proc p JOIN pg_namespace n ON n.oid = p.pronamespace" + " WHERE n.nspname = 'fsma' AND p.proname LIKE 'prevent_traceability_event_%'" + ) + ).scalars() + ) + assert functions == set() + finally: + with upgraded_engine.begin() as conn: + for stmt in (mod._ROW_FUNCTION_SQL, mod._TRUNCATE_FUNCTION_SQL, mod._TRIGGERS_SQL): + conn.execute(text(stmt)) + with upgraded_engine.connect() as conn: + assert len(set(conn.execute(query).scalars())) == 2 diff --git a/tests/shared/test_exception_waiver_replay_live.py b/tests/shared/test_exception_waiver_replay_live.py new file mode 100644 index 000000000..230bb78af --- /dev/null +++ b/tests/shared/test_exception_waiver_replay_live.py @@ -0,0 +1,117 @@ +"""Live-DB regression: a replayed waiver does not sign off or comment twice. + +``PATCH /api/v1/exceptions/{case_id}/waive`` records the authenticated +principal as the approver (S5, #3717). The route answers 200 whether or not +the case changed, so a retried or replayed request must be a no-op at the +service: ``ExceptionQueueService.waive_exception`` guards the UPDATE with +``status NOT IN ('resolved', 'waived')`` and only writes the signoff and the +status-change comment when a row actually moved. That guard lives in SQL, +which is why the router unit tests (MagicMock service) cannot prove it. + +Drives the real router and the real service against a schema-at-head +PostgreSQL inside one transaction that is rolled back at the end. Requires +a designated live DB and skips without it (same gate as +test_exception_queue_uuid_arrays_live.py). +""" + +from __future__ import annotations + +import uuid +from datetime import datetime, timezone + +import pytest +from shared.db_url import sync_postgres_url + +from tests.migrations.rls_db import guarded_live_database_url + +DATABASE_URL = guarded_live_database_url() +pytestmark = pytest.mark.skipif( + not DATABASE_URL, + reason="no designated live test DB (CI DATABASE_URL or REGENGINE_MIGRATION_TEST_DB) — skipping live waiver replay test", +) + + +def _count(session, table: str, case_id: str, extra: str = "") -> int: + from sqlalchemy import text + + return session.execute( + text(f"SELECT count(*) FROM fsma.{table} WHERE case_id = CAST(:cid AS uuid) {extra}"), # noqa: S608 - fixed table names + {"cid": case_id}, + ).scalar_one() + + +def test_waiver_replay_does_not_duplicate_signoff_or_comment(monkeypatch) -> None: + from fastapi import FastAPI + from fastapi.testclient import TestClient + from sqlalchemy import create_engine, text + from sqlalchemy.orm import Session + + from services.ingestion.app import authz, exception_router + from services.ingestion.app.authz import IngestionPrincipal, get_ingestion_principal + from services.shared.exception_queue import ExceptionQueueService + + monkeypatch.setattr(authz, "consume_tenant_rate_limit", lambda **_kw: (True, 99)) + + engine = create_engine(sync_postgres_url(DATABASE_URL)) + tenant_id = str(uuid.uuid4()) + approver = f"session:{uuid.uuid4()}" + principal = IngestionPrincipal( + key_id=approver, + tenant_id=tenant_id, + scopes=["exceptions.read", "exceptions.write"], + auth_mode="test", + ) + + with Session(engine) as session: + session.execute( + text("INSERT INTO tenants (id, name, slug, created_at) VALUES (CAST(:id AS uuid), :name, :slug, :now)"), + { + "id": tenant_id, + "name": "waiver-replay-live-test", + "slug": f"waiver-replay-live-{tenant_id[:8]}", + "now": datetime.now(timezone.utc), + }, + ) + svc = ExceptionQueueService(session) + case = svc.create_exception( + tenant_id=tenant_id, + severity="warning", + rule_category="kde_completeness", + source_supplier="Live Replay Supplier", + ) + case_id = case.case_id + comments_before = _count(session, "exception_comments", case_id) + assert _count(session, "exception_signoffs", case_id) == 0 + + app = FastAPI() + app.include_router(exception_router.router) + app.dependency_overrides[get_ingestion_principal] = lambda: principal + app.dependency_overrides[exception_router._get_db_session] = lambda: session + client = TestClient(app) + + body = {"waiver_reason": "Low-risk supplier", "waiver_approved_by": "someone-else"} + first = client.patch(f"/api/v1/exceptions/{case_id}/waive", json=body) + assert first.status_code == 200, first.text + replay = client.patch(f"/api/v1/exceptions/{case_id}/waive", json=body) + assert replay.status_code == 200, replay.text + + row = session.execute( + text( + "SELECT status, waiver_approved_by, waiver_reason FROM fsma.exception_cases WHERE case_id = CAST(:cid AS uuid)" + ), + {"cid": case_id}, + ).one() + assert tuple(row) == ("waived", approver, "Low-risk supplier") + + # Exactly one signoff, signed by the principal, never by the body. + signoffs = session.execute( + text("SELECT signoff_type, signed_by FROM fsma.exception_signoffs WHERE case_id = CAST(:cid AS uuid)"), + {"cid": case_id}, + ).all() + assert [tuple(r) for r in signoffs] == [("waive", approver)] + + # Exactly one status-change comment on top of the creation comment. + assert _count(session, "exception_comments", case_id) == comments_before + 1 + assert _count(session, "exception_comments", case_id, "AND comment_type = 'status_change'") == 1 + + session.rollback() diff --git a/tests/shared/test_postgres_evidence_workflow_pg.py b/tests/shared/test_postgres_evidence_workflow_pg.py index 963720050..9265f8156 100644 --- a/tests/shared/test_postgres_evidence_workflow_pg.py +++ b/tests/shared/test_postgres_evidence_workflow_pg.py @@ -237,7 +237,10 @@ def test_canonical_content_tampering_is_rejected(workflow, tmp_path, column): _assert_package(w, tmp_path, verified=True) # Simulate corruption by a cluster administrator; runtime ACLs stay intact. expression = "quantity + 0.25" if column == "quantity" else "event_timestamp + interval '1 minute'" - with w.admin.cursor() as cur: + # v141 makes the canonical table append-only; the simulated administrator + # uses the break-glass GUC inside one transaction, as a real one would. + with w.admin.transaction(), w.admin.cursor() as cur: + cur.execute("SET LOCAL fsma.allow_mutation = 'true'") cur.execute(f"UPDATE fsma.traceability_events SET {column}={expression} WHERE tenant_id=%s", (w.tenant,)) _assert_package(w, tmp_path, verified=False) @@ -556,7 +559,7 @@ def rejected(match, statement, params=None, call=None): blob_insert, {"run": other_blobless_run, "tenant": w.tenant, "sha": "0" * 64}, ) - # INSERT is column-scoped: created_at still cannot be forged. v138 grants + # INSERT is column-scoped: created_at still cannot be forged. v140 grants # storage_url INSERT for partner-uploads pointer-only rows. rejected( "permission denied",