#!/usr/bin/env python3 """ AEGIS CYBER — Audit Chain Independent Verification Walker Specification: docs/regulatory/AEGIS_CYBER_AUDIT_CHAIN_VERIFICATION_SPEC.md v1.0-RATIFIED Verifies the integrity of an AEGIS CYBER audit chain export file. Standard library only — no external dependencies. Usage: python3 scripts/audit-chain-walker.py Exit codes: 0 — PASS or PARTIAL_MODERN (chain integrity confirmed within the method's scope) 1 — FAIL (at least one integrity check failed) 2 — EMPTY (export contains no entries) 3 — Input error (file not found, invalid JSON, missing required fields) """ import hashlib import json import sys from datetime import datetime, timezone GENESIS_LITERAL = "GENESIS_BLOCK_AEGIS_CYBER_2026" SPEC_VERSION = "1.0" WALKER_VERSION = "1.0" METHOD_BOUNDARY_NOTE = ( "\n---\n" "Method boundary (every run): this verification confirms genesis anchor, linkage, and content\n" "integrity. It does not detect consistent-downstream-re-hashing (§6.1): a chain tampered and\n" "fully re-hashed from the tampered entry forward will pass this verification. See §6.1 for the\n" "limitation's scope and the controls that bear on the exploitation path." ) def compute_entry_hash(entry: dict) -> str: """ Recompute the SHA-256 hash for a modern chain entry. Canonical payload (§2.2): seven fields in fixed order, compact JSON (no whitespace — matches JavaScript JSON.stringify output). Field notes: - 'id' is source_audit_id (string), NOT the integer row id - 'userId' is null-preserved (JSON null if user_id is None) - 'timestamp' must end in 'Z' (UTC ISO 8601) """ payload = json.dumps( { "id": entry["source_audit_id"], # string, from source_audit_id column "previousHash": entry["previous_hash"], "timestamp": entry["entry_timestamp"], # ISO 8601 UTC ending in "Z" "action": entry["action"], "userId": entry["user_id"], # None → JSON null (not omitted) "resource": entry["resource"], "details": entry["details"], }, separators=(",", ":"), # compact — matches JSON.stringify ) return hashlib.sha256(payload.encode("utf-8")).hexdigest() def is_legacy_entry(entry: dict) -> bool: """ Derive legacy/modern status from authoritative nullable fields. IMPORTANT: the export's 'is_legacy' convenience flag is NOT used here. A dishonest exporter could set is_legacy=true on a modern row to suppress content recomputation. Legacy status is derived independently from source_audit_id and entry_timestamp nullness (§3.3, §4.2). """ return entry.get("source_audit_id") is None or entry.get("entry_timestamp") is None def validate_export_structure(export: dict) -> list[str]: """Return a list of structural errors in the export file, or [] if clean.""" errors = [] if "export_metadata" not in export: errors.append("Missing required field: export_metadata") if "entries" not in export: errors.append("Missing required field: entries") elif not isinstance(export["entries"], list): errors.append("'entries' must be a JSON array") if errors: return errors for i, entry in enumerate(export["entries"]): for field in ["id", "source_audit_id", "hash", "previous_hash", "action", "user_id", "resource", "details", "entry_timestamp"]: if field not in entry: errors.append(f"Entry at index {i} missing required field: {field}") return errors def walk_chain(export_file: str) -> int: """ Walk the audit chain export and print the verification result. Returns the process exit code. """ # --- Load and validate --- try: with open(export_file, "r", encoding="utf-8") as f: export = json.load(f) except FileNotFoundError: print(f"ERROR: File not found: {export_file}", file=sys.stderr) return 3 except json.JSONDecodeError as e: print(f"ERROR: Invalid JSON in {export_file}: {e}", file=sys.stderr) return 3 structural_errors = validate_export_structure(export) if structural_errors: print("ERROR: Export file structural validation failed:", file=sys.stderr) for err in structural_errors: print(f" - {err}", file=sys.stderr) return 3 metadata = export["export_metadata"] entries = export["entries"] print(f"AEGIS CYBER Audit Chain Walker \u2014 v{WALKER_VERSION}") print(f"Export file: {export_file}") print(f"Generated: {metadata.get('generated_at', '(not specified)')}") # --- EMPTY case --- if len(entries) == 0: print(f"Entry count: 0") print() print("Verification result: EMPTY") print(METHOD_BOUNDARY_NOTE) return 2 # --- Count legacy vs modern --- legacy_entries = [e for e in entries if is_legacy_entry(e)] modern_entries = [e for e in entries if not is_legacy_entry(e)] total = len(entries) legacy_count = len(legacy_entries) modern_count = len(modern_entries) print(f"Entry count: {total} ({legacy_count} legacy / {modern_count} modern)") print() # --- Walk the chain --- prev_hash = None failed_at = None failed_reason = None failed_detail = None for i, entry in enumerate(entries): entry_id = entry["id"] # Step 1: Genesis check (first entry only) if i == 0: expected_prev = GENESIS_LITERAL if entry["previous_hash"] != GENESIS_LITERAL: failed_at = entry_id failed_reason = "genesis" failed_detail = ( f"stored previous_hash = {entry['previous_hash']!r}\n" f" expected (genesis literal) = {GENESIS_LITERAL!r}" ) break # Step 2: Linkage check (all entries after the first) if i > 0: expected_prev = prev_hash if entry["previous_hash"] != expected_prev: failed_at = entry_id failed_reason = "linkage" failed_detail = ( f"stored previous_hash = {entry['previous_hash']!r}\n" f" expected = {expected_prev!r}" ) break # Step 3: Content check — modern entries only # Skip condition: source_audit_id is null OR entry_timestamp is null. # The 'is_legacy' export flag is NOT consulted here (§3.3, §4.2). if not is_legacy_entry(entry): recomputed = compute_entry_hash(entry) if recomputed != entry["hash"]: failed_at = entry_id failed_reason = "content" failed_detail = ( f"stored hash = {entry['hash']!r}\n" f" recomputed hash = {recomputed!r}" ) break prev_hash = entry["hash"] # --- Result --- if failed_at is not None: fail_index = next( (i for i, e in enumerate(entries) if e["id"] == failed_at), "??" ) print(f"Verification result: FAIL") print() print(f" FAILED at entry id={failed_at} (entry index {fail_index + 1} of {total})") print(f" Check failed: {failed_reason}") print(f" Detail: {failed_detail}") print(f" Entries verified before failure: {fail_index}") print(METHOD_BOUNDARY_NOTE) return 1 # Determine result code if legacy_count > 0: result_code = "PARTIAL_MODERN" else: result_code = "PASS" print(f"Verification result: {result_code}") print() print(f" Genesis anchor: CONFIRMED ({GENESIS_LITERAL})") print(f" Linkage: {total} entries verified") print(f" Content: {modern_count} modern entries recomputed and matched") if legacy_count > 0: print(f" Legacy: {legacy_count} entries linkage-only (pre-2026-05-12 \u2014 see spec \u00a72.4)") print(METHOD_BOUNDARY_NOTE) return 0 if __name__ == "__main__": if len(sys.argv) != 2: print(f"Usage: python3 {sys.argv[0]} ", file=sys.stderr) print() print("Test fixtures (T1\u2013T7):", file=sys.stderr) print(" python3 scripts/audit-chain-walker.py scripts/test-fixtures/chain-export-good.json", file=sys.stderr) sys.exit(3) sys.exit(walk_chain(sys.argv[1]))