From 2bc20af441f6ec14cea37522c436beb6740ef3c8 Mon Sep 17 00:00:00 2001 From: abrichr Date: Tue, 8 Sep 2026 22:56:24 -0400 Subject: [PATCH 1/2] feat: bind measured Flow evidence to existing admission issuance Signed-off-by: abrichr --- .../flow-1.35.1-measured/README.md | 70 + .../flow-1.35.1-measured/issue.py | 1316 +++++++++++++++++ tests/test_measured_flow_admission.py | 1135 ++++++++++++++ 3 files changed, 2521 insertions(+) create mode 100755 local-candidates/flow-1.35.1-measured/issue.py create mode 100644 tests/test_measured_flow_admission.py diff --git a/local-candidates/flow-1.35.1-measured/README.md b/local-candidates/flow-1.35.1-measured/README.md index 870afcb..bede392 100644 --- a/local-candidates/flow-1.35.1-measured/README.md +++ b/local-candidates/flow-1.35.1-measured/README.md @@ -39,3 +39,73 @@ protected-main qualification and publication steps. A changed published wheel requires new exact-byte evidence. The existing issuer must then bind the reviewed release and workflow contracts before any signed admission or live projection changes. The published Flow admission still names 1.34.0. + +## Issue a measured admission + +`issue.py` accepts a separate one-task, one-bundle campaign and a reviewed private +mapping. The comparative study above stays supplementary. Keep the mapping, +contract openings, encrypted bundle, and trial evidence in their private +location. Each file reference binds its relative path, SHA-256, and byte size. +The adapter checks the published wheel and sdist against both publishers. + +Prepare one phase at a time: `receipt`, `workflow`, `manifest`, `summary`, then +`release`. Each phase request supplies `issuer_source_commit`, `issued_at`, +`expires_at`, and a unique `request_handle` (`qair_` followed by 43 URL-safe +characters). Set `references` to the required registered regular-object +references. The manifest phase also needs `acceptance_issuer_source_commit`. +The issuer derives workflow and release expiry from their signed dependencies. + +From the repository root, set these variables to your private paths and the +reviewed mapping digest, including its `sha256:` prefix: + +```bash +python local-candidates/flow-1.35.1-measured/issue.py \ + --mapping "$MEASURED_MAPPING" --mapping-sha256 "$MAPPING_SHA256" \ + --phase-request "$PHASE_REQUEST" --output "$UNSIGNED_PLAN" +``` + +The default writes an unsigned plan. Review its exact bytes and retain the +printed `plan_sha256`. To sign that plan with the existing Keychain key, repeat +those inputs and choose a permanent state directory and a separate output +directory: + +```bash +python local-candidates/flow-1.35.1-measured/issue.py \ + --mapping "$MEASURED_MAPPING" --mapping-sha256 "$MAPPING_SHA256" \ + --phase-request "$PHASE_REQUEST" --output "$SIGNED_OUTPUT" \ + --sign --reviewed-plan-sha256 "$PLAN_SHA256" --state-dir "$ADMISSION_STATE" +``` + +Add `--stage-registry /path/to/evidence-registry.json` to append the signed pair +through the existing staging tool. Commit the pair before using its reference. +The script doesn't commit or push. + +Use current `.github` main for receipt, workflow, release, and outer signing +sources. The receipt must reach `.github` main before workflow issuance, and the +summary must reach main before release issuance. Intermediate workflow and +manifest storage references can name committed branch objects; preserve those +commits in the reviewed merge. Acceptance uses the reviewed `openadapt-evals` +main source. A storage commit never substitutes for an issuer source. + +Collect fresh publication staging after the campaign, receipt, and workflow +are complete, just before preparing the acceptance manifest. Preserve the +frozen campaign records and earlier signed objects. The manifest and summary +bind that staging digest, so refreshing only the release request won't extend +its window. + +Keep issue times within the last hour and finish release issuance within the +publication observation's one-hour window. If it expires, stop for review of a +new acceptance issuance; preserve all signed objects and one-use state. Before +consumption, changed inputs require another unsigned plan and review. + +Keep the state directory and its request journals permanently. After an +interruption, read the original result without a key or network call: + +```bash +python local-candidates/flow-1.35.1-measured/issue.py \ + --reconcile-journal "$REQUEST_JOURNAL" --state-dir "$ADMISSION_STATE" \ + --output "$RECONCILIATION_OUTPUT" +``` + +Reconciliation returns the consumed unsigned result. An unknown result requires +investigation; don't delete the journal or retry with a new handle. diff --git a/local-candidates/flow-1.35.1-measured/issue.py b/local-candidates/flow-1.35.1-measured/issue.py new file mode 100755 index 0000000..ea0273f --- /dev/null +++ b/local-candidates/flow-1.35.1-measured/issue.py @@ -0,0 +1,1316 @@ +#!/usr/bin/env python3 +"""Prepare, then explicitly sign, a measured one-bundle Flow admission. + +The private mapping and bundle verification proof are reviewed inputs, not a +public qualification harness. Every file reference is relative to its owning +JSON file and binds SHA-256 and size. Unsigned preparation is the default and +never reads a key, consumes a request, or changes a registry. Each later phase +requires the SHA-256 of its exact unsigned plan. Dependency references must name +pairs at actual containing commits. The immediate receipt/workflow and +summary/release edges require protected main. Other storage references may name +committed branches that the reviewed merge preserves. The caller keeps the state directory permanently, including unknown +outcomes. Re-running an interrupted request only reconciles the original result. +""" + +from __future__ import annotations + +import argparse +import base64 +import copy +import hashlib +import importlib.util +import json +import os +import sqlite3 +import stat +import subprocess +import sys +import zipfile +from datetime import datetime, timezone +from pathlib import Path +from urllib.parse import urlsplit + +ROOT = Path(__file__).resolve().parents[2] +sys.path.insert(0, str(ROOT / "scripts")) +import production_trust as trust +import public_trust_kms as public_trust +import qualification_issuer as issuer +import qualification_software_ed25519 as software +import stage_production_evidence as stage +import validate_evidence_registry as evidence +import verify_production_release_admission as verifier + +_SPEC = importlib.util.spec_from_file_location( + "measured_flow_verifier", ROOT / "local-candidates/flow-1.35.0-measured/prepare.py" +) +assert _SPEC and _SPEC.loader +measured = importlib.util.module_from_spec(_SPEC) +_SPEC.loader.exec_module(measured) + +SUBJECT_FIELDS = { + "bundle_content_digest", + "workflow_contract_sha256", + "parameter_schema_sha256", + "governed_policy_contract_sha256", + "governed_runtime_inputs_digest", + "governed_qualification_project_contract_sha256", + "observed_application_sha256", + "observed_application_version_sha256", + "observed_environment_digest", + "observed_environment_binding_sha256", + "qualification_environment_observer_id", + "qualification_environment_observer_contract_sha256", +} +# These are byte commitments. Existing authority, signer and revocation identities +# instead come from verified current signed state; they are not new measurements. +FILE_COMMITMENTS = { + "decision_identity_sha256", + "decision_commitment_sha256", + "evidence_manifest_sha256", + "evidence_manifest_readback_sha256", + "campaign_artifact_sha256", + "organization_id_sha256", + "workflow_id_sha256", + "workflow_version_id_sha256", + "bundle_sha256", + "admitted_runtime_sha256", + "application_contract_sha256", + "environment_contract_sha256", + "input_contract_sha256", + "action_contract_sha256", + "identity_contract_sha256", + "effect_contract_sha256", + "policy_contract_sha256", + "campaign_permit_sha256", +} +KINDS = { + "receipt": "qualification-evidence-decision-receipt", + "workflow": "qualification-admission", + "manifest": "production-acceptance-manifest", + "summary": "production-acceptance-summary", + "release": "qualification-release", +} + + +def sha(raw: bytes) -> str: + return "sha256:" + hashlib.sha256(raw).hexdigest() + + +def canonical(value) -> bytes: + return evidence.canonical(value) + b"\n" + + +def fail(message: str) -> None: + raise ValueError(message) + + +def checked_file(owner: Path, reference: dict) -> tuple[Path, bytes]: + """Read an exact retained file without allowing traversal or symlink aliases.""" + if not isinstance(reference, dict): + fail("file reference must be an object") + relative = Path(reference.get("path", "")) + base = owner.resolve().parent + if not reference.get("path") or relative.is_absolute() or ".." in relative.parts: + fail("file reference must remain inside its owning directory") + path = base / relative + if any(part.is_symlink() for part in (path, *path.parents) if part != base.parent): + fail("file reference contains a symlink") + if not path.resolve().is_relative_to(base) or not path.is_file(): + fail("file reference is not a retained regular file") + raw = path.read_bytes() + if ( + type(reference.get("size_bytes")) is not int + or reference["size_bytes"] != len(raw) + or reference.get("sha256") != sha(raw) + ): + fail("retained file bytes differ from their reviewed commitment") + return path, raw + + +def checked_json(owner: Path, reference: dict) -> tuple[Path, dict]: + path, raw = checked_file(owner, reference) + return path, json.loads(raw) + + +def subject(value: dict) -> dict: + trust.closed(value, SUBJECT_FIELDS, "native admitted subject") + if any(not isinstance(v, str) or not v for v in value.values()): + fail("every native subject field must be explicit and nonempty") + return value + + +def verify_bundle( + owner: Path, reference: dict, proof_reference: dict, expected: dict +) -> dict: + archive_path, archive_raw = checked_file(owner, reference) + proof_path, proof = checked_json(owner, proof_reference) + if ( + proof.get("schema_version") != "openadapt.measured-bundle-verification/v1" + or proof.get("verified") is not True + or proof.get("bundle_archive_sha256") != sha(archive_raw) + or subject(proof.get("admitted_subject")) != expected + ): + fail("reviewed encrypted-bundle verification does not bind this subject") + checked_file(proof_path, proof.get("verifier")) + members = proof.get("bundle_members", []) + indexed = {item["path"]: item for item in members} + if not indexed or len(indexed) != len(members): + fail("encrypted bundle member inventory is empty or duplicated") + with zipfile.ZipFile(archive_path) as archive: + infos = archive.infolist() + if len(infos) != len(indexed) or {i.filename for i in infos} != set(indexed): + fail("encrypted archive members differ from reviewed proof") + for info in infos: + path = Path(info.filename) + if ( + path.is_absolute() + or ".." in path.parts + or info.is_dir() + or stat.S_ISLNK(info.external_attr >> 16) + ): + fail("encrypted bundle archive member is not a safe regular file") + raw = archive.read(info) + item = indexed[info.filename] + if sha(raw) != item["sha256"] or len(raw) != item["size_bytes"]: + fail("encrypted bundle member bytes differ") + return proof + + +def replay_ledger_record(raw: bytes, *, key: str, primary: dict) -> tuple: + """Read the verified snapshot bytes without opening or migrating a live ledger.""" + if not raw.startswith(b"SQLite format 3\x00"): + fail("replay ledger snapshot must contain the actual SQLite v3 ledger") + run_id = primary.get("run_id_sha256") + if ( + not isinstance(run_id, str) + or not run_id + or primary.get("transaction_outcome") != "VERIFIED" + ): + fail("replay proof needs the actual verified initial run identity") + try: + connection = sqlite3.connect(":memory:") + try: + connection.deserialize(raw) + connection.execute("PRAGMA query_only = ON") + metadata = connection.execute( + "SELECT schema_version, namespace, owner_path FROM ledger_metadata WHERE singleton = 1" + ).fetchall() + if ( + len(metadata) != 1 + or metadata[0][0] != "openadapt.idempotency-ledger/v3" + or any(not isinstance(v, str) or not v for v in metadata[0]) + ): + fail( + "retained replay ledger metadata differs from the native v3 schema" + ) + rows = connection.execute( + "SELECT run_id, reserved_at, outcome FROM reservations WHERE namespace = ? AND reservation_key = ?", + (metadata[0][1], key), + ).fetchall() + if ( + len(rows) != 1 + or not isinstance(rows[0][0], str) + or not rows[0][0] + or hashlib.sha256(rows[0][0].encode("utf-8")).hexdigest() != run_id + or rows[0][2] != "VERIFIED" + or not isinstance(rows[0][1], str) + or not rows[0][1] + ): + fail( + "retained replay ledger does not bind the same key to the verified initial run" + ) + return (*metadata[0], *rows[0]) + finally: + connection.close() + except sqlite3.Error as exc: + raise ValueError( + "retained replay ledger is not a readable native SQLite snapshot" + ) from exc + + +def verify_subjects( + manifest_path: Path, manifest: dict, expected: dict, prior: dict | None +) -> None: + if subject(manifest.get("admitted_subject")) != expected: + fail("measured manifest admitted subject differs from mapping") + tasks = {t["task_id"] for t in manifest["trials"]} + if len(tasks) != 1: + fail("this admission requires one task and one admitted bundle") + inventory = {item["path"]: item for item in manifest["artifacts"]} + seen_primary = set() + + def inventoried(ref): + item = inventory.get(ref.get("path")) + if item is None or item["sha256"] != ref.get("sha256"): + fail("replay proof reference is absent from verified inventory") + return checked_file(manifest_path, item) + + for trial in manifest["trials"]: + _, observation = checked_json( + manifest_path, inventory[trial["observation"]["path"]] + ) + counter_source = observation.get("counter_source", {}) + _, source_raw = inventoried(counter_source.get("observation", {})) + source_rows = json.loads(source_raw) + row_index = counter_source.get("row_index") + if ( + not isinstance(source_rows, list) + or type(row_index) is not int + or row_index < 0 + or row_index >= len(source_rows) + ): + fail("counter source must select an actual retained observation row") + row = source_rows[row_index] + if ( + row.get("passed") is not True + or row.get("errors") != [] + or row.get("runtime_version") != manifest["runtime"]["version"] + or any( + row.get(k) != trial[k] + for k in ("class", "task_id", "condition", "trial", "counters") + ) + ): + fail( + "normalized counters or trial identity differ from the actual passing source row" + ) + + def report_bindings(reports): + if not isinstance(reports, list) or not reports: + fail("source row must retain its actual native reports") + bindings = [] + for item in reports: + if not isinstance(item, dict) or any( + not isinstance(item.get(k), str) or not item[k] + for k in ("path", "role", "bundle_role") + ): + fail("native report source binding is incomplete") + # Producer paths are campaign-root relative, unchanged by normalization. + path = Path(item["path"]) + if ( + path.is_absolute() + or ".." in path.parts + or str(path) != item["path"] + ): + fail("native report source path is not canonical campaign-relative") + bindings.append((item["path"], item["role"], item["bundle_role"])) + if len({item[0] for item in bindings}) != len(bindings): + fail("native report source binding contains duplicate paths") + return sorted(bindings) + + if report_bindings(row.get("native_reports")) != report_bindings( + observation["reports"] + ): + fail("normalized native reports do not belong to the counted source row") + if trial["class"] == "governed_repair": + if prior is None: + fail("governed repair requires an explicit prior bundle subject") + _, transition_raw = inventoried( + observation.get("repair_transition_verification", {}) + ) + transition = json.loads(transition_raw) + _, approval_raw = inventoried(transition["candidate_approval"]) + _, pointer_raw = inventoried(transition["active_pointer"]) + approval, pointer = json.loads(approval_raw), json.loads(pointer_raw) + prior_digest = prior["bundle_content_digest"] + admitted_digest = expected["bundle_content_digest"] + if ( + transition.get("prior_bundle_content_digest") != prior_digest + or transition.get("proposed_bundle_content_digest") != admitted_digest + or transition.get("active_bundle_content_digest") != admitted_digest + or approval.get("prior_content_digest") != prior_digest + or approval.get("proposed_content_digest") != admitted_digest + or not isinstance(approval.get("approved_by"), str) + or not approval["approved_by"] + or pointer.get("mode") != "active" + or pointer.get("active_digest") != admitted_digest + ): + fail( + "actual repair approval and ACTIVE pointer do not bind prior and admitted bundles" + ) + for ref in observation["reports"]: + _, report = checked_json(manifest_path, inventory[ref["path"]]) + role = ref.get("bundle_role") + if role == "admitted": + bound = expected + elif ( + role == "repair-prior" + and trial["class"] == "governed_repair" + and ref.get("role") != "primary" + and prior is not None + ): + bound = prior + else: + fail("native report bundle role is missing or is outside its class") + if ref.get("role") == "replay" and trial["class"] == "idempotency_replay": + base_fields = { + "bundle_content_digest", + "workflow_contract_sha256", + "parameter_schema_sha256", + } + if ( + role != "admitted" + or report.get("idempotent_replay") is not True + or report.get("success") is not False + or any(report.get(k) != expected[k] for k in base_fields) + or any( + report.get(k) is not None for k in SUBJECT_FIELDS - base_fields + ) + ): + fail("early replay refusal does not bind the admitted bundle") + _, proof_raw = inventoried( + observation.get("replay_refusal_verification", {}) + ) + proof = json.loads(proof_raw) + primary = [ + r for r in observation["reports"] if r.get("role") == "primary" + ] + if ( + proof.get("schema_version") + != "openadapt.idempotent-replay-refusal-verification/v1" + or proof.get("report_sha256") != ref["sha256"] + or len(primary) != 1 + or proof.get("primary_report_sha256") != primary[0]["sha256"] + or proof.get("same_key") is not True + or proof.get("no_new_input") is not True + ): + fail("early replay refusal has no bound same-key/no-input proof") + _, primary_raw = inventoried(primary[0]) + primary_report = json.loads(primary_raw) + key = report.get("idempotency_key") + if ( + not isinstance(key, str) + or not key + or primary_report.get("idempotency_key") != key + ): + fail( + "primary and replay native idempotency keys must be nonempty and equal" + ) + ledger_records = [ + replay_ledger_record( + inventoried(proof[field])[1], key=key, primary=primary_report + ) + for field in ("ledger_before", "ledger_after") + ] + if ledger_records[0] != ledger_records[1]: + fail("replay changed its initial reservation or ledger authority") + _, before = inventoried(proof["input_events_before"]) + _, after = inventoried(proof["input_events_after"]) + if before != after: + fail("replay refusal produced new input events") + elif any(report.get(key) != value for key, value in bound.items()): + fail( + "selected native report does not bind its exact declared bundle tuple" + ) + if ref.get("role") == "primary": + if role != "admitted" or ref["sha256"] in seen_primary: + fail("primary report must be distinct and bind the admitted bundle") + seen_primary.add(ref["sha256"]) + + +def gh(path: str): + return json.loads(subprocess.check_output(["gh", "api", path], text=True)) + + +def verify_publication( + candidate: dict, staging: dict, files: dict, *, api=gh, fetch=verifier.fetch +) -> None: + """Compare actual public bytes and current release controls with the retained observation.""" + observed = candidate["release_observation"] + if observed.get("published") is not True: + fail("an unpublished candidate cannot be admitted") + trust.validate_staging(staging) + repo = "OpenAdaptAI/openadapt-flow" + version = observed["version"] + tag = f"v{version}" + release = api(f"repos/{repo}/releases/tags/{tag}") + if ( + release["draft"] + or release["prerelease"] + or release["tag_name"] != tag + or str(release["id"]) != staging["draft_release_id"] + or release["author"]["login"] != staging["release_author_login"] + or staging["target_commitish"] != observed["source_commit"] + or staging["repository"] != repo + ): + fail("retained publication differs from the stable GitHub release") + tag_object = api(f"repos/{repo}/git/ref/tags/{tag}")["object"] + for _ in range(4): + if tag_object["type"] != "tag": + break + tag_object = api(f"repos/{repo}/git/tags/{tag_object['sha']}")["object"] + if tag_object["type"] != "commit" or tag_object["sha"] != observed["source_commit"]: + fail("published tag source differs") + immutable = api(f"repos/{repo}/immutable-releases") + if {key: immutable[key] for key in staging["immutable_releases"]} != staging[ + "immutable_releases" + ]: + fail("immutable release controls changed") + for expected in staging["tag_rulesets"]: + live = api(f"repos/{repo}/rulesets/{expected['ruleset_id']}") + projected = { + "ruleset_id": str(live["id"]), + **{ + k: live[k] + for k in ("name", "target", "enforcement", "conditions", "rules") + }, + "bypass_actors": [ + {**actor, "actor_id": str(actor["actor_id"])} + for actor in live["bypass_actors"] + ], + } + if any(expected[key] != value for key, value in projected.items()): + fail("retained tag ruleset differs from current controls") + pypi = json.loads(fetch(f"https://pypi.org/pypi/openadapt-flow/{version}/json")) + indexed = {item["filename"]: item for item in pypi["urls"]} + if len(indexed) != 2 or len(pypi["urls"]) != 2 or set(indexed) != set(files): + fail("PyPI must expose the exact wheel and sdist") + for item in observed["artifacts"]: + name = item["name"] + published = indexed[name] + assets = [a for a in release["assets"] if a["name"] == name] + staged = [a for a in staging["assets"] if a["name"] == name] + if len(assets) != 1 or len(staged) != 1: + fail("publication artifact is absent or duplicated") + asset = assets[0] + if ( + published["yanked"] + or "sha256:" + published["digests"]["sha256"] != item["sha256"] + or published["size"] != item["size_bytes"] + or asset["state"] != "uploaded" + or asset.get("digest") != item["sha256"] + or asset["size"] != item["size_bytes"] + or str(asset["id"]) != staged[0]["asset_id"] + or asset["uploader"]["login"] != staged[0]["uploader_login"] + or str(asset["uploader"]["id"]) != staged[0]["uploader_id"] + ): + fail("published artifact identity differs from the candidate") + for url in (published["url"], asset["browser_download_url"]): + parsed = urlsplit(url) + if parsed.scheme != "https" or parsed.hostname not in { + "files.pythonhosted.org", + "github.com", + }: + fail("publication download URL is outside the expected hosts") + if fetch(url) != files[name]: + fail("actual published artifact bytes differ from retained bytes") + + +def prepare_inputs( + mapping_path: Path, mapping_sha256: str, *, publication_check=verify_publication +) -> dict: + raw = mapping_path.read_bytes() + if sha(raw) != mapping_sha256: + fail("private mapping bytes differ from reviewed hash") + mapping = json.loads(raw) + if mapping.get("schema_version") != "openadapt.measured-admission-mapping/v1": + fail("unsupported measured admission mapping") + expected = subject(mapping["admitted_subject"]) + prior = mapping.get("repair_prior_subject") + if prior is not None: + subject(prior) + if prior == expected: + fail("repair prior must differ from admitted subject") + candidate_path, candidate = checked_json(mapping_path, mapping["candidate"]) + if ( + candidate.get("schema_version") + != "openadapt.unsigned-measured-release-candidate/v1" + or candidate.get("evidence_class") != "remote-safe-synthetic" + or candidate.get("target") != "flow" + or candidate.get("admission_issued") is not False + or candidate.get("state") != "ready-for-review" + or candidate.get("validation_errors") + ): + fail("wrong unsigned measured Flow candidate") + trust.closed( + mapping["commitments"], FILE_COMMITMENTS, "retained receipt commitments" + ) + paths, commitments = {}, {} + for field, reference in mapping["commitments"].items(): + path, data = checked_file(mapping_path, reference) + paths[field] = path + commitments[field] = sha(data) + manifest_path = paths["evidence_manifest_sha256"] + manifest = json.loads(manifest_path.read_bytes()) + if manifest.get("candidate_ready") is not True: + fail("measured manifest has unresolved admission requirements") + observation = candidate["release_observation"] + artifacts = observation["artifacts"] + if len(artifacts) != 2 or {a["kind"] for a in artifacts} != { + "python-wheel", + "python-sdist", + }: + fail("candidate needs exactly one wheel and one sdist") + files = {} + for item in artifacts: + _, data = checked_file(mapping_path, mapping["publication_files"][item["name"]]) + if sha(data) != item["sha256"] or len(data) != item["size_bytes"]: + fail("retained publication file differs from candidate") + files[item["name"]] = data + wheel = next(a for a in artifacts if a["kind"] == "python-wheel") + if commitments["admitted_runtime_sha256"] != wheel["sha256"]: + fail("admitted runtime must commit the actual published wheel bytes") + summary, measured_evidence = measured.summarize( + manifest_path, wheel["sha256"], version=observation["version"] + ) + if ( + candidate["campaign_summary"] != summary + or candidate["measured_evidence"] != measured_evidence + or candidate["measured_manifest_sha256"] + != commitments["evidence_manifest_sha256"] + or measured_evidence["task_count"] != 1 + ): + fail( + "candidate measured counts or manifest differ from the new one-task campaign" + ) + verify_subjects(manifest_path, manifest, expected, prior) + verify_bundle( + mapping_path, + mapping["commitments"]["bundle_sha256"], + mapping["bundle_verification"], + expected, + ) + bridge_path, bridge = checked_json(mapping_path, mapping["contract_verification"]) + if ( + bridge.get("schema_version") != "openadapt.measured-contract-verification/v1" + or bridge.get("verified") is not True + or bridge.get("admitted_subject") != expected + or bridge.get("bundle_verification_sha256") + != mapping["bundle_verification"]["sha256"] + or bridge.get("receipt_commitments") != commitments + ): + fail("reviewed contract mapping does not bind every actual receipt commitment") + checked_file(bridge_path, bridge.get("verifier")) + # The retained readback and campaign are separately hashed documents, and + # must explicitly bind the same final manifest and native subject. + for field in ( + "evidence_manifest_readback_sha256", + "campaign_artifact_sha256", + "decision_commitment_sha256", + ): + document = json.loads(paths[field].read_bytes()) + if ( + document.get("evidence_manifest_sha256") + != commitments["evidence_manifest_sha256"] + or document.get("admitted_subject") != expected + ): + fail( + "retained campaign, decision or readback binds another manifest or subject" + ) + _, staging = checked_json(mapping_path, mapping["publication_staging"]) + publication_check(candidate, staging, files) + inventory = { + "schema_version": "openadapt.production-release-artifact-inventory/v1", + "target": "flow", + "claim_scope": "production_flow", + "artifacts": sorted( + [ + { + **{k: a[k] for k in ("name", "kind", "sha256", "size_bytes")}, + "media_type": "application/zip" + if a["kind"] == "python-wheel" + else "application/gzip", + "publish_destinations": ["github-release", "pypi"], + } + for a in artifacts + ], + key=lambda a: (a["kind"], a["name"], a["sha256"]), + ), + } + trust.validate_artifact_inventory(inventory) + release = { + "schema_version": "openadapt.production-release-candidate/v1", + "kind": "package", + "source_repository": "OpenAdaptAI/openadapt-flow", + "source_repository_id": "1291376938", + "source_commit": observation["source_commit"], + "version": observation["version"], + "tag": f"v{observation['version']}", + "deployment_id": None, + "deployment_sha256": None, + "artifacts": inventory["artifacts"], + } + return { + "mapping_sha256": mapping_sha256, + "candidate_sha256": sha(candidate_path.read_bytes()), + "commitments": commitments, + "campaign_summary": summary, + "bundle_version": mapping["bundle_version"], + "decision_revision": mapping["decision_revision"], + "release_identity": candidate["proposed_release_identity"], + "release": release, + "artifact_inventory": inventory, + "publication_staging": staging, + } + + +class PreviewConsumer: + """Evaluate the existing issuer without persisting or claiming issuance.""" + + def commit_once(self, **_kwargs): + return {"unsigned_preview": True} + + +def current_context(source_commit: str, now: datetime) -> dict: + if verifier.protected_main_commit() != source_commit: + fail("issuer source must be actual current protected main") + policy_raw = verifier.fetch( + verifier.raw_url(source_commit, "production-evidence-policy.json") + ) + lifecycle_raw = verifier.fetch( + verifier.raw_url(source_commit, "production-lifecycle-policy.json") + ) + if ( + policy_raw != (ROOT / "production-evidence-policy.json").read_bytes() + or lifecycle_raw != (ROOT / "production-lifecycle-policy.json").read_bytes() + ): + fail("local reviewed policies differ from current protected main") + resolver = issuer.GitHubEvidenceResolver(json.loads(policy_raw)) + authority_evidence, revocation_evidence = resolver.current_trust_state( + registry_source_commit=source_commit + ) + registry, authority, revocation, expiry = issuer._active_trust_state( + current_signer_registry=authority_evidence["current_signer_registry"], + authority_evidence=authority_evidence, + revocation_evidence=revocation_evidence, + now=now, + ) + return { + "resolver": resolver, + "registry": registry, + "authority": authority, + "revocation": revocation, + "expiry": expiry, + "authority_reference": authority_evidence["reference"], + "revocation_reference": revocation_evidence["reference"], + "acceptance_policy_sha256": sha(policy_raw), + "lifecycle_policy_sha256": sha(lifecycle_raw), + } + + +def check_release_identity(inputs: dict, source: str) -> None: + ledger = json.loads( + verifier.fetch(verifier.raw_url(source, "production-lifecycle-admissions.json")) + ) + previous = [] + for ref in ledger["admissions"]: + value = verifier.verify_bytes( + verifier.fetch(verifier.raw_url(source, ref["object_path"])), + ref, + "ledger admission", + ) + if value.get("target") == "flow": + previous.append(value) + if not previous: + fail("current Flow admission history is missing") + last = max(previous, key=lambda value: value["release_identity"]["sequence"]) + expected = { + "schema_version": "openadapt.monotonic-production-release/v1", + "channel": "production", + "sequence": last["release_identity"]["sequence"] + 1, + "previous_admission_sha256": last["admission_id_sha256"], + } + if inputs["release_identity"] != expected: + fail("proposed release identity differs from the current committed ledger") + + +def resolve_dependency(context: dict, reference: dict, phase: str) -> dict: + # The canonical resolver fetches the pair and containing registry at the + # exact actual storage commit. A storage commit need not assert issuer main. + return context["resolver"].resolve(reference, kind=KINDS[phase]) + + +def receipt_binding(value: dict, inputs: dict, context: dict) -> None: + for key, digest in inputs["commitments"].items(): + if value.get(key) != digest: + fail("committed decision receipt differs from reviewed measured inputs") + if ( + value["campaign_summary"]["classes"] != inputs["campaign_summary"] + or value["campaign_summary"]["task_count"] != 1 + or value["bundle_version"] != inputs["bundle_version"] + or value["evidence_authority_contract_sha256"] + != context["authority"]["evidence_authority_sha256"] + or value["signer_registry_sha256"] + != evidence.signer_registry_identity_digest(context["registry"]) + or value["revocation_state_sha256"] + != context["revocation"]["revocation_state_sha256"] + ): + fail("committed decision receipt tuple or current authority differs") + + +def issuer_identity(source: str, phase: str) -> dict: + workflow, environment = { + "receipt": ( + "issue-synthetic-qualification-evidence-decision", + "synthetic-qualification-evidence-decision", + ), + "manifest": ("issue-production-acceptance", "production-acceptance"), + }[phase] + is_acceptance = phase == "manifest" + return { + "repository": "OpenAdaptAI/openadapt-evals" + if is_acceptance + else "OpenAdaptAI/.github", + "repository_id": "1135998197" if is_acceptance else "858454062", + "repository_owner_id": "132681217", + "workflow": f".github/workflows/{workflow}.yml", + "ref": "refs/heads/main", + "source_commit": source, + "environment": environment, + } + + +def phase_object( + inputs: dict, request: dict, context: dict, *, consumer=None +) -> tuple[dict, dict | None]: + phase = request["phase"] + source = request["issuer_source_commit"] + now = trust.require_timestamp(request["issued_at"], "phase issued_at") + expiry = trust.optional_timestamp(request["expires_at"], "phase expires_at") + if not trust.expires_contained(expiry, context["expiry"]): + fail("requested phase validity exceeds current trust") + window = {k: request[k] for k in ("issued_at", "expires_at")} + window["not_before"] = request["issued_at"] + trust.validate_window(window, now=now) + registry_identity = evidence.signer_registry_identity_digest(context["registry"]) + state = { + "signer_registry_sha256": registry_identity, + "authority_state_sha256": context["authority"]["authority_state_sha256"], + "revocation_state_sha256": context["revocation"]["revocation_state_sha256"], + } + resolver = context["resolver"] + references = request.get("references", {}) + if phase == "receipt": + inner = [ + s + for s in context["registry"]["signers"] + if s["status"] == "active" + and s["algorithm"] == "ed25519" + and "qualification-evidence-decision-receipt" in s["allowed_usages"] + and "https://github.com/OpenAdaptAI/.github/.github/workflows/issue-synthetic-qualification-evidence-decision.yml@refs/heads/main" + in s["allowed_workflows"] + ] + if len(inner) != 1: + fail("current authority must select exactly one existing decision signer") + value = { + "schema_version": "openadapt.qualification-evidence-decision-receipt/v2", + "evidence_class": "remote-safe-synthetic", + **inputs["commitments"], + "decision_revision": inputs["decision_revision"], + "bundle_version": inputs["bundle_version"], + "evidence_authority_contract_sha256": context["authority"][ + "evidence_authority_sha256" + ], + "signer_registry_sha256": registry_identity, + "revocation_state_sha256": state["revocation_state_sha256"], + "entity_class": "record", + "campaign_summary": { + "schema_version": "openadapt.qualification-evidence-decision-campaign-summary/v1", + "minimum_trials_per_task_condition": 3, + "task_count": 1, + "classes": inputs["campaign_summary"], + }, + "verdict": "ADMIT", + **window, + "issuer_key_id": inner[0]["key_id"], + "algorithm": "ed25519", + "signature": "", + "signing_statement": {}, + "issuer": issuer_identity(source, "receipt"), + } + value["signing_statement"] = trust.signing_statement( + value, + object_schema_version=value["schema_version"], + signature_domain=trust.DECISION_RECEIPT_SIGNATURE_DOMAIN, + ) + trust.require_positive_int(value["decision_revision"], "decision revision") + if ( + not isinstance(value["bundle_version"], str) + or trust.BUNDLE_VERSION.fullmatch(value["bundle_version"]) is None + ): + fail("bundle version must be canonical") + if ( + len( + { + value[k] + for k in ( + "decision_commitment_sha256", + "evidence_manifest_sha256", + "evidence_manifest_readback_sha256", + "campaign_artifact_sha256", + ) + } + ) + != 4 + ): + fail( + "decision, manifest, readback and campaign byte commitments must differ" + ) + # Signature structure is checked only after real signing. The preview + # deliberately contains an empty signature, never a fabricated one. + return value, None + if phase in {"workflow", "release"}: + dependency = "receipt" if phase == "workflow" else "summary" + resolved = resolve_dependency(context, references[dependency], dependency) + if resolved["reference"]["registry_source_commit"] != source: + fail( + "immediate issuer dependency must already be committed to protected main" + ) + if phase == "workflow": + receipt_binding(resolved["value"], inputs, context) + else: + summary = resolved["value"] + manifest = resolve_dependency( + context, summary["production_acceptance_manifest_reference"], "manifest" + )["value"] + receipt = resolve_dependency( + context, + summary["qualification_evidence_decision_receipt_reference"], + "receipt", + )["value"] + receipt_binding(receipt, inputs, context) + if ( + manifest["release"] != inputs["release"] + or summary["release_identity"] != inputs["release_identity"] + ): + fail("committed acceptance names another release") + issue_request = { + "schema_version": "openadapt.qualification-admission-issue-request/v1" + if phase == "workflow" + else "openadapt.qualification-release-issue-request/v1", + "request_handle": request["request_handle"], + "evidence_class": "remote-safe-synthetic", + "decision_receipt_reference" + if phase == "workflow" + else "production_acceptance_summary_reference": resolved["reference"], + } + issue = ( + issuer.issue_workflow_admission + if phase == "workflow" + else issuer.issue_release_admission + ) + value = issue( + issue_request, + resolver=resolver, + issuer_source_commit=source, + now=now, + consumer=consumer or PreviewConsumer(), + ) + # Issuer derives expiry from its signed dependencies; never override it. + if value["expires_at"] != request["expires_at"]: + fail("reviewed expiry differs from the existing issuer's dependency window") + return value, issue_request + if phase == "manifest": + receipt = resolve_dependency(context, references["receipt"], "receipt") + admission = resolve_dependency(context, references["workflow"], "workflow") + receipt_binding(receipt["value"], inputs, context) + if admission["value"]["decision_receipt_reference"] != receipt["reference"]: + fail( + "workflow admission must preserve its exact original receipt storage reference" + ) + evals_source = request["acceptance_issuer_source_commit"] + if ( + gh("repos/OpenAdaptAI/openadapt-evals/git/ref/heads/main")["object"]["sha"] + != evals_source + ): + fail("acceptance issuer must be actual reviewed evals protected main") + value = { + "schema_version": "openadapt.production-acceptance/v3", + "target": "flow", + "verdict": "accepted", + "claim_scope": "production_flow", + **{ + k: context[k] + for k in ("acceptance_policy_sha256", "lifecycle_policy_sha256") + }, + "release_identity": inputs["release_identity"], + "release": inputs["release"], + "release_sha256": trust.digest_bytes( + trust.RELEASE_DOMAIN, + { + "target": "flow", + "claim_scope": "production_flow", + "release": inputs["release"], + }, + ), + "artifact_inventory": inputs["artifact_inventory"], + "artifact_inventory_sha256": trust.artifact_inventory_digest( + inputs["artifact_inventory"] + ), + "publication_staging": inputs["publication_staging"], + "publication_staging_sha256": trust.staging_digest( + inputs["publication_staging"] + ), + "qualification_evidence_decision_receipt_reference": receipt["reference"], + "qualification_evidence_decision_receipt_bundle_reference": receipt[ + "bundle_reference" + ], + "qualification_admission_reference": admission["reference"], + "qualification_admission_bundle_reference": admission["bundle_reference"], + "campaign_summary": inputs["campaign_summary"], + **state, + **window, + "issuer": issuer_identity(evals_source, "manifest"), + } + trust.validate_acceptance_manifest( + value, + receipt=receipt["value"], + qualification_admission=admission["value"], + receipt_signer_registry=receipt["bound_signer_registry"], + now=now, + ) + return value, None + if phase == "summary": + manifest = resolve_dependency(context, references["manifest"], "manifest") + value = copy.deepcopy(manifest["value"]) + receipt = resolve_dependency( + context, + value["qualification_evidence_decision_receipt_reference"], + "receipt", + ) + admission = resolve_dependency( + context, value["qualification_admission_reference"], "workflow" + ) + receipt_binding(receipt["value"], inputs, context) + if ( + value["release"] != inputs["release"] + or value["release_identity"] != inputs["release_identity"] + or any(value[k] != v for k, v in state.items()) + ): + fail("committed acceptance manifest release or current trust differs") + for field in ("release", "artifact_inventory"): + value.pop(field) + value.update( + schema_version="openadapt.production-lifecycle-evidence-summary/v3", + production_acceptance_manifest_reference=manifest["reference"], + production_acceptance_manifest_bundle_reference=manifest[ + "bundle_reference" + ], + **window, + ) + value["evidence_identity_sha256"] = trust.acceptance_summary_identity(value) + trust.validate_acceptance_summary( + value, + manifest=manifest["value"], + receipt=receipt["value"], + qualification_admission=admission["value"], + receipt_signer_registry=receipt["bound_signer_registry"], + now=now, + ) + return value, None + fail("unsupported phase") + + +def write_exclusive(path: Path, raw: bytes) -> bool: + """Durably create once; an existing file is never truncated or replaced.""" + path.parent.mkdir(parents=True, exist_ok=True) + try: + descriptor = os.open(path, os.O_WRONLY | os.O_CREAT | os.O_EXCL, 0o600) + except FileExistsError: + if path.is_symlink() or path.read_bytes() != raw: + fail("existing durable file differs; reconcile the original operation") + return False + with os.fdopen(descriptor, "wb") as stream: + stream.write(raw) + stream.flush() + os.fsync(stream.fileno()) + directory = os.open(path.parent, os.O_RDONLY) + try: + os.fsync(directory) + finally: + os.close(directory) + return True + + +def persist_once(plan: dict, state_dir: Path, perform) -> dict: + """Unknown results never call the issuer again, even with another handle.""" + request = plan["phase_request"] + issue_request = plan["issue_request"] + if issue_request: + dependency = next( + v for k, v in issue_request.items() if k.endswith("_reference") + ) + semantic = { + "phase": request["phase"], + "dependency_identity_sha256": dependency["semantic_identity_sha256"], + } + else: + semantic = { + "phase": request["phase"], + "decision_identity_sha256": plan["inputs"]["commitments"][ + "decision_identity_sha256" + ], + } + identity = sha(canonical(semantic)).removeprefix("sha256:") + journal = state_dir / f"{identity}.request.json" + first = write_exclusive(journal, canonical(plan)) + consumer = issuer.SqliteOneUseConsumer(state_dir / "one-use.sqlite3") + handle = request["request_handle"] + if first: + value = perform(consumer) + if not issue_request: + consumer.commit_once( + request_handle=handle, + operation=f"measured-{request['phase']}", + request_sha256=sha(canonical(plan)), + effect_sha256="sha256:" + identity, + result=canonical(value), + ) + # Reconcile after every call, including success. A missing durable row is an + # unknown outcome. Do not delete the journal or generate another request. + record = consumer.reconcile(request_handle=handle) + expected_request = ( + issuer._request_digest(issue_request) if issue_request else sha(canonical(plan)) + ) + if record["request_sha256"] != expected_request: + fail("persistent consumption record belongs to another request") + value = json.loads(base64.b64decode(record["result_base64"], validate=True)) + if value != plan["unsigned_object"]: + fail("persisted result differs from the reviewed unsigned plan") + return value + + +def sign_pair( + value: dict, plan: dict, context: dict, *, private_key +) -> tuple[bytes, bytes]: + request = plan["phase_request"] + phase = request["phase"] + now = trust.require_timestamp(request["issued_at"], "phase issued_at") + if phase == "receipt": + value = software.sign_receipt( + value, private_key=private_key, signer_registry=context["registry"] + ) + trust.validate_receipt(value, signer_registry=context["registry"], now=now) + key_id = public_trust.software_public_key_id(private_key.public_key()) + signers = [s for s in context["registry"]["signers"] if s["key_id"] == key_id] + if len(signers) != 1: + fail("existing Keychain key is absent from current public trust registry") + kind = KINDS[phase] + raw = canonical(value) + entry = stage.entry_for(raw, kind=kind, subject=None) + statement = { + "schema_version": public_trust.STATEMENT_SCHEMA, + "object_kind": kind, + "object_schema_version": entry["object_schema_version"], + "object_media_type": entry["object_media_type"], + "object_sha256": sha(raw), + "object_size_bytes": len(raw), + "semantic_identity_sha256": entry["semantic_identity_sha256"], + "source_issuer": { + field: value["issuer"][field] for field in public_trust.ISSUER_FIELDS + }, + "signer_registry_sha256": evidence.signer_registry_identity_digest( + context["registry"] + ), + "authority_state_sha256": context["authority"]["authority_state_sha256"], + "revocation_state_sha256": context["revocation"]["revocation_state_sha256"], + "issued_at": request["issued_at"], + "not_before": request["issued_at"], + "expires_at": request["expires_at"], + "request_id_sha256": sha(canonical(plan)), + "signing_authority": { + "repository": "OpenAdaptAI/.github", + "repository_id": "858454062", + "repository_owner_id": "132681217", + "workflow": public_trust.SOFTWARE_WORKFLOW_PATH, + "ref": "refs/heads/main", + "source_commit": request["issuer_source_commit"], + "environment": public_trust.SOFTWARE_ENVIRONMENT, + "key_origin": "software", + }, + "key_id": key_id, + "signature_profile": public_trust.SOFTWARE_SIGNATURE_PROFILE, + } + public_trust.validate_statement_object_binding( + statement, + object_raw=raw, + object_value=value, + object_kind=kind, + object_schema_version=entry["object_schema_version"], + object_media_type=entry["object_media_type"], + semantic_identity_sha256=entry["semantic_identity_sha256"], + expected_signer_registry_sha256=statement["signer_registry_sha256"], + expected_authority_state_sha256=statement["authority_state_sha256"], + expected_revocation_state_sha256=statement["revocation_state_sha256"], + ) + bundle = public_trust.sign_software_bundle( + statement, private_key=private_key, signer=signers[0], now=now + ) + return raw, canonical(bundle) + + +def reconcile_journal(journal: Path, state_dir: Path) -> dict: + """Read a persisted outcome without keys, network access, or issuer calls.""" + if journal.is_symlink() or not journal.is_file(): + fail("reconciliation requires the retained original request journal") + plan = json.loads(journal.read_bytes()) + record = issuer.SqliteOneUseConsumer(state_dir / "one-use.sqlite3").reconcile( + request_handle=plan["phase_request"]["request_handle"] + ) + request = plan["issue_request"] + expected = issuer._request_digest(request) if request else sha(canonical(plan)) + if ( + record["request_sha256"] != expected + or json.loads(base64.b64decode(record["result_base64"], validate=True)) + != plan["unsigned_object"] + ): + fail("reconciled result differs from the original reviewed request") + return { + "schema_version": "openadapt.measured-admission-reconciliation/v1", + "unsigned_result": True, + "consumption_record": record, + } + + +def validate_staging_registry(path: Path, source: str, context: dict) -> None: + current = json.loads( + verifier.fetch(verifier.raw_url(source, "evidence-registry.json")) + ) + proposed = json.loads(path.read_bytes()) + evidence.validate_registry(current) + evidence.validate_registry(proposed, root=path.resolve().parent) + prefix = current["entries"] + if ( + proposed["entries"][: len(prefix)] != prefix + or proposed["signer_registry"] != current["signer_registry"] + or proposed["signer_registry_history"] != current["signer_registry_history"] + or proposed["signer_registry"]["registry_identity_sha256"] + != evidence.signer_registry_identity_digest(context["registry"]) + ): + fail("staging registry does not append to the verified current trust state") + + +def main(argv=None) -> int: + parser = argparse.ArgumentParser(description=__doc__) + parser.add_argument("--mapping", type=Path) + parser.add_argument("--mapping-sha256") + parser.add_argument("--phase-request", type=Path) + parser.add_argument( + "--reconcile-journal", + type=Path, + help="Read the original durable result only; no signing or retry", + ) + parser.add_argument("--output", type=Path, required=True) + parser.add_argument( + "--sign", + action="store_true", + help="Explicitly use the existing Keychain signer after exact-plan review", + ) + parser.add_argument("--reviewed-plan-sha256") + parser.add_argument("--state-dir", type=Path) + parser.add_argument( + "--stage-registry", + type=Path, + help="Explicit append-only staging after signing; never commits or pushes", + ) + args = parser.parse_args(argv) + try: + if args.reconcile_journal: + if args.sign or args.stage_registry or not args.state_dir: + fail("reconciliation requires permanent state and cannot sign or stage") + result = reconcile_journal(args.reconcile_journal, args.state_dir) + write_exclusive(args.output, canonical(result)) + print(json.dumps({"reconciled": True, "unsigned_result": True})) + return 0 + if not args.mapping or not args.mapping_sha256 or not args.phase_request: + fail("preparation requires mapping, mapping hash, and phase request") + if not args.sign and ( + args.stage_registry or args.state_dir or args.reviewed_plan_sha256 + ): + fail("unsigned prepare cannot consume state, stage, or sign") + if args.sign and (not args.state_dir or not args.reviewed_plan_sha256): + fail( + "signing requires exact reviewed plan hash and permanent one-use state directory" + ) + request = json.loads(args.phase_request.read_bytes()) + if request.get("phase") not in KINDS: + fail("phase must be receipt, workflow, manifest, summary, or release") + issuer._request_handle(request["request_handle"]) + now = trust.require_timestamp(request["issued_at"], "phase issued_at") + live_now = datetime.now(timezone.utc) + if now > live_now or (live_now - now).total_seconds() > 3600: + fail("reviewed issue time must be within the last hour") + inputs = prepare_inputs(args.mapping, args.mapping_sha256) + context = current_context(request["issuer_source_commit"], live_now) + check_release_identity(inputs, request["issuer_source_commit"]) + value, issue_request = phase_object(inputs, request, context) + plan = { + "schema_version": "openadapt.measured-admission-unsigned-plan/v1", + "inputs": inputs, + "phase_request": request, + "issue_request": issue_request, + "unsigned_object": value, + "authority_reference": context["authority_reference"], + "revocation_reference": context["revocation_reference"], + } + raw = canonical(plan) + if not args.sign: + write_exclusive(args.output, raw) + print( + json.dumps( + { + "unsigned": True, + "plan_sha256": sha(raw), + "phase": request["phase"], + } + ) + ) + return 0 + if sha(raw) != args.reviewed_plan_sha256: + fail("unsigned plan changed after review; signing refused") + if verifier.protected_main_commit() != request["issuer_source_commit"]: + fail("protected main moved after verification; prepare and review again") + if args.stage_registry: + validate_staging_registry( + args.stage_registry, request["issuer_source_commit"], context + ) + value = persist_once( + plan, + args.state_dir, + lambda consumer: phase_object(inputs, request, context, consumer=consumer)[ + 0 + ], + ) + private_key = software.load_private_key(software.keychain_read()) + object_raw, bundle_raw = sign_pair( + value, plan, context, private_key=private_key + ) + object_path, bundle_path = ( + args.output / "object.json", + args.output / "bundle.json", + ) + write_exclusive(object_path, object_raw) + write_exclusive(bundle_path, bundle_raw) + if args.stage_registry and stage.main( + [ + "--registry", + str(args.stage_registry), + "--kind", + KINDS[request["phase"]], + "--object", + str(object_path), + "--bundle", + str(bundle_path), + ] + ): + fail( + "append-only stage failed; reconcile the existing files before continuing" + ) + print( + json.dumps( + { + "object_sha256": sha(object_raw), + "bundle_sha256": sha(bundle_raw), + "phase": request["phase"], + "storage_reference": None, + "next": "Commit the staged pair; use its actual containing commit for the next dependency reference.", + } + ) + ) + return 0 + except (OSError, ValueError, KeyError, TypeError, zipfile.BadZipFile) as exc: + print(f"REFUSED: {exc}", file=sys.stderr) + return 1 + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/tests/test_measured_flow_admission.py b/tests/test_measured_flow_admission.py new file mode 100644 index 0000000..fae6b37 --- /dev/null +++ b/tests/test_measured_flow_admission.py @@ -0,0 +1,1135 @@ +"""Admission adapter tests use temporary parser fixtures and test-only keys.""" + +import copy +import importlib.util +import json +import sqlite3 +import sys +import tempfile +import unittest +import zipfile +from pathlib import Path +from unittest import mock + +ROOT = Path(__file__).resolve().parents[1] +sys.path.insert(0, str(ROOT / "tests")) +import test_measured_flow_candidate as samples +import test_qualification_issuer as signed_samples + +SPEC = importlib.util.spec_from_file_location( + "measured_flow_admission", ROOT / "local-candidates/flow-1.35.1-measured/issue.py" +) +adapter = importlib.util.module_from_spec(SPEC) +SPEC.loader.exec_module(adapter) + + +def retain(path): + return { + "path": path.name, + "sha256": adapter.sha(path.read_bytes()), + "size_bytes": path.stat().st_size, + } + + +def write(path, value): + path.write_bytes(adapter.canonical(value)) + return retain(path) + + +def inputs_fixture(root): + wheel, sdist = samples.archives(root) + manifest_path, manifest = samples.manifest(root, wheel=wheel, version="1.35.1") + subject = { + key: adapter.sha(key.encode()).removeprefix("sha256:") + for key in adapter.SUBJECT_FIELDS + } + manifest["admitted_subject"] = subject + manifest["candidate_ready"] = True + for trial in manifest["trials"]: + observed_path = root / trial["observation"]["path"] + observation = json.loads(observed_path.read_bytes()) + for ref in observation["reports"]: + report_path = root / ref["path"] + report = json.loads(report_path.read_bytes()) + report.update(subject) + write(report_path, report) + samples.update_references(observation, report_path) + samples.update_references(manifest, report_path) + ref["bundle_role"] = "admitted" + source_ref = write( + root / f"source-{trial['class']}-{trial['trial']}.json", + [ + { + **{ + k: trial[k] + for k in ("class", "task_id", "condition", "trial", "counters") + }, + "runtime_version": "1.35.1", + "native_reports": [ + {k: ref[k] for k in ("path", "role", "bundle_role")} + for ref in observation["reports"] + ], + "passed": True, + "errors": [], + } + ], + ) + manifest["artifacts"].append(source_ref) + observation["counter_source"] = {"observation": source_ref, "row_index": 0} + if trial["class"] == "governed_repair": + prior_digest = "e" * 64 + approval = write( + root / f"repair-approval-{trial['trial']}.json", + { + "prior_content_digest": prior_digest, + "proposed_content_digest": subject["bundle_content_digest"], + "approved_by": "test-only-reviewer", + }, + ) + pointer = write( + root / f"repair-pointer-{trial['trial']}.json", + {"mode": "active", "active_digest": subject["bundle_content_digest"]}, + ) + transition = write( + root / f"repair-transition-{trial['trial']}.json", + { + "prior_bundle_content_digest": prior_digest, + "proposed_bundle_content_digest": subject["bundle_content_digest"], + "active_bundle_content_digest": subject["bundle_content_digest"], + "candidate_approval": approval, + "active_pointer": pointer, + }, + ) + manifest["artifacts"].extend((approval, pointer, transition)) + observation["repair_transition_verification"] = transition + write(observed_path, observation) + samples.update_references(manifest, observed_path) + write(manifest_path, manifest) + archive = root / "encrypted-bundle.zip" + with zipfile.ZipFile(archive, "w") as output: + output.writestr("workflow.json.enc", b"test-only-encrypted-fixture") + verifier_ref = write(root / "reviewed-verifier.json", {"test_only": True}) + proof = { + "schema_version": "openadapt.measured-bundle-verification/v1", + "verified": True, + "bundle_archive_sha256": retain(archive)["sha256"], + "admitted_subject": subject, + "bundle_members": [ + { + "path": "workflow.json.enc", + "size_bytes": 27, + "sha256": adapter.sha(b"test-only-encrypted-fixture"), + } + ], + "verifier": verifier_ref, + } + proof["bundle_members"][0]["size_bytes"] = len(b"test-only-encrypted-fixture") + proof_ref = write(root / "bundle-verification.json", proof) + commitments = {} + for field in adapter.FILE_COMMITMENTS: + if field == "bundle_sha256": + ref = retain(archive) + elif field == "admitted_runtime_sha256": + ref = retain(wheel) + elif field == "evidence_manifest_sha256": + ref = retain(manifest_path) + else: + ref = write( + root / f"{field}.json", + { + "test_only": True, + "field": field, + "evidence_manifest_sha256": retain(manifest_path)["sha256"], + "admitted_subject": subject, + }, + ) + commitments[field] = ref + bridge_ref = write( + root / "contract-verification.json", + { + "schema_version": "openadapt.measured-contract-verification/v1", + "verified": True, + "admitted_subject": subject, + "bundle_verification_sha256": proof_ref["sha256"], + "receipt_commitments": { + key: value["sha256"] for key, value in commitments.items() + }, + "verifier": verifier_ref, + }, + ) + summary, measured = adapter.measured.summarize( + manifest_path, retain(wheel)["sha256"], version="1.35.1" + ) + candidate = { + "schema_version": "openadapt.unsigned-measured-release-candidate/v1", + "target": "flow", + "evidence_class": "remote-safe-synthetic", + "admission_issued": False, + "state": "ready-for-review", + "campaign_summary": summary, + "measured_evidence": measured, + "measured_manifest_sha256": retain(manifest_path)["sha256"], + "proposed_release_identity": { + "schema_version": "openadapt.monotonic-production-release/v1", + "channel": "production", + "sequence": 2, + "previous_admission_sha256": adapter.sha(b"test-previous"), + }, + "release_observation": { + "published": True, + "version": "1.35.1", + "source_commit": "a" * 40, + "artifacts": [ + { + "kind": kind, + "name": path.name, + **{key: retain(path)[key] for key in ("sha256", "size_bytes")}, + } + for kind, path in (("python-wheel", wheel), ("python-sdist", sdist)) + ], + }, + } + mapping = { + "schema_version": "openadapt.measured-admission-mapping/v1", + "admitted_subject": subject, + "repair_prior_subject": {**subject, "bundle_content_digest": "e" * 64}, + "candidate": write(root / "candidate.json", candidate), + "commitments": commitments, + "bundle_verification": proof_ref, + "contract_verification": bridge_ref, + "publication_files": {p.name: retain(p) for p in (wheel, sdist)}, + "publication_staging": write(root / "staging.json", {"test_only": True}), + "bundle_version": "1.0.1", + "decision_revision": 1, + } + path = root / "mapping.json" + write(path, mapping) + return path, mapping, manifest_path, manifest + + +def trust_context(): + fixture = signed_samples.trust_fixture(decision_origin="software", expires_at=None) + registry = fixture["registry"] + registry["signers"].append( + adapter.public_trust.software_public_signer( + fixture["decision_key"].public_key() + ) + ) + identity = adapter.evidence.signer_registry_identity_digest(registry) + authority = fixture["authority"] + authority["signer_registry_sha256"] = adapter.sha(adapter.canonical(registry)) + authority["signer_registry_identity_sha256"] = identity + projection = { + k: v + for k, v in authority.items() + if k not in {"authority_state_sha256", "signature", "signing_statement"} + } + authority["authority_state_sha256"] = adapter.trust.digest_bytes( + adapter.trust.AUTHORITY_STATE_IDENTITY_DOMAIN, projection + ) + signed_samples.sign_embedded( + authority, + fixture["authority_key"], + schema=authority["schema_version"], + domain=adapter.trust.AUTHORITY_STATE_SIGNATURE_DOMAIN, + ) + revocation = fixture["revocation"] + revocation.update( + signer_registry_sha256=identity, + authority_state_sha256=authority["authority_state_sha256"], + ) + projection = { + k: v + for k, v in revocation.items() + if k not in {"revocation_state_sha256", "signature", "signing_statement"} + } + revocation["revocation_state_sha256"] = adapter.trust.digest_bytes( + adapter.trust.REVOCATION_STATE_IDENTITY_DOMAIN, projection + ) + signed_samples.sign_embedded( + revocation, + fixture["revocation_key"], + schema=revocation["schema_version"], + domain=adapter.trust.REVOCATION_STATE_SIGNATURE_DOMAIN, + ) + fixture["receipt"].update( + signer_registry_sha256=identity, + revocation_state_sha256=revocation["revocation_state_sha256"], + ) + fixture["receipt"] = adapter.software.sign_receipt( + fixture["receipt"], + private_key=fixture["decision_key"], + signer_registry=registry, + ) + fixture["receipt_ref"], fixture["receipt_bundle_ref"] = ( + signed_samples.reference_pair(adapter.KINDS["receipt"], fixture["receipt"]) + ) + resolver = signed_samples.RecordingResolver(fixture) + return fixture, { + "resolver": resolver, + "registry": registry, + "authority": authority, + "revocation": revocation, + "expiry": None, + "acceptance_policy_sha256": adapter.sha(b"test policy"), + "lifecycle_policy_sha256": adapter.sha(b"test lifecycle"), + "authority_reference": resolver.authority["reference"], + "revocation_reference": resolver.revocation["reference"], + } + + +def request(phase, source="1" * 40, **references): + return { + "phase": phase, + "issuer_source_commit": source, + "issued_at": signed_samples.utc(signed_samples.NOW), + "expires_at": None, + "references": references, + "request_handle": "qair_" + phase[0] * 43, + } + + +class MeasuredAdmissionTests(unittest.TestCase): + def setUp(self): + temporary = tempfile.TemporaryDirectory() + self.addCleanup(temporary.cleanup) + self.root = Path(temporary.name) + self.path, self.mapping, self.manifest_path, self.manifest = inputs_fixture( + self.root + ) + + def prepare(self): + return adapter.prepare_inputs( + self.path, + adapter.sha(self.path.read_bytes()), + publication_check=lambda *_args: None, + ) + + def test_one_task_counts_and_distinct_digest_meanings(self): + value = self.prepare() + self.assertEqual( + value["campaign_summary"]["healthy"]["observed_trial_count"], 3 + ) + self.assertEqual( + value["campaign_summary"]["governed_repair"]["approved_repair_count"], 3 + ) + self.assertNotEqual( + value["commitments"]["bundle_sha256"], + "sha256:" + self.mapping["admitted_subject"]["bundle_content_digest"], + ) + self.assertNotIn("evidence_authority_contract_sha256", value["commitments"]) + + def test_mutated_actual_contract_bytes_refuse(self): + ref = self.mapping["commitments"]["action_contract_sha256"] + (self.root / ref["path"]).write_bytes(b"changed") + with self.assertRaisesRegex(ValueError, "bytes differ"): + self.prepare() + + def test_mutated_mapping_hash_refuses(self): + with self.assertRaisesRegex(ValueError, "mapping bytes"): + adapter.prepare_inputs(self.path, adapter.sha(b"wrong")) + + def test_semantically_changed_contract_rehash_still_requires_reviewed_bridge(self): + ref = self.mapping["commitments"]["action_contract_sha256"] + path = self.root / ref["path"] + self.mapping["commitments"]["action_contract_sha256"] = write( + path, {"different": True} + ) + write(self.path, self.mapping) + with self.assertRaisesRegex(ValueError, "contract mapping"): + self.prepare() + + def test_second_task_cannot_reuse_old_campaign_counts(self): + self.manifest["trials"][0]["task_id"] = "another-task" + with self.assertRaisesRegex(ValueError, "one task"): + adapter.verify_subjects( + self.manifest_path, + self.manifest, + self.mapping["admitted_subject"], + self.mapping["repair_prior_subject"], + ) + + def test_wrong_native_tuple_refuses_even_when_rehashed(self): + trial = self.manifest["trials"][0] + observation = json.loads( + (self.root / trial["observation"]["path"]).read_bytes() + ) + path = self.root / observation["reports"][0]["path"] + report = json.loads(path.read_bytes()) + report["bundle_content_digest"] = "f" * 64 + write(path, report) + samples.update_references(self.manifest, path) + with self.assertRaisesRegex(ValueError, "exact declared bundle tuple"): + adapter.verify_subjects( + self.manifest_path, + self.manifest, + self.mapping["admitted_subject"], + self.mapping["repair_prior_subject"], + ) + + def bind_source_reports(self, observation): + ref = observation["counter_source"]["observation"] + path = self.root / ref["path"] + rows = json.loads(path.read_bytes()) + rows[observation["counter_source"]["row_index"]]["native_reports"] = [ + {k: r[k] for k in ("path", "role", "bundle_role")} + for r in observation["reports"] + ] + write(path, rows) + samples.update_references(observation, path) + samples.update_references(self.manifest, path) + + def test_auxiliary_prior_is_only_allowed_for_governed_repair(self): + trial = next( + t for t in self.manifest["trials"] if t["class"] == "governed_repair" + ) + path = self.root / trial["observation"]["path"] + observation = json.loads(path.read_bytes()) + prior = {**self.mapping["admitted_subject"], "bundle_content_digest": "e" * 64} + ref = write( + self.root / "prior.json", + { + **prior, + "success": False, + "transaction_outcome": "RECONCILIATION_REQUIRED", + "model_calls": 0, + }, + ) + self.manifest["artifacts"].append(ref) + observation["reports"].append( + {**ref, "role": "discovery", "bundle_role": "repair-prior"} + ) + self.bind_source_reports(observation) + write(path, observation) + samples.update_references(self.manifest, path) + adapter.verify_subjects( + self.manifest_path, self.manifest, self.mapping["admitted_subject"], prior + ) + observation["reports"][0]["bundle_role"] = "repair-prior" + self.bind_source_reports(observation) + write(path, observation) + samples.update_references(self.manifest, path) + with self.assertRaisesRegex(ValueError, "bundle role"): + adapter.verify_subjects( + self.manifest_path, + self.manifest, + self.mapping["admitted_subject"], + prior, + ) + + def test_symlink_and_traversal_refuse(self): + file = self.root / "real" + file.write_bytes(b"actual") + link = self.root / "link" + link.symlink_to(file) + for ref in ( + {**retain(file), "path": "link"}, + {**retain(file), "path": "../real"}, + ): + with self.assertRaises(ValueError): + adapter.checked_file(self.path, ref) + + def test_plain_prepare_never_reads_key_or_changes_registry(self): + with ( + mock.patch.object( + adapter.software, + "keychain_read", + side_effect=AssertionError("secret access"), + ), + mock.patch.object( + adapter.stage, "main", side_effect=AssertionError("registry mutation") + ), + ): + self.prepare() + + def test_default_cli_rejects_signing_options(self): + result = adapter.main( + [ + "--mapping", + str(self.path), + "--mapping-sha256", + adapter.sha(self.path.read_bytes()), + "--phase-request", + str(self.path), + "--output", + str(self.root / "plan"), + "--state-dir", + str(self.root / "state"), + ] + ) + self.assertEqual(result, 1) + self.assertFalse((self.root / "state").exists()) + + def test_branch_cannot_be_claimed_as_current_main(self): + with ( + mock.patch.object( + adapter.verifier, "protected_main_commit", return_value="1" * 40 + ), + mock.patch.object( + adapter.verifier, + "fetch", + side_effect=AssertionError("must refuse before read"), + ), + self.assertRaisesRegex(ValueError, "actual current protected main"), + ): + adapter.current_context("2" * 40, signed_samples.NOW) + + def test_receipt_signature_is_empty_until_real_test_key_signs(self): + fixture, context = trust_context() + value, issue = adapter.phase_object(self.prepare(), request("receipt"), context) + self.assertEqual(value["signature"], "") + self.assertIsNone(issue) + plan = {"phase_request": request("receipt"), "unsigned_object": value} + raw, bundle = adapter.sign_pair( + value, plan, context, private_key=fixture["decision_key"] + ) + signed = json.loads(raw) + adapter.trust.validate_receipt( + signed, signer_registry=fixture["registry"], now=signed_samples.NOW + ) + self.assertEqual(signed["campaign_summary"]["task_count"], 1) + self.assertEqual( + signed["evidence_authority_contract_sha256"], + context["authority"]["evidence_authority_sha256"], + ) + self.assertTrue(json.loads(bundle)["dsseEnvelope"]["signatures"]) + + def test_persistent_result_recovers_without_reissuing(self): + plan = { + "phase_request": request("receipt"), + "issue_request": None, + "inputs": self.prepare(), + "unsigned_object": {"test_only": "result"}, + } + calls = [] + + def perform(_consumer): + calls.append(1) + return plan["unsigned_object"] + + state = self.root / "state" + self.assertEqual( + adapter.persist_once(plan, state, perform), plan["unsigned_object"] + ) + self.assertEqual( + adapter.persist_once(plan, state, lambda _: self.fail("retry")), + plan["unsigned_object"], + ) + self.assertEqual(calls, [1]) + changed = copy.deepcopy(plan) + changed["phase_request"]["request_handle"] = "qair_" + "Z" * 43 + with self.assertRaisesRegex(ValueError, "existing durable file differs"): + adapter.persist_once( + changed, state, lambda _: self.fail("retry with new handle") + ) + + def test_unknown_result_never_retries_or_deletes_state(self): + plan = { + "phase_request": request("receipt"), + "issue_request": None, + "inputs": self.prepare(), + "unsigned_object": {"test_only": "result"}, + } + state = self.root / "state" + with self.assertRaisesRegex(RuntimeError, "interruption"): + adapter.persist_once( + plan, + state, + lambda _: (_ for _ in ()).throw(RuntimeError("interruption")), + ) + before = list(state.glob("*.request.json")) + with self.assertRaises(adapter.issuer.IssuerError): + adapter.persist_once( + plan, state, lambda _: self.fail("unknown must not retry") + ) + self.assertEqual(before, list(state.glob("*.request.json"))) + + def test_normalizer_cannot_invent_counter_missing_from_actual_source(self): + trial = self.manifest["trials"][0] + observation_path = self.root / trial["observation"]["path"] + observation = json.loads(observation_path.read_bytes()) + source_path = self.root / observation["counter_source"]["observation"]["path"] + rows = json.loads(source_path.read_bytes()) + rows[0]["counters"].pop("silent_incorrect_success_count") + observation["counter_source"]["observation"] = write(source_path, rows) + samples.update_references(self.manifest, source_path) + write(observation_path, observation) + samples.update_references(self.manifest, observation_path) + with self.assertRaisesRegex(ValueError, "normalized counters"): + adapter.verify_subjects( + self.manifest_path, + self.manifest, + self.mapping["admitted_subject"], + self.mapping["repair_prior_subject"], + ) + + def test_repair_labels_cannot_hide_wrong_actual_active_pointer(self): + trial = next( + t for t in self.manifest["trials"] if t["class"] == "governed_repair" + ) + observation_path = self.root / trial["observation"]["path"] + observation = json.loads(observation_path.read_bytes()) + transition_path = ( + self.root / observation["repair_transition_verification"]["path"] + ) + transition = json.loads(transition_path.read_bytes()) + pointer_path = self.root / transition["active_pointer"]["path"] + transition["active_pointer"] = write( + pointer_path, + { + "mode": "active", + "active_digest": self.mapping["repair_prior_subject"][ + "bundle_content_digest" + ], + }, + ) + samples.update_references(self.manifest, pointer_path) + observation["repair_transition_verification"] = write( + transition_path, transition + ) + samples.update_references(self.manifest, transition_path) + write(observation_path, observation) + samples.update_references(self.manifest, observation_path) + with self.assertRaisesRegex( + ValueError, "actual repair approval and ACTIVE pointer" + ): + adapter.verify_subjects( + self.manifest_path, + self.manifest, + self.mapping["admitted_subject"], + self.mapping["repair_prior_subject"], + ) + + def replay_fixture(self): + trial = next( + t for t in self.manifest["trials"] if t["class"] == "idempotency_replay" + ) + path = self.root / trial["observation"]["path"] + observation = json.loads(path.read_bytes()) + primary_path = self.root / observation["reports"][0]["path"] + primary_report = json.loads(primary_path.read_bytes()) + primary_report.update( + idempotency_key="test-same-key", + run_id_sha256=adapter.sha(b"test-initial-run").removeprefix("sha256:"), + ) + write(primary_path, primary_report) + samples.update_references(observation, primary_path) + samples.update_references(self.manifest, primary_path) + base_fields = { + "bundle_content_digest", + "workflow_contract_sha256", + "parameter_schema_sha256", + } + replay = { + key: value if key in base_fields else None + for key, value in self.mapping["admitted_subject"].items() + } + replay.update( + idempotent_replay=True, + idempotency_key="test-same-key", + run_id_sha256=adapter.sha(b"test-replay-run").removeprefix("sha256:"), + success=False, + transaction_outcome="HALTED", + model_calls=0, + ) + ref = write(self.root / "replay.json", replay) + self.manifest["artifacts"].append(ref) + proof = { + "schema_version": "openadapt.idempotent-replay-refusal-verification/v1", + "report_sha256": ref["sha256"], + "primary_report_sha256": observation["reports"][0]["sha256"], + "same_key": True, + "no_new_input": True, + } + for field in ( + "ledger_before", + "ledger_after", + "input_events_before", + "input_events_after", + ): + snapshot_path = self.root / f"{field}.json" + if field.startswith("ledger_"): + with sqlite3.connect(snapshot_path) as connection: + connection.execute( + "CREATE TABLE ledger_metadata(singleton INTEGER PRIMARY KEY, schema_version TEXT, namespace TEXT, owner_path TEXT)" + ) + connection.execute( + "INSERT INTO ledger_metadata VALUES (1, 'openadapt.idempotency-ledger/v3', 'openadapt-flow-runtime/v1', '/test-only/original-ledger')" + ) + connection.execute( + "CREATE TABLE reservations(namespace TEXT, reservation_key TEXT, run_id TEXT, reserved_at TEXT, outcome TEXT)" + ) + connection.execute( + "INSERT INTO reservations VALUES ('openadapt-flow-runtime/v1', 'test-same-key', 'test-initial-run', '2026-08-27T12:00:00Z', 'VERIFIED')" + ) + snapshot = retain(snapshot_path) + else: + snapshot = write(snapshot_path, {"fixture_events": []}) + self.manifest["artifacts"].append(snapshot) + proof[field] = snapshot + proof_ref = write(self.root / "replay-proof.json", proof) + self.manifest["artifacts"].append(proof_ref) + observation["replay_refusal_verification"] = proof_ref + observation["reports"].append( + {**ref, "role": "replay", "bundle_role": "admitted"} + ) + self.bind_source_reports(observation) + write(path, observation) + samples.update_references(self.manifest, path) + return path, observation, proof + + def update_replay_proof(self, observation_path, observation, proof): + proof_path = self.root / observation["replay_refusal_verification"]["path"] + observation["replay_refusal_verification"] = write(proof_path, proof) + samples.update_references(self.manifest, proof_path) + write(observation_path, observation) + samples.update_references(self.manifest, observation_path) + + def test_replay_keys_must_be_present_nonempty_and_equal_in_native_reports(self): + observation_path, observation, proof = self.replay_fixture() + ref = next(r for r in observation["reports"] if r["role"] == "replay") + path = self.root / ref["path"] + original = json.loads(path.read_bytes()) + for value in (None, "", "different-native-key"): + with self.subTest(key=value): + report = {**original, "idempotency_key": value} + write(path, report) + samples.update_references(observation, path) + samples.update_references(self.manifest, path) + proof["report_sha256"] = retain(path)["sha256"] + self.update_replay_proof(observation_path, observation, proof) + with self.assertRaisesRegex(ValueError, "native idempotency keys"): + adapter.verify_subjects( + self.manifest_path, + self.manifest, + self.mapping["admitted_subject"], + self.mapping["repair_prior_subject"], + ) + + def test_replay_primary_native_key_cannot_be_missing(self): + observation_path, observation, proof = self.replay_fixture() + path = self.root / observation["reports"][0]["path"] + report = json.loads(path.read_bytes()) + report.pop("idempotency_key") + write(path, report) + samples.update_references(observation, path) + samples.update_references(self.manifest, path) + proof["primary_report_sha256"] = retain(path)["sha256"] + self.update_replay_proof(observation_path, observation, proof) + with self.assertRaisesRegex(ValueError, "native idempotency keys"): + adapter.verify_subjects( + self.manifest_path, + self.manifest, + self.mapping["admitted_subject"], + self.mapping["repair_prior_subject"], + ) + + def test_replay_ledger_must_bind_the_actual_initial_run_digest(self): + observation_path, observation, proof = self.replay_fixture() + path = self.root / proof["ledger_after"]["path"] + with sqlite3.connect(path) as connection: + connection.execute("UPDATE reservations SET run_id = 'another-initial-run'") + proof["ledger_after"] = retain(path) + samples.update_references(self.manifest, path) + self.update_replay_proof(observation_path, observation, proof) + with self.assertRaisesRegex(ValueError, "same key to the verified initial run"): + adapter.verify_subjects( + self.manifest_path, + self.manifest, + self.mapping["admitted_subject"], + self.mapping["repair_prior_subject"], + ) + + def test_same_subject_reports_cannot_move_between_counted_source_rows(self): + trials = [ + next(t for t in self.manifest["trials"] if t["class"] == name) + for name in ("healthy", "declared_attended") + ] + paths = [self.root / t["observation"]["path"] for t in trials] + observations = [json.loads(p.read_bytes()) for p in paths] + observations[0]["reports"], observations[1]["reports"] = ( + observations[1]["reports"], + observations[0]["reports"], + ) + for path, observation in zip(paths, observations): + write(path, observation) + samples.update_references(self.manifest, path) + inventory = {item["path"]: item for item in self.manifest["artifacts"]} + for trial in trials: + adapter.measured.verify_trial_observation( + trial, + inventory, + self.root, + "1.35.1", + self.manifest["runtime"]["wheel_sha256"], + set(), + ) + with self.assertRaisesRegex( + ValueError, "do not belong to the counted source row" + ): + adapter.verify_subjects( + self.manifest_path, + self.manifest, + self.mapping["admitted_subject"], + self.mapping["repair_prior_subject"], + ) + + def test_early_replay_refusal_preserves_real_missing_observations(self): + self.replay_fixture() + adapter.verify_subjects( + self.manifest_path, + self.manifest, + self.mapping["admitted_subject"], + self.mapping["repair_prior_subject"], + ) + + def test_replay_no_input_label_cannot_hide_actual_new_events(self): + _, _, proof = self.replay_fixture() + path = self.root / proof["input_events_after"]["path"] + write(path, {"fixture_events": ["new input"]}) + samples.update_references(self.manifest, path) + proof["input_events_after"] = retain(path) + proof_path = self.root / "replay-proof.json" + write(proof_path, proof) + samples.update_references(self.manifest, proof_path) + trial = next( + t for t in self.manifest["trials"] if t["class"] == "idempotency_replay" + ) + observation_path = self.root / trial["observation"]["path"] + observation = json.loads(observation_path.read_bytes()) + observation["replay_refusal_verification"] = retain(proof_path) + write(observation_path, observation) + samples.update_references(self.manifest, observation_path) + with self.assertRaisesRegex(ValueError, "new input events"): + adapter.verify_subjects( + self.manifest_path, + self.manifest, + self.mapping["admitted_subject"], + self.mapping["repair_prior_subject"], + ) + + def test_replay_exception_does_not_allow_missing_primary_tuple(self): + _, observation, _ = self.replay_fixture() + path = self.root / observation["reports"][0]["path"] + report = json.loads(path.read_bytes()) + report["observed_environment_digest"] = None + write(path, report) + samples.update_references(self.manifest, path) + with self.assertRaisesRegex(ValueError, "exact declared bundle tuple"): + adapter.verify_subjects( + self.manifest_path, + self.manifest, + self.mapping["admitted_subject"], + self.mapping["repair_prior_subject"], + ) + + def test_reconciliation_cli_never_reads_keys_network_or_calls_issuer(self): + plan = { + "phase_request": request("receipt"), + "issue_request": None, + "inputs": self.prepare(), + "unsigned_object": {"test_only": "result"}, + } + state = self.root / "state" + adapter.persist_once(plan, state, lambda _: plan["unsigned_object"]) + journal = next(state.glob("*.request.json")) + with ( + mock.patch.object( + adapter.software, + "keychain_read", + side_effect=AssertionError("key read"), + ), + mock.patch.object( + adapter.verifier, "fetch", side_effect=AssertionError("network") + ), + mock.patch.object( + adapter, "phase_object", side_effect=AssertionError("issuer") + ), + ): + status = adapter.main( + [ + "--reconcile-journal", + str(journal), + "--state-dir", + str(state), + "--output", + str(self.root / "reconciled.json"), + ] + ) + self.assertEqual(status, 0) + self.assertTrue( + json.loads((self.root / "reconciled.json").read_bytes())["unsigned_result"] + ) + + def publication_fixture(self): + fixture, context = trust_context() + resolver = context["resolver"] + admission = adapter.issuer.issue_workflow_admission( + signed_samples.workflow_request(fixture), + resolver=resolver, + issuer_source_commit="1" * 40, + now=signed_samples.NOW, + consumer=adapter.PreviewConsumer(), + ) + signed_samples.flow_release_inputs(fixture, admission, resolver) + template = next( + v["value"] + for v in resolver.objects.values() + if v["value"].get("schema_version") == "openadapt.production-acceptance/v3" + ) + staging = copy.deepcopy(template["publication_staging"]) + inputs = self.prepare() + staging.update( + publication_mode="already-published-pypi", + draft=False, + target_commitish=inputs["release"]["source_commit"], + tag=inputs["release"]["tag"], + ) + staging["assets"] = [ + { + "asset_id": str(i + 30), + **a, + "uploader_id": "321543906", + "uploader_login": "openadapt-release[bot]", + } + for i, a in enumerate( + sorted( + inputs["artifact_inventory"]["artifacts"], key=lambda a: a["name"] + ) + ) + ] + staging["pypi_files"] = adapter.trust.pypi_files_from_assets(staging["assets"]) + staging["tag_ref_state"] = {"ref": "refs/tags/v1.35.1", "exists": True} + staging["tag_ref_state_sha256"] = adapter.trust.digest_bytes( + adapter.trust.TAG_REF_STATE_DOMAIN, staging["tag_ref_state"] + ) + candidate = json.loads( + (self.root / self.mapping["candidate"]["path"]).read_bytes() + ) + files = { + name: (self.root / ref["path"]).read_bytes() + for name, ref in self.mapping["publication_files"].items() + } + repo = "repos/OpenAdaptAI/openadapt-flow" + responses = { + f"{repo}/releases/tags/v1.35.1": { + "id": 20, + "draft": False, + "prerelease": False, + "tag_name": "v1.35.1", + "author": {"login": "openadapt-release[bot]"}, + "assets": [], + }, + f"{repo}/git/ref/tags/v1.35.1": { + "object": {"type": "commit", "sha": inputs["release"]["source_commit"]} + }, + f"{repo}/immutable-releases": staging["immutable_releases"], + } + downloads = {} + pypi = {"urls": []} + for asset in staging["assets"]: + name = asset["name"] + github_url = f"https://github.com/OpenAdaptAI/openadapt-flow/releases/download/v1.35.1/{name}" + pypi_url = f"https://files.pythonhosted.org/packages/{name}" + responses[f"{repo}/releases/tags/v1.35.1"]["assets"].append( + { + "name": name, + "id": int(asset["asset_id"]), + "size": asset["size_bytes"], + "state": "uploaded", + "digest": asset["sha256"], + "uploader": { + "login": asset["uploader_login"], + "id": int(asset["uploader_id"]), + }, + "browser_download_url": github_url, + } + ) + pypi["urls"].append( + { + "filename": name, + "size": asset["size_bytes"], + "yanked": False, + "digests": {"sha256": asset["sha256"].removeprefix("sha256:")}, + "url": pypi_url, + } + ) + downloads[github_url] = downloads[pypi_url] = files[name] + for rule in staging["tag_rulesets"]: + responses[f"{repo}/rulesets/{rule['ruleset_id']}"] = { + "id": int(rule["ruleset_id"]), + **{ + k: rule[k] + for k in ( + "name", + "target", + "enforcement", + "conditions", + "rules", + "bypass_actors", + ) + }, + } + downloads["https://pypi.org/pypi/openadapt-flow/1.35.1/json"] = ( + adapter.canonical(pypi) + ) + return candidate, staging, files, responses, downloads + + def test_publication_checks_actual_bytes_from_both_publishers(self): + candidate, staging, files, api, downloads = self.publication_fixture() + adapter.verify_publication( + candidate, staging, files, api=api.__getitem__, fetch=downloads.__getitem__ + ) + key = next(key for key in downloads if key.startswith("https://github.com/")) + downloads[key] = b"wrong immutable content" + with self.assertRaisesRegex(ValueError, "actual published artifact bytes"): + adapter.verify_publication( + candidate, + staging, + files, + api=api.__getitem__, + fetch=downloads.__getitem__, + ) + + def test_publication_rejects_unpublished_and_changed_tag_source(self): + candidate, staging, files, api, downloads = self.publication_fixture() + candidate["release_observation"]["published"] = False + with self.assertRaisesRegex(ValueError, "unpublished"): + adapter.verify_publication( + candidate, + staging, + files, + api=api.__getitem__, + fetch=downloads.__getitem__, + ) + candidate["release_observation"]["published"] = True + api["repos/OpenAdaptAI/openadapt-flow/git/ref/tags/v1.35.1"]["object"][ + "sha" + ] = "b" * 40 + with self.assertRaisesRegex(ValueError, "tag source"): + adapter.verify_publication( + candidate, + staging, + files, + api=api.__getitem__, + fetch=downloads.__getitem__, + ) + + def test_all_five_objects_use_existing_signers_issuers_and_real_storage_edges(self): + inputs = self.prepare() + fixture, context = trust_context() + resolver = context["resolver"] + value, _ = adapter.phase_object(inputs, request("receipt"), context) + raw, _ = adapter.sign_pair( + value, + {"phase_request": request("receipt")}, + context, + private_key=fixture["decision_key"], + ) + fixture["receipt"] = json.loads(raw) + receipt_ref, receipt_bundle = signed_samples.reference_pair( + adapter.KINDS["receipt"], fixture["receipt"] + ) + fixture["receipt_ref"], fixture["receipt_bundle_ref"] = ( + receipt_ref, + receipt_bundle, + ) + resolver.add( + adapter.KINDS["receipt"], fixture["receipt"], receipt_ref, receipt_bundle + ) + workflow_request = request("workflow", receipt=receipt_ref) + workflow, _ = adapter.phase_object(inputs, workflow_request, context) + # Existing fixtures supply policy-complete staging. This test replaces + # every package artifact with the temporary byte-verified candidate. + signed_samples.flow_release_inputs(fixture, workflow, resolver) + template = next( + v["value"] + for v in resolver.objects.values() + if v["value"].get("schema_version") == "openadapt.production-acceptance/v3" + ) + staging = copy.deepcopy(template["publication_staging"]) + staging.update( + publication_mode="already-published-pypi", + draft=False, + target_commitish=inputs["release"]["source_commit"], + tag=inputs["release"]["tag"], + ) + staging["assets"] = [ + { + "asset_id": str(i + 30), + **a, + "uploader_id": "321543906", + "uploader_login": "openadapt-release[bot]", + } + for i, a in enumerate( + sorted( + inputs["artifact_inventory"]["artifacts"], key=lambda a: a["name"] + ) + ) + ] + staging["pypi_files"] = adapter.trust.pypi_files_from_assets(staging["assets"]) + staging["tag_ref_state"] = {"ref": "refs/tags/v1.35.1", "exists": True} + staging["tag_ref_state_sha256"] = adapter.trust.digest_bytes( + adapter.trust.TAG_REF_STATE_DOMAIN, staging["tag_ref_state"] + ) + inputs["publication_staging"] = staging + workflow_ref, workflow_bundle = signed_samples.reference_pair( + adapter.KINDS["workflow"], workflow, registry_source_commit="2" * 40 + ) + resolver.add(adapter.KINDS["workflow"], workflow, workflow_ref, workflow_bundle) + manifest_request = request( + "manifest", receipt=receipt_ref, workflow=workflow_ref + ) + manifest_request["acceptance_issuer_source_commit"] = "a" * 40 + with mock.patch.object( + adapter, "gh", return_value={"object": {"sha": "a" * 40}} + ): + manifest, _ = adapter.phase_object(inputs, manifest_request, context) + manifest_ref, manifest_bundle = signed_samples.reference_pair( + adapter.KINDS["manifest"], manifest, registry_source_commit="3" * 40 + ) + resolver.add(adapter.KINDS["manifest"], manifest, manifest_ref, manifest_bundle) + summary_request = request("summary", manifest=manifest_ref) + summary, _ = adapter.phase_object(inputs, summary_request, context) + summary_ref, summary_bundle = signed_samples.reference_pair( + adapter.KINDS["summary"], summary, registry_source_commit="4" * 40 + ) + resolver.add(adapter.KINDS["summary"], summary, summary_ref, summary_bundle) + release_request = request("release", source="4" * 40, summary=summary_ref) + release, _ = adapter.phase_object(inputs, release_request, context) + for phase, value, phase_request in ( + ("workflow", workflow, workflow_request), + ("manifest", manifest, manifest_request), + ("summary", summary, summary_request), + ("release", release, release_request), + ): + raw, bundle = adapter.sign_pair( + value, + {"phase_request": phase_request}, + context, + private_key=fixture["decision_key"], + ) + self.assertTrue(json.loads(bundle)["dsseEnvelope"]["signatures"]) + self.assertEqual(json.loads(raw)["schema_version"], value["schema_version"]) + self.assertEqual( + manifest["qualification_admission_reference"]["registry_source_commit"], + "2" * 40, + ) + self.assertEqual( + summary["production_acceptance_manifest_reference"][ + "registry_source_commit" + ], + "3" * 40, + ) + self.assertEqual(release["issuer"]["source_commit"], "4" * 40) + self.assertEqual(release["release"], inputs["release"]) + with self.assertRaisesRegex(ValueError, "immediate issuer dependency"): + adapter.phase_object( + inputs, + request("workflow", source="2" * 40, receipt=receipt_ref), + context, + ) + + +if __name__ == "__main__": + unittest.main() From 6eef685308a8c581905135e3e845c93b4a6754cc Mon Sep 17 00:00:00 2001 From: abrichr Date: Tue, 8 Sep 2026 23:49:36 -0400 Subject: [PATCH 2/2] fix: verify registered software DSSE release evidence Signed-off-by: abrichr --- .../verify_production_release_admission.py | 323 +++++++++++++++++- tests/test_registered_software_verifier.py | 270 +++++++++++++++ 2 files changed, 582 insertions(+), 11 deletions(-) create mode 100644 tests/test_registered_software_verifier.py diff --git a/scripts/verify_production_release_admission.py b/scripts/verify_production_release_admission.py index 89a964c..0c6a942 100644 --- a/scripts/verify_production_release_admission.py +++ b/scripts/verify_production_release_admission.py @@ -19,6 +19,8 @@ from typing import Any import production_trust as trust +import public_trust_kms as public_trust +import public_trust_resolver as public_resolver import validate_evidence_registry as evidence ROOT = Path(__file__).resolve().parents[1] @@ -379,7 +381,13 @@ def load_signer(pointer: dict[str, Any], *, require_active: bool) -> dict[str, A return signer_value current_signer_registry = load_signer(current_pointer, require_active=True) - bound_identity = value.get("signer_registry_sha256") + # Authority v2 commits registry bytes and semantic identity separately. + bound_identity = value.get( + "signer_registry_identity_sha256" + if value.get("schema_version") + == "openadapt.qualification-authority-state-receipt/v2" + else "signer_registry_sha256" + ) if bound_identity is None: bound_pointer = current_pointer else: @@ -419,11 +427,10 @@ def resolve_pair( regular_raw, bundle_raw, value, bound_signer_registry, current_signer_registry = ( fetch_pair(regular_reference, bundle_reference) ) - verify_sigstore( - regular_raw, - bundle_raw, - kind=kind, - object_value=value, + verify_registered_signature( + regular_reference, + bundle_reference, + (regular_raw, bundle_raw, value, bound_signer_registry, current_signer_registry), policy=policy, ) return value, bound_signer_registry, current_signer_registry @@ -455,6 +462,298 @@ def derive_bundle_reference(regular_reference: dict[str, Any]) -> dict[str, Any] ) +def _registered_state_pairs( + reference: dict[str, Any], + *, + kind: str, + identity: str | None = None, + authority_identity: str | None = None, +) -> list[tuple[dict[str, Any], dict[str, Any], tuple]]: + """Resolve historical state from the object's exact containing registry.""" + + commit = reference["registry_source_commit"] + registry = json.loads(fetch(raw_url(commit, "evidence-registry.json"))) + entries = evidence.validate_registry(registry) + if ( + registry["revision"] != reference["registry_revision"] + or registry["registry_head_sha256"] != reference["registry_head_sha256"] + ): + raise trust.TrustError("state reference registry revision or head differs") + matches = [] + for entry in entries: + if entry["kind"] != kind or ( + identity is not None and entry["semantic_identity_sha256"] != identity + ): + continue + regular = { + **{field: reference[field] for field in ( + "schema_version", "repository", "repository_id", + "repository_owner_id", "registry_source_commit", + "registry_revision", "registry_head_sha256", + )}, + **entry, + } + bundle = derive_bundle_reference(regular) + pair = fetch_pair(regular, bundle) + if authority_identity is not None and ( + pair[2].get("authority_state_sha256") != authority_identity + ): + continue + matches.append((regular, bundle, pair)) + return matches + + +def _registered_state_pair(reference: dict[str, Any], **selectors: Any) -> tuple: + matches = _registered_state_pairs(reference, **selectors) + if len(matches) != 1: + raise trust.TrustError("exactly one bound state must be registered") + return matches[0] + + +def _software_signer(registry: dict[str, Any], statement: dict[str, Any]) -> dict: + matches = [signer for signer in registry["signers"] if ( + signer["key_id"] == statement["key_id"] + and signer["algorithm"] == "ed25519" + and signer.get("key_origin") == "software" + )] + if len(matches) != 1: + raise trust.TrustError("software bundle does not select one registered signer") + return matches[0] + + +def _verify_software_pair( + reference: dict[str, Any], + bundle_reference: dict[str, Any], + pair: tuple, + *, + authority: dict[str, Any], + revocation: dict[str, Any], + now: datetime, +) -> None: + raw, bundle_raw, value, bound_registry, current_registry = pair + bundle = json.loads(bundle_raw) + statement = public_trust.statement_from_bundle(bundle) + if statement["signature_profile"] != public_trust.SOFTWARE_SIGNATURE_PROFILE: + raise trust.TrustError("registered software route rejects this signature profile") + bound_raw = evidence.canonical(bound_registry) + b"\n" + bound_identity = evidence.signer_registry_identity_digest(bound_registry) + signed_at = trust.require_timestamp(statement["issued_at"], "statement issued_at") + expectations = { + "expected_signer_registry_sha256": bound_identity, + "expected_authority_state_sha256": authority["authority_state_sha256"], + "expected_revocation_state_sha256": revocation["revocation_state_sha256"], + } + if value.get("schema_version") == ( + "openadapt.qualification-authority-state-receipt/v2" + ): + # This existing schema and its outer statement bind the raw registry + # digest. Do not relabel it as the independent semantic identity. + if raw != evidence.canonical(value) + b"\n" or ( + bundle_raw != evidence.canonical(bundle) + b"\n" + ): + raise trust.TrustError("authority pair must be canonical JSON plus LF") + generated = evidence._timestamp(bound_registry["generated_at"], "registry generated_at") + expires = evidence.optional_timestamp(bound_registry["expires_at"], "registry expires_at") + if signed_at < generated or (expires is not None and signed_at >= expires): + raise trust.TrustError("authority bound signer registry is not active at statement time") + if value["signer_registry_identity_sha256"] != bound_identity: + raise trust.TrustError("authority signer registry identity differs") + expectations["expected_signer_registry_sha256"] = ( + "sha256:" + hashlib.sha256(bound_raw).hexdigest() + ) + public_trust.validate_statement_object_binding( + statement, + object_raw=raw, + object_value=value, + object_kind=reference["kind"], + object_schema_version=reference["object_schema_version"], + object_media_type=reference["object_media_type"], + semantic_identity_sha256=reference["semantic_identity_sha256"], + **expectations, + ) + public_trust.verify_bundle( + bundle, expected_statement=statement, + signer=_software_signer(bound_registry, statement), now=signed_at, + ) + else: + public_resolver.verify_registered_public_trust_pair( + object_raw=raw, object_reference=reference, + bundle_raw=bundle_raw, bundle_reference=bundle_reference, + signer_registry_raw=bound_raw, now=signed_at, **expectations, + ) + # Historical bindings do not grant a revoked key present-day authority. + public_trust.verify_bundle( + bundle, expected_statement=statement, + signer=_software_signer(current_registry, statement), now=now, + ) + + +def _verify_embedded_state( + state_pair: tuple, + *, + bound_registry: dict[str, Any], + current_registry: dict[str, Any], + now: datetime, +) -> None: + reference, _, fetched = state_pair + state, state_bound, state_current = fetched[2:] + if ( + evidence.signer_registry_identity_digest(state_bound) + != evidence.signer_registry_identity_digest(bound_registry) + or evidence.signer_registry_identity_digest(state_current) + != evidence.signer_registry_identity_digest(current_registry) + ): + raise trust.TrustError("registered trust-state signer registries differ") + if reference["kind"] == "qualification-authority-state-receipt": + validator = trust.validate_authority_state + domain = trust.AUTHORITY_STATE_SIGNATURE_DOMAIN + elif reference["kind"] == "qualification-revocation-state-receipt": + validator = trust.validate_revocation_state + domain = trust.REVOCATION_STATE_SIGNATURE_DOMAIN + else: + raise trust.TrustError("registered state has an unsupported kind") + observed_at = trust.require_timestamp(state["observed_at"], "state observed_at") + validator(state, now=observed_at) + for registry, instant in ((state_bound, observed_at), (state_current, now)): + trust.verify_embedded_signature( + state, signer_registry=registry, + object_schema_version=reference["object_schema_version"], + signature_domain=domain, usage=reference["kind"], now=instant, + ) + + +def _validate_registered_state_links( + authority: dict[str, Any], + revocation: dict[str, Any], + *, + bound_registry: dict[str, Any], + current_registry: dict[str, Any], +) -> None: + identity = evidence.signer_registry_identity_digest(bound_registry) + if ( + authority["signer_registry_identity_sha256"] != identity + or authority["signer_registry_sha256"] + != "sha256:" + hashlib.sha256(evidence.canonical(bound_registry) + b"\n").hexdigest() + or authority["signer_registry_revision"] != bound_registry["revision"] + or revocation["signer_registry_sha256"] != identity + or revocation["authority_state_sha256"] != authority["authority_state_sha256"] + ): + raise trust.TrustError("registered bound authority and signer state differ") + revoked = {(item["subject_kind"], item["subject_id"]) + for item in revocation["revocations"]} + if any(signer["status"] == "active" and ( + "qualification-signer-key", signer["public_key_sha256"] + ) in revoked for signer in current_registry["signers"]): + raise trust.TrustError("an active signer key is revoked") + + +def _verify_authority_pair( + authority_pair: tuple, + *, + bound_registry: dict[str, Any], + current_registry: dict[str, Any], + now: datetime, +) -> tuple: + """Find the independently verified historical reverse link missing in v2.""" + + reference, bundle_reference, pair = authority_pair + authority = pair[2] + matches = [] + for candidate in _registered_state_pairs( + reference, kind="qualification-revocation-state-receipt", + authority_identity=authority["authority_state_sha256"], + ): + revocation = candidate[2][2] + try: + _verify_embedded_state(candidate, bound_registry=bound_registry, + current_registry=current_registry, now=now) + _validate_registered_state_links(authority, revocation, + bound_registry=bound_registry, + current_registry=current_registry) + _verify_software_pair(*candidate, authority=authority, + revocation=revocation, now=now) + _verify_software_pair(reference, bundle_reference, pair, + authority=authority, revocation=revocation, now=now) + except (trust.TrustError, public_trust.PublicTrustKmsError, + public_resolver.PublicTrustResolutionError, + evidence.EvidenceRegistryError, KeyError, TypeError, + json.JSONDecodeError): + continue + matches.append(candidate) + if len(matches) != 1: + raise trust.TrustError("authority must bind exactly one verified registered revocation") + return matches[0] + + +def verify_registered_signature( + reference: dict[str, Any], + bundle_reference: dict[str, Any], + pair: tuple, + *, + policy: dict[str, Any], +) -> None: + """Route registered software DSSE without changing any keyless profile.""" + + raw, bundle_raw, value, bound_registry, current_registry = pair + bundle = json.loads(bundle_raw) + envelope = bundle.get("dsseEnvelope", {}) if isinstance(bundle, dict) else {} + material = bundle.get("verificationMaterial", {}) if isinstance(bundle, dict) else {} + is_public_trust = ( + isinstance(envelope, dict) + and envelope.get("payloadType") == public_trust.STATEMENT_MEDIA_TYPE + ) or (isinstance(material, dict) and "publicKey" in material) + if not is_public_trust: + verify_sigstore(raw, bundle_raw, kind=reference["kind"], + object_value=value, policy=policy) + return + try: + statement = public_trust.statement_from_bundle(bundle) + if statement["signature_profile"] != public_trust.SOFTWARE_SIGNATURE_PROFILE: + raise trust.TrustError("registered software route rejects this signature profile") + now = datetime.now(timezone.utc) + if value.get("schema_version") == ( + "openadapt.qualification-authority-state-receipt/v2" + ): + authority_pair = (reference, bundle_reference, pair) + _verify_embedded_state(authority_pair, bound_registry=bound_registry, + current_registry=current_registry, now=now) + _verify_authority_pair(authority_pair, bound_registry=bound_registry, + current_registry=current_registry, now=now) + return + # Select the target's states from registered object commitments, + # never from an outer DSSE statement. + revocation_pair = _registered_state_pair( + reference, kind="qualification-revocation-state-receipt", + identity=value.get("revocation_state_sha256"), + authority_identity=value.get("authority_state_sha256"), + ) + revocation = revocation_pair[2][2] + authority_pair = _registered_state_pair( + reference, kind="qualification-authority-state-receipt", + identity=revocation["authority_state_sha256"], + ) + authority = authority_pair[2][2] + for state_pair in (authority_pair, revocation_pair): + _verify_embedded_state(state_pair, bound_registry=bound_registry, + current_registry=current_registry, now=now) + _validate_registered_state_links(authority, revocation, + bound_registry=bound_registry, + current_registry=current_registry) + # The authority outer signature keeps its own historical reverse link + # when a later revocation retains the same authority identity. + _verify_authority_pair(authority_pair, bound_registry=bound_registry, + current_registry=current_registry, now=now) + _verify_software_pair(*revocation_pair, authority=authority, + revocation=revocation, now=now) + _verify_software_pair(reference, bundle_reference, pair, + authority=authority, revocation=revocation, now=now) + except (public_trust.PublicTrustKmsError, + public_resolver.PublicTrustResolutionError, + evidence.EvidenceRegistryError, KeyError, TypeError, + json.JSONDecodeError) as exc: + raise trust.TrustError(f"registered software DSSE verification failed: {exc}") from exc + + def verify_sigstore( regular_raw: bytes, bundle_raw: bytes, @@ -746,11 +1045,13 @@ def main(argv: list[str] | None = None) -> int: _current_signer_registry, ) = fetch_pair(reference, bundle_reference) policy = json.loads(POLICY_PATH.read_text(encoding="utf-8")) - verify_sigstore( - regular_raw, - bundle_raw, - kind="qualification-release", - object_value=admission_value, + if reference.get("kind") != "qualification-release": + raise trust.TrustError("the referenced object is not qualification-release") + verify_registered_signature( + reference, + bundle_reference, + (regular_raw, bundle_raw, admission_value, + release_signer_registry, _current_signer_registry), policy=policy, ) now = datetime.now(timezone.utc) diff --git a/tests/test_registered_software_verifier.py b/tests/test_registered_software_verifier.py new file mode 100644 index 0000000..efad71e --- /dev/null +++ b/tests/test_registered_software_verifier.py @@ -0,0 +1,270 @@ +"""Exercise the registered software route with retained, unchanged signatures.""" + +from __future__ import annotations + +import base64 +import copy +import hashlib +import json +import sys +import unittest +from datetime import datetime, timezone +from pathlib import Path +from unittest import mock + +ROOT = Path(__file__).resolve().parents[1] +sys.path.insert(0, str(ROOT / "scripts")) + +import production_trust as trust +import public_trust_kms as public_trust +import validate_evidence_registry as evidence +import verify_production_release_admission as verifier + +# These committed public bytes retain the actual provisioned-key signatures. +RELEASE_SHA = "sha256:790122a25c87e456c6e45d25ebf5cd029b21b8511265b062fd9129b74aa1dd82" +AUTHORITY_SHA = "sha256:a22f9815ec0f7c56f7629aeabfd21d14bc6d739efcfb7f7a073ce9987e19479e" +REVOCATION_SHA = "sha256:18633b8cc243f686706606162bfa249a29733811563f63b9e68cc7c4a3507676" +SIGNER_SHA = "sha256:7a81bf3d213c74673f3c6b5fa179234cbee534c9432aaea6ae09e455562f96b6" +STORAGE_COMMIT = "f" * 40 + + +def canonical(value: dict) -> bytes: + return evidence.canonical(value) + b"\n" + + +class RegisteredSoftwareVerifierTests(unittest.TestCase): + def setUp(self) -> None: + self.registry = json.loads((ROOT / "evidence-registry.json").read_bytes()) + retained = {RELEASE_SHA, AUTHORITY_SHA, REVOCATION_SHA} + self.registry["entries"] = [ + entry for entry in self.registry["entries"] + if entry["object_sha256"] in retained or entry["subject_sha256"] in retained + ] + pointer = next(pointer for pointer in self.registry["signer_registry_history"] + if pointer["object_sha256"] == SIGNER_SHA) + self.registry.update(signer_registry=pointer, signer_registry_history=[pointer], + revision=1, previous_registry_head_sha256=None) + self.refresh_registry() + self.files = {entry["object_path"]: (ROOT / entry["object_path"]).read_bytes() + for entry in self.registry["entries"]} + self.files[pointer["object_path"]] = (ROOT / pointer["object_path"]).read_bytes() + self.policy = json.loads(verifier.POLICY_PATH.read_bytes()) + self.fetch = mock.patch.object(verifier, "fetch", side_effect=self.fetch_bytes) + self.fetch.start() + self.addCleanup(self.fetch.stop) + self.keyless = mock.patch.object(verifier, "verify_sigstore") + self.keyless_mock = self.keyless.start() + self.addCleanup(self.keyless.stop) + + def refresh_registry(self) -> None: + self.registry["registry_head_sha256"] = evidence.registry_head_digest(self.registry) + evidence.validate_registry(self.registry) + + def fetch_bytes(self, url: str, **_kwargs) -> bytes: + prefix = verifier.raw_url(STORAGE_COMMIT, "") + if not url.startswith(prefix): + raise AssertionError(f"unexpected network request: {url}") + path = url.removeprefix(prefix) + return canonical(self.registry) if path == "evidence-registry.json" else self.files[path] + + def reference(self, digest: str) -> dict: + entry = next(entry for entry in self.registry["entries"] + if entry["object_sha256"] == digest) + return { + "schema_version": evidence.REFERENCE_SCHEMA, + "repository": evidence.REPOSITORY, + "repository_id": evidence.REPOSITORY_ID, + "repository_owner_id": evidence.REPOSITORY_OWNER_ID, + "registry_source_commit": STORAGE_COMMIT, + "registry_revision": self.registry["revision"], + "registry_head_sha256": self.registry["registry_head_sha256"], + **entry, + } + + def resolve(self, digest: str = RELEASE_SHA) -> tuple: + reference = self.reference(digest) + return verifier.resolve_pair(reference, verifier.derive_bundle_reference(reference), + kind=reference["kind"], policy=self.policy) + + def replace_bundle(self, digest: str, mutate) -> None: + reference = self.reference(digest) + bundle_ref = verifier.derive_bundle_reference(reference) + bundle = json.loads(self.files[bundle_ref["object_path"]]) + mutate(bundle) + raw = canonical(bundle) + changed = next(entry for entry in self.registry["entries"] + if entry["object_sha256"] == bundle_ref["object_sha256"]) + changed["object_sha256"] = "sha256:" + hashlib.sha256(raw).hexdigest() + digest_hex = changed["object_sha256"].removeprefix("sha256:") + changed["object_path"] = ( + f"production-evidence/objects/sha256/{digest_hex[:2]}/" + f"{digest_hex}.{changed['kind']}.json" + ) + changed["size_bytes"] = len(raw) + changed["semantic_identity_sha256"] = evidence.semantic_identity_digest( + kind=changed["kind"], object_schema_version=changed["object_schema_version"], + object_value=changed["subject_sha256"], object_sha256=changed["object_sha256"], + ) + changed["registry_entry_sha256"] = evidence.entry_digest(changed) + self.files[changed["object_path"]] = raw + self.refresh_registry() + + def test_actual_release_authority_and_revocation_signatures_resolve(self) -> None: + with mock.patch.object(verifier, "protected_main_commit", + side_effect=AssertionError("storage is not the live main tip")): + for digest in (RELEASE_SHA, AUTHORITY_SHA, REVOCATION_SHA): + with self.subTest(digest=digest): + value, bound, current = self.resolve(digest) + self.assertEqual(bound, current) + if digest == AUTHORITY_SHA: + self.assertEqual(value["signer_registry_sha256"], SIGNER_SHA) + self.assertEqual(value["signer_registry_identity_sha256"], + evidence.signer_registry_identity_digest(bound)) + self.assertNotEqual(value["signer_registry_sha256"], + value["signer_registry_identity_sha256"]) + self.keyless_mock.assert_not_called() + + def test_corrupt_registered_outer_signature_never_falls_back(self) -> None: + for digest in (RELEASE_SHA, AUTHORITY_SHA, REVOCATION_SHA): + with self.subTest(digest=digest): + self.replace_bundle(digest, lambda bundle: bundle["dsseEnvelope"]["signatures"][0].update( + sig=base64.b64encode(bytes(64)).decode())) + with self.assertRaises(trust.TrustError): + self.resolve(digest) + self.keyless_mock.assert_not_called() + + def test_profile_issuer_ref_environment_and_state_tampering_refuses(self) -> None: + reference = self.reference(RELEASE_SHA) + bundle_reference = verifier.derive_bundle_reference(reference) + original_pair = verifier.fetch_pair(reference, bundle_reference) + for field, replacement in ( + ("signature_profile", "unsupported-profile"), + ("authority_state_sha256", "sha256:" + "1" * 64), + ("source_issuer", {**original_pair[2]["issuer"], "ref": "refs/heads/other"}), + ("signing_authority", {"ref": "refs/heads/other"}), + ("signing_authority", {"environment": "unregistered-environment"}), + ): + with self.subTest(field=field, replacement=replacement): + bundle = json.loads(original_pair[1]) + statement = json.loads(base64.b64decode(bundle["dsseEnvelope"]["payload"])) + if field == "signing_authority": + statement[field].update(replacement) + else: + statement[field] = replacement + bundle["dsseEnvelope"]["payload"] = base64.b64encode(canonical(statement)).decode() + pair = (original_pair[0], canonical(bundle), *original_pair[2:]) + with self.assertRaises(trust.TrustError): + verifier.verify_registered_signature(reference, bundle_reference, pair, + policy=self.policy) + self.keyless_mock.assert_not_called() + + def test_corrupt_embedded_authority_signature_refuses_before_outer_validation(self) -> None: + original_fetch_pair = verifier.fetch_pair + + def corrupt(reference, bundle_reference): + pair = original_fetch_pair(reference, bundle_reference) + if reference["kind"] == "qualification-authority-state-receipt": + value = copy.deepcopy(pair[2]) + value["signature"] = base64.b64encode(bytes(64)).decode() + return canonical(value), pair[1], value, pair[3], pair[4] + return pair + + with mock.patch.object(verifier, "fetch_pair", side_effect=corrupt), \ + mock.patch.object(verifier, "_verify_software_pair") as outer: + with self.assertRaises(trust.TrustError): + self.resolve() + outer.assert_not_called() + self.keyless_mock.assert_not_called() + + def test_storage_current_revoked_outer_key_refuses(self) -> None: + original_fetch_pair = verifier.fetch_pair + + def revoked(reference, bundle_reference): + pair = original_fetch_pair(reference, bundle_reference) + current = copy.deepcopy(pair[4]) + for signer in current["signers"]: + if signer.get("key_origin") == "software": + signer.update(status="revoked", revoked_at="2026-09-03T00:00:00Z") + return *pair[:4], current + + with mock.patch.object(verifier, "fetch_pair", side_effect=revoked): + with self.assertRaises(trust.TrustError): + self.resolve() + self.keyless_mock.assert_not_called() + + def test_authority_rollover_selects_its_verified_historical_reverse_link(self) -> None: + authority_ref = self.reference(AUTHORITY_SHA) + authority_bundle = verifier.derive_bundle_reference(authority_ref) + authority_pair = (authority_ref, authority_bundle, + verifier.fetch_pair(authority_ref, authority_bundle)) + revocation_ref = self.reference(REVOCATION_SHA) + revocation_bundle = verifier.derive_bundle_reference(revocation_ref) + original = (revocation_ref, revocation_bundle, + verifier.fetch_pair(revocation_ref, revocation_bundle)) + newer = copy.deepcopy(original) + newer[2][2]["revocation_state_sha256"] = "sha256:" + "f" * 64 + newer[2][2]["revision"] = 2 + verified_candidates = [] + + def verify_outer(reference, _bundle, _pair, *, authority, revocation, now): + if reference["kind"] == "qualification-authority-state-receipt" and ( + revocation["revocation_state_sha256"] + != original[2][2]["revocation_state_sha256"] + ): + raise trust.TrustError("authority keeps its original signed reverse link") + + # This isolates reverse-link selection. The retained-byte test above + # exercises the real embedded and outer signature primitives. + with mock.patch.object(verifier, "_registered_state_pairs", return_value=[original, newer]), \ + mock.patch.object(verifier, "_verify_embedded_state", + side_effect=lambda candidate, **kwargs: verified_candidates.append(candidate)), \ + mock.patch.object(verifier, "_validate_registered_state_links"), \ + mock.patch.object(verifier, "_verify_software_pair", side_effect=verify_outer): + chosen = verifier._verify_authority_pair( + authority_pair, bound_registry=authority_pair[2][3], + current_registry=authority_pair[2][4], now=datetime.now(timezone.utc), + ) + self.assertEqual(chosen, original) + self.assertEqual(verified_candidates, [original, newer]) + + def test_authority_outer_checks_registry_window_at_statement_time(self) -> None: + reference = self.reference(AUTHORITY_SHA) + bundle_reference = verifier.derive_bundle_reference(reference) + pair = verifier.fetch_pair(reference, bundle_reference) + bundle = json.loads(pair[1]) + statement = json.loads(base64.b64decode(bundle["dsseEnvelope"]["payload"])) + statement.update(issued_at="2020-01-01T00:00:00Z", not_before="2020-01-01T00:00:00Z") + bundle["dsseEnvelope"]["payload"] = base64.b64encode(canonical(statement)).decode() + changed = (pair[0], canonical(bundle), *pair[2:]) + revocation_ref = self.reference(REVOCATION_SHA) + revocation = json.loads(self.files[revocation_ref["object_path"]]) + with self.assertRaisesRegex(trust.TrustError, "registry is not active at statement time"): + verifier._verify_software_pair( + reference, bundle_reference, changed, authority=pair[2], + revocation=revocation, now=datetime.now(timezone.utc), + ) + + def test_non_public_bundle_retains_keyless_verifier(self) -> None: + reference = self.reference(RELEASE_SHA) + bundle_reference = verifier.derive_bundle_reference(reference) + pair = verifier.fetch_pair(reference, bundle_reference) + changed = (pair[0], b"{}", *pair[2:]) + verifier.verify_registered_signature(reference, bundle_reference, changed, policy=self.policy) + self.keyless_mock.assert_called_once_with( + pair[0], b"{}", kind=reference["kind"], object_value=pair[2], policy=self.policy, + ) + + def test_keyless_cli_version_pin_is_unchanged(self) -> None: + self.keyless.stop() + self.addCleanup(lambda: None) + reference = self.reference(RELEASE_SHA) + value = json.loads(self.files[reference["object_path"]]) + result = mock.Mock(stdout="gh version 99.0.0 (2099-01-01)\n") + with mock.patch.object(verifier.subprocess, "run", return_value=result): + with self.assertRaisesRegex(trust.TrustError, "GitHub CLI version differs"): + verifier.verify_sigstore(canonical(value), b"{}", kind=reference["kind"], + object_value=value, policy=self.policy) + + +if __name__ == "__main__": + unittest.main()