diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index e8b4b9f..297e378 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -14,6 +14,8 @@ jobs: run: python3 scripts/validate_examples.py - name: Validate Phase 0 metadata schemas run: python3 scripts/validate_phase0_schemas.py + - name: Enforce zone-gate engine (CHK-01..CHK-10) + run: python3 scripts/test_exodus_gate.py - name: Validate synthetic intake path run: python3 scripts/validate_intake_fixture.py - name: Validate synthetic Exodus workroom demo diff --git a/docs/zone-gate-enforcement.md b/docs/zone-gate-enforcement.md new file mode 100644 index 0000000..d61115f --- /dev/null +++ b/docs/zone-gate-enforcement.md @@ -0,0 +1,46 @@ +# Zone-gate enforcement (CHK-01..CHK-10) + +Metadata Standards v0.1 §7 defines the WNZL Dirt-to-Diamond promotion gates as a +schema (`schemas/policy-gate.json`). This slice makes them **executable**: the +gates are now run and enforced, not merely described. + +## What this adds + +- `scripts/exodus_gate.py` — the gate engine. `evaluate_gate(artifact, policy, ctx)` + runs the policy's required fields + checks and returns a **fail-closed** + `GateDecision` plus the CustodyEvent that must be written: + - `ZonePromotion` when admitted, `PolicyException` when blocked; + - `custody_status = IntegrityViolation` whenever CHK-01 fails (recomputed + BLAKE3 ≠ stored hash); + - `auto_promote` is only true when checks pass **and** no human/legal review is + outstanding (promotions to Governed/Diamond never auto-promote). +- `examples/gate-policies/` — the first concrete gate policies (previously the + schema had zero instances): + - `GATE-Landing-Examination-v1` — CHK-01/03/09/10 + CHK-08 (non-blocking flag). + - `GATE-Examination-Integration-v1` — CHK-01/02 + the adversarial-evidence + gate: CHK-04 (counter-explanations) and CHK-05 (null hypothesis at E3+). + - `GATE-Integration-Governed-v1` — CHK-01/06/07 + human + legal review. +- `scripts/test_exodus_gate.py` — dependency-free CHK matrix (pass/fail for each + check, integrity-violation custody status, non-blocking duplicate flag, and the + review-gated Governed promotion). Wired into CI. + +## Check semantics + +| Check | Enforces | Blocking | +|---|---|---| +| CHK-01 IntegrityVerification | recomputed BLAKE3 == stored; fail-closed if bytes not recomputed | yes (→ IntegrityViolation) | +| CHK-02 TemporalConsistency | valid_from ≤ valid_to; observed_at not in the future | yes | +| CHK-03 ExhibitIdUniqueness | exhibit_id unused in corpus | yes | +| CHK-04 CounterExplanationPresent | ≥1 benign alternative before analysis | yes | +| CHK-05 NullHypothesisDefined | null hypothesis declared at evidence_grade ≥ E3 | yes | +| CHK-06 SecurityLabelSet / CHK-07 WitnessRetentionSet | disclosure controls set before Governed | yes | +| CHK-08 DuplicateDetection | byte-identical to another exhibit | no (flag + analyst ack) | +| CHK-09 SourceAccountRegistered | source_account in the registry | yes | +| CHK-10 HashAlgorithmCompliance | blake3 + sha256 present, non-zero, 64 hex | yes | + +## Still open (unchanged by this slice) + +TriTRPC `BEACON_COMMIT` sealing of CustodyEvents, HellGraph atom signing, the +working ForensicBundle export command, account-registry population, and BSM +ingest. The engine emits the CustodyEvent *shape*; cryptographic sealing is the +next dependency. diff --git a/examples/gate-policies/GATE-Examination-Integration-v1.json b/examples/gate-policies/GATE-Examination-Integration-v1.json new file mode 100644 index 0000000..95bb4d7 --- /dev/null +++ b/examples/gate-policies/GATE-Examination-Integration-v1.json @@ -0,0 +1,18 @@ +{ + "policy_id": "GATE-Examination-Integration-v1", + "from_zone": "Examination", + "to_zone": "Integration", + "applies_to_classes": ["*"], + "required_fields": ["evidence_grade", "counter_explanations"], + "required_checks": [ + {"check_id": "CHK-01", "name": "IntegrityVerification", "failure_action": "Set custody_status=IntegrityViolation; block promotion; alert custodian"}, + {"check_id": "CHK-02", "name": "TemporalConsistency", "failure_action": "Require analyst review; flag in classification block"}, + {"check_id": "CHK-04", "name": "CounterExplanationPresent", "failure_action": "Block promotion to Integration; require analyst to populate"}, + {"check_id": "CHK-05", "name": "NullHypothesisDefined", "failure_action": "Block promotion to Integration"} + ], + "human_review_required": false, + "legal_review_required": false, + "policy_version": 1, + "effective_from_txn": "txn-0", + "status": "Active" +} diff --git a/examples/gate-policies/GATE-Integration-Governed-v1.json b/examples/gate-policies/GATE-Integration-Governed-v1.json new file mode 100644 index 0000000..58240dc --- /dev/null +++ b/examples/gate-policies/GATE-Integration-Governed-v1.json @@ -0,0 +1,17 @@ +{ + "policy_id": "GATE-Integration-Governed-v1", + "from_zone": "Integration", + "to_zone": "Governed", + "applies_to_classes": ["*"], + "required_fields": ["security_label", "witness_retention"], + "required_checks": [ + {"check_id": "CHK-01", "name": "IntegrityVerification", "failure_action": "Set custody_status=IntegrityViolation; block promotion; alert custodian"}, + {"check_id": "CHK-06", "name": "SecurityLabelSet", "failure_action": "Block promotion to Governed"}, + {"check_id": "CHK-07", "name": "WitnessRetentionSet", "failure_action": "Block promotion to Governed"} + ], + "human_review_required": true, + "legal_review_required": true, + "policy_version": 1, + "effective_from_txn": "txn-0", + "status": "Active" +} diff --git a/examples/gate-policies/GATE-Landing-Examination-v1.json b/examples/gate-policies/GATE-Landing-Examination-v1.json new file mode 100644 index 0000000..0c0b094 --- /dev/null +++ b/examples/gate-policies/GATE-Landing-Examination-v1.json @@ -0,0 +1,19 @@ +{ + "policy_id": "GATE-Landing-Examination-v1", + "from_zone": "Landing", + "to_zone": "Examination", + "applies_to_classes": ["*"], + "required_fields": ["artifact_id", "exhibit_id", "hash_blake3", "hash_sha256", "source_account", "observed_at_micros"], + "required_checks": [ + {"check_id": "CHK-01", "name": "IntegrityVerification", "failure_action": "Set custody_status=IntegrityViolation; block promotion; alert custodian"}, + {"check_id": "CHK-03", "name": "ExhibitIdUniqueness", "failure_action": "Reject; return error"}, + {"check_id": "CHK-09", "name": "SourceAccountRegistered", "failure_action": "Block intake; require account registration"}, + {"check_id": "CHK-10", "name": "HashAlgorithmCompliance", "failure_action": "Block all zone transitions beyond Landing"}, + {"check_id": "CHK-08", "name": "DuplicateDetection", "failure_action": "Flag as Duplicate; require analyst acknowledgment", "blocking": false} + ], + "human_review_required": false, + "legal_review_required": false, + "policy_version": 1, + "effective_from_txn": "txn-0", + "status": "Active" +} diff --git a/scripts/exodus_gate.py b/scripts/exodus_gate.py new file mode 100644 index 0000000..8591a8f --- /dev/null +++ b/scripts/exodus_gate.py @@ -0,0 +1,250 @@ +#!/usr/bin/env python3 +"""Exodus zone-gate engine — executable enforcement of the WNZL Dirt-to-Diamond +promotion gates (Metadata Standards v0.1 Section 7). + +The policy-gate schema describes CHK-01..CHK-10; this module *runs* them and +returns a fail-closed promotion decision plus the CustodyEvent that must be +written (ZonePromotion when admitted, PolicyException when blocked, and a +custody_status of IntegrityViolation when the acquisition hash no longer +verifies). Dependency-free so it runs in CI without extra packages. + +Doctrine encoded here (the adversarial-evidence rule): an artifact cannot reach +an analysis-bearing zone without counter-explanations and, at E3+, a declared +null hypothesis. Observation is preserved separately from interpretation. +""" +from __future__ import annotations + +import argparse +import datetime as dt +import json +import sys +from dataclasses import dataclass, field +from pathlib import Path +from typing import Any, Callable + +ZERO_HASH_64 = "0" * 64 +_UNSET_ENUMS = {None, "", "Unknown"} +_GRADE_RANK = {"E1": 1, "E2": 2, "E3": 3, "E4": 4, "E5": 5} + + +def now_micros() -> int: + return int(dt.datetime.now(dt.timezone.utc).timestamp() * 1_000_000) + + +@dataclass +class GateContext: + """Out-of-band facts the checks need (kept explicit + injectable for replay).""" + recomputed_blake3: str | None = None # CHK-01: hash recomputed from bytes now + other_exhibit_ids: set[str] = field(default_factory=set) # CHK-03: exhibit_ids already in corpus + hash_to_exhibit: dict[str, str] = field(default_factory=dict) # CHK-08: hash -> exhibit_id (other artifacts) + registered_accounts: set[str] = field(default_factory=set) # CHK-09 + now_micros: int | None = None + + +@dataclass +class CheckResult: + check_id: str + name: str + passed: bool + blocking: bool + reason: str + + def to_dict(self) -> dict[str, Any]: + return {"check_id": self.check_id, "name": self.name, "passed": self.passed, "blocking": self.blocking, "reason": self.reason} + + +@dataclass +class GateDecision: + checks_passed: bool # all blocking checks + required fields satisfied + auto_promote: bool # checks_passed AND no human/legal review outstanding + requires_human_review: bool + requires_legal_review: bool + custody_event_type: str # ZonePromotion | PolicyException + custody_status: str # Intact | IntegrityViolation | PendingVerification + results: list[CheckResult] + reasons: list[str] + + def to_dict(self) -> dict[str, Any]: + return { + "checks_passed": self.checks_passed, + "auto_promote": self.auto_promote, + "requires_human_review": self.requires_human_review, + "requires_legal_review": self.requires_legal_review, + "custody_event_type": self.custody_event_type, + "custody_status": self.custody_status, + "results": [r.to_dict() for r in self.results], + "reasons": self.reasons, + } + + +# ── Individual checks (artifact, ctx) -> (passed, reason) ───────────────────── +def _grade_rank(artifact: dict[str, Any]) -> int: + return _GRADE_RANK.get(str(artifact.get("evidence_grade", "")), 0) + + +def _chk01(a, ctx): # IntegrityVerification + if ctx.recomputed_blake3 is None: + return False, "cannot verify: artifact bytes not recomputed (fail-closed)" + if str(ctx.recomputed_blake3).lower() != str(a.get("hash_blake3", "")).lower(): + return False, "recomputed BLAKE3 does not match stored hash" + return True, "hash verified" + + +def _chk02(a, ctx): # TemporalConsistency + vf, vt = a.get("valid_from_micros"), a.get("valid_to_micros") + if vf is not None and vt is not None and vf > vt: + return False, f"valid_from ({vf}) > valid_to ({vt})" + now = ctx.now_micros if ctx.now_micros is not None else now_micros() + obs = a.get("observed_at_micros") + if obs is not None and obs > now: + return False, "observed_at_micros is in the future" + return True, "temporal ordering consistent" + + +def _chk03(a, ctx): # ExhibitIdUniqueness + if a.get("exhibit_id") in ctx.other_exhibit_ids: + return False, f"exhibit_id {a.get('exhibit_id')} already used in corpus" + return True, "exhibit_id unique" + + +def _chk04(a, ctx): # CounterExplanationPresent + ce = a.get("counter_explanations") + if not isinstance(ce, list) or not ce: + return False, "counter_explanations empty — adversarial review required before analysis" + return True, f"{len(ce)} counter-explanation(s) present" + + +def _chk05(a, ctx): # NullHypothesisDefined (only E3+) + if _grade_rank(a) >= 3: + nh = a.get("null_hypothesis_ids") + if not isinstance(nh, list) or not nh: + return False, "evidence_grade >= E3 requires a declared null hypothesis" + return True, "null hypothesis present or not yet required" + + +def _chk06(a, ctx): # SecurityLabelSet + if a.get("security_label") in _UNSET_ENUMS: + return False, "security_label unset" + return True, "security_label set" + + +def _chk07(a, ctx): # WitnessRetentionSet + if a.get("witness_retention") in _UNSET_ENUMS: + return False, "witness_retention unset" + return True, "witness_retention set" + + +def _chk08(a, ctx): # DuplicateDetection (non-blocking flag by default) + h = str(a.get("hash_blake3", "")).lower() + other = ctx.hash_to_exhibit.get(h) + if other and other != a.get("exhibit_id"): + return False, f"byte-identical to exhibit {other} — classify as Duplicate; analyst ack required" + return True, "no duplicate detected" + + +def _chk09(a, ctx): # SourceAccountRegistered + if a.get("source_account") not in ctx.registered_accounts: + return False, f"source_account {a.get('source_account')!r} not in account registry" + return True, "source account registered" + + +def _chk10(a, ctx): # HashAlgorithmCompliance + for fld in ("hash_blake3", "hash_sha256"): + v = str(a.get(fld, "")) + if len(v) != 64 or v == ZERO_HASH_64: + return False, f"{fld} missing/zero/wrong-length" + return True, "blake3 + sha256 present and non-zero" + + +_CHECKS: dict[str, tuple[str, Callable[[dict, GateContext], tuple[bool, str]]]] = { + "CHK-01": ("IntegrityVerification", _chk01), + "CHK-02": ("TemporalConsistency", _chk02), + "CHK-03": ("ExhibitIdUniqueness", _chk03), + "CHK-04": ("CounterExplanationPresent", _chk04), + "CHK-05": ("NullHypothesisDefined", _chk05), + "CHK-06": ("SecurityLabelSet", _chk06), + "CHK-07": ("WitnessRetentionSet", _chk07), + "CHK-08": ("DuplicateDetection", _chk08), + "CHK-09": ("SourceAccountRegistered", _chk09), + "CHK-10": ("HashAlgorithmCompliance", _chk10), +} +# CHK-08 is a flag, not a hard block, per the spec (failure_action = "flag"). +_DEFAULT_NONBLOCKING = {"CHK-08"} + + +def _missing_required_fields(artifact: dict[str, Any], required: list[str]) -> list[str]: + return [f for f in required if artifact.get(f) in (None, "")] + + +def evaluate_gate(artifact: dict[str, Any], policy: dict[str, Any], ctx: GateContext) -> GateDecision: + """Run a zone-gate policy against an artifact. Fail-closed: any blocking check + failure or missing required field denies promotion.""" + reasons: list[str] = [] + results: list[CheckResult] = [] + + # Class scoping: a policy with applies_to_classes not matching this artifact is a no-op pass-through. + classes = policy.get("applies_to_classes") or [] + if "*" not in classes and artifact.get("artifact_class") not in classes: + reasons.append(f"policy does not apply to class {artifact.get('artifact_class')}") + return GateDecision(False, False, bool(policy.get("human_review_required")), + bool(policy.get("legal_review_required")), "PolicyException", + "PendingVerification", results, reasons) + + missing = _missing_required_fields(artifact, policy.get("required_fields") or []) + if missing: + reasons.append("missing required fields: " + ", ".join(missing)) + + integrity_violation = False + for spec in policy.get("required_checks") or []: + cid = spec.get("check_id") + name, fn = _CHECKS.get(cid, (spec.get("name", cid), None)) + blocking = spec.get("blocking", cid not in _DEFAULT_NONBLOCKING) + if fn is None: + results.append(CheckResult(cid, name, False, blocking, "unknown check id")) + reasons.append(f"{cid}: unknown check id") + continue + passed, why = fn(artifact, ctx) + results.append(CheckResult(cid, name, passed, blocking, why)) + if not passed: + if cid == "CHK-01": + integrity_violation = True + if blocking: + reasons.append(f"{cid} {name}: {why}") + + blocking_failures = [r for r in results if not r.passed and r.blocking] + checks_passed = not missing and not blocking_failures + + requires_human = bool(policy.get("human_review_required")) + requires_legal = bool(policy.get("legal_review_required")) + auto_promote = checks_passed and not requires_human and not requires_legal + if checks_passed and (requires_human or requires_legal): + reasons.append("checks passed; awaiting human/legal review before promotion") + + custody_status = "IntegrityViolation" if integrity_violation else ("Intact" if checks_passed else "PendingVerification") + custody_event_type = "ZonePromotion" if checks_passed else "PolicyException" + + return GateDecision(checks_passed, auto_promote, requires_human, requires_legal, + custody_event_type, custody_status, results, reasons) + + +def main(argv: list[str] | None = None) -> int: + p = argparse.ArgumentParser(description="Evaluate an Exodus zone-gate policy against an artifact.") + p.add_argument("--artifact", required=True) + p.add_argument("--policy", required=True) + p.add_argument("--recomputed-blake3", default=None, help="BLAKE3 recomputed from bytes now (for CHK-01)") + p.add_argument("--registered-account", action="append", default=[], help="repeatable; accounts in the registry") + args = p.parse_args(argv if argv is not None else sys.argv[1:]) + + artifact = json.loads(Path(args.artifact).read_text(encoding="utf-8")) + policy = json.loads(Path(args.policy).read_text(encoding="utf-8")) + ctx = GateContext( + recomputed_blake3=args.recomputed_blake3, + registered_accounts=set(args.registered_account) or {artifact.get("source_account")}, + ) + decision = evaluate_gate(artifact, policy, ctx) + print(json.dumps(decision.to_dict(), indent=2, sort_keys=True)) + return 0 if decision.checks_passed else 1 + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/scripts/test_exodus_gate.py b/scripts/test_exodus_gate.py new file mode 100644 index 0000000..9493113 --- /dev/null +++ b/scripts/test_exodus_gate.py @@ -0,0 +1,112 @@ +#!/usr/bin/env python3 +"""Dependency-free tests for the Exodus zone-gate engine (CHK-01..CHK-10). + +Run: python3 scripts/test_exodus_gate.py +Matches the repo's validate_*.py idiom — plain asserts, exits non-zero on failure. +""" +from __future__ import annotations + +import copy +import json +from pathlib import Path + +from exodus_gate import GateContext, evaluate_gate, now_micros + +ROOT = Path(__file__).resolve().parents[1] +ARTIFACT = json.loads((ROOT / "examples/valid-evidence-artifact.json").read_text()) +GP = ROOT / "examples/gate-policies" +LANDING = json.loads((GP / "GATE-Landing-Examination-v1.json").read_text()) +EXAM = json.loads((GP / "GATE-Examination-Integration-v1.json").read_text()) +GOVERNED = json.loads((GP / "GATE-Integration-Governed-v1.json").read_text()) + +ACCT = ARTIFACT["source_account"] +GOODHASH = ARTIFACT["hash_blake3"] + + +def ctx(**kw) -> GateContext: + base = dict(recomputed_blake3=GOODHASH, registered_accounts={ACCT}, now_micros=now_micros()) + base.update(kw) + return GateContext(**base) + + +def run() -> None: + # Landing -> Examination on a clean artifact: passes + auto-promotes. + d = evaluate_gate(ARTIFACT, LANDING, ctx()) + assert d.checks_passed and d.auto_promote, d.reasons + assert d.custody_event_type == "ZonePromotion" and d.custody_status == "Intact" + + # CHK-01 integrity failure -> blocked + IntegrityViolation custody status. + d = evaluate_gate(ARTIFACT, LANDING, ctx(recomputed_blake3="f" * 64)) + assert not d.checks_passed and d.custody_status == "IntegrityViolation" + assert d.custody_event_type == "PolicyException" + + # CHK-01 cannot verify when bytes not recomputed -> fail-closed. + d = evaluate_gate(ARTIFACT, LANDING, ctx(recomputed_blake3=None)) + assert not d.checks_passed + + # CHK-09 unregistered account -> blocked. + d = evaluate_gate(ARTIFACT, LANDING, ctx(registered_accounts=set())) + assert not d.checks_passed and any("CHK-09" in r for r in d.reasons) + + # CHK-10 zero/short hash -> blocked. + bad = copy.deepcopy(ARTIFACT); bad["hash_sha256"] = "0" * 64 + d = evaluate_gate(bad, LANDING, ctx(recomputed_blake3=bad["hash_blake3"])) + assert not d.checks_passed and any("CHK-10" in r for r in d.reasons) + + # CHK-03 exhibit collision -> blocked. + d = evaluate_gate(ARTIFACT, LANDING, ctx(other_exhibit_ids={ARTIFACT["exhibit_id"]})) + assert not d.checks_passed and any("CHK-03" in r for r in d.reasons) + + # CHK-08 duplicate is a non-blocking FLAG: failed check present, but still passes. + d = evaluate_gate(ARTIFACT, LANDING, ctx(hash_to_exhibit={GOODHASH: "AF-9999"})) + assert d.checks_passed, d.reasons + chk08 = next(r for r in d.results if r.check_id == "CHK-08") + assert not chk08.passed and not chk08.blocking + + # Examination -> Integration: missing counter_explanations -> CHK-04 blocks. + no_ce = copy.deepcopy(ARTIFACT); no_ce["counter_explanations"] = [] + d = evaluate_gate(no_ce, EXAM, ctx()) + assert not d.checks_passed and any("CHK-04" in r for r in d.reasons) + + # Examination -> Integration: E3 without null hypothesis -> CHK-05 blocks. + no_null = copy.deepcopy(ARTIFACT); no_null["null_hypothesis_ids"] = [] + d = evaluate_gate(no_null, EXAM, ctx()) + assert not d.checks_passed and any("CHK-05" in r for r in d.reasons) + + # Examination -> Integration clean -> passes. + d = evaluate_gate(ARTIFACT, EXAM, ctx()) + assert d.checks_passed and d.auto_promote, d.reasons + + # CHK-02 temporal inversion -> blocked. + inv = copy.deepcopy(ARTIFACT); inv["valid_from_micros"] = 200; inv["valid_to_micros"] = 100 + d = evaluate_gate(inv, EXAM, ctx()) + assert not d.checks_passed and any("CHK-02" in r for r in d.reasons) + + # Integration -> Governed: checks pass but human+legal review gate auto-promotion. + d = evaluate_gate(ARTIFACT, GOVERNED, ctx()) + assert d.checks_passed and not d.auto_promote + assert d.requires_human_review and d.requires_legal_review + + # Integration -> Governed: missing security_label -> blocked (CHK-06 + required field). + no_label = copy.deepcopy(ARTIFACT); no_label["security_label"] = "Unknown" + d = evaluate_gate(no_label, GOVERNED, ctx()) + assert not d.checks_passed and any("CHK-06" in r for r in d.reasons) + + print("OK: exodus_gate CHK-01..CHK-10 + policy enforcement — all assertions passed") + + +def validate_policy_shapes() -> None: + required_top = {"policy_id", "from_zone", "to_zone", "applies_to_classes", "required_fields", + "required_checks", "human_review_required", "legal_review_required", + "policy_version", "effective_from_txn"} + for pol in (LANDING, EXAM, GOVERNED): + missing = required_top - set(pol) + assert not missing, f"{pol.get('policy_id')} missing {missing}" + for c in pol["required_checks"]: + assert {"check_id", "name", "failure_action"} <= set(c), f"bad check_spec in {pol['policy_id']}" + print("OK: 3 gate policies are schema-shaped") + + +if __name__ == "__main__": + validate_policy_shapes() + run()