"""Offline, fail-closed admission guard for external frontier models. The registry is research and qualification policy, not a downloader, trainer, router, publisher, or deletion tool. Unknown repositories and unpinned revisions are denied. A permitted operation is only a preflight admission of a signed declaration after its exact evidence requirements are supplied; it never grants execution authority or implies model quality or promotion eligibility. """ from __future__ import annotations import argparse import importlib import json import re import sys from pathlib import Path from typing import Any HERE = Path(__file__).resolve().parent DEFAULT_REGISTRY = HERE / "frontier-adoption.json" SHA256_RE = re.compile(r"^[0-9a-f]{64}$") REPO_ROOT = HERE.parents[1] if str(REPO_ROOT) not in sys.path: sys.path.insert(0, str(REPO_ROOT)) EVIDENCE_PAYLOAD_TYPE = "application/vnd.szl.frontier-operation-evidence.v1+json" UNSIGNED_OPERATION_ALLOWLIST = frozenset({"READ_METADATA"}) HARD_DENIED_OPERATIONS = frozenset({ "TRAIN", "SERVE_PRODUCTION", "PROMOTE", "MERGE_WEIGHTS", "INGEST_UPSTREAM_TRAINING_TRACES", "IDENTITY_EDIT_REAL_PERSON", }) class FrontierAdmissionError(RuntimeError): """The requested external-model operation failed closed.""" def _load_object(path: Path) -> dict[str, Any]: value = json.loads(path.read_text(encoding="utf-8")) if not isinstance(value, dict): raise FrontierAdmissionError(f"expected JSON object: {path}") return value def load_registry(path: Path = DEFAULT_REGISTRY) -> dict[str, Any]: if not path.is_file(): raise FrontierAdmissionError(f"frontier adoption registry missing: {path}") registry = _load_object(path) if registry.get("schema_version") != "szl.frontier-adoption.v1": raise FrontierAdmissionError("unsupported frontier adoption registry schema") if registry.get("status") != "GOVERNED_PLAN_NOT_LOCAL_QUALIFICATION": raise FrontierAdmissionError("frontier registry honesty state changed") policy = registry.get("evidence_policy", {}) required_policy = { "local_measurements_present": False, "download_performed_by_this_contract": False, "promotion_authority": False, "unknown_repository_policy": "DENY", "zero_download_policy": "ZERO_DOWNLOADS_ALONE_NEVER_AUTHORIZES_DELETE_OR_ARCHIVE", } for field, expected in required_policy.items(): if policy.get(field) != expected: raise FrontierAdmissionError(f"registry policy mismatch: {field}") mutations = registry.get("external_mutations") if not isinstance(mutations, dict) or not mutations: raise FrontierAdmissionError("registry external_mutations contract is missing") if any(value is not False for value in mutations.values()): raise FrontierAdmissionError("registry reports an external mutation") brain = registry.get("brain_model_truth") if not isinstance(brain, dict): raise FrontierAdmissionError("registry brain_model_truth is missing") from a11oy_brain_graph import get_brain_graph expected_raw_nodes = len((get_brain_graph(refresh=True)).get("nodes") or []) if brain.get("raw_nodes_observed") != expected_raw_nodes: raise FrontierAdmissionError("registry Brain node observation changed") if brain.get("raw_nodes_available_to_retrieval_and_evaluation") != expected_raw_nodes: raise FrontierAdmissionError("registry retrieval/evaluation Brain count changed") if brain.get("raw_nodes_admitted_to_gradients") != 0: raise FrontierAdmissionError("registry reports unadmitted Brain gradient rows") github_estate = registry.get("github_estate_strategy") if not isinstance(github_estate, dict): raise FrontierAdmissionError("registry github_estate_strategy is missing") if github_estate.get("source_reported_repository_count") != 54: raise FrontierAdmissionError("registry GitHub source count changed") if github_estate.get("inventory_complete") is not False: raise FrontierAdmissionError("registry overclaims a complete GitHub inventory") if github_estate.get("public_github_repositories_observed") != 50: raise FrontierAdmissionError("registry public GitHub readback count changed") if github_estate.get("public_archived_observed") != 9: raise FrontierAdmissionError("registry public GitHub archive count changed") if github_estate.get("reconciliation_state") != ( "SOURCE_REPORT_54_PUBLIC_READBACK_50_ATTACHMENT_TRUNCATED_REVIEW_REQUIRED" ): raise FrontierAdmissionError("registry GitHub reconciliation state changed") if github_estate.get("code_to_weight_policy") != ( "CODE_REPOSITORIES_ARE_SERVICES_LIBRARIES_OR_EVIDENCE_NOT_MODEL_WEIGHTS" ): raise FrontierAdmissionError("registry code-to-weight boundary changed") if github_estate.get("unclassified_repository_policy") != ( "DISCOVER_CLASSIFY_FAIL_CLOSED_NO_ARCHIVE" ): raise FrontierAdmissionError("registry unclassified repository policy changed") layers = github_estate.get("layers") if not isinstance(layers, list) or len(layers) < 8: raise FrontierAdmissionError("registry GitHub estate layer map is incomplete") candidates = registry.get("candidates") if not isinstance(candidates, list) or len(candidates) < 11: raise FrontierAdmissionError("registry candidate inventory is incomplete") for candidate in candidates: if not isinstance(candidate, dict): raise FrontierAdmissionError("registry candidate record is malformed") required_candidate = { "id", "decision", "upstream", "runtime", "allowed_operations", "prohibited_operations", "required_evidence_by_operation", } if not required_candidate.issubset(candidate): raise FrontierAdmissionError("registry candidate record is incomplete") upstream = candidate.get("upstream") if not isinstance(upstream, dict) or not { "repository_id", "revision", "artifact_inventory" }.issubset(upstream): raise FrontierAdmissionError("registry candidate upstream record is incomplete") allowed = candidate.get("allowed_operations") prohibited = candidate.get("prohibited_operations") requirements = candidate.get("required_evidence_by_operation") if not isinstance(allowed, list) or not allowed: raise FrontierAdmissionError("registry candidate operation allowlist is missing") if not isinstance(prohibited, list) or not HARD_DENIED_OPERATIONS.issubset(prohibited): raise FrontierAdmissionError("registry candidate hard-deny boundary changed") if set(allowed) & set(prohibited): raise FrontierAdmissionError("registry candidate operation is both allowed and prohibited") if not isinstance(requirements, dict): raise FrontierAdmissionError("registry candidate evidence policy is missing") for permitted in allowed: if permitted in UNSIGNED_OPERATION_ALLOWLIST: continue fields = requirements.get(permitted) if not isinstance(fields, list) or not fields: raise FrontierAdmissionError( f"registry candidate signed evidence requirements missing: {permitted}" ) estate = registry.get("hf_estate") repositories = estate.get("repositories") if isinstance(estate, dict) else None if not isinstance(repositories, list) or len(repositories) < 15: raise FrontierAdmissionError("registry Hugging Face estate is incomplete") return registry def _candidate_for_repo(registry: dict[str, Any], repository_id: str) -> dict[str, Any]: matches = [ candidate for candidate in registry.get("candidates", []) if candidate.get("upstream", {}).get("repository_id") == repository_id ] if len(matches) != 1: raise FrontierAdmissionError( f"repository is unknown or ambiguous and is denied: {repository_id}" ) return matches[0] def _validate_evidence( candidate: dict[str, Any], operation: str, revision: str, evidence: dict[str, Any], ) -> None: requirements = candidate.get("required_evidence_by_operation", {}).get(operation, []) missing = [field for field in requirements if not evidence.get(field)] if missing: raise FrontierAdmissionError( f"{operation} missing required evidence: {','.join(sorted(missing))}" ) upstream = candidate["upstream"] runtime = candidate["runtime"] if "exact_revision" in requirements and evidence["exact_revision"] != revision: raise FrontierAdmissionError("evidence exact_revision does not match requested revision") if "method_source_revision" in requirements and evidence["method_source_revision"] != revision: raise FrontierAdmissionError("method source revision does not match the pinned model card") if "runtime_revision" in requirements: expected_runtime = runtime["revision"] if expected_runtime == "MODEL_CARD_ONLY" or evidence["runtime_revision"] != expected_runtime: raise FrontierAdmissionError("runtime revision is absent or does not match the pinned runtime") if "artifact_sha256" in requirements: allowlisted = { item["sha256"] for item in upstream.get("artifact_inventory", []) } if evidence["artifact_sha256"] not in allowlisted: raise FrontierAdmissionError("artifact digest is not in the pinned candidate inventory") for field in requirements: if field.endswith("_sha256") and not SHA256_RE.fullmatch(str(evidence[field])): raise FrontierAdmissionError(f"invalid SHA-256 evidence value: {field}") def _verified_evidence_payload( evidence_document: dict[str, Any], repository_id: str, revision: str, operation: str, ) -> tuple[dict[str, Any], dict[str, Any]]: """Verify a signed evidence envelope and bind it to this exact operation. Digest-looking strings are not receipts. Every evidence-bearing operation requires an ECDSA-P256 DSSE envelope verifiable by the checked-in SZL public key. The signed payload must bind repository, revision, operation, and the complete evidence object before any bounded preflight admission can be emitted. """ envelope = evidence_document.get("dsse") if not isinstance(envelope, dict): raise FrontierAdmissionError("signed DSSE evidence envelope is required") if str(REPO_ROOT) not in sys.path: sys.path.insert(0, str(REPO_ROOT)) try: verifier = importlib.import_module("szl_dsse") verdict = verifier.verify_envelope(envelope) except Exception as exc: raise FrontierAdmissionError( f"DSSE verifier unavailable: {type(exc).__name__}" ) from exc if verdict.get("verified") is not True: raise FrontierAdmissionError( f"DSSE evidence verification failed: {verdict.get('reason', 'signature mismatch')}" ) if verdict.get("payloadType") != EVIDENCE_PAYLOAD_TYPE: raise FrontierAdmissionError("signed evidence DSSE payloadType mismatch") signatures = verdict.get("signatures") or [] if not any( item.get("verified") is True and item.get("keyid") == verdict.get("keyid_expected") for item in signatures if isinstance(item, dict) ): raise FrontierAdmissionError("signed evidence key identity mismatch") payload = verdict.get("payload_decoded") if not isinstance(payload, dict): raise FrontierAdmissionError("signed DSSE evidence payload is not a JSON object") expected_binding = { "schema_version": "szl.frontier-operation-evidence.v1", "repository_id": repository_id, "revision": revision, "operation": operation, } for field, expected in expected_binding.items(): if payload.get(field) != expected: raise FrontierAdmissionError(f"signed evidence binding mismatch: {field}") evidence = payload.get("evidence") if not isinstance(evidence, dict): raise FrontierAdmissionError("signed evidence payload has no evidence object") return evidence, verdict def assert_operation_allowed( repository_id: str, revision: str, operation: str, evidence: dict[str, Any] | None = None, registry_path: Path = DEFAULT_REGISTRY, ) -> dict[str, Any]: """Preflight-admit one bounded operation declaration or fail closed. The return value is deliberately explicit that this check is not a model qualification, execution, or promotion decision. Metadata reads are the only directly authorized operation; signed declarations still require a separate replay-protected executor admission. """ registry = load_registry(registry_path) candidate = _candidate_for_repo(registry, repository_id) pinned_revision = candidate["upstream"]["revision"] if revision != pinned_revision: raise FrontierAdmissionError( f"revision mismatch for {repository_id}: expected {pinned_revision}" ) if operation in HARD_DENIED_OPERATIONS: raise FrontierAdmissionError( f"operation is hard-denied by executable policy: {operation}" ) if operation in candidate.get("prohibited_operations", []): raise FrontierAdmissionError( f"operation is explicitly prohibited for {repository_id}: {operation}" ) if operation not in candidate.get("allowed_operations", []): raise FrontierAdmissionError( f"operation is not allowlisted for {repository_id}: {operation}" ) requirements = candidate.get("required_evidence_by_operation", {}).get(operation, []) if operation not in UNSIGNED_OPERATION_ALLOWLIST and not requirements: raise FrontierAdmissionError( f"operation lacks mandatory signed evidence requirements: {operation}" ) evidence_verified = False dsse_pae_sha256 = None evidence_values: dict[str, Any] = {} if requirements: evidence_values, verdict = _verified_evidence_payload( evidence or {}, repository_id, revision, operation ) evidence_verified = True dsse_pae_sha256 = verdict.get("pae_sha256") _validate_evidence(candidate, operation, revision, evidence_values) metadata_read = operation in UNSIGNED_OPERATION_ALLOWLIST return { "schema_version": "szl.frontier-operation-admission.v1", "repository_id": repository_id, "revision": revision, "candidate_id": candidate["id"], "decision": candidate["decision"], "operation": operation, "operation_allowed_by_registry": True, "operation_authorized": metadata_read, "operation_preflight_admitted": metadata_read or evidence_verified, "execution_authority": False, "replay_protected": False, "execution_boundary": ( "METADATA_READ_ONLY" if metadata_read else "SIGNED_DECLARATION_ONLY_REQUIRES_RUN_BOUND_EXECUTOR_ADMISSION" ), "evidence_verified": evidence_verified, "signed_evidence_declaration_verified": evidence_verified, "underlying_receipt_content_recomputed": False, "dsse_pae_sha256": dsse_pae_sha256, "model_qualified": False, "training_authorized": False, "production_serving_authorized": False, "promotion_authorized": False, "external_mutation_performed": False, } def classify_hf_repository( repository_id: str, registry_path: Path = DEFAULT_REGISTRY, ) -> dict[str, Any]: registry = load_registry(registry_path) matches = [ item for item in registry.get("hf_estate", {}).get("repositories", []) if item.get("repository_id") == repository_id ] if len(matches) != 1: return { "repository_id": repository_id, "classification": "UNCLASSIFIED_FAIL_CLOSED", "delete_authorized": False, } return dict(matches[0]) def audit_registry(registry_path: Path = DEFAULT_REGISTRY) -> dict[str, Any]: registry = load_registry(registry_path) errors: list[str] = [] candidates = registry.get("candidates", []) ids = [candidate.get("id") for candidate in candidates] repos = [candidate.get("upstream", {}).get("repository_id") for candidate in candidates] if len(ids) != len(set(ids)): errors.append("duplicate candidate id") if len(repos) != len(set(repos)): errors.append("duplicate candidate repository") for candidate in candidates: allowed = set(candidate.get("allowed_operations", [])) prohibited = set(candidate.get("prohibited_operations", [])) if allowed & prohibited: errors.append(f"{candidate.get('id')}: operation both allowed and prohibited") if candidate.get("decision", "").startswith("QUARANTINE_") and allowed != {"READ_METADATA"}: errors.append(f"{candidate.get('id')}: quarantine may only allow metadata reads") if candidate.get("decision") in {"LOCAL_QUALIFY", "EVALUATION_ONLY_INFERENCE_ARTIFACT"}: if not candidate.get("required_evidence_by_operation", {}).get("EVALUATE_SANDBOXED"): errors.append(f"{candidate.get('id')}: sandbox evaluation lacks evidence requirements") if "TRAIN" not in prohibited or "PROMOTE" not in prohibited or "MERGE_WEIGHTS" not in prohibited: errors.append(f"{candidate.get('id')}: frontier candidates must deny train/promote/merge") estate = registry.get("hf_estate", {}).get("repositories", []) estate_ids = [item.get("repository_id") for item in estate] if len(estate_ids) != len(set(estate_ids)): errors.append("duplicate Hugging Face estate repository") if any(item.get("delete_authorized") is not False for item in estate): errors.append("an estate record authorizes deletion") policy = registry["evidence_policy"] mutations = registry["external_mutations"] github_estate = registry["github_estate_strategy"] return { "schema_version": "szl.frontier-adoption-audit.v1", "state": "PASS" if not errors else "FAIL", "candidate_count": len(candidates), "estate_repository_count": len(estate), "github_source_report_count": github_estate["source_reported_repository_count"], "github_public_readback_count": github_estate["public_github_repositories_observed"], "github_inventory_complete": github_estate["inventory_complete"], "local_measurements_present": bool(policy["local_measurements_present"]), "promotion_authority": bool(policy["promotion_authority"]), "external_mutation_performed": any(bool(value) for value in mutations.values()), "errors": errors, } def _main(argv: list[str] | None = None) -> int: parser = argparse.ArgumentParser(description=__doc__) parser.add_argument("--registry", type=Path, default=DEFAULT_REGISTRY) subparsers = parser.add_subparsers(dest="command", required=True) subparsers.add_parser("audit") check = subparsers.add_parser("check") check.add_argument("--repository", required=True) check.add_argument("--revision", required=True) check.add_argument("--operation", required=True) check.add_argument("--evidence-json", type=Path) classify = subparsers.add_parser("classify-estate") classify.add_argument("--repository", required=True) args = parser.parse_args(argv) try: if args.command == "audit": result = audit_registry(args.registry) print(json.dumps(result, indent=2, sort_keys=True)) return 0 if result["state"] == "PASS" else 2 if args.command == "classify-estate": print(json.dumps(classify_hf_repository(args.repository, args.registry), indent=2, sort_keys=True)) return 0 evidence = _load_object(args.evidence_json) if args.evidence_json else {} result = assert_operation_allowed( args.repository, args.revision, args.operation, evidence, args.registry, ) print(json.dumps(result, indent=2, sort_keys=True)) return 0 except (OSError, ValueError, KeyError, FrontierAdmissionError) as exc: print(json.dumps({ "state": "BLOCKED", "reason": str(exc), "external_mutation_performed": False, }, indent=2, sort_keys=True)) return 3 if __name__ == "__main__": raise SystemExit(_main())