feat(audit): append-only, hash-chained audit log (M3a, #107)

alembic/versions/0006_append_only_audit.py

@@ -0,0 +1,109 @@
+"""Append-only, hash-chained audit_log (M3a, #107; BR-24, BR-35).
+
+Decided 28 Oct: upgrade the existing per-tenant ``audit_log`` rather than add
+an ``audit_events`` table.
+
+- New columns: ``actor``, ``tenant_schema``, ``before_value``, ``after_value``,
+  ``prev_hash``, ``row_hash``.
+- A BEFORE INSERT trigger serialises writers with an advisory lock, gives the
+  row its id under that lock (so id order is chain order), and sets
+  ``prev_hash`` to the previous row's ``row_hash``.
+- Triggers reject UPDATE, DELETE and TRUNCATE for every role, owner included.
+  The owner can still disable a trigger; the hash chain is what exposes that,
+  and M3d adds a runtime role that can't.
+- Rows written before this revision are chained in id order during upgrade.
+
+Runs per tenant, like 0003.
+
+Revision ID: 0006
+Revises: 0005
+Create Date: 2026-10-28
+"""
+
+from alembic import op
+
+from api.config import settings
+
+revision = "0006"
+down_revision = "0005"
+branch_labels = None
+depends_on = None
+
+
+def _schema() -> str:
+    return op.get_bind().dialect.identifier_preparer.quote(settings.UPLOADS_SCHEMA)
+
+
+def upgrade() -> None:
+    s = _schema()
+    t = f"{s}.audit_log"
+    op.execute(f"""ALTER TABLE {t}
+        ADD COLUMN actor TEXT,
+        ADD COLUMN tenant_schema TEXT,
+        ADD COLUMN before_value JSONB,
+        ADD COLUMN after_value JSONB,
+        ADD COLUMN prev_hash TEXT,
+        ADD COLUMN row_hash TEXT""")
+    # The hash covers every column except row_hash. Times are hashed in UTC so
+    # the result doesn't depend on the session's TimeZone.
+    op.execute(f"""
+        CREATE FUNCTION {s}.audit_row_hash(prev TEXT, r {t}) RETURNS TEXT
+        LANGUAGE sql IMMUTABLE AS $$
+          SELECT encode(sha256(convert_to(coalesce(prev, '') || jsonb_build_object(
+            'id', r.id, 'action', r.action, 'file_id', r.file_id, 'file_name', r.file_name,
+            'table_name', r.table_name, 'mode', r.mode,
+            'performed_at', to_char(r.performed_at AT TIME ZONE 'UTC', 'YYYY-MM-DD"T"HH24:MI:SS.US'),
+            'actor', r.actor, 'tenant_schema', r.tenant_schema,
+            'before_value', r.before_value, 'after_value', r.after_value)::text, 'UTF8')), 'hex')
+        $$""")
+    # Chain the rows that already exist, oldest first.
+    op.execute(f"""DO $$
+        DECLARE r RECORD; prev TEXT := NULL;
+        BEGIN
+          FOR r IN SELECT * FROM {t} ORDER BY id LOOP
+            UPDATE {t} SET tenant_schema = TG_SCHEMA_PLACEHOLDER, prev_hash = prev
+             WHERE id = r.id;
+            SELECT {s}.audit_row_hash(prev, a) INTO prev FROM {t} a WHERE a.id = r.id;
+            UPDATE {t} SET row_hash = prev WHERE id = r.id;
+          END LOOP;
+        END $$""".replace("TG_SCHEMA_PLACEHOLDER", f"'{settings.UPLOADS_SCHEMA}'"))
+    op.execute(f"""
+        CREATE FUNCTION {s}.audit_chain() RETURNS trigger LANGUAGE plpgsql AS $$
+        BEGIN
+          PERFORM pg_advisory_xact_lock(hashtext(TG_TABLE_SCHEMA || '.audit_log'));
+          NEW.id := nextval(pg_get_serial_sequence(quote_ident(TG_TABLE_SCHEMA) || '.audit_log', 'id'));
+          NEW.tenant_schema := TG_TABLE_SCHEMA;
+          EXECUTE format('SELECT row_hash FROM %I.audit_log ORDER BY id DESC LIMIT 1', TG_TABLE_SCHEMA)
+            INTO NEW.prev_hash;
+          NEW.row_hash := {s}.audit_row_hash(NEW.prev_hash, NEW);
+          RETURN NEW;
+        END $$""")
+    op.execute(f"""
+        CREATE FUNCTION {s}.audit_reject_change() RETURNS trigger LANGUAGE plpgsql AS $$
+        BEGIN
+          RAISE EXCEPTION 'audit_log is append-only: % is not allowed', TG_OP
+            USING ERRCODE = 'insufficient_privilege';
+        END $$""")
+    # The trigger takes the id under its lock; the column default would take a second one.
+    op.execute(f"ALTER TABLE {t} ALTER COLUMN id DROP DEFAULT")
+    op.execute(f"CREATE TRIGGER audit_chain BEFORE INSERT ON {t} "
+               f"FOR EACH ROW EXECUTE FUNCTION {s}.audit_chain()")
+    op.execute(f"CREATE TRIGGER audit_no_change BEFORE UPDATE OR DELETE ON {t} "
+               f"FOR EACH ROW EXECUTE FUNCTION {s}.audit_reject_change()")
+    op.execute(f"CREATE TRIGGER audit_no_truncate BEFORE TRUNCATE ON {t} "
+               f"FOR EACH STATEMENT EXECUTE FUNCTION {s}.audit_reject_change()")
+
+
+def downgrade() -> None:
+    s = _schema()
+    t = f"{s}.audit_log"
+    for trig in ("audit_no_truncate", "audit_no_change", "audit_chain"):
+        op.execute(f"DROP TRIGGER IF EXISTS {trig} ON {t}")
+    seq = f"pg_get_serial_sequence('{t}', 'id')"
+    op.execute(f"ALTER TABLE {t} ALTER COLUMN id SET DEFAULT nextval({seq}::regclass)")
+    op.execute(f"DROP FUNCTION IF EXISTS {s}.audit_reject_change()")
+    op.execute(f"DROP FUNCTION IF EXISTS {s}.audit_chain()")
+    op.execute(f"DROP FUNCTION IF EXISTS {s}.audit_row_hash(TEXT, {t})")
+    op.execute(f"""ALTER TABLE {t}
+        DROP COLUMN actor, DROP COLUMN tenant_schema, DROP COLUMN before_value,
+        DROP COLUMN after_value, DROP COLUMN prev_hash, DROP COLUMN row_hash""")

api/auth.py

@@ -20,6 +20,7 @@
 from api.config import settings
 from api.db import Conn
 from api.services.accounts import session_for
+from api.services.audit import acting_as
 from api.tenancy import schema_for, scoped
 
 
@@ -68,10 +69,11 @@ async def tenant_scope(authorization: str = Header(default=""),
     """
     if not authorization:
         require_api_key(x_api_key)
-        yield None
+        with acting_as("api_key"):
+            yield None
         return
     session = await run_in_threadpool(_live_session, bearer_token(authorization))
-    with scoped(schema_for(session["tenant"])):
+    with scoped(schema_for(session["tenant"])), acting_as(session["username"]):
         yield session
 
 

api/routers/audit_routes.py

@@ -15,6 +15,7 @@
 from api.auth import allow, tenant_scope
 from api.tenancy import uploads_schema
 from api.db import Conn
+from api.services.audit import verify_chain
 
 router = APIRouter(
     prefix="/api/audit",
@@ -37,7 +38,7 @@ def audit_log(limit: int = 50, offset: int = 0) -> dict:
             total = cur.fetchone()[0]
             cur.execute(
                 sql.SQL(
-                    "SELECT id, action, file_id, file_name, table_name, mode, performed_at "
+                    "SELECT id, action, file_id, file_name, table_name, mode, performed_at, actor "
                     "FROM {}.audit_log ORDER BY performed_at DESC LIMIT %s OFFSET %s"
                 ).format(sql.Identifier(uploads_schema())),
                 (limit, offset),
@@ -51,7 +52,17 @@ def audit_log(limit: int = 50, offset: int = 0) -> dict:
                     "tableName": r[4],
                     "mode": r[5],
                     "performedAt": r[6].isoformat(),
+                    "actor": r[7],
                 }
                 for r in cur.fetchall()
             ]
     return {"total": total, "limit": limit, "offset": offset, "entries": entries}
+
+
+@router.get("/verify", dependencies=[Depends(allow("read_audit"))])
+def audit_verify() -> dict:
+    """Recompute this tenant's audit hash chain (M3a, #107). ``intact`` is false
+    if any row was changed, removed or inserted out of order."""
+    with Conn() as conn:
+        with conn.cursor() as cur:
+            return verify_chain(cur)

api/routers/auth_routes.py

@@ -14,7 +14,7 @@
 from api.db import Conn
 from api import oidc
 from api.services import accounts
-from api.services.audit import write_audit
+from api.services.audit import acting_as, write_audit
 from api.tenancy import schema_for, scoped
 
 router = APIRouter(prefix="/api/auth", tags=["auth"])
@@ -114,10 +114,9 @@ def reset_password(username: str, req: ResetPasswordRequest,
         with conn.cursor() as cur:
             result = accounts.reset_password(cur, session["tenantId"], username, req.newPassword)
             if result == "ok":
-                # ponytail: audit_log has file columns only; target in file_name and actor
-                # in mode until M3's audit_events (#107) adds actor and object.
-                with scoped(schema_for(session["tenant"])):
-                    write_audit(cur, "password_reset", None, username, None, f"by {session['username']}")
+                # The target user goes in file_name, the feed's object column (M3a, #107).
+                with scoped(schema_for(session["tenant"])), acting_as(session["username"]):
+                    write_audit(cur, "password_reset", None, username, None, None)
         conn.commit()
     if result == "missing":
         raise HTTPException(status_code=404, detail="No such user in this tenant")

api/services/audit.py

@@ -24,7 +24,10 @@
 
 from __future__ import annotations
 
+import json
 import logging
+from contextlib import contextmanager
+from contextvars import ContextVar
 
 from psycopg2 import sql
 
@@ -37,26 +40,58 @@
 # than an unqueryable string in the table.
 VALID_ACTIONS = frozenset({"upload", "delete", "password_reset"})
 
+# Who is acting (M3a, #107). tenant_scope sets it per request: the signed-in
+# username, or "api_key" for a request made with the shared key.
+_actor: ContextVar[str | None] = ContextVar("audit_actor", default=None)
+
+
+@contextmanager
+def acting_as(actor: str):
+    token = _actor.set(actor)
+    try:
+        yield
+    finally:
+        _actor.reset(token)
+
 
 def write_audit(cur, action: str, file_id: int | None, file_name: str | None,
-                table_name: str | None, mode: str | None) -> None:
+                table_name: str | None, mode: str | None,
+                before: dict | None = None, after: dict | None = None) -> None:
     """Insert one audit row using the caller's cursor.
 
     Use this when you already hold a transaction and want the audit row to
     commit atomically with the change it describes. The caller commits.
+    The table's trigger adds the tenant schema and the hash chain (0006).
     """
     if action not in VALID_ACTIONS:
         raise ValueError(f"Unknown audit action {action!r}; expected one of {sorted(VALID_ACTIONS)}")
     cur.execute(
         sql.SQL(
             "INSERT INTO {}.audit_log "
-            "(action, file_id, file_name, table_name, mode) "
-            "VALUES (%s, %s, %s, %s, %s)"
+            "(action, file_id, file_name, table_name, mode, actor, before_value, after_value) "
+            "VALUES (%s, %s, %s, %s, %s, %s, %s, %s)"
         ).format(sql.Identifier(uploads_schema())),
-        (action, file_id, file_name, table_name, mode),
+        (action, file_id, file_name, table_name, mode, _actor.get(),
+         None if before is None else json.dumps(before),
+         None if after is None else json.dumps(after)),
     )
 
 
+def verify_chain(cur) -> dict:
+    """Recompute the audit hash chain (M3a). Returns the row count and the
+    first row whose link or hash doesn't match, or None if the chain is intact."""
+    schema = sql.Identifier(uploads_schema())
+    cur.execute(sql.SQL(
+        "SELECT count(*), min(id) FILTER (WHERE prev_hash IS DISTINCT FROM expected_prev "
+        "  OR row_hash IS DISTINCT FROM {s}.audit_row_hash(prev_hash, r)) "
+        "FROM (SELECT a AS r, a.id, a.prev_hash, a.row_hash, "
+        "             lag(a.row_hash) OVER (ORDER BY a.id) AS expected_prev "
+        "      FROM {s}.audit_log a) t"
+    ).format(s=schema))
+    rows, first_bad = cur.fetchone()
+    return {"rows": rows, "intact": first_bad is None, "firstBadId": first_bad}
+
+
 def record_audit(action: str, file_id: int | None, file_name: str | None,
                  table_name: str | None, mode: str | None) -> bool:
     """Insert one audit row in its own transaction. Never raises.

docs/VCRM.md

@@ -154,7 +154,7 @@ Status as assessed on 2026-10-26 at `main` 032b522. None of these is "Verified"
 | BR-32 | ❌ Not started | No code | API + e2e | M6 |
 | BR-33 | ⚠️ Partial | Single-transaction load; no staging or lineage | Integration incl. failure injection | M6 |
 | BR-34 | ⚠️ Partial | Total, inserted, duplicate and failed counts persisted per import (M1: #90, `tests/test_import_counts.py`); an upload where every row fails is reported as failed and saves nothing, and partial failures are shown on the files list (#123, `tests/test_failed_uploads.py`); over-wide rows reported, not dropped (M1: #88, `tests/test_ragged_rows.py`); excluded and edited rows not yet tracked (M6) | Integration | M1, M6 |
-| BR-35 | ⚠️ Partial | Export formula injection neutralised (M1: #86, `export.test.ts`); audit table still plain, no actor, no filters | DB negative tests + unit | M1, M3, M7 |
+| BR-35 | ⚠️ Partial | Export formula injection neutralised (M1: #86, `export.test.ts`). Audit log append-only (M3a, #107): triggers reject UPDATE, DELETE and TRUNCATE for every role; actor recorded; hash chain checked by `GET /api/audit/verify` (`test_audit_append_only.py`, 7 tests). Still to do: the least-privilege runtime role (M3d) and report filters (M7) | DB negative tests + unit | M1, M3, M7 |
 | BR-36 | ⚠️ Partial | Done in M2:<br>• local accounts with lockout (#98, #103)<br>• OIDC (#105)<br>• per-tenant schemas, T&E tables included (#100, #104)<br>• role matrix in `docs/ROLES.md` (#101)<br><br>Tests: `tests/test_accounts.py`, `test_tenant_isolation.py`, `test_roles.py`, `test_oidc.py`, `test_password_reset.py`. Each failed with its control removed.<br>• frontend sign-in; the browser no longer holds the shared key (#115)<br>• password change, admin reset and break-glass reset (#117; `tests/test_password_reset.py`)<br><br>Open: the shared key still grants non-tenant access to scripts and services; security review (#112); real-IdP OIDC check (#113) | Cross-tenant negative tests; role matrix tests | M2 |
 | BR-37 | ⚠️ Partial | No external calls found; no automated check; no offline doc (#60) | Egress-blocked run + offline dry run | M7 |
 | BR-38 | ❌ Not started | No container bundle for the application yet (`infra/Dockerfile` is the schema and evals runner only) | Offline smoke test 7/7 on Windows, macOS and Linux | #128 |

tests/test_api_hardening.py

@@ -209,7 +209,8 @@ def execute(self, statement, params=None):
                 captured["params"] = params
 
         write_audit(RecordingCursor(), "upload", 7, "f.csv", "csv_abc", "dynamic")
-        self.assertEqual(captured["params"], ("upload", 7, "f.csv", "csv_abc", "dynamic"))
+        self.assertEqual(captured["params"],
+                         ("upload", 7, "f.csv", "csv_abc", "dynamic", None, None, None))
 
     def test_file_id_is_coerced_from_the_loaders_string_form(self):
         self.assertEqual(coerce_file_id("42"), 42)

tests/test_api_integration.py

@@ -414,6 +414,17 @@ def test_audit_row_fields_match_the_upload_response(self):
         self.assertEqual(str(row["fileId"]), str(body["fileId"]))
         self.assertTrue(row["performedAt"], "performedAt must not be empty")
 
+    def test_audit_row_names_the_actor_and_the_chain_verifies(self):
+        """M3a (#107): who acted is recorded, and the hash chain checks out."""
+        up = self.client.post("/api/csv/upload",
+                              json={"fileName": self.name, "content": self.content})
+        self.assertEqual(up.json()["status"], "ok", up.text)
+        row = next(e for e in self._entries_for(self.name) if e["action"] == "upload")
+        self.assertEqual(row["actor"], "api_key")
+        check = self.client.get("/api/audit/verify").json()
+        self.assertTrue(check["intact"], check)
+        self.assertGreaterEqual(check["rows"], 1)
+
     def test_upload_then_delete_leaves_both_rows(self):
         """The full lifecycle. Before O-04 only the second half was recorded,
         so a table could be seen to vanish with no record of its arrival."""