#!/usr/bin/env python3 """Fail-closed validator for the durable-approval HOW article fixture.""" from __future__ import annotations import argparse import hashlib import json import re import sys from datetime import datetime, timedelta, timezone from pathlib import Path from typing import Any HEX64 = re.compile(r"^[0-9a-f]{64}$") ZERO_HASH = "0" * 64 class Blocked(Exception): def __init__(self, code: str, detail: str): self.code = code self.detail = detail super().__init__(detail) def raw_sha256(path: Path) -> str: return hashlib.sha256(path.read_bytes()).hexdigest() def load_object(path: Path, label: str) -> dict[str, Any]: try: value = json.loads(path.read_text(encoding="utf-8")) except (OSError, UnicodeError, json.JSONDecodeError) as exc: raise Blocked(f"{label}_unreadable", str(exc)) from exc if not isinstance(value, dict): raise Blocked(f"{label}_schema", "top level must be a JSON object") return value def require(record: dict[str, Any], key: str, kind: type, label: str) -> Any: value = record.get(key) if not isinstance(value, kind) or (kind is str and not value.strip()): raise Blocked("schema", f"{label}.{key} must be a non-empty {kind.__name__}") return value def parse_time(value: Any, label: str) -> datetime: if not isinstance(value, str) or not re.fullmatch(r"\d{4}-\d{2}-\d{2}T\d{2}:\d{2}:\d{2}Z", value): raise Blocked("timestamp", f"{label} must use UTC YYYY-MM-DDTHH:MM:SSZ") return datetime.strptime(value, "%Y-%m-%dT%H:%M:%SZ").replace(tzinfo=timezone.utc) def event_sha256(event: dict[str, Any]) -> str: payload = {key: value for key, value in event.items() if key != "event_sha256"} encoded = json.dumps(payload, sort_keys=True, separators=(",", ":"), ensure_ascii=True).encode() return hashlib.sha256(encoded).hexdigest() def approval_payload_sha256(approval: dict[str, Any]) -> str: """Bind every authority-bearing field without creating a hash cycle.""" payload = {key: value for key, value in approval.items() if key != "audit_events"} encoded = json.dumps(payload, sort_keys=True, separators=(",", ":"), ensure_ascii=True).encode() return hashlib.sha256(encoded).hexdigest() def validate_audit_chain( approval: dict[str, Any], approver: str, proposal_author: str, requested: datetime, decided: datetime, ) -> None: events = require(approval, "audit_events", list, "approval") if len(events) < 2: raise Blocked("audit_chain", "approval.audit_events needs request and decision events") previous = ZERO_HASH prior_time: datetime | None = None for index, event in enumerate(events): if not isinstance(event, dict): raise Blocked("audit_chain", f"audit event {index} must be an object") if event.get("previous_event_sha256") != previous: raise Blocked("audit_chain", f"audit event {index} has the wrong predecessor") actual = event_sha256(event) if event.get("event_sha256") != actual: raise Blocked("audit_chain", f"audit event {index} digest does not match its content") event_time = parse_time(event.get("timestamp"), f"audit_events[{index}].timestamp") if prior_time is not None and event_time < prior_time: raise Blocked("audit_chain", "audit event timestamps must be nondecreasing") prior_time = event_time previous = actual first = events[0] last = events[-1] if (first.get("event") != "approval_requested" or first.get("actor") != proposal_author or parse_time(first.get("timestamp"), "audit_events[0].timestamp") != requested): raise Blocked("audit_chain", "first audit event must bind the author's request timestamp") if last.get("event") != "decision_recorded" or last.get("actor") != approver: raise Blocked("audit_chain", "last audit event must record the approver's decision") if parse_time(last.get("timestamp"), "last audit event timestamp") != decided: raise Blocked("audit_chain", "last audit event timestamp must equal decided_at") if last.get("approval_payload_sha256") != approval_payload_sha256(approval): raise Blocked("audit_chain", "decision event does not bind the current approval payload") def validate_evidence( evidence: Any, approval_path: Path, criterion_id: str, requested: datetime, decided: datetime, ) -> None: if not isinstance(evidence, dict) or set(evidence) != {"path", "sha256", "method", "observed_at"}: raise Blocked("criterion_evidence", f"criterion {criterion_id} needs exact inspectable evidence") rel = evidence.get("path") if not isinstance(rel, str) or not rel or Path(rel).is_absolute() or ".." in Path(rel).parts: raise Blocked("criterion_evidence", f"criterion {criterion_id} evidence.path must be a safe relative path") digest = evidence.get("sha256") if not isinstance(digest, str) or not HEX64.fullmatch(digest): raise Blocked("criterion_evidence", f"criterion {criterion_id} evidence.sha256 must be lowercase SHA-256") method = evidence.get("method") if not isinstance(method, str) or len(method.strip()) < 12: raise Blocked("criterion_evidence", f"criterion {criterion_id} needs a concrete verification method") observed = parse_time(evidence.get("observed_at"), f"criterion {criterion_id} evidence.observed_at") if not requested <= observed <= decided: raise Blocked("criterion_evidence", f"criterion {criterion_id} evidence must be current within the decision window") evidence_path = (approval_path.parent / rel).resolve() try: evidence_path.relative_to(approval_path.parent.resolve()) except ValueError as exc: raise Blocked("criterion_evidence", f"criterion {criterion_id} evidence escapes the approval directory") from exc if not evidence_path.is_file() or raw_sha256(evidence_path) != digest: raise Blocked("criterion_evidence", f"criterion {criterion_id} evidence file is absent or stale") def validate(policy_path: Path, proposal_path: Path, candidate_path: Path, approval_path: Path, at: datetime) -> str: policy = load_object(policy_path, "policy") proposal = load_object(proposal_path, "proposal") approval = load_object(approval_path, "approval") policy_id = require(policy, "policy_id", str, "policy") policy_version = require(policy, "version", str, "policy") action_class = require(policy, "action_class", str, "policy") allowed = require(policy, "allowed_approvers", list, "policy") required_criteria = require(policy, "required_criteria", list, "policy") max_age_hours = policy.get("max_approval_age_hours") if not isinstance(max_age_hours, int) or max_age_hours <= 0: raise Blocked("policy_schema", "policy.max_approval_age_hours must be a positive integer") if policy.get("silence_default") != "BLOCKED": raise Blocked("policy_schema", "policy.silence_default must be BLOCKED") if policy.get("authority_model") != "allowed-approver-registry": raise Blocked("policy_schema", "policy.authority_model must be allowed-approver-registry") re_review = policy.get("re_review_triggers") if (not isinstance(re_review, list) or not re_review or not all(isinstance(item, str) and item.strip() for item in re_review)): raise Blocked("policy_schema", "policy.re_review_triggers must name material changes") proposal_author = require(proposal, "author", str, "proposal") proposal_scope = require(proposal, "scope", str, "proposal") if proposal.get("action_class") != action_class: raise Blocked("action_class", "proposal action class does not match policy") candidate_ref = require(proposal, "candidate", dict, "proposal") if candidate_ref.get("path") != candidate_path.name: raise Blocked("candidate_scope", "proposal candidate.path must name the checked candidate file") candidate_sha = candidate_ref.get("sha256") if not isinstance(candidate_sha, str) or not HEX64.fullmatch(candidate_sha): raise Blocked("candidate_hash", "proposal candidate.sha256 must be lowercase SHA-256") actual_candidate_sha = raw_sha256(candidate_path) if candidate_sha != actual_candidate_sha: raise Blocked("stale_candidate", "proposal does not bind the release candidate bytes being checked") if approval.get("schema_version") != "durable-approval/1.0": raise Blocked("approval_schema", "approval.schema_version must be durable-approval/1.0") decision = require(approval, "decision", str, "approval") deadline = parse_time(approval.get("response_deadline"), "approval.response_deadline") if decision == "PENDING": code = "silence_deadline_elapsed" if at >= deadline else "decision_pending" raise Blocked(code, "silence grants no authority; approval remains BLOCKED") if decision != "APPROVED": raise Blocked(f"decision_{decision.lower()}", "only an explicit APPROVED record can pass") status = require(approval, "status", str, "approval") if status != "active": raise Blocked(f"status_{status}", "revoked or superseded approval cannot authorize action") if approval.get("revoked_at") is not None: raise Blocked("revoked", "approval carries a revocation timestamp") approver = require(approval, "approver", str, "approval") if approver not in allowed: raise Blocked("wrong_authority", "approver is not allowed by the bound policy") artifact = require(approval, "artifact", dict, "approval") recorded_artifact_sha = require(artifact, "sha256", str, "approval.artifact") if not HEX64.fullmatch(recorded_artifact_sha): raise Blocked("artifact_hash", "artifact.sha256 must be lowercase SHA-256") actual_artifact_sha = raw_sha256(proposal_path) if recorded_artifact_sha != actual_artifact_sha: raise Blocked("stale_artifact", "approval does not bind the proposal bytes being checked") if artifact.get("path") != proposal_path.name: raise Blocked("artifact_scope", "artifact.path must name the checked proposal file") policy_ref = require(approval, "policy", dict, "approval") if policy_ref.get("id") != policy_id or policy_ref.get("version") != policy_version: raise Blocked("stale_policy", "approval does not bind the current policy identity") if policy_ref.get("sha256") != raw_sha256(policy_path): raise Blocked("stale_policy", "approval does not bind the current policy bytes") if approval.get("action_class") != action_class or approval.get("scope") != proposal_scope: raise Blocked("approval_scope", "approval action class or scope does not match proposal") if approval.get("candidate") != candidate_ref: raise Blocked("approval_candidate", "approval does not bind the proposal's checked release candidate") requested = parse_time(approval.get("requested_at"), "approval.requested_at") decided = parse_time(approval.get("decided_at"), "approval.decided_at") expires = parse_time(approval.get("expires_at"), "approval.expires_at") if not requested <= decided < expires: raise Blocked("timestamp_order", "timestamps must be requested <= decided < expires") if deadline < requested: raise Blocked("timestamp_order", "response deadline cannot precede request") if decided > deadline: raise Blocked("decision_after_deadline", "decision was recorded after the response deadline") if expires > decided + timedelta(hours=max_age_hours): raise Blocked("expiry_policy", "approval expiry exceeds policy maximum age") if at < decided: raise Blocked("not_yet_effective", "approval cannot authorize action before decided_at") if at >= expires: raise Blocked("expired", "approval has expired") criteria = require(approval, "criteria", list, "approval") by_id: dict[str, dict[str, Any]] = {} for item in criteria: if not isinstance(item, dict) or not isinstance(item.get("id"), str): raise Blocked("criteria_schema", "each criterion needs an id, result, and evidence") if item["id"] in by_id: raise Blocked("criteria_schema", f"duplicate criterion {item['id']}") by_id[item["id"]] = item if set(by_id) != set(required_criteria): raise Blocked("criteria_set", "approval criteria must exactly match the current policy") for criterion_id in required_criteria: item = by_id[criterion_id] if item.get("result") != "PASS": raise Blocked("criterion_failed", f"criterion {criterion_id} lacks PASS evidence") validate_evidence(item.get("evidence"), approval_path, criterion_id, requested, decided) options = require(approval, "alternatives_considered", list, "approval") if len(options) < 2 or approval.get("selected_option") not in options: raise Blocked("alternatives", "record at least two options and the selected option") if proposal.get("irreversible") is True: premortem = approval.get("premortem") if not isinstance(premortem, list) or not premortem: raise Blocked("premortem", "irreversible action needs a forward-failure test") if approval.get("evidence_settled") is False and not approval.get("reversible_probe"): raise Blocked("reversible_probe", "unsettled evidence needs a named reversible probe") validate_audit_chain(approval, approver, proposal_author, requested, decided) return ( f"PASS approval_id={approval['record_id']} decision=APPROVED " f"candidate_sha256={actual_candidate_sha} artifact_sha256={actual_artifact_sha} " f"policy_sha256={raw_sha256(policy_path)}" ) def main(argv: list[str] | None = None) -> int: parser = argparse.ArgumentParser(description=__doc__) parser.add_argument("--policy", type=Path, required=True) parser.add_argument("--proposal", type=Path, required=True) parser.add_argument("--candidate", type=Path, required=True) parser.add_argument("--approval", type=Path, required=True) parser.add_argument("--test-mode", action="store_true", help="allow a deterministic test clock") parser.add_argument("--at", help="UTC evaluation time; accepted only with --test-mode") args = parser.parse_args(argv) try: if args.at and not args.test_mode: raise Blocked("test_time_override", "--at is prohibited outside --test-mode") if args.test_mode and not args.at: raise Blocked("test_time_missing", "--test-mode requires --at") at = parse_time(args.at, "--at") if args.test_mode else datetime.now(timezone.utc) print(validate(args.policy, args.proposal, args.candidate, args.approval, at)) except Blocked as exc: print(f"BLOCKED code={exc.code} detail={exc.detail}") return 2 except Exception as exc: # fail closed on an unexpected validator error print(f"BLOCKED code=validator_error detail={exc}") return 2 return 0 if __name__ == "__main__": sys.exit(main())