From 62bb09f183e5bff57af0a4ebb99ed577f0e56404 Mon Sep 17 00:00:00 2001 From: Tyler Date: Sat, 20 Jun 2026 23:45:43 -0600 Subject: [PATCH] feat(sp11): tamper-evident hash-chained audit_log MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - New audit_log table (migration 0009, user_version=9): * id, event_type, entity_type, entity_id, actor, payload_json, created_at, prev_hash, hash * Indexes on (entity_type, entity_id), event_type, created_at - New cyclone.audit_log module: * append_event(session, AuditEvent) — appends one chained row * verify_chain(session) — walks the chain, returns first bad id * SHA-256 hash over canonical row form (unit-separator delimited) * Genesis prev_hash = 64 zeros (Bitcoin-style sentinel) - New AuditLog ORM model - New admin API endpoints: * GET /api/admin/audit-log (paginated, filterable) * GET /api/admin/audit-log/verify (returns ok/first_bad_id/reason) - Hooked into existing endpoints to append events: * /api/parse-999 → 'claim.rejected' per matched claim * /api/parse-277ca → 'claim.payer_rejected' per matched claim * /api/clearhouse/submit → 'clearhouse.submitted' per claim - HIPAA §164.316(b)(2) compliance note in docs Tests: 688 -> 705 (9 audit + 8 audit-API). All 705 backend tests pass. --- backend/src/cyclone/api.py | 110 ++++++++ backend/src/cyclone/audit_log.py | 254 ++++++++++++++++++ backend/src/cyclone/db.py | 39 +++ .../src/cyclone/migrations/0009_audit_log.sql | 33 +++ backend/tests/test_acks.py | 9 +- backend/tests/test_api_audit_log.py | 159 +++++++++++ backend/tests/test_audit_log.py | 187 +++++++++++++ docs/reference/co-medicaid.md | 31 +++ 8 files changed, 818 insertions(+), 4 deletions(-) create mode 100644 backend/src/cyclone/audit_log.py create mode 100644 backend/src/cyclone/migrations/0009_audit_log.sql create mode 100644 backend/tests/test_api_audit_log.py create mode 100644 backend/tests/test_audit_log.py diff --git a/backend/src/cyclone/api.py b/backend/src/cyclone/api.py index ec23894..df95cb6 100644 --- a/backend/src/cyclone/api.py +++ b/backend/src/cyclone/api.py @@ -35,6 +35,7 @@ from cyclone import __version__, db from cyclone.db import Claim, ClaimState, Remittance from cyclone.inbox_state import apply_999_rejections from cyclone.inbox_state_277ca import apply_277ca_rejections +from cyclone.audit_log import AuditEvent, append_event, verify_chain from cyclone.parsers.exceptions import CycloneParseError from cyclone.parsers.models import BatchSummary, ClaimOutput, Envelope, ParseResult from cyclone.parsers.models_270 import ( @@ -663,6 +664,20 @@ async def parse_999_endpoint( raw_json=json.loads(result.model_dump_json()), ) + # SP11: append one audit row per rejected claim. Each row chains + # to the previous one — see cyclone.audit_log. + if _rejection_result.matched: + with db.SessionLocal()() as audit_s: + for cid in _rejection_result.matched: + append_event(audit_s, AuditEvent( + event_type="claim.rejected", + entity_type="claim", + entity_id=cid, + payload={"source_batch_id": synthetic_id, "ack_id": row.id}, + actor="999-parser", + )) + audit_s.commit() + return JSONResponse(content={ "ack": { "id": row.id, @@ -918,6 +933,20 @@ async def parse_277ca_endpoint( bus = request.app.state.event_bus for cid in apply_result.matched: await bus.publish("claim.payer_rejected", {"claim_id": cid}) + # SP11: audit trail for each payer-rejected claim. + with db.SessionLocal()() as audit_s: + for cid in apply_result.matched: + append_event(audit_s, AuditEvent( + event_type="claim.payer_rejected", + entity_type="claim", + entity_id=cid, + payload={ + "source_batch_id": synthetic_id, + "277ca_id": row.id, + }, + actor="277ca-parser", + )) + audit_s.commit() if apply_result.orphans: log.warning( "277CA had %d orphan status entries (no matching claim): %s", @@ -2359,6 +2388,21 @@ def submit_to_clearhouse(body: dict): "staging_path": str(staging_path), "remote_path": remote, }) + # SP11: audit trail for each successful clearhouse submission. + with db.SessionLocal()() as audit_s: + append_event(audit_s, AuditEvent( + event_type="clearhouse.submitted", + entity_type="claim", + entity_id=cid, + payload={ + "filename": filename, + "remote_path": remote, + "tpid": ch.tpid, + "stub": ch.sftp_block.stub, + }, + actor="clearhouse-submit", + )) + audit_s.commit() return {"ok": True, "submitted": results, "stub": ch.sftp_block.stub} @@ -2407,6 +2451,72 @@ def list_configured_providers(is_active: bool | None = Query(default=True)): return [json.loads(p.model_dump_json()) for p in store.list_providers(is_active=is_active)] +# --------------------------------------------------------------------------- # +# SP11: tamper-evident audit log (admin) +# --------------------------------------------------------------------------- # + + +@app.get("/api/admin/audit-log") +def list_audit_log_endpoint( + entity_type: str | None = Query(default=None), + entity_id: str | None = Query(default=None), + event_type: str | None = Query(default=None), + limit: int = Query(default=100, ge=1, le=1000), +) -> Any: + """List audit-log rows, newest first, with optional filters. + + Filters match the (entity_type, entity_id) pair (typical use: + "show me everything that happened to claim C-123") or a single + event_type (typical use: "show me all clearhouse.submitted + events today"). + """ + with db.SessionLocal()() as s: + q = s.query(db.AuditLog) + if entity_type: + q = q.filter(db.AuditLog.entity_type == entity_type) + if entity_id: + q = q.filter(db.AuditLog.entity_id == entity_id) + if event_type: + q = q.filter(db.AuditLog.event_type == event_type) + rows = q.order_by(db.AuditLog.id.desc()).limit(limit).all() + return { + "total": len(rows), + "items": [ + { + "id": r.id, + "event_type": r.event_type, + "entity_type": r.entity_type, + "entity_id": r.entity_id, + "actor": r.actor, + "payload": json.loads(r.payload_json) if r.payload_json else None, + "created_at": r.created_at.isoformat() if r.created_at else None, + "prev_hash": r.prev_hash, + "hash": r.hash, + } + for r in rows + ], + } + + +@app.get("/api/admin/audit-log/verify") +def verify_audit_log_endpoint() -> Any: + """Walk the audit-log chain and verify every row's hash. + + Returns ``{"ok": true, "checked": N}`` for a clean chain, or + ``{"ok": false, "checked": K, "first_bad_id": X, "reason": "..."}`` + for a broken chain. This is the operator's "did anyone tamper?" + endpoint; run it on demand or via a nightly cron job. + """ + with db.SessionLocal()() as s: + result = verify_chain(s) + return { + "ok": result.ok, + "checked": result.checked, + "first_bad_id": result.first_bad_id, + "reason": result.reason, + } + + @app.get("/api/config/providers/{npi}") def get_configured_provider(npi: str): p = store.get_provider(npi) diff --git a/backend/src/cyclone/audit_log.py b/backend/src/cyclone/audit_log.py new file mode 100644 index 0000000..8cbe3ac --- /dev/null +++ b/backend/src/cyclone/audit_log.py @@ -0,0 +1,254 @@ +"""Tamper-evident hash-chained audit_log. + +SP11. + +Each row's hash is SHA-256 of +``(id, event_type, entity_type, entity_id, actor, payload_json, +created_at, prev_hash)`` and ``prev_hash`` is the previous row's hash. +That forms a chain: changing any row's payload invalidates every +subsequent row's hash. :func:`verify_chain` walks the chain and +returns the first mismatch index (or ``None`` for a clean chain). + +We use SHA-256 (FIPS-approved, fast on commodity hardware) instead +of a Merkle tree because the chain is linear: every row depends on +exactly one prior row. A Merkle tree would let us prove individual +membership with O(log n) witnesses, but the chain's whole point is +end-to-end integrity, not selective disclosure. + +Append-only by convention: the application MUST NOT call +``session.delete(row)`` or modify an existing row. Doing so is +auditable via :func:`verify_chain`. We deliberately do not enforce +this at the DB level (no triggers, no revoked UPDATE permission) +because that breaks the test fixtures that recreate the DB. +""" +from __future__ import annotations + +import hashlib +import json +import logging +from dataclasses import dataclass, field +from datetime import datetime, timezone +from typing import Any + +from sqlalchemy.orm import Session + +from cyclone.db import AuditLog + +log = logging.getLogger(__name__) + +# 64 hex chars = 256 bits. Constant for easy comparison. +HASH_LEN = 64 + +# Genesis row's prev_hash — a fixed "all zeros" sentinel so the first +# row in the chain has a deterministic predecessor. This is the same +# convention Bitcoin and other ledgers use. +GENESIS_PREV_HASH = "0" * HASH_LEN + + +# --------------------------------------------------------------------------- # +# Hashing +# --------------------------------------------------------------------------- # + + +def _hash_row( + *, + row_id: int, + event_type: str, + entity_type: str, + entity_id: str, + actor: str, + payload_json: str | None, + created_at: datetime, + prev_hash: str, +) -> str: + """Compute SHA-256 hex of a row's canonical form. + + The fields are concatenated with a separator that cannot appear + inside any field (``\\x1f`` — the ASCII unit separator). Using a + delimiter avoids length-ambiguity attacks where two different + payloads with the same string-joined form would hash to the same + digest. + """ + sep = "\x1f" + # Normalize the timestamp to an ISO 8601 UTC string so the hash is + # stable across timezone-aware and timezone-naive datetimes (the + # DB may give us either depending on the SQLite build). + if created_at.tzinfo is None: + created_at = created_at.replace(tzinfo=timezone.utc) + created_at_iso = created_at.astimezone(timezone.utc).isoformat() + payload = payload_json or "" + canonical = sep.join([ + str(row_id), + event_type, + entity_type, + entity_id, + actor, + created_at_iso, + payload, + prev_hash, + ]) + return hashlib.sha256(canonical.encode("utf-8")).hexdigest() + + +# --------------------------------------------------------------------------- # +# Append +# --------------------------------------------------------------------------- # + + +@dataclass +class AuditEvent: + """An audit event ready to be appended. + + Mirrors the ``AuditLog`` row shape minus the auto-assigned id and + computed hash. Payload must be JSON-serializable; the audit_log + module handles the encoding so callers don't need to think about + canonical form. + """ + + event_type: str + entity_type: str + entity_id: str + payload: dict[str, Any] = field(default_factory=dict) + actor: str = "system" + created_at: datetime | None = None + + +def append_event( + session: Session, + event: AuditEvent, +) -> AuditLog: + """Append one event to the audit_log chain and return the row. + + The caller is responsible for ``session.commit()`` — this lets + callers batch multiple appends into one transaction (e.g., a + parser that appends one event per parsed claim). + """ + # Read the latest hash within the same session so concurrent + # appends don't see stale state. SQLite default isolation level + # gives us serializable reads for this query; for Postgres we'd + # need SELECT ... FOR UPDATE but that's overkill for v1. + latest = ( + session.query(AuditLog) + .order_by(AuditLog.id.desc()) + .first() + ) + prev_hash = latest.hash if latest is not None else GENESIS_PREV_HASH + + created_at = event.created_at or datetime.now(timezone.utc) + if created_at.tzinfo is None: + created_at = created_at.replace(tzinfo=timezone.utc) + + # Canonical payload form: sort_keys + compact separators. This + # makes the hash independent of dict insertion order across + # Python versions and across API runs. + payload_json = json.dumps(event.payload, sort_keys=True, separators=(",", ":")) if event.payload else None + + # Insert the row with a placeholder hash, then UPDATE once we + # know the auto-assigned id. SQLite + SQLAlchemy gives us the id + # after the INSERT, so we can compute the real hash then. + row = AuditLog( + event_type=event.event_type, + entity_type=event.entity_type, + entity_id=event.entity_id, + actor=event.actor, + payload_json=payload_json, + created_at=created_at, + prev_hash=prev_hash, + hash=GENESIS_PREV_HASH, # placeholder; updated below + ) + session.add(row) + session.flush() # populate row.id + row.hash = _hash_row( + row_id=row.id, + event_type=row.event_type, + entity_type=row.entity_type, + entity_id=row.entity_id, + actor=row.actor, + payload_json=row.payload_json, + created_at=row.created_at, + prev_hash=row.prev_hash, + ) + session.flush() + return row + + +# --------------------------------------------------------------------------- # +# Verify +# --------------------------------------------------------------------------- # + + +@dataclass +class VerifyResult: + """Outcome of :func:`verify_chain`.""" + + ok: bool + checked: int + first_bad_id: int | None = None + reason: str | None = None + + +def verify_chain(session: Session) -> VerifyResult: + """Walk the audit_log and verify every row's hash. Returns the first mismatch. + + A clean chain returns ``VerifyResult(ok=True, checked=N)``. A + broken chain returns ``ok=False, first_bad_id=X, reason='...'`` + describing what went wrong (hash mismatch, prev_hash mismatch, + or non-monotonic id). + + This is intended to be called by the operator (e.g., a nightly + cron job or the admin UI's "Verify Audit Chain" button). It is + NOT a fast operation — for a 6-year-old chain with millions of + rows, expect seconds-to-minutes. Call it rarely. + """ + rows = session.query(AuditLog).order_by(AuditLog.id.asc()).all() + if not rows: + return VerifyResult(ok=True, checked=0) + + expected_prev = GENESIS_PREV_HASH + last_id = 0 + for i, row in enumerate(rows): + # Monotonic id check — covers attempted inserts with a + # custom id, or accidental out-of-order rows. + if row.id <= last_id: + return VerifyResult( + ok=False, checked=i, first_bad_id=row.id, + reason=f"non-monotonic id (previous={last_id}, this={row.id})", + ) + last_id = row.id + + # Recompute the hash from the row's content and compare. + expected_hash = _hash_row( + row_id=row.id, + event_type=row.event_type, + entity_type=row.entity_type, + entity_id=row.entity_id, + actor=row.actor, + payload_json=row.payload_json, + created_at=row.created_at, + prev_hash=row.prev_hash, + ) + if expected_hash != row.hash: + return VerifyResult( + ok=False, checked=i, first_bad_id=row.id, + reason=f"hash mismatch (stored={row.hash[:16]}…, computed={expected_hash[:16]}…)", + ) + + # Check prev_hash linkage. + if row.prev_hash != expected_prev: + return VerifyResult( + ok=False, checked=i, first_bad_id=row.id, + reason=f"prev_hash mismatch (stored={row.prev_hash[:16]}…, expected={expected_prev[:16]}…)", + ) + expected_prev = row.hash + + return VerifyResult(ok=True, checked=len(rows)) + + +__all__ = [ + "AuditEvent", + "GENESIS_PREV_HASH", + "HASH_LEN", + "VerifyResult", + "append_event", + "verify_chain", +] diff --git a/backend/src/cyclone/db.py b/backend/src/cyclone/db.py index 7e2893f..33ccf31 100644 --- a/backend/src/cyclone/db.py +++ b/backend/src/cyclone/db.py @@ -570,6 +570,45 @@ class Two77caAck(Base): ) +# --------------------------------------------------------------------------- +# SP11: tamper-evident hash-chained audit_log +# --------------------------------------------------------------------------- + + +class AuditLog(Base): + """One row per audit event. Append-only by convention. + + Each row's :attr:`hash` is SHA-256 of + ``(id, event_type, entity_type, entity_id, actor, payload_json, + created_at, prev_hash)`` — and :attr:`prev_hash` is the previous + row's ``hash``. That forms a tamper-evident chain: changing any + row's payload invalidates every subsequent row's hash. + + See ``cyclone.audit_log.append_event`` and ``verify_chain`` for + the append + verify operations. The application code MUST NOT + UPDATE or DELETE rows; doing so breaks the chain and is + auditable via :func:`verify_chain`. + """ + + __tablename__ = "audit_log" + + id: Mapped[int] = mapped_column(Integer, primary_key=True, autoincrement=True) + event_type: Mapped[str] = mapped_column(String(64), nullable=False) + entity_type: Mapped[str] = mapped_column(String(64), nullable=False) + entity_id: Mapped[str] = mapped_column(String(64), nullable=False) + actor: Mapped[str] = mapped_column(String(64), nullable=False, default="system") + payload_json: Mapped[Optional[str]] = mapped_column(Text, nullable=True) + created_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), nullable=False) + prev_hash: Mapped[str] = mapped_column(String(64), nullable=False) + hash: Mapped[str] = mapped_column(String(64), nullable=False) + + __table_args__ = ( + Index("idx_audit_log_entity", "entity_type", "entity_id"), + Index("idx_audit_log_event_type", "event_type"), + Index("idx_audit_log_created_at", "created_at"), + ) + + # --------------------------------------------------------------------------- # SP9: providers, payers, payer_configs, clearhouse # --------------------------------------------------------------------------- diff --git a/backend/src/cyclone/migrations/0009_audit_log.sql b/backend/src/cyclone/migrations/0009_audit_log.sql new file mode 100644 index 0000000..7b76788 --- /dev/null +++ b/backend/src/cyclone/migrations/0009_audit_log.sql @@ -0,0 +1,33 @@ +-- version: 9 +-- SP11: tamper-evident hash-chained audit_log +-- +-- Each row carries a SHA-256 hash of (id, event_type, entity_type, +-- entity_id, actor, payload_json, created_at, prev_hash). The prev_hash +-- field chains the row to the previous row's hash — a tamper-evident +-- Merkle-like chain (no tree, just a list). +-- +-- Append-only by convention: no UPDATE/DELETE in the application code. +-- Compliance: HIPAA §164.316(b)(2) requires 6-year retention. We don't +-- enforce retention in the schema (no TTL), but a separate vacuum +-- job (out of scope here) can prune rows older than 6 years after +-- exporting them to cold storage. +-- +-- Indexes: (entity_type, entity_id) for "show me the audit trail for +-- this claim"; (event_type) for "show me all clearhouse.submitted +-- events"; (created_at) for time-range scans. + +CREATE TABLE audit_log ( + id INTEGER PRIMARY KEY AUTOINCREMENT, + event_type TEXT NOT NULL, + entity_type TEXT NOT NULL, + entity_id TEXT NOT NULL, + actor TEXT NOT NULL DEFAULT 'system', + payload_json TEXT, + created_at TEXT NOT NULL, + prev_hash TEXT NOT NULL, + hash TEXT NOT NULL +); + +CREATE INDEX idx_audit_log_entity ON audit_log(entity_type, entity_id); +CREATE INDEX idx_audit_log_event_type ON audit_log(event_type); +CREATE INDEX idx_audit_log_created_at ON audit_log(created_at); diff --git a/backend/tests/test_acks.py b/backend/tests/test_acks.py index faaf7d3..ed75b6e 100644 --- a/backend/tests/test_acks.py +++ b/backend/tests/test_acks.py @@ -51,17 +51,18 @@ def test_migration_0002_creates_acks_table(): def test_migration_latest_idempotent_on_fresh_db(): """Re-running the migration on the same DB must be a no-op (PRAGMA - user_version already at the latest version — currently 8 after + user_version already at the latest version — currently 9 after 0004-0006 line_reconciliation, 0005 ta1_acks, SP9's 0007 - providers/payers/clearhouse, and SP10's 0008 payer_rejected).""" + providers/payers/clearhouse, SP10's 0008 payer_rejected, and + SP11's 0009 audit_log).""" with db.engine().begin() as c: v1 = c.exec_driver_sql("PRAGMA user_version").scalar() or 0 - assert v1 == 8 + assert v1 == 9 # A second run should not raise and should not bump the version. db_migrate.run(db.engine()) with db.engine().begin() as c: v2 = c.exec_driver_sql("PRAGMA user_version").scalar() or 0 - assert v2 == 8 + assert v2 == 9 def test_add_ack_persists_row(): diff --git a/backend/tests/test_api_audit_log.py b/backend/tests/test_api_audit_log.py new file mode 100644 index 0000000..371966e --- /dev/null +++ b/backend/tests/test_api_audit_log.py @@ -0,0 +1,159 @@ +"""Tests for the audit-log admin API endpoints. SP11 T3.""" +from __future__ import annotations + +import json + +import pytest +from fastapi.testclient import TestClient + +from cyclone import db +from cyclone.audit_log import AuditEvent, append_event +from cyclone.api import app +from cyclone.db import AuditLog, init_db + + +@pytest.fixture(autouse=True) +def _fresh_db(): + init_db() + yield + + +@pytest.fixture +def client() -> TestClient: + return TestClient(app) + + +def _seed_audit(rows: list[AuditEvent]) -> None: + with db.SessionLocal()() as s: + for ev in rows: + append_event(s, ev) + s.commit() + + +# --------------------------------------------------------------------------- # +# List endpoint +# --------------------------------------------------------------------------- # + + +class TestListAuditLog: + def test_empty_returns_zero_items(self, client: TestClient): + r = client.get("/api/admin/audit-log") + assert r.status_code == 200 + body = r.json() + assert body["total"] == 0 + assert body["items"] == [] + + def test_returns_newest_first(self, client: TestClient): + _seed_audit([ + AuditEvent(event_type="a.first", entity_type="claim", entity_id="c1"), + AuditEvent(event_type="a.second", entity_type="claim", entity_id="c1"), + AuditEvent(event_type="a.third", entity_type="claim", entity_id="c1"), + ]) + r = client.get("/api/admin/audit-log") + assert r.status_code == 200 + items = r.json()["items"] + assert [i["event_type"] for i in items] == ["a.third", "a.second", "a.first"] + + def test_filter_by_entity(self, client: TestClient): + _seed_audit([ + AuditEvent(event_type="x", entity_type="claim", entity_id="c1"), + AuditEvent(event_type="x", entity_type="claim", entity_id="c2"), + AuditEvent(event_type="x", entity_type="batch", entity_id="b1"), + ]) + r = client.get("/api/admin/audit-log", params={"entity_type": "claim", "entity_id": "c1"}) + items = r.json()["items"] + assert len(items) == 1 + assert items[0]["entity_id"] == "c1" + + def test_filter_by_event_type(self, client: TestClient): + _seed_audit([ + AuditEvent(event_type="claim.parsed", entity_type="claim", entity_id="c1"), + AuditEvent(event_type="claim.rejected", entity_type="claim", entity_id="c1"), + ]) + r = client.get("/api/admin/audit-log", params={"event_type": "claim.rejected"}) + items = r.json()["items"] + assert len(items) == 1 + assert items[0]["event_type"] == "claim.rejected" + + +# --------------------------------------------------------------------------- # +# Verify endpoint +# --------------------------------------------------------------------------- # + + +class TestVerifyAuditLog: + def test_empty_chain_returns_ok(self, client: TestClient): + r = client.get("/api/admin/audit-log/verify") + assert r.status_code == 200 + body = r.json() + assert body["ok"] is True + assert body["checked"] == 0 + + def test_clean_chain_returns_ok(self, client: TestClient): + _seed_audit([ + AuditEvent(event_type="x", entity_type="y", entity_id=f"id-{i}") + for i in range(10) + ]) + r = client.get("/api/admin/audit-log/verify") + body = r.json() + assert body["ok"] is True + assert body["checked"] == 10 + + def test_tampered_chain_returns_failure(self, client: TestClient): + _seed_audit([ + AuditEvent(event_type="x", entity_type="y", entity_id=f"id-{i}") + for i in range(5) + ]) + with db.SessionLocal()() as s: + row = s.query(AuditLog).filter(AuditLog.id == 3).first() + row.payload_json = json.dumps({"evil": True}) + s.commit() + r = client.get("/api/admin/audit-log/verify") + body = r.json() + assert body["ok"] is False + assert body["first_bad_id"] == 3 + assert body["reason"] + + +# --------------------------------------------------------------------------- # +# End-to-end: 999 → audit row appears in list +# --------------------------------------------------------------------------- # + + +class TestAuditLogHookedIntoEndpoints: + def test_parse_999_creates_audit_rows(self, client: TestClient): + """Upload a 999 with one rejected set, see an audit row appear.""" + # Seed a claim so the 999 rejection matches. + from cyclone.db import Claim, ClaimState + with db.SessionLocal()() as s: + s.add(Claim( + id="c1", batch_id="B-1", patient_control_number="0001", + state=ClaimState.SUBMITTED, + )) + s.commit() + + text = ( + "ISA*00* *00* *ZZ*AAAAAAAAAAAAAAA*ZZ*BBBBBBBBBBBBBBB*240620*1200*^*00501*000000001*0*P*:~" + "GS*HN*A*B*20240620*1200*1*X*005010X231A1~" + "ST*999*0001*005010X231A1~" + "AK1*HC*0001~" + "AK2*837*0001~" + "AK5*R~" + "AK9*R*1*0*1~" + "SE*7*0001~" + "GE*1*1~" + "IEA*1*000000001~" + ) + r = client.post( + "/api/parse-999", + files={"file": ("r.txt", text, "text/plain")}, + ) + assert r.status_code == 200, r.text + + # One audit row should have appeared for the rejected claim. + r2 = client.get( + "/api/admin/audit-log", + params={"entity_type": "claim", "entity_id": "c1"}, + ) + items = r2.json()["items"] + assert any(i["event_type"] == "claim.rejected" for i in items) diff --git a/backend/tests/test_audit_log.py b/backend/tests/test_audit_log.py new file mode 100644 index 0000000..76c8e3e --- /dev/null +++ b/backend/tests/test_audit_log.py @@ -0,0 +1,187 @@ +"""Tests for the tamper-evident hash-chained audit_log. + +SP11 T1. +""" +from __future__ import annotations + +from datetime import datetime, timezone + +import json +import pytest + +from cyclone import db +from cyclone.audit_log import ( + GENESIS_PREV_HASH, + HASH_LEN, + AuditEvent, + append_event, + verify_chain, +) +from cyclone.db import AuditLog, init_db + + +@pytest.fixture(autouse=True) +def _fresh_db(): + init_db() + yield + + +def _row_count() -> int: + with db.SessionLocal()() as s: + return s.query(AuditLog).count() + + +# --------------------------------------------------------------------------- # +# Append +# --------------------------------------------------------------------------- # + + +class TestAppendEvent: + def test_first_row_has_genesis_prev_hash(self): + """The first row in a fresh chain uses the all-zeros prev_hash.""" + with db.SessionLocal()() as s: + row = append_event(s, AuditEvent( + event_type="claim.parsed", + entity_type="claim", + entity_id="c1", + payload={"patient_control_number": "PCN-1"}, + )) + s.commit() + assert row.id == 1 + assert row.prev_hash == GENESIS_PREV_HASH + assert len(row.hash) == HASH_LEN + + def test_second_row_chains_to_first(self): + """Row N's prev_hash == row N-1's hash.""" + with db.SessionLocal()() as s: + r1 = append_event(s, AuditEvent( + event_type="claim.parsed", entity_type="claim", entity_id="c1", + )) + s.commit() + r2 = append_event(s, AuditEvent( + event_type="claim.rejected", entity_type="claim", entity_id="c1", + )) + s.commit() + assert r2.prev_hash == r1.hash + assert r2.id == r1.id + 1 + + def test_payload_canonicalization(self): + """Two payloads with the same content but different key order hash the same. + + We rely on sort_keys=True to make the canonical form independent + of dict insertion order. + """ + with db.SessionLocal()() as s: + r1 = append_event(s, AuditEvent( + event_type="x", entity_type="y", entity_id="z", + payload={"a": 1, "b": 2}, + )) + s.commit() + # New session to drop any in-memory ordering cache. + with db.SessionLocal()() as s2: + r2 = append_event(s2, AuditEvent( + event_type="x", entity_type="y", entity_id="z", + payload={"b": 2, "a": 1}, + )) + s2.commit() + # The two rows have different IDs and different prev_hash + # inputs, but their hash *recipe* (modulo id + prev_hash) is + # the same. Just confirm the rows are different. + assert r1.id != r2.id + # The payload_json fields ARE byte-identical (canonical form). + assert r1.payload_json == r2.payload_json + + +# --------------------------------------------------------------------------- # +# Verify +# --------------------------------------------------------------------------- # + + +class TestVerifyChain: + def test_empty_chain_is_ok(self): + """No rows = nothing to verify = ok.""" + with db.SessionLocal()() as s: + result = verify_chain(s) + assert result.ok is True + assert result.checked == 0 + + def test_single_row_chain_is_ok(self): + with db.SessionLocal()() as s: + append_event(s, AuditEvent("x", "y", "z")) + s.commit() + result = verify_chain(s) + assert result.ok is True + assert result.checked == 1 + + def test_long_chain_is_ok(self): + with db.SessionLocal()() as s: + for i in range(50): + append_event(s, AuditEvent("x", "y", f"id-{i}")) + s.commit() + result = verify_chain(s) + assert result.ok is True + assert result.checked == 50 + + def test_tampered_payload_detected(self): + """Modifying a row's payload_json breaks the chain at that row.""" + with db.SessionLocal()() as s: + for i in range(5): + append_event(s, AuditEvent("x", "y", f"id-{i}")) + s.commit() + # Tamper with row #3's payload (the first 3 rows are still + # valid; row 3 will fail). + with db.SessionLocal()() as s: + row = s.query(AuditLog).filter(AuditLog.id == 3).first() + row.payload_json = json.dumps({"evil": "tampered"}) + s.commit() + with db.SessionLocal()() as s: + result = verify_chain(s) + assert result.ok is False + assert result.first_bad_id == 3 + assert "hash" in (result.reason or "").lower() + + def test_tampered_prev_hash_detected(self): + """Modifying a row's prev_hash invalidates that row's own hash. + + Because the row's hash is computed from prev_hash (and the + other fields), changing prev_hash changes the row's own hash, + so the verifier detects the breakage at the tampered row + itself (not the row after). The next-row check (which compares + prev_hash to the previous row's hash) would catch a different + class of tampering: a row whose prev_hash was set to a value + matching the previous row's hash but whose own hash was also + regenerated — but that requires recomputing the hash to match, + which is what verify_chain would not detect by itself (the + content-vs-hash check still catches content tampering). + """ + with db.SessionLocal()() as s: + for i in range(5): + append_event(s, AuditEvent("x", "y", f"id-{i}")) + s.commit() + with db.SessionLocal()() as s: + row = s.query(AuditLog).filter(AuditLog.id == 2).first() + row.prev_hash = "f" * 64 # fake + s.commit() + with db.SessionLocal()() as s: + result = verify_chain(s) + assert result.ok is False + # The tampered row itself fails its hash check first. + assert result.first_bad_id == 2 + assert "hash" in (result.reason or "").lower() + + def test_deleted_row_detected(self): + """Deleting a middle row breaks the chain at the row after it.""" + with db.SessionLocal()() as s: + for i in range(5): + append_event(s, AuditEvent("x", "y", f"id-{i}")) + s.commit() + with db.SessionLocal()() as s: + row = s.query(AuditLog).filter(AuditLog.id == 3).first() + s.delete(row) + s.commit() + with db.SessionLocal()() as s: + result = verify_chain(s) + assert result.ok is False + # Row 3 is gone; row 4's prev_hash now points at row 2's hash, + # which doesn't match row 4's stored prev_hash. + assert result.first_bad_id == 4 diff --git a/docs/reference/co-medicaid.md b/docs/reference/co-medicaid.md index 5972514..0ef8eb1 100644 --- a/docs/reference/co-medicaid.md +++ b/docs/reference/co-medicaid.md @@ -147,6 +147,37 @@ All CMS POS codes `01`–`99` are accepted. The canonical list lives in `cyclone/parsers/payer.py` as `CMS_PLACE_OF_SERVICE_CODES` and is the source of truth for validation and any UI dropdowns. +## Audit log (SP11) + +Cyclone persists every state-changing event to a tamper-evident +hash-chained audit log. Each row carries a SHA-256 hash of +`(id, event_type, entity_type, entity_id, actor, payload_json, +created_at, prev_hash)`, where `prev_hash` is the previous row's +hash. Modifying any row's payload invalidates every subsequent row's +hash. + +To list events: + +```bash +curl 'http://localhost:8000/api/admin/audit-log?entity_type=claim&entity_id=C-123' +``` + +To verify the chain (run nightly or on demand): + +```bash +curl http://localhost:8000/api/admin/audit-log/verify +# → {"ok": true, "checked": 1234} +# → {"ok": false, "checked": 1180, "first_bad_id": 1181, "reason": "hash mismatch ..."} +``` + +Events written today: +- `claim.rejected` (999 ACK AK5 R/E) +- `claim.payer_rejected` (277CA STC A4/A6/A7) +- `clearhouse.submitted` (SFTP submit) + +Compliance: HIPAA §164.316(b)(2) requires 6-year retention. The +schema doesn't enforce retention — that's a separate vacuum job. + ## 277CA Claim Acknowledgment (SP10) After Gainwell accepts our 837P file (999 AK5=A) and adjudicates the