Implement AUDIT-WP-0007 hash-chain integrity.

Accept now extends a single-schema chain. Verify walks it; a rewritten
payload_hash is a break. Tamper evidence is that detector plus an
external chain-head attestation, not WORM.
This commit is contained in:
tegwick 2026-08-16 01:18:30 +02:00
parent 5faede18fc
commit 5fd04e2095
17 changed files with 696 additions and 29 deletions

View file

@ -107,6 +107,34 @@ def build_parser() -> argparse.ArgumentParser:
)
migrate_parser.set_defaults(func=migrate_store)
verify_parser = sub.add_parser(
"verify-chain",
help="Walk the custody hash chain; exit 0 only if intact.",
)
verify_parser.add_argument(
"--to-postgres", dest="destination", default=None,
help="DSN; defaults to AUDIT_CORE_DATABASE_URL or a mounted credential dir.",
)
verify_parser.add_argument("--schema", default="audit_core")
verify_parser.add_argument(
"--against",
default=None,
help="Chain-head attestation JSON. A cited head missing from the live chain is a break.",
)
verify_parser.set_defaults(func=verify_chain)
attest_parser = sub.add_parser(
"attest-chain",
help="Write the live chain head to a file outside the database.",
)
attest_parser.add_argument(
"--to-postgres", dest="destination", default=None,
help="DSN; defaults to AUDIT_CORE_DATABASE_URL or a mounted credential dir.",
)
attest_parser.add_argument("--schema", default="audit_core")
attest_parser.add_argument("--output", required=True)
attest_parser.set_defaults(func=attest_chain)
return parser
@ -158,6 +186,32 @@ def replay_event(args: argparse.Namespace) -> int:
return 0 if result.duplicate else 2
def verify_chain(args: argparse.Namespace) -> int:
from audit_core.integrity import load_attestation
backend = _postgres_backend(args.destination, args.schema, migrate=False)
try:
cited = load_attestation(args.against) if args.against else None
report = backend.verify_chain(cited)
finally:
backend.close()
print(json.dumps(report.as_dict(), sort_keys=True))
return 0 if report.intact else 1
def attest_chain(args: argparse.Namespace) -> int:
from audit_core.integrity import write_attestation
backend = _postgres_backend(args.destination, args.schema, migrate=False)
try:
report = backend.verify_chain()
body = write_attestation(args.output, report)
finally:
backend.close()
print(json.dumps({"ok": report.intact, "path": args.output, **body}, sort_keys=True))
return 0 if report.intact else 1
def migrate_store(args: argparse.Namespace) -> int:
import os