Generating an Inspection-Ready Audit-Trail Export
When an inspector asks for the history of one submission, the answer has to arrive as a self-contained package they can open, read, and independently verify — not a live query against a production database they will never be granted access to. This walkthrough builds that package: pull every audit event tied to one submission identifier, verify the hash chain across the full sequence, redact anything that must not leave the compliance boundary, and render a timestamped, attributable report bundled with a manifest that proves the export itself has not been altered since it was generated. It is the concrete implementation behind Inspection Readiness & Audit-Trail Reporting, part of the Core Architecture & Regulatory Mapping for Clinical Trials domain, and the export process itself is recorded in the append-only audit log it draws from.
Why naive approaches fail
The first instinct is to write a database query, dump the rows to a spreadsheet, and call it an export. That approach fails an inspection in several specific ways:
- No proof the rows are complete. A spreadsheet export cannot show that no event was silently excluded from the query — only a verified hash chain, where each link depends on the one before it, can demonstrate that the sequence handed over is the whole sequence.
- No proof the rows are unaltered. Without recomputing each event’s hash against its stored value, a single edited field in a spreadsheet is indistinguishable from an unedited one.
- The export itself is unverifiable after the fact. If the exported file has no integrity proof of its own, there is no way to confirm — a week or a year later — that the file an inspector received is the same one that was generated, rather than a version edited after export.
- PHI leaks through an untyped dump. A raw
SELECT *frequently pulls fields that were never meant to leave the compliance boundary; an export pipeline that formats rows directly from a query has no redaction step to catch that.
Architecture overview
The export pipeline runs the same five stages regardless of the reporting tool wrapped around it: pull the ordered event sequence for one subject, verify its hash chain, redact anything sensitive, render a human-readable report, and package the result with a manifest that proves its own integrity.
Setup and configuration
Everything the export needs is in the standard library — hashlib for chain verification and manifest hashing, json for the raw record dump, zoneinfo for rendering timestamps in a stated timezone. No third-party dependency is required for the integrity-critical path.
python -m venv .venv && source .venv/bin/activateConfiguration — where completed exports are written and which timezone the human-readable report renders timestamps in — is read from the environment, never hardcoded, so the same script runs unchanged in a validated production environment and a local reproduction:
"""Environment-driven configuration for the audit export tool."""
from __future__ import annotations
import os
from dataclasses import dataclass
@dataclass(frozen=True)
class ExportConfig:
output_dir: str
display_timezone: str
@classmethod
def from_env(cls) -> "ExportConfig":
def required(name: str) -> str:
value = os.environ.get(name)
if not value:
raise RuntimeError(f"Missing required environment variable: {name}")
return value
return cls(
output_dir=required("AUDIT_EXPORT_OUTPUT_DIR"),
display_timezone=os.environ.get("AUDIT_EXPORT_DISPLAY_TZ", "UTC"),
)Full working implementation
The export function ties together chain verification, redaction, rendering, and packaging into one auditable call. Each event is modeled as a frozen dataclass so nothing downstream can mutate a record while the export is in progress.
"""Generate a self-contained, integrity-proven audit export for one subject."""
from __future__ import annotations
import hashlib
import json
from dataclasses import dataclass
from datetime import datetime, timezone
from pathlib import Path
from zoneinfo import ZoneInfo
@dataclass(frozen=True)
class AuditEvent:
event_id: str
subject_id: str
ts: datetime
actor: str
action: str
prev_hash: str
entry_hash: str
fields: dict
@dataclass(frozen=True)
class ChainBreak:
index: int
event_id: str
reason: str
class ExportIntegrityError(RuntimeError):
"""Raised when the chain fails verification and export must not proceed silently."""
def _recompute_entry_hash(event: AuditEvent) -> str:
body = {
"event_id": event.event_id,
"subject_id": event.subject_id,
"ts": event.ts.isoformat(),
"actor": event.actor,
"action": event.action,
"prev_hash": event.prev_hash,
"fields": event.fields,
}
serialized = json.dumps(body, sort_keys=True, separators=(",", ":"))
return hashlib.sha256(serialized.encode()).hexdigest()
def verify_chain(events: list[AuditEvent]) -> list[ChainBreak]:
breaks: list[ChainBreak] = []
previous_hash = ""
for index, event in enumerate(events):
if event.prev_hash != previous_hash:
breaks.append(ChainBreak(index, event.event_id,
f"prev_hash mismatch at position {index}"))
if _recompute_entry_hash(event) != event.entry_hash:
breaks.append(ChainBreak(index, event.event_id,
"entry_hash does not match recomputed value"))
previous_hash = event.entry_hash
return breaks
_PHI_KEYS = frozenset({"patient_name", "dob", "mrn", "ssn", "address"})
def redact(fields: dict) -> dict:
return {
key: ("[REDACTED]" if key in _PHI_KEYS else value)
for key, value in fields.items()
}
def render_report(subject_id: str, events: list[AuditEvent],
breaks: list[ChainBreak], tz_name: str) -> str:
tz = ZoneInfo(tz_name)
lines = [
f"Inspection audit trail — subject {subject_id}",
f"Chain verified: {'YES' if not breaks else 'NO — see breaks below'}",
f"Event count: {len(events)}",
"",
]
for event in events:
local_ts = event.ts.astimezone(tz).isoformat()
lines.append(f"[{local_ts}] {event.actor} — {event.action} (event_id={event.event_id})")
if breaks:
lines.append("")
lines.append("CHAIN INTEGRITY BREAKS:")
for brk in breaks:
lines.append(f" index={brk.index} event_id={brk.event_id}: {brk.reason}")
return "\n".join(lines)
def export_audit_trail(
subject_id: str,
events: list[AuditEvent],
config: "ExportConfig",
generated_by: str,
) -> Path:
"""Write a self-contained export package and return its directory path.
Raises ExportIntegrityError if the chain does not verify — an export
must never be produced over an unverified chain without that fact
being the headline of the resulting package.
"""
breaks = verify_chain(events)
redacted_events = [
AuditEvent(
event_id=e.event_id, subject_id=e.subject_id, ts=e.ts,
actor=e.actor, action=e.action, prev_hash=e.prev_hash,
entry_hash=e.entry_hash, fields=redact(e.fields),
)
for e in events
]
report_text = render_report(subject_id, redacted_events, breaks, config.display_timezone)
export_dir = Path(config.output_dir) / f"export-{subject_id}"
export_dir.mkdir(parents=True, exist_ok=True)
events_path = export_dir / "events.jsonl"
with events_path.open("w", encoding="utf-8") as fh:
for e in redacted_events:
fh.write(json.dumps({
"event_id": e.event_id, "ts": e.ts.isoformat(),
"actor": e.actor, "action": e.action,
"prev_hash": e.prev_hash, "entry_hash": e.entry_hash,
"fields": e.fields,
}, sort_keys=True) + "\n")
report_path = export_dir / "report.txt"
report_path.write_text(report_text, encoding="utf-8")
manifest = {
"subject_id": subject_id,
"generated_by": generated_by,
"generated_at": datetime.now(timezone.utc).isoformat(),
"event_count": len(events),
"chain_verified": not breaks,
"events_jsonl_sha256": hashlib.sha256(events_path.read_bytes()).hexdigest(),
"report_txt_sha256": hashlib.sha256(report_path.read_bytes()).hexdigest(),
}
(export_dir / "manifest.json").write_text(
json.dumps(manifest, sort_keys=True, indent=2), encoding="utf-8"
)
if breaks:
raise ExportIntegrityError(
f"exported {subject_id} with {len(breaks)} chain break(s); "
"see report.txt and manifest.json for detail"
)
return export_dirThe manifest is generated last, after both other files are written to disk, and records the SHA-256 of each — an inspector (or a later internal reviewer) can recompute those two hashes independently and confirm the files have not changed since the export ran. The function still raises ExportIntegrityError on a broken chain rather than swallowing the problem, but it writes the full package first, so the evidence of what broke and where is itself preserved on disk rather than lost to an exception.
Validation and edge-case handling
Three situations deserve explicit handling beyond the happy path:
- A subject identifier with no matching events. Treat this as a distinct, clearly labeled outcome — “zero events found for this identifier” — rather than producing an empty report indistinguishable from a verified-empty history. The caller should decide whether that reflects a typo in the identifier or a genuinely uneventful record.
- A chain break partway through a long sequence. The export still completes and still writes every file, including the events after the break — an inspector needs to see the entire sequence to judge the scope of the problem, not just the events that verified cleanly before it.
- Re-running the export for the same subject later. Because the manifest records a generation timestamp and is written fresh each run, two exports of the same subject taken at different times are expected to differ in their manifest even if the underlying events are identical — only
events_jsonl_sha256andreport_txt_sha256need to match if nothing changed in between, and a mismatch there is itself worth investigating.
Testing and verification
The test suite proves the two properties an inspector will actually probe: that a clean chain produces a verified export, and that a single altered field anywhere in the sequence is caught rather than silently exported.
"""Tests for the inspection-ready audit export pipeline."""
from __future__ import annotations
import hashlib
import json
from datetime import datetime, timezone
from pathlib import Path
import pytest
def _make_event(event_id: str, prev_hash: str, action: str) -> AuditEvent:
body = {
"event_id": event_id, "subject_id": "SEQ-0007",
"ts": datetime(2026, 7, 10, 12, 0, 0, tzinfo=timezone.utc).isoformat(),
"actor": "j.chen", "action": action, "prev_hash": prev_hash, "fields": {},
}
entry_hash = hashlib.sha256(
json.dumps(body, sort_keys=True, separators=(",", ":")).encode()
).hexdigest()
return AuditEvent(
event_id=event_id, subject_id="SEQ-0007",
ts=datetime(2026, 7, 10, 12, 0, 0, tzinfo=timezone.utc),
actor="j.chen", action=action, prev_hash=prev_hash,
entry_hash=entry_hash, fields={},
)
def test_clean_chain_exports_without_raising(tmp_path: Path) -> None:
e1 = _make_event("EVT-1", "", "sequence_assembled")
e2 = _make_event("EVT-2", e1.entry_hash, "sequence_signed")
config = ExportConfig(output_dir=str(tmp_path), display_timezone="UTC")
export_dir = export_audit_trail("SEQ-0007", [e1, e2], config, generated_by="j.chen")
manifest = json.loads((export_dir / "manifest.json").read_text())
assert manifest["chain_verified"] is True
def test_tampered_event_is_detected(tmp_path: Path) -> None:
e1 = _make_event("EVT-1", "", "sequence_assembled")
e2 = _make_event("EVT-2", e1.entry_hash, "sequence_signed")
tampered = AuditEvent(
event_id=e2.event_id, subject_id=e2.subject_id, ts=e2.ts,
actor="unauthorized", action=e2.action, prev_hash=e2.prev_hash,
entry_hash=e2.entry_hash, fields=e2.fields,
)
config = ExportConfig(output_dir=str(tmp_path), display_timezone="UTC")
with pytest.raises(ExportIntegrityError):
export_audit_trail("SEQ-0007", [e1, tampered], config, generated_by="j.chen")The second test changes only the actor field on the second event without recomputing its entry_hash — exactly the mistake an attacker (or a careless manual edit) would make — and confirms the export still runs to completion, still writes every file, and still raises so the caller cannot mistake a broken chain for a clean one.
FAQ
Why does the export still write files even when the chain fails verification?
An inspector investigating a possible integrity problem needs the full, honest picture of what was found, including the events surrounding a break — silently discarding output on failure would destroy exactly the evidence needed to investigate. Raising an exception after writing ensures the caller cannot treat a failed export as if it had succeeded, while the on-disk package still preserves everything for review.
What does the manifest actually prove?
It records a SHA-256 of the events file and the rendered report at the moment of export, so anyone who later receives the package can recompute both hashes and confirm neither file has changed since generation. It does not, on its own, prove the underlying audit events were never tampered with before export — that proof comes from the hash-chain verification recorded inside the manifest, not from the manifest’s own file hashes.
Can this export be re-run safely for the same subject?
Yes. Each run is independent and side-effect-free against the source log — it only reads events and writes a fresh export directory — so re-running after new events have been appended simply produces a larger, still-verifiable export. Comparing two exports taken at different times is a reasonable way to confirm that no historical event was altered in between, since the hash of any unchanged prefix of the chain will be identical.