diff --git a/changelog.d/611-schema4-verifier.added.md b/changelog.d/611-schema4-verifier.added.md new file mode 100644 index 00000000..21fb5dcf --- /dev/null +++ b/changelog.d/611-schema4-verifier.added.md @@ -0,0 +1 @@ +The release contract verifies schema-4 gate-battery reports: exact-k UK releases dispatch on the report's own schema version (3 keeps the legacy aggregator checker untouched, the published June release keeps its grandfathered path), and the new checker pins the executor's identity and the committed UK spec's digests (policy, manifest, fingerprint, entry membership) as hand-mirrored vintage constants, recomputes shippability from the recorded outcomes instead of trusting the `shippable` flag, requires the release-candidate posture, re-applies every legacy observable detail check through a projection onto the shared gate names (including the one-exclusion-clock rule), links `release_evidence.calibration_diagnostics_sha256` to the shipped diagnostics bytes, and authenticates the complete report with the executor's `MICROCOSM_UK_TERMINAL_GATE_SIGNING_KEY` trust root. Build manifests may carry either evidence vocabulary — legacy stage keys or battery entry ids — never a mixture. Cross-shard sync tests hold every mirror equal to the live producer. diff --git a/packages/microcosm-build/tests/test_gate_battery_contract_pins.py b/packages/microcosm-build/tests/test_gate_battery_contract_pins.py new file mode 100644 index 00000000..85179880 --- /dev/null +++ b/packages/microcosm-build/tests/test_gate_battery_contract_pins.py @@ -0,0 +1,169 @@ +"""The schema-4 verifier's mirrors, held to the producer they mirror. + +microcosm-data deliberately does not import microcosm-build, so every +expectation its schema-4 checker enforces is a hand-mirrored constant. +These tests are the lockstep: they import both shards (tests may) and hold +each mirror equal to the live producer — the executor's constants, the +committed UK spec's digests, and the canonical-JSON signature scheme. +""" + +from __future__ import annotations + +import base64 +import hashlib +import hmac +import inspect +import json + +import pytest + +from microcosm.build import load_country_spec +from microcosm.build.gate_battery import ( + GATE_BATTERY_ATTESTATION_SCHEMA_VERSION, + GATE_BATTERY_PRODUCER, + GATE_BATTERY_SCHEMA_VERSION, + EvidenceContext, + GateBatteryRun, + gate_signing_key_env, +) +from microcosm.build.uk_runtime.battery_bindings import UK_GATE_REGISTRY +from microcosm.build.uk_runtime.weighted_integrity import ( + UK_INPUT_MASS_REFERENCE_EVIDENCE_SHA256, +) +from microcosm.data import contract as data_contract + +KEY = base64.b64encode(b"\x07" * 32).decode("ascii") + + +def _uk_run(tmp_path, **overrides) -> GateBatteryRun: + arguments = { + "release_id": "populace-uk-2023-frs-k535080", + "report_path": tmp_path / "terminal_gates.json", + "release_candidate": False, + "registry": UK_GATE_REGISTRY, + **overrides, + } + return GateBatteryRun(load_country_spec("uk").gates, **arguments) + + +class TestMirrorConstants: + def test_executor_identity_mirrors(self) -> None: + assert ( + data_contract._UK_GATE_BATTERY_SCHEMA_VERSION == GATE_BATTERY_SCHEMA_VERSION + ) + assert ( + data_contract._UK_GATE_BATTERY_ATTESTATION_SCHEMA_VERSION + == GATE_BATTERY_ATTESTATION_SCHEMA_VERSION + ) + assert data_contract._UK_GATE_BATTERY_PRODUCER == GATE_BATTERY_PRODUCER + assert data_contract._UK_GATE_BATTERY_SIGNING_KEY_ENV == ( + gate_signing_key_env("uk") + ) + + def test_vintage_pins_mirror_the_committed_spec(self, tmp_path) -> None: + run = _uk_run(tmp_path) + payload = run.report_payload() + assert data_contract._UK_GATE_BATTERY_POLICY_SHA256 == payload["policy_sha256"] + assert ( + data_contract._UK_GATE_BATTERY_GATES_MANIFEST_SHA256 + == payload["gates_manifest_sha256"] + ) + assert ( + data_contract._UK_GATE_BATTERY_SPEC_FINGERPRINT + == payload["spec_fingerprint"] + ) + + def test_entry_membership_mirrors_the_committed_spec(self) -> None: + spec = load_country_spec("uk") + assert data_contract._UK_GATE_BATTERY_ENTRY_IDS == { + entry.id for entry in spec.gates.gates + } + assert set(data_contract._UK_GATE_BATTERY_ENTRY_LEGACY_NAMES) <= { + entry.id for entry in spec.gates.gates + } + + def test_input_mass_evidence_pin_mirrors_the_wrapped_reference(self) -> None: + from microcosm.build.gate_battery import _canonical_sha256 + + assert data_contract._UK_GATE_BATTERY_INPUT_MASS_EVIDENCE_SHA256 == ( + _canonical_sha256( + {"reference_evidence_sha256": (UK_INPUT_MASS_REFERENCE_EVIDENCE_SHA256)} + ) + ) + + +class TestProducerRoundTrip: + """A real executor report against the real verifier. + + The report is produced unarmed (no build evidence), so verdict-driven + refusals are expected; what must NOT appear is any identity, pin, or + signature failure — those would mean a mirror drifted from the producer. + """ + + MIRROR_DRIFT_NEEDLES = ( + "policy_sha256 does not match", + "gates_manifest_sha256 does not match", + "spec_fingerprint does not match", + "producer must name", + "signature does not authenticate", + "signing_key_sha256 does not identify", + "gate_outcomes_sha256 does not match", + "exactly the declared UK entry ids", + ) + + def test_signature_scheme_is_verifiable_across_the_shards( + self, tmp_path, monkeypatch + ) -> None: + monkeypatch.setenv(gate_signing_key_env("uk"), KEY) + run = _uk_run(tmp_path) + run.run_phase("preflight", EvidenceContext()) + run.run_phase("terminal", EvidenceContext()) + report = json.loads((tmp_path / "terminal_gates.json").read_text()) + + signature = report["attestation"]["signature"] + assert signature is not None + report["attestation"]["signature"] = None + recomputed = hmac.new( + base64.b64decode(KEY), + data_contract._canonical_json_bytes(report), + hashlib.sha256, + ).hexdigest() + assert recomputed == signature, ( + "the data shard's canonical JSON must reproduce the producer's signed bytes" + ) + + def test_a_real_report_survives_every_mirror_check( + self, tmp_path, monkeypatch + ) -> None: + monkeypatch.setenv(gate_signing_key_env("uk"), KEY) + arguments: dict = {} + if "release_evidence" in inspect.signature(GateBatteryRun.__init__).parameters: + arguments["release_evidence"] = {"calibration_diagnostics_sha256": "c" * 64} + else: # pragma: no cover - pre-consumer-flip executors only + pytest.skip( + "GateBatteryRun has no release_evidence slot yet; flip when " + "the consumer PR (uk-battery-consumer-a2) merges." + ) + run = _uk_run(tmp_path, release_candidate=True, **arguments) + run.run_phase("preflight", EvidenceContext()) + run.run_phase("terminal", EvidenceContext()) + report = json.loads((tmp_path / "terminal_gates.json").read_text()) + + failures: list[str] = [] + data_contract._check_uk_gate_battery_report( + report, + release_id="populace-uk-2023-frs-k535080", + calibration_diagnostics_sha256="c" * 64, + build_manifest=None, + calibration_diagnostics=None, + failures=failures, + ) + # Unarmed evidence: every blocking entry is a named gap, so verdict + # refusals are the honest outcome. Mirror drift is not. + assert failures, "an unarmed candidate report cannot verify clean" + drifted = [ + line + for line in failures + if any(needle in line for needle in self.MIRROR_DRIFT_NEEDLES) + ] + assert drifted == [], drifted diff --git a/packages/microcosm-data/src/microcosm/data/contract.py b/packages/microcosm-data/src/microcosm/data/contract.py index f48fae75..2f993f32 100644 --- a/packages/microcosm-data/src/microcosm/data/contract.py +++ b/packages/microcosm-data/src/microcosm/data/contract.py @@ -307,6 +307,78 @@ {"national", "region", "country", "local_authority", "constituency"} ) +# --------------------------------------------------------------------------- +# Schema-4 gate-battery verification. Every constant here mirrors the shared +# executor (microcosm.build.gate_battery) or the committed UK spec by name; +# the data shard deliberately does not import the build shard, and the +# build-shard sync tests hold each mirror in lockstep. +# --------------------------------------------------------------------------- +_UK_GATE_BATTERY_SCHEMA_VERSION = 4 +_UK_GATE_BATTERY_ATTESTATION_SCHEMA_VERSION = 6 +_UK_GATE_BATTERY_PRODUCER = "microcosm.build.gate_battery" +# gate_signing_key_env("uk") in the build shard; the legacy POPULACE variable +# stays with the schema-3 path above. +_UK_GATE_BATTERY_SIGNING_KEY_ENV = "MICROCOSM_UK_TERMINAL_GATE_SIGNING_KEY" +_UK_GATE_BATTERY_PHASES = ("preflight", "terminal") +_UK_GATE_BATTERY_STATUSES = frozenset( + {"passed", "failed", "not_applicable", "evidence_absent", "unreached"} +) +_UK_GATE_BATTERY_CRITICALITIES = frozenset({"release_blocking", "diagnostic"}) +_UK_GATE_BATTERY_SHIPPABLE_STATUSES = frozenset({"passed", "not_applicable"}) +# Vintage pins over the committed uk/gates.json: the manifest digest covers +# phase order and notes, the policy digest the reviewed thresholds, and the +# fingerprint derives from the manifest digest. Editing the spec moves all +# three here in the same reviewed change. +_UK_GATE_BATTERY_POLICY_SHA256 = ( + "544cb5475e355abc978b205443bbe49c3af498990fe85cfce6a2a794bf7d179e" +) +_UK_GATE_BATTERY_GATES_MANIFEST_SHA256 = ( + "2ae8bbf57ce0e2cda2380a8db38278752453692f3e4c0005d5a886de1e9711d3" +) +_UK_GATE_BATTERY_SPEC_FINGERPRINT = ( + "ee4b88f939167dc91dc99f79d5f7b8ee776731e0cbedf75bfb78a12efe8789dc" +) +#: Spec entry id -> the legacy gate name whose observable detail checks +#: apply unchanged (the battery re-keys the report by entry id; the gate +#: implementations and their detail schemas are the same code). +_UK_GATE_BATTERY_ENTRY_LEGACY_NAMES = { + "uk_release_input_coverage": "uk_release_input_coverage", + "uk_degenerate_release_surface": "degenerate_release_surface", + "uk_zero_weight_strata": "zero_weight_strata", + "uk_weight_ess": "weight_ess", + "uk_weight_ratio": "weight_ratio", + "uk_weights_audit": "weights_audit", + "uk_export_surface": "export_surface", + "uk_target_surface": "target_surface", + "uk_target_fit": "target_fit", + "uk_input_mass_parity": "input_mass_parity", + "uk_qrf_tail_concentration": "qrf_tail_concentration", +} +_UK_GATE_BATTERY_ENTRY_IDS = frozenset( + { + "uk_release_input_coverage_manifest_current", + "uk_release_family_build_stages", + *_UK_GATE_BATTERY_ENTRY_LEGACY_NAMES, + } +) +#: The entries whose bindings contribute an evidence digest; their keys are +#: the only ones a schema-4 ``evidence_sha256`` may carry, and each appears +#: exactly when its entry evaluated. +_UK_GATE_BATTERY_EVIDENCE_IDS = frozenset( + { + "uk_release_family_build_stages", + "uk_degenerate_release_surface", + "uk_input_mass_parity", + } +) +# The input-mass binding's evidence payload wraps the reviewed reference +# digest as {"reference_evidence_sha256": ...} before the executor's +# canonical hash; this pins the wrapped digest so the entry's evidence line +# still binds the enhanced-FRS incumbent totals. +_UK_GATE_BATTERY_INPUT_MASS_EVIDENCE_SHA256 = ( + "87eb7fa51d826bbe95c9b4d218d82dbd2795b3916469615d38557472209d1e4b" +) + def required_release_files(release_id: str) -> tuple[str, ...]: """Files required for a release id's country-specific contract.""" @@ -374,33 +446,44 @@ def _canonical_json_bytes(value: object) -> bytes: ).encode("utf-8") -def _uk_terminal_verification_key(failures: list[str]) -> bytes | None: - """Load the out-of-band trust root used to authenticate UK reports.""" +def _uk_release_key_from_env(env_var: str, failures: list[str]) -> bytes | None: + """Load an out-of-band trust root used to authenticate UK reports.""" - encoded = os.environ.get(_UK_TERMINAL_GATE_SIGNING_KEY_ENV) + encoded = os.environ.get(env_var) if not encoded: failures.append( f"{_UK_TERMINAL_GATE_REPORT_FILE} verification requires " - f"{_UK_TERMINAL_GATE_SIGNING_KEY_ENV} to contain the release key." + f"{env_var} to contain the release key." ) return None try: key = base64.b64decode(encoded, validate=True) except (binascii.Error, ValueError): failures.append( - f"{_UK_TERMINAL_GATE_SIGNING_KEY_ENV} must be valid base64 to verify " - f"{_UK_TERMINAL_GATE_REPORT_FILE}." + f"{env_var} must be valid base64 to verify {_UK_TERMINAL_GATE_REPORT_FILE}." ) return None if len(key) != 32: failures.append( - f"{_UK_TERMINAL_GATE_SIGNING_KEY_ENV} must decode to exactly 32 bytes " + f"{env_var} must decode to exactly 32 bytes " f"to verify {_UK_TERMINAL_GATE_REPORT_FILE}." ) return None return key +def _uk_terminal_verification_key(failures: list[str]) -> bytes | None: + """The legacy aggregator's trust root (schema-3 reports).""" + + return _uk_release_key_from_env(_UK_TERMINAL_GATE_SIGNING_KEY_ENV, failures) + + +def _uk_gate_battery_verification_key(failures: list[str]) -> bytes | None: + """The shared executor's trust root (schema-4 reports).""" + + return _uk_release_key_from_env(_UK_GATE_BATTERY_SIGNING_KEY_ENV, failures) + + def _reject_json_constant(token: str) -> None: raise ValueError(f"non-standard JSON constant {token}") @@ -577,21 +660,37 @@ def _check_uk_terminal_build_manifest( "'terminal_gate_evidence' object." ) else: - stages = set(evidence) - missing = sorted({"release_dataset"} - stages) - unexpected = sorted( - str(stage) for stage in stages - _UK_TERMINAL_EVIDENCE_STAGES - ) - if missing: - failures.append( - "build_manifest.json terminal_gate_evidence is missing " - f"always-applicable stage(s): {missing}." - ) - if unexpected: - failures.append( - "build_manifest.json terminal_gate_evidence has unknown " - f"stage(s): {unexpected}." - ) + # Two evidence vocabularies, never mixed: the legacy aggregator keys + # by evidence stage (release_dataset always present); the gate + # battery keys by evidence-bearing spec entry id. The report checker + # for the matching schema holds the manifest and the report equal. + stages = set(map(str, evidence)) + if stages and stages <= _UK_GATE_BATTERY_EVIDENCE_IDS: + pass # battery vocabulary; membership is entry-conditional + else: + missing = sorted({"release_dataset"} - stages) + unexpected = sorted( + str(stage) + for stage in stages + - _UK_TERMINAL_EVIDENCE_STAGES + - _UK_GATE_BATTERY_EVIDENCE_IDS + ) + mixed = sorted(stages & _UK_GATE_BATTERY_EVIDENCE_IDS) + if missing: + failures.append( + "build_manifest.json terminal_gate_evidence is missing " + f"always-applicable stage(s): {missing}." + ) + if unexpected: + failures.append( + "build_manifest.json terminal_gate_evidence has unknown " + f"stage(s): {unexpected}." + ) + if mixed: + failures.append( + "build_manifest.json terminal_gate_evidence mixes the " + f"legacy stage vocabulary with battery entry ids: {mixed}." + ) for stage, digest in evidence.items(): _check_sha256_field( filename="build_manifest.json", @@ -961,9 +1060,7 @@ def _check_release_manifest_package( ) if expected_names is not None and name not in expected_names: rendered = " or ".join(repr(n) for n in expected_names) - failures.append( - f"release_manifest.json '{field}.name' must be {rendered}." - ) + failures.append(f"release_manifest.json '{field}.name' must be {rendered}.") elif not name: failures.append(f"release_manifest.json '{field}.name' is required.") version = package.get("version") @@ -1936,6 +2033,363 @@ def _check_uk_terminal_gate_report( ) +def _check_uk_gate_battery_report( + report: Mapping, + *, + release_id: str, + calibration_diagnostics_sha256: str | None, + build_manifest: Mapping | None, + calibration_diagnostics: Mapping | None, + failures: list[str], +) -> None: + """Independently verify an exact-k schema-4 gate-battery report. + + The battery producer is not imported into microcosm-data. This verifier + pins the producer and the reviewed spec identities (policy digest, + manifest digest, fingerprint, entry membership), recomputes shippability + from the recorded outcomes instead of trusting the ``shippable`` flag, + re-applies the legacy observable detail checks through a projection onto + the shared gate implementations' names, and authenticates the complete + report with the executor's out-of-band release key. + """ + + file = _UK_TERMINAL_GATE_REPORT_FILE + required_report_fields = { + "schema_version", + "country", + "release_id", + "release_candidate", + "spec_fingerprint", + "gates_manifest_sha256", + "phases", + "phases_evaluated", + "blocked_at_phase", + "shippable", + "gates", + "policy_sha256", + "release_evidence", + "evidence_sha256", + "attestation", + } + if set(report) != required_report_fields: + failures.append( + f"{file} (schema 4) must contain exactly " + f"{sorted(required_report_fields)}, got {sorted(map(str, report))}." + ) + if report.get("schema_version") != _UK_GATE_BATTERY_SCHEMA_VERSION: + failures.append( + f"{file} schema_version must be {_UK_GATE_BATTERY_SCHEMA_VERSION}." + ) + if report.get("country") != "uk": + failures.append(f"{file} country must be 'uk'.") + if report.get("release_id") != release_id: + failures.append( + f"{file} release_id must match the release being validated; " + f"expected {release_id!r}, got {report.get('release_id')!r}." + ) + if report.get("release_candidate") is not True: + failures.append( + f"{file} release_candidate must be true: a report produced off " + "the candidate posture excused absent evidence and cannot be " + "promoted into a release." + ) + if report.get("blocked_at_phase") is not None: + failures.append(f"{file} blocked_at_phase must be null for a release.") + if list(report.get("phases") or ()) != list(_UK_GATE_BATTERY_PHASES): + failures.append(f"{file} phases must be {list(_UK_GATE_BATTERY_PHASES)}.") + if list(report.get("phases_evaluated") or ()) != list(_UK_GATE_BATTERY_PHASES): + failures.append(f"{file} phases_evaluated must cover every declared phase.") + if report.get("shippable") is not True: + failures.append(f"{file} shippable must be true.") + + if report.get("policy_sha256") != _UK_GATE_BATTERY_POLICY_SHA256: + failures.append( + f"{file} policy_sha256 does not match the certified UK gate " + "policy for this spec vintage." + ) + if report.get("gates_manifest_sha256") != _UK_GATE_BATTERY_GATES_MANIFEST_SHA256: + failures.append( + f"{file} gates_manifest_sha256 does not match the committed " + "uk/gates.json for this spec vintage." + ) + if report.get("spec_fingerprint") != _UK_GATE_BATTERY_SPEC_FINGERPRINT: + failures.append( + f"{file} spec_fingerprint does not match the committed UK spec " + "for this vintage." + ) + + gates = report.get("gates") + valid_gates: dict[str, Mapping] = {} + if not isinstance(gates, Mapping): + failures.append(f"{file} gates must be an object keyed by entry id.") + gates = {} + if set(map(str, gates)) != set(_UK_GATE_BATTERY_ENTRY_IDS): + failures.append( + f"{file} gates must contain exactly the declared UK entry ids; " + f"expected {sorted(_UK_GATE_BATTERY_ENTRY_IDS)}, got " + f"{sorted(map(str, gates))}." + ) + required_entry_fields = { + "gate", + "phase", + "criticality", + "status", + "failures", + "details", + "reason", + } + for entry_id, outcome in gates.items(): + owner = f"{file} gates[{entry_id!r}]" + if not isinstance(outcome, Mapping): + failures.append(f"{owner} must be an object.") + continue + if set(outcome) != required_entry_fields: + failures.append( + f"{owner} must contain exactly {sorted(required_entry_fields)}." + ) + continue + status = outcome.get("status") + if status not in _UK_GATE_BATTERY_STATUSES: + failures.append(f"{owner}.status {status!r} is outside the taxonomy.") + continue + if status == "unreached": + failures.append( + f"{owner} is unreached, which contradicts a complete, " + "unblocked evaluation." + ) + criticality = outcome.get("criticality") + if criticality not in _UK_GATE_BATTERY_CRITICALITIES: + failures.append(f"{owner}.criticality {criticality!r} is unknown.") + elif ( + criticality == "release_blocking" + and status not in _UK_GATE_BATTERY_SHIPPABLE_STATUSES + ): + # Shippability is recomputed here, per entry, instead of + # trusting the report's own shippable flag. + failures.append( + f"{owner} is release-blocking with status {status!r}; the " + "release cannot ship it." + ) + if not isinstance(outcome.get("details"), Mapping): + failures.append(f"{owner}.details must be an object.") + continue + if not isinstance(outcome.get("failures"), list): + failures.append(f"{owner}.failures must be a list.") + continue + valid_gates[str(entry_id)] = outcome + + preflight_coverage = valid_gates.get("uk_release_input_coverage_manifest_current") + if preflight_coverage is not None: + if preflight_coverage.get("status") != "passed": + failures.append(f"{file} the manifest-currency preflight must have passed.") + elif dict(preflight_coverage.get("details", {})) != { + "check": "manifest_current" + }: + failures.append( + f"{file} the manifest-currency preflight details must be " + "exactly {'check': 'manifest_current'}." + ) + roster = valid_gates.get("uk_release_family_build_stages") + if roster is not None and set(roster.get("details", {})) != {"stage_names"}: + failures.append( + f"{file} the build-stage roster details must carry exactly 'stage_names'." + ) + + # The observable detail schemas are the same gate implementations the + # legacy report carried, re-keyed by entry id; project the evaluated + # entries back onto the legacy names and reuse the checks verbatim. + projected = { + _UK_GATE_BATTERY_ENTRY_LEGACY_NAMES[entry_id]: { + "passed": outcome.get("status") == "passed", + "failures": list(outcome.get("failures", ())), + "details": dict(outcome.get("details", {})), + } + for entry_id, outcome in valid_gates.items() + if entry_id in _UK_GATE_BATTERY_ENTRY_LEGACY_NAMES + and outcome.get("status") in ("passed", "failed") + } + _check_uk_terminal_gate_observables( + projected, + calibration_diagnostics=calibration_diagnostics, + failures=failures, + ) + + release_evidence = report.get("release_evidence") + if not isinstance(release_evidence, Mapping) or set(release_evidence) != { + "calibration_diagnostics_sha256" + }: + failures.append( + f"{file} release_evidence must carry exactly " + "'calibration_diagnostics_sha256'." + ) + else: + _check_sha256_field( + filename=file, + owner="release_evidence.calibration_diagnostics_sha256", + value=release_evidence.get("calibration_diagnostics_sha256"), + failures=failures, + ) + if ( + calibration_diagnostics_sha256 is not None + and release_evidence.get("calibration_diagnostics_sha256") + != calibration_diagnostics_sha256 + ): + failures.append( + f"{file} release_evidence.calibration_diagnostics_sha256 " + "must match the local calibration_diagnostics.json bytes." + ) + + evidence = report.get("evidence_sha256") + if not isinstance(evidence, Mapping): + failures.append(f"{file} evidence_sha256 must be an object.") + evidence = {} + unexpected_evidence = sorted( + str(key) for key in set(evidence) - _UK_GATE_BATTERY_EVIDENCE_IDS + ) + if unexpected_evidence: + failures.append( + f"{file} evidence_sha256 has keys outside the evidence-bearing " + f"entries: {unexpected_evidence}." + ) + for entry_id in sorted(_UK_GATE_BATTERY_EVIDENCE_IDS): + evaluated = valid_gates.get(entry_id, {}).get("status") in ( + "passed", + "failed", + ) + if evaluated and entry_id not in evidence: + failures.append( + f"{file} evidence_sha256 is missing the evaluated " + f"evidence-bearing entry {entry_id!r}." + ) + if not evaluated and entry_id in evidence: + failures.append( + f"{file} evidence_sha256 carries {entry_id!r} although the " + "entry did not evaluate." + ) + for entry_id, digest in evidence.items(): + _check_sha256_field( + filename=file, + owner=f"evidence_sha256[{entry_id!r}]", + value=digest, + failures=failures, + ) + if ( + "uk_input_mass_parity" in evidence + and evidence.get("uk_input_mass_parity") + != _UK_GATE_BATTERY_INPUT_MASS_EVIDENCE_SHA256 + ): + failures.append( + f"{file} uk_input_mass_parity evidence digest must bind the " + "reviewed enhanced-FRS incumbent totals." + ) + if build_manifest is not None: + build_evidence = build_manifest.get("terminal_gate_evidence") + if isinstance(build_evidence, Mapping) and dict(build_evidence) != dict( + evidence + ): + failures.append( + f"{file} evidence_sha256 must exactly match " + "build_manifest.json terminal_gate_evidence." + ) + + attestation = report.get("attestation") + if not isinstance(attestation, Mapping): + failures.append(f"{file} attestation must be an object.") + return + required_attestation_fields = { + "schema_version", + "producer", + "country", + "release_id", + "release_candidate", + "spec_fingerprint", + "gates_manifest_sha256", + "policy_sha256", + "phases", + "phases_evaluated", + "blocked_at_phase", + "release_evidence", + "evidence_sha256", + "gate_outcomes_sha256", + "signature_algorithm", + "signing_key_sha256", + "signature", + } + if set(attestation) != required_attestation_fields: + # signing_error is deliberately outside the set: an unsigned report + # records the hole there and can never verify as a release. + failures.append( + f"{file} attestation must contain exactly " + f"{sorted(required_attestation_fields)}, got " + f"{sorted(map(str, attestation))}." + ) + if attestation.get("schema_version") != _UK_GATE_BATTERY_ATTESTATION_SCHEMA_VERSION: + failures.append( + f"{file} attestation.schema_version must be " + f"{_UK_GATE_BATTERY_ATTESTATION_SCHEMA_VERSION}." + ) + if attestation.get("producer") != _UK_GATE_BATTERY_PRODUCER: + failures.append( + f"{file} attestation.producer must name the shared gate-battery executor." + ) + for field in ( + "country", + "release_id", + "release_candidate", + "spec_fingerprint", + "gates_manifest_sha256", + "policy_sha256", + "phases", + "phases_evaluated", + "blocked_at_phase", + "release_evidence", + "evidence_sha256", + ): + if attestation.get(field) != report.get(field): + failures.append( + f"{file} attestation.{field} must equal the report body's {field}." + ) + if attestation.get("signature_algorithm") != _UK_TERMINAL_GATE_SIGNATURE_ALGORITHM: + failures.append( + f"{file} attestation.signature_algorithm must be " + f"{_UK_TERMINAL_GATE_SIGNATURE_ALGORITHM!r}." + ) + expected_gate_outcomes_sha = _canonical_sha256( + {str(entry_id): outcome for entry_id, outcome in gates.items()} + ) + if attestation.get("gate_outcomes_sha256") != expected_gate_outcomes_sha: + failures.append( + f"{file} attestation.gate_outcomes_sha256 does not match gates." + ) + + verification_key = _uk_gate_battery_verification_key(failures) + if verification_key is not None: + expected_key_sha256 = hashlib.sha256(verification_key).hexdigest() + if attestation.get("signing_key_sha256") != expected_key_sha256: + failures.append( + f"{file} attestation.signing_key_sha256 does not identify " + "the trusted release key." + ) + unsigned_report = dict(report) + unsigned_report["attestation"] = { + **{str(key): value for key, value in attestation.items()}, + "signature": None, + } + expected_signature = hmac.new( + verification_key, + _canonical_json_bytes(unsigned_report), + hashlib.sha256, + ).hexdigest() + signature = attestation.get("signature") + if not isinstance(signature, str) or not hmac.compare_digest( + signature, expected_signature + ): + failures.append( + f"{file} attestation.signature does not authenticate the " + "complete report with the trusted release key." + ) + + def _check_calibration_diagnostics( diagnostics: Mapping, failures: list[str], @@ -3291,14 +3745,35 @@ def validate_release_dir(release_dir: Path | str) -> None: ) terminal_gate_report = _load_json(terminal_gate_path, failures) if terminal_gate_report is not None: - _check_uk_terminal_gate_report( - terminal_gate_report, - release_id=release_id, - calibration_diagnostics_sha256=calibration_diagnostics_sha256, - build_manifest=build_manifest, - calibration_diagnostics=calibration_diagnostics, - failures=failures, - ) + # Vintage dispatch on the report's own schema: 3 is the legacy + # aggregator, 4 the shared gate battery. Anything else is not a + # UK terminal report. + report_schema = terminal_gate_report.get("schema_version") + if report_schema == _UK_GATE_BATTERY_SCHEMA_VERSION: + _check_uk_gate_battery_report( + terminal_gate_report, + release_id=release_id, + calibration_diagnostics_sha256=calibration_diagnostics_sha256, + build_manifest=build_manifest, + calibration_diagnostics=calibration_diagnostics, + failures=failures, + ) + elif report_schema == _UK_TERMINAL_GATE_SCHEMA_VERSION: + _check_uk_terminal_gate_report( + terminal_gate_report, + release_id=release_id, + calibration_diagnostics_sha256=calibration_diagnostics_sha256, + build_manifest=build_manifest, + calibration_diagnostics=calibration_diagnostics, + failures=failures, + ) + else: + failures.append( + f"{_UK_TERMINAL_GATE_REPORT_FILE} schema_version must be " + f"{_UK_TERMINAL_GATE_SCHEMA_VERSION} (legacy aggregator) " + f"or {_UK_GATE_BATTERY_SCHEMA_VERSION} (gate battery), " + f"got {report_schema!r}." + ) _check_cross_manifest_consistency( build_manifest, diff --git a/packages/microcosm-data/tests/test_contract.py b/packages/microcosm-data/tests/test_contract.py index 50801705..8423dcbd 100644 --- a/packages/microcosm-data/tests/test_contract.py +++ b/packages/microcosm-data/tests/test_contract.py @@ -93,6 +93,64 @@ def _trusted_terminal_gate_signing_key(monkeypatch) -> None: ) +# --------------------------------------------------------------------------- +# Schema-4 gate-battery mirrors. Local copies in the schema-3 style; the +# lockstep test below holds them equal to the contract module's pins, and the +# build-shard sync tests hold the contract module equal to the producer. +# --------------------------------------------------------------------------- +UK_GATE_BATTERY_PRODUCER = "microcosm.build.gate_battery" +UK_GATE_BATTERY_SIGNING_KEY_ENV = "MICROCOSM_UK_TERMINAL_GATE_SIGNING_KEY" +UK_GATE_BATTERY_POLICY_SHA256 = ( + "544cb5475e355abc978b205443bbe49c3af498990fe85cfce6a2a794bf7d179e" +) +UK_GATE_BATTERY_GATES_MANIFEST_SHA256 = ( + "2ae8bbf57ce0e2cda2380a8db38278752453692f3e4c0005d5a886de1e9711d3" +) +UK_GATE_BATTERY_SPEC_FINGERPRINT = ( + "ee4b88f939167dc91dc99f79d5f7b8ee776731e0cbedf75bfb78a12efe8789dc" +) +#: Spec entry id -> (neutral gate name, phase, legacy detail-schema name). +UK_GATE_BATTERY_ENTRIES = { + "uk_release_input_coverage_manifest_current": ( + "release_input_coverage", + "preflight", + None, + ), + "uk_release_family_build_stages": ("source_coverage", "preflight", None), + "uk_release_input_coverage": ( + "release_input_coverage", + "terminal", + "uk_release_input_coverage", + ), + "uk_degenerate_release_surface": ( + "degenerate_release_surface", + "terminal", + "degenerate_release_surface", + ), + "uk_zero_weight_strata": ("zero_weight_strata", "terminal", "zero_weight_strata"), + "uk_weight_ess": ("weight_ess", "terminal", "weight_ess"), + "uk_weight_ratio": ("weight_ratio", "terminal", "weight_ratio"), + "uk_weights_audit": ("weights_audit", "terminal", "weights_audit"), + "uk_export_surface": ("export_surface", "terminal", "export_surface"), + "uk_target_surface": ("target_surface", "terminal", "target_surface"), + "uk_target_fit": ("target_fit", "terminal", "target_fit"), + "uk_input_mass_parity": ("input_mass_parity", "terminal", "input_mass_parity"), + "uk_qrf_tail_concentration": ( + "tail_concentration", + "terminal", + "qrf_tail_concentration", + ), +} + + +@pytest.fixture(autouse=True) +def _trusted_gate_battery_signing_key(monkeypatch) -> None: + monkeypatch.setenv( + UK_GATE_BATTERY_SIGNING_KEY_ENV, + TEST_UK_TERMINAL_GATE_SIGNING_KEY, + ) + + DEDUCTION_CRITICAL_TARGETS = ( ( "irs_soi.ty2022.historic_table_2.us.all.itemized_deductions_amount@2024", @@ -834,6 +892,136 @@ def _refresh_terminal_manifest_hashes(release_dir: Path) -> None: release_path.write_text(json.dumps(release)) +def _resign_gate_battery( + payload: dict, + *, + signing_key: bytes = TEST_UK_TERMINAL_GATE_SIGNING_KEY_BYTES, +) -> None: + """Rebuild the schema-4 attestation from the body and sign, as the + producer does: the signature covers the canonical payload with the + signature slot nulled.""" + + attestation = { + "schema_version": 6, + "producer": UK_GATE_BATTERY_PRODUCER, + "country": payload["country"], + "release_id": payload["release_id"], + "release_candidate": payload["release_candidate"], + "spec_fingerprint": payload["spec_fingerprint"], + "gates_manifest_sha256": payload["gates_manifest_sha256"], + "policy_sha256": payload["policy_sha256"], + "phases": payload["phases"], + "phases_evaluated": payload["phases_evaluated"], + "blocked_at_phase": payload["blocked_at_phase"], + "release_evidence": payload["release_evidence"], + "evidence_sha256": payload["evidence_sha256"], + "gate_outcomes_sha256": _canonical_sha256(payload["gates"]), + "signature_algorithm": UK_TERMINAL_GATE_SIGNATURE_ALGORITHM, + "signing_key_sha256": hashlib.sha256(signing_key).hexdigest(), + "signature": None, + } + payload["attestation"] = attestation + attestation["signature"] = hmac.new( + signing_key, _canonical_json_bytes(payload), hashlib.sha256 + ).hexdigest() + + +def _gate_battery_payload( + *, + release_id: str, + calibration_diagnostics_sha256: str, + signing_key: bytes = TEST_UK_TERMINAL_GATE_SIGNING_KEY_BYTES, +) -> tuple[dict, dict[str, str]]: + """A fully-armed, all-passing, signed schema-4 battery report.""" + + stage_names = ["frs_hmrc_retained_leaves", "hmrc_spi_income"] + gates: dict[str, dict] = {} + for entry_id, (gate, phase, detail_name) in UK_GATE_BATTERY_ENTRIES.items(): + if entry_id == "uk_release_input_coverage_manifest_current": + details: dict = {"check": "manifest_current"} + elif entry_id == "uk_release_family_build_stages": + details = {"stage_names": list(stage_names)} + else: + details = _terminal_gate_details(detail_name) + gates[entry_id] = { + "gate": gate, + "phase": phase, + "criticality": "release_blocking", + "status": "passed", + "failures": [], + "details": details, + "reason": None, + } + evidence = { + "uk_release_family_build_stages": _canonical_sha256( + {"stage_names": list(stage_names)} + ), + "uk_degenerate_release_surface": _canonical_sha256( + {"exclusions_register": "committed"} + ), + "uk_input_mass_parity": _canonical_sha256( + {"reference_evidence_sha256": UK_INPUT_MASS_REFERENCE_EVIDENCE_SHA256} + ), + } + payload = { + "schema_version": 4, + "country": "uk", + "release_id": release_id, + "release_candidate": True, + "spec_fingerprint": UK_GATE_BATTERY_SPEC_FINGERPRINT, + "gates_manifest_sha256": UK_GATE_BATTERY_GATES_MANIFEST_SHA256, + "phases": ["preflight", "terminal"], + "phases_evaluated": ["preflight", "terminal"], + "blocked_at_phase": None, + "shippable": True, + "gates": gates, + "policy_sha256": UK_GATE_BATTERY_POLICY_SHA256, + "release_evidence": { + "calibration_diagnostics_sha256": calibration_diagnostics_sha256 + }, + "evidence_sha256": evidence, + } + _resign_gate_battery(payload, signing_key=signing_key) + return payload, evidence + + +def _upgrade_release_to_gate_battery(directory: Path) -> dict: + """Swap a fixture release's schema-3 report for a valid schema-4 one.""" + + payload, evidence = _gate_battery_payload( + release_id=UK_EXACT_K_RELEASE_ID, + calibration_diagnostics_sha256=_sha256( + directory / "calibration_diagnostics.json" + ), + ) + build_path = directory / "build_manifest.json" + build = json.loads(build_path.read_text()) + build["terminal_gate_evidence"] = evidence + build_path.write_text(json.dumps(build)) + _write_terminal_and_refresh_manifest_hashes(directory, payload) + return payload + + +def _write_battery_release(tmp_path: Path) -> tuple[Path, dict]: + directory = _write_uk_release_dir(tmp_path, UK_EXACT_K_RELEASE_ID, tier="frs") + payload = _upgrade_release_to_gate_battery(directory) + return directory, payload + + +def _rewrite_battery_report( + directory: Path, payload: dict, *, resign: bool = True +) -> None: + if resign: + _resign_gate_battery(payload) + _write_terminal_and_refresh_manifest_hashes(directory, payload) + + +def _battery_failures(directory: Path) -> str: + with pytest.raises(ReleaseContractError) as excinfo: + validate_release_dir(directory) + return "\n".join(excinfo.value.failures) + + def _write_uk_release_dir( tmp_path: Path, release_id: str, @@ -3449,3 +3637,235 @@ def test_exact_k_uk_terminal_rejects_invalid_qrf_observable_values( failures = "\n".join(excinfo.value.failures) assert match in failures assert "attestation.signature does not authenticate" not in failures + + +# --------------------------------------------------------------------------- +# Schema-4 gate-battery reports (the #611 executor). +# --------------------------------------------------------------------------- + + +def test_uk_gate_battery_pins_are_in_lockstep_with_the_contract() -> None: + from microcosm.data import contract as contract_module + + assert contract_module._UK_GATE_BATTERY_PRODUCER == UK_GATE_BATTERY_PRODUCER + assert ( + contract_module._UK_GATE_BATTERY_SIGNING_KEY_ENV + == UK_GATE_BATTERY_SIGNING_KEY_ENV + ) + assert ( + contract_module._UK_GATE_BATTERY_POLICY_SHA256 == UK_GATE_BATTERY_POLICY_SHA256 + ) + assert ( + contract_module._UK_GATE_BATTERY_GATES_MANIFEST_SHA256 + == UK_GATE_BATTERY_GATES_MANIFEST_SHA256 + ) + assert ( + contract_module._UK_GATE_BATTERY_SPEC_FINGERPRINT + == UK_GATE_BATTERY_SPEC_FINGERPRINT + ) + assert contract_module._UK_GATE_BATTERY_ENTRY_IDS == frozenset( + UK_GATE_BATTERY_ENTRIES + ) + + +def test_exact_k_uk_gate_battery_report_validates_end_to_end(tmp_path: Path) -> None: + directory, _payload = _write_battery_release(tmp_path) + + validate_release_dir(directory) + + +def test_exact_k_uk_gate_battery_rejects_unknown_report_schema( + tmp_path: Path, +) -> None: + directory, payload = _write_battery_release(tmp_path) + payload["schema_version"] = 5 + _rewrite_battery_report(directory, payload, resign=False) + + assert "schema_version must be 3" in _battery_failures(directory) + + +def test_exact_k_uk_gate_battery_rejects_a_non_candidate_report( + tmp_path: Path, +) -> None: + # A report produced off the candidate posture excused absent evidence; + # promoting it into a release dir must fail even when honestly signed. + directory, payload = _write_battery_release(tmp_path) + payload["release_candidate"] = False + _rewrite_battery_report(directory, payload) + + assert "release_candidate must be true" in _battery_failures(directory) + + +def test_exact_k_uk_gate_battery_rejects_a_blocked_report(tmp_path: Path) -> None: + directory, payload = _write_battery_release(tmp_path) + payload["blocked_at_phase"] = "terminal" + _rewrite_battery_report(directory, payload) + + assert "blocked_at_phase must be null" in _battery_failures(directory) + + +def test_exact_k_uk_gate_battery_recomputes_shippability(tmp_path: Path) -> None: + # shippable: true is asserted but never trusted — a failed blocking + # entry inside an honestly re-signed report still refuses. + directory, payload = _write_battery_release(tmp_path) + entry = payload["gates"]["uk_weight_ratio"] + entry["status"] = "failed" + entry["failures"] = ["seeded ratio failure"] + _rewrite_battery_report(directory, payload) + + failures = _battery_failures(directory) + assert "release-blocking with status 'failed'" in failures + + +def test_exact_k_uk_gate_battery_rejects_excused_absent_evidence( + tmp_path: Path, +) -> None: + directory, payload = _write_battery_release(tmp_path) + entry = payload["gates"]["uk_input_mass_parity"] + entry["status"] = "evidence_absent" + entry["details"] = {} + entry["reason"] = "missing evidence: input_mass_policy" + del payload["evidence_sha256"]["uk_input_mass_parity"] + _rewrite_battery_report(directory, payload) + build_path = directory / "build_manifest.json" + build = json.loads(build_path.read_text()) + build["terminal_gate_evidence"] = dict(payload["evidence_sha256"]) + build_path.write_text(json.dumps(build)) + _refresh_terminal_manifest_hashes(directory) + + failures = _battery_failures(directory) + assert "release-blocking with status 'evidence_absent'" in failures + + +def test_exact_k_uk_gate_battery_rejects_a_missing_entry(tmp_path: Path) -> None: + directory, payload = _write_battery_release(tmp_path) + del payload["gates"]["uk_weight_ess"] + _rewrite_battery_report(directory, payload) + + assert "exactly the declared UK entry ids" in _battery_failures(directory) + + +def test_exact_k_uk_gate_battery_rejects_an_uncertified_policy( + tmp_path: Path, +) -> None: + directory, payload = _write_battery_release(tmp_path) + payload["policy_sha256"] = "b" * 64 + _rewrite_battery_report(directory, payload) + + assert "certified UK gate policy" in _battery_failures(directory) + + +def test_exact_k_uk_gate_battery_rejects_a_moved_manifest_digest( + tmp_path: Path, +) -> None: + directory, payload = _write_battery_release(tmp_path) + payload["gates_manifest_sha256"] = "b" * 64 + _rewrite_battery_report(directory, payload) + + assert "committed uk/gates.json" in _battery_failures(directory) + + +def test_exact_k_uk_gate_battery_rejects_a_moved_spec_fingerprint( + tmp_path: Path, +) -> None: + directory, payload = _write_battery_release(tmp_path) + payload["spec_fingerprint"] = "b" * 64 + _rewrite_battery_report(directory, payload) + + assert "spec_fingerprint does not match" in _battery_failures(directory) + + +def test_exact_k_uk_gate_battery_rejects_a_broken_diagnostics_link( + tmp_path: Path, +) -> None: + directory, payload = _write_battery_release(tmp_path) + payload["release_evidence"] = {"calibration_diagnostics_sha256": "b" * 64} + _rewrite_battery_report(directory, payload) + + assert ( + "release_evidence.calibration_diagnostics_sha256 must match" + in _battery_failures(directory) + ) + + +def test_exact_k_uk_gate_battery_rejects_a_mixed_exclusion_clock( + tmp_path: Path, +) -> None: + directory, payload = _write_battery_release(tmp_path) + details = payload["gates"]["uk_degenerate_release_surface"]["details"] + details["exclusions_evaluated_on"] = "2026-01-01" + _rewrite_battery_report(directory, payload) + + assert "exclusion-consuming gates must" in _battery_failures(directory) + + +def test_exact_k_uk_gate_battery_no_resign_tamper_fails_authentication( + tmp_path: Path, +) -> None: + directory, payload = _write_battery_release(tmp_path) + payload["gates"]["uk_weight_ratio"]["details"]["max_weight"] = 1.0e9 + _rewrite_battery_report(directory, payload, resign=False) + + failures = _battery_failures(directory) + assert "attestation.signature does not authenticate" in failures + assert "attestation.gate_outcomes_sha256 does not match" in failures + + +def test_exact_k_uk_gate_battery_rejects_a_forged_key(tmp_path: Path) -> None: + directory, payload = _write_battery_release(tmp_path) + _resign_gate_battery(payload, signing_key=FORGED_UK_TERMINAL_GATE_SIGNING_KEY_BYTES) + _rewrite_battery_report(directory, payload, resign=False) + + failures = _battery_failures(directory) + assert "signing_key_sha256 does not identify the trusted release key" in failures + + +def test_exact_k_uk_gate_battery_requires_the_executor_key_env( + tmp_path: Path, monkeypatch +) -> None: + directory, _payload = _write_battery_release(tmp_path) + monkeypatch.delenv(UK_GATE_BATTERY_SIGNING_KEY_ENV) + + assert UK_GATE_BATTERY_SIGNING_KEY_ENV in _battery_failures(directory) + + +def test_exact_k_uk_gate_battery_rejects_a_recorded_signing_error( + tmp_path: Path, +) -> None: + directory, payload = _write_battery_release(tmp_path) + _resign_gate_battery(payload) + payload["attestation"]["signing_error"] = "key was absent at build time" + _rewrite_battery_report(directory, payload, resign=False) + + assert "attestation must contain exactly" in _battery_failures(directory) + + +def test_exact_k_uk_gate_battery_rejects_an_unpinned_input_mass_evidence( + tmp_path: Path, +) -> None: + directory, payload = _write_battery_release(tmp_path) + payload["evidence_sha256"]["uk_input_mass_parity"] = "b" * 64 + _rewrite_battery_report(directory, payload) + build_path = directory / "build_manifest.json" + build = json.loads(build_path.read_text()) + build["terminal_gate_evidence"] = dict(payload["evidence_sha256"]) + build_path.write_text(json.dumps(build)) + _refresh_terminal_manifest_hashes(directory) + + assert "bind the reviewed enhanced-FRS" in _battery_failures(directory) + + +def test_exact_k_uk_gate_battery_rejects_mixed_manifest_evidence_vocabulary( + tmp_path: Path, +) -> None: + directory, payload = _write_battery_release(tmp_path) + build_path = directory / "build_manifest.json" + build = json.loads(build_path.read_text()) + build["terminal_gate_evidence"] = { + "release_dataset": "a" * 64, + **dict(payload["evidence_sha256"]), + } + build_path.write_text(json.dumps(build)) + + failures = _battery_failures(directory) + assert "mixes the legacy stage vocabulary" in failures