tyler/cyclone
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat(sp11): tamper-evident hash-chained audit_log

Browse Source 
- 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.
This commit is contained in:
Tyler
2026-06-20 23:45:43 -06:00
parent 9acdcb8dbd
commit 62bb09f183
8 changed files with 818 additions and 4 deletions
Download Patch File Download Diff File Expand all files Collapse all files
backend/src/cyclone/api.py
+110
View File
@@ -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)
backend/src/cyclone/audit_log.py
+254
View File
@@ -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",
]
backend/src/cyclone/db.py
+39
View File
@@ -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
# ---------------------------------------------------------------------------
backend/src/cyclone/migrations/0009_audit_log.sql
+33
View File
@@ -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);
backend/tests/test_acks.py
+5 -4
View File
@@ -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():
backend/tests/test_api_audit_log.py
+159
View File
@@ -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)
backend/tests/test_audit_log.py
+187
View File
@@ -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