diff --git a/local-candidates/cloud-measured/README.md b/local-candidates/cloud-measured/README.md new file mode 100644 index 0000000..d36c199 --- /dev/null +++ b/local-candidates/cloud-measured/README.md @@ -0,0 +1,83 @@ +# Measured Cloud admission + +`issue.py` prepares the existing five admission objects from one hosted campaign +and its exact deployment. Run it from the reviewed canonical checkout with the +GitHub CLI version pinned in `production-evidence-policy.json`. Preparation is +unsigned by default. + +```sh +python local-candidates/cloud-measured/issue.py \ + --mapping /secure/evidence/admission-mapping.json \ + --mapping-sha256 sha256: \ + --phase-request /secure/evidence/receipt-request.json \ + --output /secure/evidence/receipt-unsigned-plan.json +``` + +The mapping uses `openadapt.measured-cloud-admission-mapping/v1` and has exactly +`candidate`, `derivative`, `provenance`, `retained_files`, +`publication_staging`, `publication_observation`, and `schema_version`. +Each file reference contains +`path`, `sha256` (with the `sha256:` prefix), and `size_bytes`. Paths are relative +to the mapping's directory. Keep the mapping above its retained files; traversal +and symlinks are refused. + +The candidate uses `openadapt.measured-cloud-release-candidate/v1` with +`target: cloud`, `state: ready-for-review`, the canonical `release` and +`artifact_inventory`, and `proposed_release_identity` from the current Cloud +ledger. It supplies no trial totals. The only release artifact is the exact +signed deployment manifest. Its `deployment_id` is the completed protected +deployment workflow's run ID; the manifest also binds the run attempt, source, +and provider identities. Package `version` and `tag` are null. The staging tag +`v0.0.0-deployment.` is schema metadata. Don't create or look up a Git or +PyPI release tag for it. + +The fixed Internal workflow attests the closed derivative's exact bytes using +standard GitHub SLSA provenance. `provenance` references its `attestation`, +`workflow_source`, `workflow_run`, and `protected_main`. The verifier checks the +subject, certificate, repository IDs, source, workflow, run and attempt, then +compares retained records with GitHub. This authenticates evidence derivation. +The current canonical authority still controls admission issuance. + +Keep the raw evidence in private storage. The derivative binds all 18 receipt +opening files and every referenced native phase and observer record. The private +producer verifies the original contracts, signatures, dispatch and effect +semantics before it emits the derivative. Contract-opening wrappers identify +their semantic components; their file hashes must not be substituted for native +contract identities. The public adapter opens retained bytes and recomputes all +17 campaign counters from the closed groups. Cell and trial totals are derived. +All six canonical classes must pass; each declared cell needs three trials. + +Required named raw roles include `runtime_build_identity`, `evidence_identity`, +`runtime_version`, `deployment_readback`, `deployment_workflow_run`, and +`deployment_workflow_source`. The completed deployment run response is collected +after the deploy job. Its in-job provider readback cannot prove workflow +completion. The native bundle digest, sealed archive hash, runtime-build domain +hash, component-manifest domain hash, and runtime-version canonical JSON hash +remain separate. The adapter checks those bindings before it prepares an issuer +request. + +For the original observation, set `publication_observation` to null and use +the original provider readback time. A later observation uses a separate +`{artifact, provenance, provider_readback}` selection. Its attested +`openadapt.hosted-publication-observation/v1` metadata binds the original +derivative hash, the same subject, and fresh provider bytes. The observer time +must fall within its own authenticated workflow attempt. Staging uses that time +at whole-second precision. Preserve the original campaign and provenance files; +a refresh neither changes an activation nor issues another workflow admission. + +Use the existing receipt, workflow, manifest, summary and release request shapes +in the [shared issuer](../flow-1.35.1-measured/issue.py). Each next phase names +object/bundle pairs at their actual containing commits. The immediate receipt +reference for workflow issuance and summary reference for release issuance must +already be on protected main. Issuer source commits always name actual reviewed +main. Register one pair per registry revision and preserve referenced commits +through the merge. Collect fresh staging before the acceptance manifest, then +finish summary and release issuance within its unchanged observation window. + +After review of the exact unsigned plan, use the same command with `--sign`, +`--reviewed-plan-sha256`, and the existing permanent `--state-dir`. Signing uses +the current Keychain key and authority. Keep that directory across every phase +and attempt. If a result is unknown, use `--reconcile-journal` with the same state +directory and a new output path. Don't delete state or retry an unknown effect. +`--stage-registry` explicitly stages an append after signing; this script never +commits, pushes, merges or deploys. diff --git a/local-candidates/cloud-measured/issue.py b/local-candidates/cloud-measured/issue.py new file mode 100755 index 0000000..01ea412 --- /dev/null +++ b/local-candidates/cloud-measured/issue.py @@ -0,0 +1,1259 @@ +#!/usr/bin/env python3 +"""Verify measured hosted evidence before using the existing admission issuer. + +The fixed producer identity authenticates evidence derivation, not admission +issuance. Raw evidence and oracle recipes stay in private retained storage. +""" + +from __future__ import annotations + +import base64 +import importlib.util +import json +import math +from pathlib import Path + +ROOT = Path(__file__).resolve().parents[2] +_SPEC = importlib.util.spec_from_file_location( + "measured_cloud_shared", ROOT / "local-candidates/flow-1.35.1-measured/issue.py" +) +assert _SPEC and _SPEC.loader +shared = importlib.util.module_from_spec(_SPEC) +_SPEC.loader.exec_module(shared) +trust = shared.trust +PRODUCER_REPOSITORY = "OpenAdaptAI/openadapt-internal" +PRODUCER_REPOSITORY_ID = "1170060695" +PRODUCER_OWNER_ID = "132681217" +PRODUCER_WORKFLOW = ".github/workflows/production-qualification-admission.yml" +PRODUCER_REF = "refs/heads/main" +PRODUCER_IDENTITY = ( + f"https://github.com/{PRODUCER_REPOSITORY}/{PRODUCER_WORKFLOW}@{PRODUCER_REF}" +) +PRODUCER_FIELDS = { + "repository", + "repository_id", + "repository_owner_id", + "workflow", + "ref", + "source_commit", + "run_id", + "run_attempt", +} + + +def strict_json(raw: bytes): + """Reject alternate interpretations of hash-bound JSON bytes.""" + + def pairs(values): + result = {} + for key, value in values: + if key in result: + shared.fail("duplicate JSON key in retained evidence") + result[key] = value + return result + + def invalid_constant(_value): + shared.fail("non-finite JSON value in retained evidence") + + def finite_float(value): + result = float(value) + if not math.isfinite(result): + shared.fail("non-finite JSON value in retained evidence") + return result + + return json.loads( + raw, + object_pairs_hook=pairs, + parse_constant=invalid_constant, + parse_float=finite_float, + ) + + +def checked_json(owner: Path, reference: dict): + path, raw = shared.checked_file(owner, reference) + return path, strict_json(raw) + + +def validate_producer(value: dict) -> dict: + trust.closed(value, PRODUCER_FIELDS, "hosted evidence producer") + if any( + value[key] != expected + for key, expected in { + "repository": PRODUCER_REPOSITORY, + "repository_id": PRODUCER_REPOSITORY_ID, + "repository_owner_id": PRODUCER_OWNER_ID, + "workflow": PRODUCER_WORKFLOW, + "ref": PRODUCER_REF, + }.items() + ): + shared.fail("hosted evidence producer differs from the fixed workflow") + if not isinstance(value["source_commit"], str) or not trust.HEX40.fullmatch( + value["source_commit"] + ): + shared.fail("hosted producer source must be exact") + for key in ("run_id", "run_attempt"): + trust.require_decimal_id(value[key], f"producer {key}") + return value + + +def validate_provenance_result(result: dict, producer: dict) -> None: + """Check additional fields only in an already verified SLSA statement.""" + validate_producer(producer) + predicate = result["statement"]["predicate"] + definition = predicate["buildDefinition"] + expected_workflow = { + "repository": f"https://github.com/{PRODUCER_REPOSITORY}", + "path": PRODUCER_WORKFLOW, + "ref": PRODUCER_REF, + } + if ( + definition["buildType"] != "https://actions.github.io/buildtypes/workflow/v1" + or definition["externalParameters"] != {"workflow": expected_workflow} + or definition["internalParameters"] + != { + "github": { + "event_name": "workflow_dispatch", + "repository_id": PRODUCER_REPOSITORY_ID, + "repository_owner_id": PRODUCER_OWNER_ID, + "runner_environment": "github-hosted", + } + } + or predicate["runDetails"]["builder"] != {"id": PRODUCER_IDENTITY} + or predicate["runDetails"]["metadata"]["invocationId"] + != ( + f"https://github.com/{PRODUCER_REPOSITORY}/actions/runs/" + f"{producer['run_id']}/attempts/{producer['run_attempt']}" + ) + ): + shared.fail("verified hosted provenance workflow or invocation differs") + + +def validate_producer_run(run: dict, producer: dict) -> None: + validate_producer(producer) + if ( + type(run.get("id")) is not int + or str(run["id"]) != producer["run_id"] + or type(run.get("run_attempt")) is not int + or str(run["run_attempt"]) != producer["run_attempt"] + or run.get("head_sha") != producer["source_commit"] + or run.get("head_branch") != "main" + or run.get("path") != PRODUCER_WORKFLOW + or run.get("event") != "workflow_dispatch" + or run.get("status") != "completed" + or run.get("conclusion") != "success" + or run.get("repository", {}).get("full_name") != PRODUCER_REPOSITORY + or type(run.get("repository", {}).get("id")) is not int + or str(run["repository"]["id"]) != PRODUCER_REPOSITORY_ID + or type(run.get("repository", {}).get("owner", {}).get("id")) is not int + or str(run["repository"]["owner"]["id"]) != PRODUCER_OWNER_ID + ): + shared.fail("actual hosted producer run or attempt differs") + + +def verify_derivative_provenance( + owner: Path, + derivative_raw: bytes, + producer: dict, + references: dict, + *, + gh=shared.gh, +) -> dict: + """Use fixed-workflow provenance and actual retained and live source readback.""" + validate_producer(producer) + trust.closed( + references, + { + "attestation", + "workflow_source", + "workflow_run", + "protected_main", + }, + "hosted derivative provenance references", + ) + _, bundle_raw = shared.checked_file(owner, references["attestation"]) + _, workflow_raw = shared.checked_file(owner, references["workflow_source"]) + _, retained_run = checked_json(owner, references["workflow_run"]) + _, retained_main = checked_json(owner, references["protected_main"]) + validate_producer_run(retained_run, producer) + if ( + retained_main.get("ref") != PRODUCER_REF + or retained_main.get("object", {}).get("sha") != producer["source_commit"] + ): + shared.fail("retained producer main differs from the reviewed source") + sigstore = strict_json((ROOT / "production-evidence-policy.json").read_bytes())[ + "sigstore" + ] + verified = shared.verifier.verify_github_attestation( + derivative_raw, + bundle_raw, + identity={ + "issuer_repository": PRODUCER_REPOSITORY, + "certificate_identity": PRODUCER_IDENTITY, + }, + issuer_identity={ + "source_commit": producer["source_commit"], + "ref": PRODUCER_REF, + }, + sigstore=sigstore, + ) + validate_provenance_result(verified, producer) + live_main = gh(f"repos/{PRODUCER_REPOSITORY}/git/ref/heads/main") + if live_main.get("object", {}).get("sha") != producer["source_commit"]: + shared.fail("actual producer protected main changed after review") + run = gh( + f"repos/{PRODUCER_REPOSITORY}/actions/runs/{producer['run_id']}" + f"/attempts/{producer['run_attempt']}" + ) + validate_producer_run(run, producer) + source = gh( + f"repos/{PRODUCER_REPOSITORY}/contents/{PRODUCER_WORKFLOW}" + f"?ref={producer['source_commit']}" + ) + if ( + source.get("encoding") != "base64" + or source.get("type") != "file" + or source.get("path") != PRODUCER_WORKFLOW + or base64.b64decode(source["content"].replace("\n", ""), validate=True) + != workflow_raw + ): + shared.fail("actual producer workflow bytes differ from the reviewed file") + return run + + +DERIVATIVE_FIELDS = { + "schema_version", + "target", + "scope", + "producer", + "validator", + "subject", + "campaign_id", + "receipt_commitments", + "raw_evidence_refs", + "groups", +} +SUBJECT_FIELDS = { + "source_commit", + "deployment_manifest_sha256", + "runtime_build_identity_sha256", + "runtime_manifest_sha256", + "environment_digest", + "evidence_identity_sha256", + "bundle_artifact_sha256", + "bundle_content_digest", + "runtime_wheel_sha256", +} +GROUP_FIELDS = { + "trial_id", + "task", + "condition", + "campaign_class", + "ordinal", + "phases", + "observer_before_sha256", + "observer_after_sha256", + "counts", +} +PHASE_FIELDS = { + "phase", + "qualification_run_id_sha256", + "run_report_sha256", + "runner_receipt_sha256", + "input_sha256", + "auxiliary_artifacts", +} +GROUP_COUNT_FIELDS = trust.CAMPAIGN_COUNT_FIELDS - { + "task_condition_cell_count", + "minimum_trials_per_cell", + "observed_trial_count", +} +PHASES = { + "healthy": ({"primary"}, "primary"), + "safe_halt": ({"primary"}, "primary"), + "idempotency_replay": ({"primary", "replay"}, "primary"), + "uncertain_delivery": ({"primary"}, "primary"), + "declared_attended": ( + {"primary", "attended_continuation"}, + "attended_continuation", + ), + "governed_repair": ({"repair_prior", "repair_canary"}, "repair_canary"), +} + + +def raw_digest(value, label: str) -> str: + # The hosted grammar uses plain digests; canonical receipts add their prefix. + if ( + not isinstance(value, str) + or len(value) != 64 + or any(c not in "0123456789abcdef" for c in value) + ): + shared.fail(f"{label} must be a plain lowercase SHA-256 digest") + return value + + +def strict_count(value, label: str, *, minimum=0) -> int: + if type(value) is not int or not minimum <= value <= 9007199254740991: + shared.fail(f"{label} must be an explicit safe integer") + return value + + +def identifier(value, label: str) -> str: + if not isinstance(value, str) or not value or len(value) > 128: + shared.fail(f"{label} must be explicit and bounded") + return value + + +def uuid_identifier(value, label: str) -> str: + from uuid import UUID + + if not isinstance(value, str) or str(UUID(value)) != value: + shared.fail(f"{label} must be a canonical UUID") + return value + + +def evidence_references(value: list) -> list[dict]: + if not isinstance(value, list): + shared.fail("evidence references must be an explicit list") + keys = [] + for item in value: + trust.closed(item, {"role", "sha256"}, "raw evidence reference") + keys.append( + ( + identifier(item["role"], "evidence role"), + raw_digest(item["sha256"], "evidence hash"), + ) + ) + if keys != sorted(set(keys)): + shared.fail("evidence references must be sorted and unique") + return value + + +def retained_inventory(owner: Path, references: list) -> dict[str, tuple[Path, bytes]]: + if not isinstance(references, list) or not references: + shared.fail("retained evidence inventory must be nonempty") + result, paths = {}, set() + for reference in references: + trust.closed( + reference, {"path", "sha256", "size_bytes"}, "retained evidence file" + ) + path, raw = shared.checked_file(owner, reference) + digest = shared.sha(raw).removeprefix("sha256:") + if digest in result or path in paths: + shared.fail("retained inventory must resolve each digest exactly once") + paths.add(path) + result[digest] = (path, raw) + return result + + +def evidence_json(inventory: dict, digest: str): + return strict_json(inventory[raw_digest(digest, "evidence file")][1]) + + +def verify_derivative(derivative: dict, inventory: dict) -> dict: + """Recompute six-class counts from the authenticated, byte-bound groups. + + The fixed private verifier proves scenario, signature and oracle semantics. + This adapter checks its closed derivative and underlying byte inventory; + it neither executes that verifier nor supplies any empirical rule or recipe. + """ + trust.closed(derivative, DERIVATIVE_FIELDS, "hosted qualification derivative") + if ( + derivative["schema_version"] != "openadapt.hosted-qualification-derivative/v1" + or derivative["target"] != "cloud" + or derivative["scope"] != "hosted-synthetic-qualification" + ): + shared.fail("unsupported hosted derivative target or scope") + validate_producer(derivative["producer"]) + uuid_identifier(derivative["campaign_id"], "campaign id") + subject = trust.closed(derivative["subject"], SUBJECT_FIELDS, "hosted subject") + if not isinstance(subject["source_commit"], str) or not trust.HEX40.fullmatch( + subject["source_commit"] + ): + shared.fail("hosted deployment source must be exact") + for key in SUBJECT_FIELDS - {"source_commit"}: + raw_digest(subject[key], key) + validator = trust.closed( + derivative["validator"], + { + "repository", + "source_commit", + "path", + "sha256", + "source_inventory_sha256", + }, + "private verifier source identity", + ) + if ( + validator["repository"] != trust.TARGET_CONTRACTS["cloud"]["repository"] + or validator["source_commit"] != subject["source_commit"] + or validator["path"] != "runner/qualification_issuer.py" + ): + shared.fail("private verifier does not bind the actual Cloud source") + commitments = trust.closed( + derivative["receipt_commitments"], + shared.FILE_COMMITMENTS, + "actual receipt byte commitments", + ) + for key, value in commitments.items(): + raw_digest(value, key) + if ( + commitments["bundle_sha256"] != subject["bundle_artifact_sha256"] + or commitments["admitted_runtime_sha256"] != subject["runtime_wheel_sha256"] + ): + shared.fail("admitted bundle or runtime differs from actual retained bytes") + refs = evidence_references(derivative["raw_evidence_refs"]) + raw_hashes = {item["sha256"] for item in refs} + if raw_hashes != set(inventory): + shared.fail("retained evidence inventory differs from the attested inventory") + required = set(commitments.values()) | { + subject["deployment_manifest_sha256"], + validator["sha256"], + validator["source_inventory_sha256"], + } + groups = derivative["groups"] + if not isinstance(groups, list) or not groups: + shared.fail("hosted derivative has no actual groups") + grouped, tasks, trial_ids, reports, group_order = {}, set(), set(), set(), [] + for group in groups: + trust.closed(group, GROUP_FIELDS, "hosted trial group") + trial_id = uuid_identifier(group["trial_id"], "trial id") + task = identifier(group["task"], "task") + condition = identifier(group["condition"], "condition") + campaign_class = group["campaign_class"] + if campaign_class not in PHASES or trial_id in trial_ids: + shared.fail("duplicate group or unsupported campaign class") + ordinal = strict_count(group["ordinal"], "trial ordinal", minimum=1) + group_order.append((task, condition, ordinal)) + trial_ids.add(trial_id) + tasks.add(task) + counts = trust.closed( + group["counts"], GROUP_COUNT_FIELDS, "derived per-group counts" + ) + for key, value in counts.items(): + strict_count(value, key) + if campaign_class == "declared_attended" and any( + counts[key] != 1 + for key in ( + "authenticated_bound_decision_count", + "live_target_revalidation_count", + ) + ): + shared.fail( + "each attended group requires its own bound decision and revalidation" + ) + if campaign_class == "governed_repair" and any( + counts[key] != 1 + for key in ( + "policy_approved_repair_count", + "approved_repair_count", + "retained_repair_evidence_count", + "live_target_revalidation_count", + ) + ): + shared.fail( + "each repair group requires its own approval and retained evidence" + ) + phases = group["phases"] + if not isinstance(phases, list): + shared.fail("native phases must be an explicit list") + expected_phases, principal = PHASES[campaign_class] + seen_phases, native = set(), {} + for phase in phases: + trust.closed(phase, PHASE_FIELDS, "native phase reference") + name = phase["phase"] + if name in seen_phases or name not in expected_phases: + shared.fail("missing, duplicate or unsupported native phase") + seen_phases.add(name) + for key in PHASE_FIELDS - {"phase", "auxiliary_artifacts"}: + raw_digest(phase[key], key) + for key in ("run_report_sha256", "runner_receipt_sha256", "input_sha256"): + required.add(phase[key]) + required.update( + item["sha256"] + for item in evidence_references(phase["auxiliary_artifacts"]) + ) + report_hash = phase["run_report_sha256"] + if report_hash in reports: + shared.fail("native report is reused across measured phases") + reports.add(report_hash) + report = evidence_json(inventory, report_hash) + same_bundle = ( + report.get("bundle_content_digest") == subject["bundle_content_digest"] + ) + if same_bundle != (name != "repair_prior"): + shared.fail( + "native phase binds the wrong principal or repair-prior bundle" + ) + if report.get("run_id_sha256") != phase["qualification_run_id_sha256"]: + shared.fail("native phase run differs from its retained report") + native[name] = (phase, report) + if seen_phases != expected_phases: + shared.fail("native phase membership is incomplete") + principal_report = native[principal][1] + if ( + principal_report.get("qualification_evidence_only") is not True + or principal_report.get("production_eligible") is not False + or principal_report.get("run_id_sha256") + != native[principal][0]["qualification_run_id_sha256"] + ): + shared.fail("principal report lacks exact qualification-only run binding") + expected_outcome = { + "safe_halt": "HALTED_BEFORE_EFFECT", + "uncertain_delivery": "RECONCILIATION_REQUIRED", + }.get(campaign_class, "VERIFIED") + if principal_report.get( + "transaction_outcome" + ) != expected_outcome or principal_report.get("success") is not ( + expected_outcome == "VERIFIED" + ): + shared.fail("actual principal outcome does not satisfy its measured class") + if counts["reconciliation_required_count"] != int( + principal_report.get("transaction_outcome") == "RECONCILIATION_REQUIRED" + ): + shared.fail("reconciliation count differs from the actual principal report") + if campaign_class == "idempotency_replay" and ( + native["primary"][0]["input_sha256"] != native["replay"][0]["input_sha256"] + or native["primary"][0]["qualification_run_id_sha256"] + == native["replay"][0]["qualification_run_id_sha256"] + ): + shared.fail( + "replay must retain the same input and distinct native invocation" + ) + if campaign_class == "idempotency_replay" and ( + not isinstance(native["primary"][1].get("idempotency_key"), str) + or not native["primary"][1]["idempotency_key"] + or native["primary"][1]["idempotency_key"] + != native["replay"][1].get("idempotency_key") + or native["replay"][1].get("idempotent_replay") is not True + or native["replay"][1].get("success") is not False + ): + shared.fail( + "actual replay lacks the same native idempotency key or refusal" + ) + if campaign_class == "declared_attended" and ( + native["primary"][0]["qualification_run_id_sha256"] + != native["attended_continuation"][0]["qualification_run_id_sha256"] + ): + shared.fail("attended continuation must bind the same durable run") + for key in ("observer_before_sha256", "observer_after_sha256"): + required.add(raw_digest(group[key], key)) + if group["observer_before_sha256"] == group["observer_after_sha256"]: + shared.fail("one observer envelope cannot prove both before and after") + grouped.setdefault(campaign_class, {}).setdefault((task, condition), []).append( + group + ) + if len(tasks) != 1 or set(grouped) != set(trust.CAMPAIGN_CLASSES): + shared.fail("measured Cloud adapter requires one task and all six classes") + if group_order != sorted(set(group_order)): + shared.fail( + "hosted groups must follow unique frozen task/condition/ordinal order" + ) + if not required <= raw_hashes: + shared.fail("attested inventory omits a referenced evidence or commitment file") + summary = {} + for name, cells in grouped.items(): + rows = [row for values in cells.values() for row in values] + for values in cells.values(): + if [row["ordinal"] for row in values] != list(range(1, len(values) + 1)): + shared.fail("trial cell ordinals are not contiguous") + counts = { + key: sum(row["counts"][key] for row in rows) for key in GROUP_COUNT_FIELDS + } + counts.update( + task_condition_cell_count=len(cells), + minimum_trials_per_cell=min(map(len, cells.values())), + observed_trial_count=len(rows), + ) + for key, value in counts.items(): + strict_count(value, key) + summary[name] = counts + return trust.validate_campaign_summary(summary) + + +def canonical_json(value) -> bytes: + return json.dumps( + value, + sort_keys=True, + separators=(",", ":"), + ensure_ascii=False, + allow_nan=False, + ).encode("utf-8") + + +def semantic_digest(value, domain=b"") -> str: + return shared.sha(domain + canonical_json(value)).removeprefix("sha256:") + + +def role_file(derivative: dict, inventory: dict, role: str) -> tuple[Path, bytes]: + matches = [ + item["sha256"] + for item in derivative["raw_evidence_refs"] + if item["role"] == role + ] + if len(matches) != 1: + shared.fail(f"retained evidence must contain exactly one {role}") + return inventory[matches[0]] + + +def role_json(derivative: dict, inventory: dict, role: str) -> dict: + return strict_json(role_file(derivative, inventory, role)[1]) + + +def verify_subject_openings(derivative: dict, inventory: dict) -> tuple[str, int]: + """Open public identity interfaces; the attested verifier checks private contracts.""" + subject, commitments = derivative["subject"], derivative["receipt_commitments"] + build = role_json(derivative, inventory, "runtime_build_identity") + identity = role_json(derivative, inventory, "evidence_identity") + runtime = role_json(derivative, inventory, "runtime_version") + if ( + build.get("schema_version") != "openadapt.admitted-runtime-build/v1" + or identity.get("schema_version") + != "openadapt.production-acceptance-evidence-identity/v2" + or semantic_digest(build, b"openadapt-admitted-runtime-build-v1\0") + != subject["runtime_build_identity_sha256"] + or semantic_digest( + identity, b"OpenAdapt production acceptance evidence identity v2\0" + ) + != subject["evidence_identity_sha256"] + or semantic_digest(runtime) != subject["runtime_manifest_sha256"] + or identity.get("runtime_build_identity") != build + ): + shared.fail( + "semantic runtime or evidence identity differs from its retained opening" + ) + if ( + build.get("substrate") != "web" + or build.get("substrate_runtime", {}).get("transport") != "browser" + or build.get("managed_browser") + != { + "playwright_version": runtime.get("playwright"), + "browser_base_image": runtime.get("browser_base_image"), + } + or build.get("flow_wheel_sha256") != subject["runtime_wheel_sha256"] + or build.get("flow_version") != runtime.get("openadapt_flow") + or build.get("flow_release_commit") != runtime.get("release_commit") + or build.get("flow_wheel_sha256") != runtime.get("wheel_sha256") + or build.get("runner_build") != runtime.get("runner_build") + or build.get("runner_artifact_sha256") != runtime.get("runner_artifact_sha256") + or build.get("runtime_manifest_sha256") + != semantic_digest( + build.get("runtime_manifest"), b"openadapt-runtime-component-manifest-v1\0" + ) + ): + shared.fail("hosted runtime does not bind its actual Flow and browser build") + for key in ( + "deployment_manifest_sha256", + "bundle_artifact_sha256", + "bundle_content_digest", + "environment_digest", + ): + if identity.get(key) != subject[key]: + shared.fail(f"expanded evidence identity differs at {key}") + if identity.get("campaign_id") != derivative["campaign_id"]: + shared.fail("expanded evidence identity names another campaign") + for commitment, field in ( + ("organization_id_sha256", "tenant_id"), + ("workflow_id_sha256", "workflow_id"), + ("workflow_version_id_sha256", "workflow_version_id"), + ): + opening = evidence_json(inventory, commitments[commitment]) + if ( + uuid_identifier(opening.get("id"), field) != identity.get(field) + or opening.get("admitted_subject") != subject + ): + shared.fail("identity opening differs from the exact hosted subject") + version = evidence_json(inventory, commitments["workflow_version_id_sha256"])[ + "bundle_version" + ] + if ( + not isinstance(version, str) + or len(version) > 64 + or not trust.BUNDLE_VERSION.fullmatch(version) + ): + shared.fail("declared sealed bundle version is invalid") + decision = evidence_json(inventory, commitments["decision_identity_sha256"]) + uuid_identifier(decision.get("id"), "decision identity") + revision = strict_count( + decision.get("decision_revision"), "decision revision", minimum=1 + ) + for name in ("decision_commitment_sha256", "evidence_manifest_readback_sha256"): + opening = evidence_json(inventory, commitments[name]) + if ( + opening.get("evidence_manifest_sha256") + != commitments["evidence_manifest_sha256"] + or opening.get("admitted_subject") != subject + ): + shared.fail( + "decision or readback differs from the actual manifest and subject" + ) + if name == "decision_commitment_sha256" and ( + opening.get("decision_revision") != revision + or opening.get("decision_id") != decision["id"] + ): + shared.fail( + "decision commitment differs from the immutable decision identity" + ) + campaign = evidence_json(inventory, commitments["campaign_artifact_sha256"]) + if ( + campaign.get("schema_version") != "openadapt.qualification-campaign/v2" + or campaign.get("campaign_id") != derivative["campaign_id"] + or campaign.get("evidence_identity_sha256") + != subject["evidence_identity_sha256"] + ): + shared.fail("retained campaign artifact names another hosted subject") + return version, revision + + +DEPLOYMENT_WORKFLOW = ".github/workflows/deploy.yml" +DEPLOYMENT_REPOSITORY = trust.TARGET_CONTRACTS["cloud"]["repository"] + + +def validate_deployment_run(run: dict, authority: dict) -> None: + if ( + type(run.get("id")) is not int + or str(run["id"]) != authority["run_id"] + or type(run.get("run_attempt")) is not int + or str(run["run_attempt"]) != authority["run_attempt"] + or run.get("head_sha") != authority["source_commit"] + or run.get("head_branch") != "main" + or run.get("path") != DEPLOYMENT_WORKFLOW + or run.get("event") != authority["event_name"] + or run.get("status") != "completed" + or run.get("conclusion") != "success" + or run.get("repository", {}).get("full_name") != DEPLOYMENT_REPOSITORY + or type(run.get("repository", {}).get("id")) is not int + or str(run["repository"]["id"]) + != trust.TARGET_CONTRACTS["cloud"]["repository_id"] + or type(run.get("repository", {}).get("owner", {}).get("id")) is not int + or str(run["repository"]["owner"]["id"]) != PRODUCER_OWNER_ID + ): + shared.fail( + "deployment must bind the actual successful protected workflow attempt" + ) + + +def verify_deployment( + derivative: dict, inventory: dict, release: dict, staging: dict, *, gh=shared.gh +) -> None: + """Match attested deployment/provider bytes and the actual protected run. + + The fixed private producer verifies deployment signatures and provider + readback. This consumer opens the committed identities and independently + checks GitHub's completed run and source bytes. It does not execute a + provider recipe or accept a caller's successful-deployment flag. + """ + subject = derivative["subject"] + manifest_raw = inventory[subject["deployment_manifest_sha256"]][1] + manifest = strict_json(manifest_raw) + if ( + manifest.get("schema_version") + != "openadapt.cloud-production-deployment-manifest/v1" + or canonical_json(manifest) != manifest_raw + ): + shared.fail( + "deployment manifest must retain its exact canonical artifact bytes" + ) + authority = manifest["build_authority"] + if any( + authority.get(key) != value + for key, value in { + "repository": DEPLOYMENT_REPOSITORY, + "workflow": DEPLOYMENT_WORKFLOW, + "source_ref": "refs/heads/main", + "source_commit": subject["source_commit"], + "workflow_ref": f"{DEPLOYMENT_REPOSITORY}/{DEPLOYMENT_WORKFLOW}@refs/heads/main", + "certificate_identity": f"https://github.com/{DEPLOYMENT_REPOSITORY}/{DEPLOYMENT_WORKFLOW}@refs/heads/main", + "oidc_issuer": "https://token.actions.githubusercontent.com", + "job": "deploy", + "environment": "production", + "environment_scope": "openadapt-cloud-production-v1", + }.items() + ) or authority.get("event_name") not in {"workflow_run", "workflow_dispatch"}: + shared.fail( + "deployment build authority differs from the fixed protected source" + ) + for key in ("run_id", "run_attempt"): + trust.require_decimal_id(authority.get(key), f"deployment {key}") + if ( + manifest["source"] + != {"repository": DEPLOYMENT_REPOSITORY, "commit": subject["source_commit"]} + or release["source_commit"] != subject["source_commit"] + or release["deployment_id"] != authority["run_id"] + or release["deployment_sha256"] + != "sha256:" + subject["deployment_manifest_sha256"] + or manifest["target"]["environment_digest"] + != "sha256:" + subject["environment_digest"] + ): + shared.fail("deployment release, source or environment identity differs") + runtime = role_json(derivative, inventory, "runtime_version") + expected_runtime = { + "runtime_manifest_sha256": "sha256:" + subject["runtime_manifest_sha256"], + "runner_source_artifact_sha256": "sha256:" + runtime["runner_artifact_sha256"], + "runner_build": runtime["runner_build"], + "modal_sdk_version": runtime["modal_sdk"], + "fastapi_version": runtime["fastapi"], + "starlette_version": runtime["starlette"], + "sandbox_network_policy": runtime["sandbox_network_policy"], + "flow": { + "version": runtime["openadapt_flow"], + "release_commit": runtime["release_commit"], + "wheel_url": runtime["wheel_url"], + "wheel_sha256": "sha256:" + runtime["wheel_sha256"], + "sdist_sha256": "sha256:" + runtime["sdist_sha256"], + }, + "browser": { + "playwright_version": runtime["playwright"], + "browser_base_image": runtime["browser_base_image"], + "runtime_contract_sha256": "sha256:" + + semantic_digest( + { + "python_base_image": runtime["browser_base_image"], + "playwright_version": runtime["playwright"], + "browser_install_command": "python -m playwright install --with-deps chromium", + }, + b"OpenAdapt managed browser image contract v1\0", + ), + }, + } + if manifest["runtime"] != expected_runtime: + shared.fail("deployment runtime differs from the measured build and wheel") + readback = role_json(derivative, inventory, "deployment_readback") + if ( + readback.get("schema_version") + != "openadapt.cloud-production-deployment-readback/v1" + or readback.get("source_commit") != subject["source_commit"] + or readback.get("manifest_sha256") != release["deployment_sha256"] + or readback.get("manifest_bytes_sha256") != release["deployment_sha256"] + or readback.get("target_attestation_sha256") != manifest["target"]["sha256"] + or readback.get("admission_activated") is not False + ): + shared.fail("provider readback differs from the actual deployment manifest") + observed = readback["provider_observation"] + github = observed["github"] + if any( + github.get(key) != authority[other] + for key, other in ( + ("run_id", "run_id"), + ("run_attempt", "run_attempt"), + ("source_commit", "source_commit"), + ("event", "event_name"), + ) + ): + shared.fail("in-job readback differs from the final deployment invocation") + host, runner = manifest["deployment"]["host"], manifest["deployment"]["runner"] + if ( + any( + observed["host"].get(key) != host[key] + for key in ( + "deploy_id", + "immutable_url", + "created_at", + "published_at", + ) + ) + or "sha256:" + + semantic_digest( + observed["host"].get("site_id"), + b"OpenAdapt Netlify production site identity v1\0", + ) + != host["site_identity_sha256"] + ): + shared.fail("actual host provider identity differs from the signed manifest") + if any( + observed["modal"].get(key) != runner[key] + for key in ( + "environment", + "app_id", + "app_name", + "app_version", + "deployed_at", + ) + ) or ( + observed["modal"].get("source_commit") != subject["source_commit"] + or observed["modal"].get("source_dirty") is not False + or observed["modal"].get("sdk_version") != runtime["modal_sdk"] + or "sha256:" + + semantic_digest( + observed["modal"].get("endpoint_origin"), + b"OpenAdapt Modal runner endpoint origin v1\0", + ) + != runner["endpoint_origin_sha256"] + ): + shared.fail("actual runner provider identity differs from the signed manifest") + context = readback["observed_runtime_context"] + if ( + context.get("deployment_manifest_sha256") + != subject["deployment_manifest_sha256"] + or context.get("runtime_build_identity") + != role_json(derivative, inventory, "runtime_build_identity") + or observed.get("environment_contract_sha256") + != host["environment_contract_sha256"] + ): + shared.fail("provider runtime context differs from the measured deployment") + trust.validate_staging(staging) + if ( + staging["publication_mode"] != "already-published-deployment" + or staging["repository"] != DEPLOYMENT_REPOSITORY + or staging["repository_id"] != trust.TARGET_CONTRACTS["cloud"]["repository_id"] + or staging["target_commitish"] != subject["source_commit"] + or staging["deployment_id"] != authority["run_id"] + or staging["tag"] != f"v0.0.0-deployment.{authority['run_id']}" + or staging["deployment_url"] != host["immutable_url"] + or staging["assets"] + != [ + { + **release["artifacts"][0], + "asset_id": None, + "uploader_id": None, + "uploader_login": None, + } + ] + ): + shared.fail("publication staging differs from the actual deployed artifact") + retained_run = role_json(derivative, inventory, "deployment_workflow_run") + validate_deployment_run(retained_run, authority) + validate_deployment_run( + gh( + f"repos/{DEPLOYMENT_REPOSITORY}/actions/runs/{authority['run_id']}" + f"/attempts/{authority['run_attempt']}" + ), + authority, + ) + for path, raw in ( + ( + DEPLOYMENT_WORKFLOW, + role_file(derivative, inventory, "deployment_workflow_source")[1], + ), + ( + "runner/runtime-version.json", + role_file(derivative, inventory, "runtime_version")[1], + ), + ( + derivative["validator"]["path"], + inventory[derivative["validator"]["sha256"]][1], + ), + ): + source = gh( + f"repos/{DEPLOYMENT_REPOSITORY}/contents/{path}?ref={subject['source_commit']}" + ) + if ( + source.get("encoding") != "base64" + or source.get("type") != "file" + or source.get("path") != path + or base64.b64decode(source["content"].replace("\n", ""), validate=True) + != raw + ): + shared.fail("actual deployed source bytes differ from retained evidence") + if ( + path == DEPLOYMENT_WORKFLOW + and shared.sha(raw) != authority["workflow_sha256"] + ): + shared.fail( + "deployment workflow source hash differs from its signed authority" + ) + + +PUBLICATION_OBSERVATION_FIELDS = { + "schema_version", + "producer", + "qualification_derivative_sha256", + "subject", + "observed_at", + "provider_readback_sha256", +} + + +def observation_time(value: str): + from datetime import datetime, timezone + + if not isinstance(value, str): + shared.fail("provider observation time must be actual UTC milliseconds") + try: + moment = datetime.fromisoformat(value.replace("Z", "+00:00")) + except ValueError: + shared.fail("provider observation time must be actual UTC milliseconds") + if ( + moment.tzinfo != timezone.utc + or moment.isoformat(timespec="milliseconds").replace("+00:00", "Z") != value + ): + shared.fail("provider observation time must be actual UTC milliseconds") + return moment + + +def verify_publication_observation( + owner: Path, + reference: dict | None, + derivative: dict, + derivative_raw: bytes, + inventory: dict, + staging: dict, + *, + gh=shared.gh, +) -> None: + """Bind staging time to original evidence or a separately attested fresh read. + + The refresh authenticates a read-only observation through the fixed producer. + It leaves the original campaign, derivative and activation byte commitments + intact. Original evidence can support only its original observation time. + """ + original = role_json(derivative, inventory, "deployment_readback") + if reference is None: + observed = observation_time(original["observed_at"]) + else: + trust.closed( + reference, + {"artifact", "provenance", "provider_readback"}, + "publication observation references", + ) + _, proof_raw = shared.checked_file(owner, reference["artifact"]) + proof = trust.closed( + strict_json(proof_raw), + PUBLICATION_OBSERVATION_FIELDS, + "publication observation", + ) + _, raw = shared.checked_file(owner, reference["provider_readback"]) + readback = trust.closed( + strict_json(raw), + { + "observed_at", + "app_id", + "app_version", + "source_commit", + "netlify_deploy_id", + "runtime_boundary_id", + "deployment_manifest_sha256", + "control_image_id", + "sandbox_image_id", + "provider_observation", + "health", + }, + "fresh provider observation", + ) + if ( + proof["schema_version"] != "openadapt.hosted-publication-observation/v1" + or proof["qualification_derivative_sha256"] + != shared.sha(derivative_raw)[7:] + or proof["subject"] != derivative["subject"] + or proof["provider_readback_sha256"] != shared.sha(raw)[7:] + or proof["observed_at"] != readback["observed_at"] + ): + shared.fail( + "fresh observation does not bind the original derivative and actual readback" + ) + observed = observation_time(readback["observed_at"]) + run = verify_derivative_provenance( + owner, + proof_raw, + proof["producer"], + reference["provenance"], + gh=gh, + ) + started = trust.require_timestamp(run["run_started_at"], "producer run start") + completed = trust.require_timestamp( + run["updated_at"], "producer run completion" + ) + # GitHub's API bounds have second precision. Compare the observation's + # containing second while preserving its exact milliseconds in evidence. + if not started <= observed.replace(microsecond=0) <= completed: + shared.fail( + "fresh observation time falls outside its authenticated workflow attempt" + ) + subject = derivative["subject"] + prior = original["provider_observation"] + context = original["observed_runtime_context"] + expected = { + "app_id": prior["modal"]["app_id"], + "app_version": prior["modal"]["app_version"], + "source_commit": subject["source_commit"], + "netlify_deploy_id": prior["host"]["deploy_id"], + "deployment_manifest_sha256": subject["deployment_manifest_sha256"], + "control_image_id": context["control_image_id"], + "sandbox_image_id": context["modal_image_id"], + "provider_observation": {key: prior[key] for key in ("host", "modal")}, + } + if any(readback[key] != value for key, value in expected.items()): + shared.fail( + "fresh provider, source or image differs from the qualified deployment" + ) + runtime = role_json(derivative, inventory, "runtime_version") + health = readback["health"] + health_expected = { + "ready": True, + "service": "runner", + "mode": "live", + "boundary_id": readback["runtime_boundary_id"], + "flow_version": runtime["openadapt_flow"], + "modal_sdk": runtime["modal_sdk"], + "fastapi": runtime["fastapi"], + "starlette": runtime["starlette"], + "runner_build": runtime["runner_build"], + "runner_artifact_sha256": runtime["runner_artifact_sha256"], + "sandbox_network_policy": runtime["sandbox_network_policy"], + "deployment_manifest_sha256": subject["deployment_manifest_sha256"], + "runtime_build_identity": role_json( + derivative, inventory, "runtime_build_identity" + ), + "runtime_environment_sha256": context["runtime_environment_sha256"], + } + if ( + health.get("ready") is not True + or context["runtime_environment_sha256"] + != health_expected["runtime_build_identity"]["substrate_runtime"][ + "runtime_boundary_sha256" + ] + or any(health.get(key) != value for key, value in health_expected.items()) + or semantic_digest({"environment": readback["runtime_boundary_id"]}) + != context["runtime_environment_sha256"] + ): + shared.fail( + "fresh runner health or runtime differs from the qualified deployment" + ) + functions = original.get("function_readbacks") + if ( + not isinstance(functions, list) + or len(functions) != 3 + or {row.get("function_tag") for row in functions if isinstance(row, dict)} + != {"enqueue", "run_flow", "run_teach"} + or health.get("function_readbacks") != functions + ): + shared.fail( + "fresh runner function inventory differs from the qualified deployment" + ) + function = health.get("deployment_readback", {}) + expected_function = { + "app_id": prior["modal"]["app_id"], + "app_version": prior["modal"]["app_version"], + "function_id": prior["modal"]["function_id"], + "function_definition_id": prior["modal"]["function_definition_id"], + "control_image_id": context["control_image_id"], + "sandbox_image_id": context["modal_image_id"], + } + if any(function.get(key) != value for key, value in expected_function.items()): + shared.fail( + "fresh runner function or image differs from the qualified deployment" + ) + if staging["observed_at"] != observed.strftime("%Y-%m-%dT%H:%M:%SZ"): + shared.fail("staging time differs from its authenticated provider observation") + + +def prepare_inputs(mapping_path: Path, mapping_sha256: str, *, gh=shared.gh) -> dict: + raw = mapping_path.read_bytes() + if shared.sha(raw) != mapping_sha256: + shared.fail("Cloud mapping changed after review") + mapping = trust.closed( + strict_json(raw), + { + "schema_version", + "candidate", + "derivative", + "provenance", + "retained_files", + "publication_staging", + "publication_observation", + }, + "measured Cloud input mapping", + ) + if mapping["schema_version"] != "openadapt.measured-cloud-admission-mapping/v1": + shared.fail("unsupported Cloud input mapping") + candidate_path, candidate = checked_json(mapping_path, mapping["candidate"]) + trust.closed( + candidate, + { + "schema_version", + "state", + "target", + "release", + "artifact_inventory", + "proposed_release_identity", + }, + "measured Cloud candidate", + ) + if ( + candidate["schema_version"] != "openadapt.measured-cloud-release-candidate/v1" + or candidate["state"] != "ready-for-review" + or candidate["target"] != "cloud" + ): + shared.fail("unsupported measured Cloud candidate target or state") + inputs = { + "release": candidate["release"], + "artifact_inventory": candidate["artifact_inventory"], + } + if shared.prepared_target(inputs) != ("cloud", "production_cloud"): + shared.fail("Cloud candidate inventory must select only the Cloud deployment") + release = trust.closed( + inputs["release"], + { + "schema_version", + "kind", + "source_repository", + "source_repository_id", + "source_commit", + "version", + "tag", + "deployment_id", + "deployment_sha256", + "artifacts", + }, + "Cloud release candidate", + ) + if ( + release["version"] is not None + or release["tag"] is not None + or len(release["artifacts"]) != 1 + ): + shared.fail("Cloud deployment cannot imply a package version or release tag") + _, derivative_raw = shared.checked_file(mapping_path, mapping["derivative"]) + derivative = strict_json(derivative_raw) + inventory = retained_inventory(mapping_path, mapping["retained_files"]) + summary = verify_derivative(derivative, inventory) + artifact = release["artifacts"][0] + manifest_raw = inventory[derivative["subject"]["deployment_manifest_sha256"]][1] + if artifact["sha256"] != shared.sha(manifest_raw) or artifact["size_bytes"] != len( + manifest_raw + ): + shared.fail( + "Cloud artifact inventory differs from the actual signed manifest bytes" + ) + bundle_version, revision = verify_subject_openings(derivative, inventory) + # Provenance is required before any closed group can become issuer input. + verify_derivative_provenance( + mapping_path, + derivative_raw, + derivative["producer"], + mapping["provenance"], + gh=gh, + ) + _, staging = checked_json(mapping_path, mapping["publication_staging"]) + verify_deployment(derivative, inventory, release, staging, gh=gh) + verify_publication_observation( + mapping_path, + mapping["publication_observation"], + derivative, + derivative_raw, + inventory, + staging, + gh=gh, + ) + inputs.update( + mapping_sha256=mapping_sha256, + candidate_sha256=shared.sha(candidate_path.read_bytes()), + derivative_sha256=shared.sha(derivative_raw), + commitments={ + key: "sha256:" + value + for key, value in derivative["receipt_commitments"].items() + }, + campaign_summary=summary, + bundle_version=bundle_version, + decision_revision=revision, + release_identity=candidate["proposed_release_identity"], + publication_staging=staging, + ) + return inputs + + +def main(argv=None) -> int: + return shared.main(argv, input_preparer=prepare_inputs, expected_target="cloud") + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/local-candidates/flow-1.35.1-measured/issue.py b/local-candidates/flow-1.35.1-measured/issue.py index ea0273f..305079b 100755 --- a/local-candidates/flow-1.35.1-measured/issue.py +++ b/local-candidates/flow-1.35.1-measured/issue.py @@ -687,10 +687,38 @@ def current_context(source_commit: str, now: datetime) -> dict: } +def prepared_target(inputs: dict) -> tuple[str, str]: + """Derive the supported target from the verified inventory and release. + + This shared phase machinery has two explicit input preparers. A caller + cannot select another product by adding a target string to a phase request. + Each preparer verifies its own retained measurements and publication bytes. + """ + inventory = trust.validate_artifact_inventory(inputs["artifact_inventory"]) + target = inventory["target"] + if target not in {"flow", "cloud"}: + fail("measured adapter does not support this target") + contract = trust.TARGET_CONTRACTS[target] + release = inputs["release"] + if ( + inventory["claim_scope"] != contract["claim_scope"] + or release["schema_version"] != "openadapt.production-release-candidate/v1" + or release["kind"] != contract["release_kind"] + or release["source_repository"] != contract["repository"] + or release["source_repository_id"] != contract["repository_id"] + or not isinstance(release["source_commit"], str) + or trust.HEX40.fullmatch(release["source_commit"]) is None + or release["artifacts"] != inventory["artifacts"] + ): + fail("measured release differs from the validated target inventory") + return target, contract["claim_scope"] + + def check_release_identity(inputs: dict, source: str) -> None: ledger = json.loads( verifier.fetch(verifier.raw_url(source, "production-lifecycle-admissions.json")) ) + target, _ = prepared_target(inputs) previous = [] for ref in ledger["admissions"]: value = verifier.verify_bytes( @@ -698,10 +726,10 @@ def check_release_identity(inputs: dict, source: str) -> None: ref, "ledger admission", ) - if value.get("target") == "flow": + if value.get("target") == target: previous.append(value) if not previous: - fail("current Flow admission history is missing") + fail("current measured target admission history is missing") last = max(previous, key=lambda value: value["release_identity"]["sequence"]) expected = { "schema_version": "openadapt.monotonic-production-release/v1", @@ -762,6 +790,7 @@ def issuer_identity(source: str, phase: str) -> dict: def phase_object( inputs: dict, request: dict, context: dict, *, consumer=None ) -> tuple[dict, dict | None]: + target, claim_scope = prepared_target(inputs) phase = request["phase"] source = request["issuer_source_commit"] now = trust.require_timestamp(request["issued_at"], "phase issued_at") @@ -915,9 +944,9 @@ def phase_object( fail("acceptance issuer must be actual reviewed evals protected main") value = { "schema_version": "openadapt.production-acceptance/v3", - "target": "flow", + "target": target, "verdict": "accepted", - "claim_scope": "production_flow", + "claim_scope": claim_scope, **{ k: context[k] for k in ("acceptance_policy_sha256", "lifecycle_policy_sha256") @@ -927,8 +956,8 @@ def phase_object( "release_sha256": trust.digest_bytes( trust.RELEASE_DOMAIN, { - "target": "flow", - "claim_scope": "production_flow", + "target": target, + "claim_scope": claim_scope, "release": inputs["release"], }, ), @@ -1179,7 +1208,7 @@ def validate_staging_registry(path: Path, source: str, context: dict) -> None: fail("staging registry does not append to the verified current trust state") -def main(argv=None) -> int: +def main(argv=None, *, input_preparer=None, expected_target="flow") -> int: parser = argparse.ArgumentParser(description=__doc__) parser.add_argument("--mapping", type=Path) parser.add_argument("--mapping-sha256") @@ -1229,7 +1258,11 @@ def main(argv=None) -> int: 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) + # The default executable remains Flow-only. The Cloud executable uses + # its separately reviewed verifier; JSON cannot choose a preparer. + inputs = (input_preparer or prepare_inputs)(args.mapping, args.mapping_sha256) + if prepared_target(inputs)[0] != expected_target: + fail("measured input target differs from this executable") 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) diff --git a/schemas/qualification-release-verification-receipt-v2.schema.json b/schemas/qualification-release-verification-receipt-v2.schema.json index 14f446c..4f0ccd0 100644 --- a/schemas/qualification-release-verification-receipt-v2.schema.json +++ b/schemas/qualification-release-verification-receipt-v2.schema.json @@ -66,7 +66,7 @@ "tag": {"oneOf": [{"$ref": "#/$defs/tag"}, {"type": "null"}]}, "deployment_id": {"oneOf": [{"$ref": "#/$defs/decimal_id"}, {"type": "null"}]}, "deployment_sha256": {"oneOf": [{"$ref": "#/$defs/digest"}, {"type": "null"}]}, - "draft_release_id": {"$ref": "#/$defs/decimal_id"}, + "draft_release_id": {"oneOf": [{"$ref": "#/$defs/decimal_id"}, {"type": "null"}]}, "publication_staging_sha256": {"$ref": "#/$defs/digest"}, "authority_state_sha256": {"$ref": "#/$defs/digest"}, "revocation_state_sha256": {"$ref": "#/$defs/digest"}, @@ -105,6 +105,7 @@ "package_identity": { "properties": { "release_kind": {"const": "package"}, + "draft_release_id": {"$ref": "#/$defs/decimal_id"}, "version": {"$ref": "#/$defs/version"}, "tag": {"$ref": "#/$defs/tag"}, "deployment_id": {"type": "null"}, @@ -123,6 +124,7 @@ "hybrid_identity": { "properties": { "release_kind": {"const": "hybrid"}, + "draft_release_id": {"$ref": "#/$defs/decimal_id"}, "version": {"$ref": "#/$defs/version"}, "tag": {"$ref": "#/$defs/tag"}, "deployment_id": {"$ref": "#/$defs/decimal_id"}, diff --git a/scripts/verify_production_release_admission.py b/scripts/verify_production_release_admission.py index 0c6a942..debafc1 100644 --- a/scripts/verify_production_release_admission.py +++ b/scripts/verify_production_release_admission.py @@ -152,6 +152,20 @@ def _timestamp(value: datetime) -> str: return value.replace(microsecond=0).strftime("%Y-%m-%dT%H:%M:%SZ") +def verification_draft_release_id(admission: dict[str, Any]) -> str | None: + """A published deployment has no GitHub draft release identity.""" + staging = admission["publication_staging"] + if staging["publication_mode"] == trust.PUBLICATION_MODE_ALREADY_PUBLISHED_DEPLOYMENT: + trust.validate_staging(staging) + if ( + admission["release"]["kind"] != "deployment" + or admission["release"]["deployment_id"] != staging["deployment_id"] + ): + raise trust.TrustError("verified deployment staging identity differs") + return None + return trust.require_decimal_id(staging["draft_release_id"], "verified draft release id") + + def verification_receipt( *, admission: dict[str, Any], @@ -182,7 +196,7 @@ def verification_receipt( "source_commit": admission["release"]["source_commit"], "version": admission["release"]["version"], "tag": admission["release"]["tag"], - "draft_release_id": admission["publication_staging"]["draft_release_id"], + "draft_release_id": verification_draft_release_id(admission), "publication_staging_sha256": admission["publication_staging_sha256"], "authority_state_sha256": admission["authority_state_sha256"], "revocation_state_sha256": admission["revocation_state_sha256"], @@ -791,6 +805,30 @@ def verify_sigstore( return if profile != "github-attestation": raise trust.TrustError(f"{kind} has an unsupported Sigstore profile") + verify_github_attestation( + regular_raw, + bundle_raw, + identity=identity, + issuer_identity=object_value.get("issuer"), + sigstore=sigstore, + ) + + +def verify_github_attestation( + regular_raw: bytes, + bundle_raw: bytes, + *, + identity: dict[str, Any], + issuer_identity: dict[str, Any] | None, + sigstore: dict[str, Any], +) -> dict[str, Any]: + """Verify retained provenance with an independently selected identity. + + Admission callers select identity from current policy. Input adapters can + select a separate, code-fixed producer identity without granting that + producer admission authority. Return only the cryptographically verified + result so those adapters can bind additional signed provenance fields. + """ version = subprocess.run( ["gh", "--version"], check=True, capture_output=True, text=True ).stdout.splitlines()[0] @@ -851,7 +889,7 @@ def verify_sigstore( != {"sha256": hashlib.sha256(regular_raw).hexdigest()} ): raise trust.TrustError("Sigstore statement subject or predicate differs") - issuer = object_value.get("issuer") + issuer = issuer_identity if not isinstance(issuer, dict): raise trust.TrustError("signed object has no closed issuer identity") source_commit = issuer.get("source_commit") @@ -873,6 +911,8 @@ def verify_sigstore( if dependencies != [expected_dependency]: raise trust.TrustError("Sigstore resolved source dependency differs") + return verification + def _canonical_base64(value: Any, *, label: str) -> bytes: if not isinstance(value, str): @@ -1213,7 +1253,7 @@ def main(argv: list[str] | None = None) -> int: admission["publication_staging"] ).decode("utf-8"), "publication_staging_sha256": admission["publication_staging_sha256"], - "draft_release_id": admission["publication_staging"]["draft_release_id"], + "draft_release_id": receipt_output["draft_release_id"] or "", "version": release["version"] or "", "tag": release["tag"] or "", "deployment_id": release["deployment_id"] or "", diff --git a/tests/test_deployment_verification_receipt.py b/tests/test_deployment_verification_receipt.py new file mode 100644 index 0000000..9634f67 --- /dev/null +++ b/tests/test_deployment_verification_receipt.py @@ -0,0 +1,173 @@ +"""Test-only deployment output fixtures; no actual admission is issued.""" + +import copy +import json +import sys +import unittest +from pathlib import Path + +from jsonschema import Draft202012Validator, ValidationError +from referencing import Registry, Resource + +ROOT = Path(__file__).resolve().parents[1] +sys.path.insert(0, str(ROOT / "tests")) +import test_qualification_issuer as samples + + +def receipt_fixture(*, target="cloud", published=True): + fixture = samples.trust_fixture() + resolver = samples.RecordingResolver(fixture) + workflow = samples.issuer.issue_workflow_admission( + samples.workflow_request(fixture), + resolver=resolver, + issuer_source_commit=samples.WORKFLOW_REGISTRY_COMMIT, + now=samples.NOW, + consumer=samples.RecordingConsumer(), + ) + request = samples.flow_release_inputs(fixture, workflow, resolver, target=target) + admission = samples.issuer.issue_release_admission( + request, + resolver=resolver, + issuer_source_commit=samples.RELEASE_REGISTRY_COMMIT, + now=samples.NOW, + consumer=samples.RecordingConsumer(), + ) + if published: + staging = { + key: value + for key, value in admission["publication_staging"].items() + if key in samples.trust.DEPLOYMENT_STAGING_FIELDS + } + staging.update( + publication_mode="already-published-deployment", + draft=False, + pypi_files=None, + deployment_id=admission["release"]["deployment_id"], + deployment_url="https://example.com/deployment/42", + ) + samples.trust.validate_staging(staging) + admission["publication_staging"] = staging + admission["publication_staging_sha256"] = samples.trust.staging_digest(staging) + projection = { + key: value + for key, value in admission.items() + if key != "admission_id_sha256" + } + admission["admission_id_sha256"] = samples.trust.digest_bytes( + samples.trust.RELEASE_ADMISSION_DOMAIN, + projection, + ) + samples.trust.validate_release(admission, now=samples.NOW) + ref, bundle = samples.reference_pair("qualification-release", admission) + receipt = samples.release_verifier.verification_receipt( + admission=admission, + admission_reference=ref, + admission_bundle_reference=bundle, + summary=resolver.objects[ + request["production_acceptance_summary_reference"]["object_sha256"] + ]["value"], + qualification_admission=workflow, + verified_at=samples.NOW, + trust_state_source_commit=samples.RELEASE_REGISTRY_COMMIT, + ) + return admission, receipt + + +class DeploymentReceiptTests(unittest.TestCase): + def validator(self): + schemas = [ + json.loads(path.read_bytes()) + for path in (ROOT / "schemas").glob("*.schema.json") + ] + registry = Registry().with_resources( + (item["$id"], Resource.from_contents(item)) for item in schemas + ) + schema = next( + item + for item in schemas + if item["$id"] + == "qualification-release-verification-receipt-v2.schema.json" + ) + return Draft202012Validator(schema, registry=registry) + + def test_actual_deployment_shape_needs_no_invented_draft_id(self): + for target in ("cloud", "docs"): + with self.subTest(target=target): + admission, receipt = receipt_fixture(target=target) + self.assertNotIn("draft_release_id", admission["publication_staging"]) + self.assertIsNone(receipt["draft_release_id"]) + self.assertEqual( + receipt["deployment_id"], admission["release"]["deployment_id"] + ) + self.validator().validate(receipt) + projection = { + key: value + for key, value in receipt.items() + if key != "verification_id_sha256" + } + self.assertEqual( + receipt["verification_id_sha256"], + samples.trust.digest_bytes( + samples.release_verifier.VERIFICATION_RECEIPT_V2_DOMAIN, + projection, + ), + ) + + def test_legacy_deployment_draft_id_is_preserved(self): + admission, receipt = receipt_fixture(published=False) + self.assertEqual( + receipt["draft_release_id"], + admission["publication_staging"]["draft_release_id"], + ) + self.validator().validate(receipt) + + def test_package_and_hybrid_still_require_decimal_draft_id(self): + _, package = receipt_fixture(target="agent", published=False) + # Exercise the existing hybrid receipt grammar as a schema-only fixture. + # Current Desktop target policy is package, so it cannot issue a hybrid. + hybrid = { + **package, + "target": "desktop", + "claim_scope": "production_desktop", + "source_repository": "OpenAdaptAI/openadapt-desktop", + "source_repository_id": "1171291730", + "release_kind": "hybrid", + "deployment_id": "42", + "deployment_sha256": "sha256:" + "a" * 64, + } + for receipt in (package, hybrid): + self.validator().validate(receipt) + receipt["draft_release_id"] = None + with ( + self.subTest(kind=receipt["release_kind"]), + self.assertRaises(ValidationError), + ): + self.validator().validate(receipt) + _, deployment = receipt_fixture() + for value in (True, 0, "0", "", "not-a-release"): + with self.subTest(value=value), self.assertRaises(ValidationError): + self.validator().validate({**deployment, "draft_release_id": value}) + + def test_null_requires_valid_published_deployment_in_builder(self): + admission, _ = receipt_fixture() + for kind in ("package", "hybrid"): + changed = copy.deepcopy(admission) + changed["release"]["kind"] = kind + with self.subTest(kind=kind), self.assertRaises(samples.trust.TrustError): + samples.release_verifier.verification_draft_release_id(changed) + changed = copy.deepcopy(admission) + changed["release"]["deployment_id"] = "43" + with self.assertRaises(samples.trust.TrustError): + samples.release_verifier.verification_draft_release_id(changed) + changed = copy.deepcopy(admission) + changed["publication_staging"]["draft_release_id"] = "20" + with self.assertRaises(samples.trust.TrustError): + samples.release_verifier.verification_draft_release_id(changed) + changed, _ = receipt_fixture(published=False) + changed["publication_staging"]["draft_release_id"] = None + with self.assertRaises(samples.trust.TrustError): + samples.release_verifier.verification_draft_release_id(changed) + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/test_measured_attestation_provenance.py b/tests/test_measured_attestation_provenance.py new file mode 100644 index 0000000..5326d3b --- /dev/null +++ b/tests/test_measured_attestation_provenance.py @@ -0,0 +1,206 @@ +"""Test-only subprocess results exercise provenance checks without signatures.""" + +import copy +import hashlib +import json +import subprocess +import sys +import unittest +from pathlib import Path +from unittest import mock + +ROOT = Path(__file__).resolve().parents[1] +sys.path.insert(0, str(ROOT / "scripts")) +import verify_production_release_admission as verifier + + +def verified_result(raw, repository, source, ref): + return { + "statement": { + "subject": [ + { + "name": "test-only.json", + "digest": { + "sha256": hashlib.sha256(raw).hexdigest(), + }, + } + ], + "predicateType": "https://slsa.dev/provenance/v1", + "predicate": { + "buildDefinition": { + "resolvedDependencies": [ + { + "uri": f"git+https://github.com/{repository}@{ref}", + "digest": {"gitCommit": source}, + } + ] + } + }, + }, + "signature": { + "certificate": { + "githubWorkflowSHA": source, + "sourceRepositoryDigest": source, + "githubWorkflowRef": ref, + "githubWorkflowRepository": repository, + } + }, + } + + +class AttestationExtractionTests(unittest.TestCase): + def setUp(self): + self.raw = b'{"test_only":true}\n' + self.bundle = b'{"test_only": "mocked subprocess; not a signature"}' + self.policy = json.loads( + (ROOT / "production-evidence-policy.json").read_bytes() + ) + self.identity = next( + x + for x in self.policy["sigstore"]["certificate_identities"] + if x["kind"] == "qualification-release" + ) + self.issuer = {"source_commit": "a" * 40, "ref": "refs/heads/main"} + self.result = verified_result( + self.raw, + self.identity["issuer_repository"], + self.issuer["source_commit"], + self.issuer["ref"], + ) + + def runner(self, results=None, version="gh version 2.98.0 (2026-08-20)", status=0): + self.commands = [] + + def run(command, **kwargs): + self.commands.append(command) + if command == ["gh", "--version"]: + return subprocess.CompletedProcess(command, 0, version + "\n", "") + return subprocess.CompletedProcess( + command, + status, + json.dumps( + results + if results is not None + else [{"verificationResult": self.result}] + ), + "test-only verification refusal" if status else "", + ) + + return run + + def verify(self): + return verifier.verify_github_attestation( + self.raw, + self.bundle, + identity=self.identity, + issuer_identity=self.issuer, + sigstore=self.policy["sigstore"], + ) + + def test_existing_policy_route_keeps_none_return_and_exact_flags(self): + with mock.patch.object(verifier.subprocess, "run", side_effect=self.runner()): + result = verifier.verify_sigstore( + self.raw, + self.bundle, + kind="qualification-release", + object_value={"issuer": self.issuer}, + policy=self.policy, + ) + self.assertIsNone(result) + command = self.commands[1] + self.assertEqual(command[:3], ["gh", "attestation", "verify"]) + self.assertEqual( + command[6:], + [ + "--repo", + self.identity["issuer_repository"], + "--cert-identity", + self.identity["certificate_identity"], + "--cert-oidc-issuer", + "https://token.actions.githubusercontent.com", + "--deny-self-hosted-runners", + "--no-public-good", + "--format", + "json", + ], + ) + + def test_helper_returns_only_the_verified_result(self): + with mock.patch.object(verifier.subprocess, "run", side_effect=self.runner()): + self.assertEqual(self.verify(), self.result) + + def test_wrong_subject_predicate_or_source_refuses(self): + mutations = ( + lambda r: r["statement"]["subject"][0]["digest"].update(sha256="b" * 64), + lambda r: r["statement"].update(predicateType="test-only/wrong"), + lambda r: r["signature"]["certificate"].update(githubWorkflowSHA="b" * 40), + lambda r: r["signature"]["certificate"].update( + sourceRepositoryDigest="b" * 40 + ), + lambda r: r["signature"]["certificate"].update( + githubWorkflowRef="refs/heads/other" + ), + lambda r: r["signature"]["certificate"].update( + githubWorkflowRepository="other/repo" + ), + lambda r: r["statement"]["predicate"]["buildDefinition"].update( + resolvedDependencies=[] + ), + ) + for mutate in mutations: + changed = copy.deepcopy(self.result) + mutate(changed) + with ( + self.subTest(mutation=mutate), + mock.patch.object( + verifier.subprocess, + "run", + side_effect=self.runner([{"verificationResult": changed}]), + ), + self.assertRaises(verifier.trust.TrustError), + ): + self.verify() + + def test_multiple_results_failed_crypto_and_wrong_cli_refuse(self): + for kwargs in ( + {"results": [{"verificationResult": self.result}] * 2}, + {"status": 1}, + {"version": "gh version 2.67.0 (2025-02-11)"}, + ): + with ( + self.subTest(kwargs=kwargs), + mock.patch.object( + verifier.subprocess, + "run", + side_effect=self.runner(**kwargs), + ), + self.assertRaises(verifier.trust.TrustError), + ): + self.verify() + self.assertEqual(self.commands, [["gh", "--version"]]) + + def test_existing_message_signature_route_is_unchanged(self): + with ( + mock.patch.object(verifier, "verify_message_signature") as message, + mock.patch.object( + verifier.subprocess, + "run", + side_effect=AssertionError("wrong profile"), + ), + ): + verifier.verify_sigstore( + self.raw, + self.bundle, + kind="qualification-evidence-decision-receipt", + object_value={"evidence_class": "private-customer"}, + policy=self.policy, + ) + message.assert_called_once() + self.assertEqual( + message.call_args.kwargs["identity"]["bundle_profile"], + "sigstore-message-signature", + ) + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/test_measured_cloud_admission.py b/tests/test_measured_cloud_admission.py new file mode 100644 index 0000000..55be25d --- /dev/null +++ b/tests/test_measured_cloud_admission.py @@ -0,0 +1,1366 @@ +"""Test-only hosted provenance. No real campaign, signature, or live call.""" + +import base64 +import copy +import importlib.util +import json +import subprocess +import sys +import tempfile +import unittest +from pathlib import Path +from unittest import mock + +ROOT = Path(__file__).resolve().parents[1] +sys.path.insert(0, str(ROOT / "tests")) +from test_measured_attestation_provenance import verified_result + +SPEC = importlib.util.spec_from_file_location( + "measured_cloud_admission", ROOT / "local-candidates/cloud-measured/issue.py" +) +adapter = importlib.util.module_from_spec(SPEC) +SPEC.loader.exec_module(adapter) + + +def producer(): + return { + "repository": adapter.PRODUCER_REPOSITORY, + "repository_id": adapter.PRODUCER_REPOSITORY_ID, + "repository_owner_id": adapter.PRODUCER_OWNER_ID, + "workflow": adapter.PRODUCER_WORKFLOW, + "ref": adapter.PRODUCER_REF, + "source_commit": "a" * 40, + "run_id": "123", + "run_attempt": "2", + } + + +class CloudProvenanceTests(unittest.TestCase): + def setUp(self): + self.producer = p = producer() + self.raw = b'{"test_only":true}\n' + self.result = verified_result( + self.raw, p["repository"], p["source_commit"], p["ref"] + ) + predicate = self.result["statement"]["predicate"] + predicate["buildDefinition"].update( + buildType="https://actions.github.io/buildtypes/workflow/v1", + externalParameters={ + "workflow": { + "repository": "https://github.com/" + p["repository"], + "path": p["workflow"], + "ref": p["ref"], + } + }, + internalParameters={ + "github": { + "repository_id": p["repository_id"], + "repository_owner_id": p["repository_owner_id"], + "event_name": "workflow_dispatch", + "runner_environment": "github-hosted", + } + }, + ) + predicate["runDetails"] = { + "builder": {"id": adapter.PRODUCER_IDENTITY}, + "metadata": { + "invocationId": ( + "https://github.com/" + + p["repository"] + + "/actions/runs/123/attempts/2" + ) + }, + } + self.run = { + "id": 123, + "run_attempt": 2, + "head_sha": p["source_commit"], + "head_branch": "main", + "path": p["workflow"], + "event": "workflow_dispatch", + "status": "completed", + "conclusion": "success", + "repository": { + "full_name": p["repository"], + "id": int(p["repository_id"]), + "owner": {"id": int(p["repository_owner_id"])}, + }, + } + self.head = {"ref": p["ref"], "object": {"sha": p["source_commit"]}} + self.source = b"# Test-only inert workflow source\n" + temporary = tempfile.TemporaryDirectory() + self.addCleanup(temporary.cleanup) + self.owner = Path(temporary.name) / "mapping.json" + self.refs = {} + for name, raw in { + "attestation": b'{"test_only":"mocked cryptographic process"}', + "workflow_source": self.source, + "workflow_run": json.dumps(self.run).encode(), + "protected_main": json.dumps(self.head).encode(), + }.items(): + (self.owner.parent / name).write_bytes(raw) + self.refs[name] = { + "path": name, + "sha256": adapter.shared.sha(raw), + "size_bytes": len(raw), + } + + def gh(self, endpoint): + if "/git/ref/heads/main" in endpoint: + return self.head + if "/actions/runs/123/attempts/2" in endpoint: + return self.run + if "/contents/" in endpoint: + return { + "encoding": "base64", + "type": "file", + "path": self.producer["workflow"], + "content": base64.b64encode(self.source).decode(), + } + raise AssertionError("unexpected live interface") + + def process(self, command, **_kwargs): + if command == ["gh", "--version"]: + return subprocess.CompletedProcess( + command, 0, "gh version 2.98.0 (2026-08-20)\n", "" + ) + for required in ( + adapter.PRODUCER_IDENTITY, + "--deny-self-hosted-runners", + "--no-public-good", + ): + self.assertIn(required, command) + return subprocess.CompletedProcess( + command, 0, json.dumps([{"verificationResult": self.result}]), "" + ) + + def verify(self, gh=None): + with mock.patch.object( + adapter.shared.verifier.subprocess, "run", side_effect=self.process + ): + return adapter.verify_derivative_provenance( + self.owner, + self.raw, + self.producer, + self.refs, + gh=gh or self.gh, + ) + + def test_exact_fixed_workflow_passes_without_network(self): + self.assertEqual(self.verify(), self.run) + + def test_verified_repo_workflow_ids_builder_run_attempt_refuse(self): + original = copy.deepcopy(self.result) + paths = ( + ("buildDefinition", "externalParameters", "workflow", "repository"), + ("buildDefinition", "externalParameters", "workflow", "path"), + ("buildDefinition", "externalParameters", "workflow", "ref"), + ("buildDefinition", "internalParameters", "github", "repository_id"), + ("buildDefinition", "internalParameters", "github", "repository_owner_id"), + ("buildDefinition", "internalParameters", "github", "runner_environment"), + ("buildDefinition", "internalParameters", "github", "event_name"), + ("runDetails", "builder", "id"), + ("runDetails", "metadata", "invocationId"), + ) + for path in paths: + self.result = copy.deepcopy(original) + current = self.result["statement"]["predicate"] + for part in path[:-1]: + current = current[part] + current[path[-1]] = "wrong" + with self.subTest(path=path), self.assertRaises(ValueError): + self.verify() + for suffix in ("124/attempts/2", "123/attempts/1"): + self.result = copy.deepcopy(original) + self.result["statement"]["predicate"]["runDetails"]["metadata"][ + "invocationId" + ] = ( + "https://github.com/" + + self.producer["repository"] + + "/actions/runs/" + + suffix + ) + with self.subTest(suffix=suffix), self.assertRaises(ValueError): + self.verify() + + def test_verified_subject_and_source_refuse(self): + self.result["signature"]["certificate"]["githubWorkflowSHA"] = "b" * 40 + with self.assertRaisesRegex(ValueError, "source identity"): + self.verify() + self.result["signature"]["certificate"]["githubWorkflowSHA"] = self.producer[ + "source_commit" + ] + self.result["statement"]["subject"][0]["digest"]["sha256"] = "b" * 64 + with self.assertRaisesRegex(ValueError, "subject"): + self.verify() + + def test_actual_run_attempt_source_success_and_main_must_match(self): + for key, value in ( + ("id", 124), + ("run_attempt", 1), + ("run_attempt", True), + ("head_sha", "b" * 40), + ("head_branch", "other"), + ("path", "other.yml"), + ("conclusion", "failure"), + ): + + def gh(endpoint, key=key, value=value): + return ( + {**self.run, key: value} + if "/actions/runs/" in endpoint + else self.gh(endpoint) + ) + + with ( + self.subTest(key=key), + self.assertRaisesRegex(ValueError, "actual hosted"), + ): + self.verify(gh) + with self.assertRaisesRegex(ValueError, "protected main changed"): + self.verify(lambda _endpoint: {"object": {"sha": "b" * 40}}) + + def test_actual_workflow_byte_change_refuses(self): + def gh(endpoint): + value = self.gh(endpoint) + if "/contents/" in endpoint: + value["content"] = base64.b64encode(b"changed").decode() + return value + + with self.assertRaisesRegex(ValueError, "workflow bytes"): + self.verify(gh) + + def test_caller_identity_and_ambiguous_json_refuse(self): + for key in ( + "repository", + "repository_id", + "repository_owner_id", + "workflow", + "ref", + ): + with ( + self.subTest(key=key), + self.assertRaisesRegex(ValueError, "fixed workflow"), + ): + adapter.validate_producer({**self.producer, key: "other"}) + for raw in (b'{"x":1,"x":2}', b'{"x":NaN}', b'{"x":Infinity}'): + with self.subTest(raw=raw), self.assertRaises(ValueError): + adapter.strict_json(raw) + + def test_exponent_overflow_refuses_and_finite_json_numbers_pass(self): + for raw in (b'{"x":1e999}', b'{"x":-1e999}'): + with ( + self.subTest(raw=raw), + self.assertRaisesRegex(ValueError, "non-finite JSON"), + ): + adapter.strict_json(raw) + for raw, expected in ( + (b'{"x":1.5}', 1.5), + (b'{"x":-2e2}', -200.0), + (b'{"x":0}', 0), + ): + self.assertEqual(adapter.strict_json(raw), {"x": expected}) + + +class MeasuredFixture: + """Inert interface fixtures, not a native campaign or an admitted deployment.""" + + def __init__(self, directory): + from uuid import UUID + + self.directory = Path(directory).resolve() + self.inventory = {} + self.refs = [] + self.uu = lambda n: str(UUID(int=n)) + self.digest = lambda text: adapter.shared.sha(text.encode())[7:] + self.source = "c" * 40 + self.wheel = self.add("runtime-wheel", b"test-only inert wheel") + self.bundle = self.add("sealed-bundle", b"test-only inert sealed bundle") + self.runtime = { + "openadapt_flow": "1.35.1", + "release_commit": "d" * 40, + "wheel_sha256": self.wheel, + "sdist_sha256": self.digest("sdist"), + "wheel_url": "https://files.pythonhosted.org/test-only.whl", + "runner_build": "report-v5", + "runner_artifact_sha256": self.digest("runner"), + "modal_sdk": "1.5.5", + "fastapi": "0.141.1", + "starlette": "1.6.0", + "playwright": "1.62.0", + "browser_base_image": "python:test@sha256:" + "a" * 64, + "sandbox_network_policy": "modal-domain-allowlist-v1", + } + self.add("runtime_version", self.runtime) + components = { + "schema_version": "openadapt.runtime-component-manifest/v1", + "components": [ + {"name": "test-only-runtime", "artifact_sha256": self.wheel}, + ], + } + self.build = { + "schema_version": "openadapt.admitted-runtime-build/v1", + "substrate": "web", + "flow_version": self.runtime["openadapt_flow"], + "flow_release_commit": self.runtime["release_commit"], + "flow_wheel_sha256": self.wheel, + "runner_build": self.runtime["runner_build"], + "runner_artifact_sha256": self.runtime["runner_artifact_sha256"], + "runtime_manifest": components, + "runtime_manifest_sha256": adapter.semantic_digest( + components, b"openadapt-runtime-component-manifest-v1\0" + ), + "managed_browser": { + "playwright_version": self.runtime["playwright"], + "browser_base_image": self.runtime["browser_base_image"], + }, + "substrate_runtime": { + "transport": "browser", + "os_family": "linux", + "runtime_boundary_sha256": adapter.semantic_digest( + {"environment": "test-only-boundary"} + ), + }, + } + self.add("runtime_build_identity", self.build) + self.workflow_raw = b"# Inert test-only deployment workflow\n" + self.add("deployment_workflow_source", self.workflow_raw) + self.validator_raw = b"# Inert test-only private verifier source\n" + validator_hash = self.add("validator-source", self.validator_raw) + source_inventory = self.add( + "source-inventory", + { + "test_only": True, + "files": [ + {"path": "runner/qualification_issuer.py", "sha256": validator_hash} + ], + }, + ) + authority = { + "repository": adapter.DEPLOYMENT_REPOSITORY, + "workflow": adapter.DEPLOYMENT_WORKFLOW, + "source_ref": "refs/heads/main", + "source_commit": self.source, + "workflow_ref": f"{adapter.DEPLOYMENT_REPOSITORY}/{adapter.DEPLOYMENT_WORKFLOW}@refs/heads/main", + "certificate_identity": f"https://github.com/{adapter.DEPLOYMENT_REPOSITORY}/{adapter.DEPLOYMENT_WORKFLOW}@refs/heads/main", + "workflow_sha256": adapter.shared.sha(self.workflow_raw), + "oidc_issuer": "https://token.actions.githubusercontent.com", + "job": "deploy", + "environment": "production", + "environment_scope": "openadapt-cloud-production-v1", + "event_name": "workflow_dispatch", + "run_id": "456", + "run_attempt": "2", + } + host = { + "provider": "netlify", + "deploy_id": "a" * 24, + "immutable_url": "https://" + "a" * 24 + "--test.netlify.app/", + "created_at": "2026-09-11T00:00:00.000Z", + "published_at": "2026-09-11T00:01:00.000Z", + "site_identity_sha256": "sha256:" + + adapter.semantic_digest( + self.uu(1), b"OpenAdapt Netlify production site identity v1\0" + ), + "environment_contract_sha256": "sha256:" + self.digest("host environment"), + } + runner = { + "provider": "modal", + "environment": "production", + "app_id": "ap-" + "a" * 22, + "app_name": "test-only-runner", + "app_version": "v2", + "deployed_at": "2026-09-11T00:01:00.000Z", + "endpoint_origin_sha256": "sha256:" + + adapter.semantic_digest( + "https://test-only.modal.run/", + b"OpenAdapt Modal runner endpoint origin v1\0", + ), + } + runtime = self.runtime + self.manifest = { + "schema_version": "openadapt.cloud-production-deployment-manifest/v1", + "source": { + "repository": adapter.DEPLOYMENT_REPOSITORY, + "commit": self.source, + }, + "build_authority": authority, + "deployment": {"host": host, "runner": runner}, + "target": { + "environment_digest": "sha256:" + self.digest("environment"), + "sha256": "sha256:" + self.digest("target"), + }, + "runtime": { + "runtime_manifest_sha256": "sha256:" + adapter.semantic_digest(runtime), + "runner_source_artifact_sha256": "sha256:" + + runtime["runner_artifact_sha256"], + "runner_build": runtime["runner_build"], + "modal_sdk_version": runtime["modal_sdk"], + "fastapi_version": runtime["fastapi"], + "starlette_version": runtime["starlette"], + "sandbox_network_policy": runtime["sandbox_network_policy"], + "flow": { + "version": runtime["openadapt_flow"], + "release_commit": runtime["release_commit"], + "wheel_url": runtime["wheel_url"], + "wheel_sha256": "sha256:" + runtime["wheel_sha256"], + "sdist_sha256": "sha256:" + runtime["sdist_sha256"], + }, + "browser": { + "playwright_version": runtime["playwright"], + "browser_base_image": runtime["browser_base_image"], + "runtime_contract_sha256": "sha256:" + + adapter.semantic_digest( + { + "python_base_image": runtime["browser_base_image"], + "playwright_version": runtime["playwright"], + "browser_install_command": "python -m playwright install --with-deps chromium", + }, + b"OpenAdapt managed browser image contract v1\0", + ), + }, + }, + "signature": "TEST ONLY: the attested private verifier is mocked", + } + manifest_hash = self.add("deployment-manifest", self.manifest) + self.identity = { + "schema_version": "openadapt.production-acceptance-evidence-identity/v2", + "runtime_build_identity": self.build, + "deployment_manifest_sha256": manifest_hash, + "bundle_artifact_sha256": self.bundle, + "bundle_content_digest": self.digest("native bundle"), + "environment_digest": self.digest("environment"), + "campaign_id": self.uu(2), + "tenant_id": self.uu(3), + "workflow_id": self.uu(4), + "workflow_version_id": self.uu(5), + } + self.add("evidence_identity", self.identity) + self.subject = { + "source_commit": self.source, + "deployment_manifest_sha256": manifest_hash, + "runtime_build_identity_sha256": adapter.semantic_digest( + self.build, b"openadapt-admitted-runtime-build-v1\0" + ), + "runtime_manifest_sha256": adapter.semantic_digest(runtime), + "evidence_identity_sha256": adapter.semantic_digest( + self.identity, b"OpenAdapt production acceptance evidence identity v2\0" + ), + "bundle_artifact_sha256": self.bundle, + "bundle_content_digest": self.identity["bundle_content_digest"], + "environment_digest": self.identity["environment_digest"], + "runtime_wheel_sha256": self.wheel, + } + self.run = { + "id": 456, + "run_attempt": 2, + "head_sha": self.source, + "head_branch": "main", + "path": adapter.DEPLOYMENT_WORKFLOW, + "event": "workflow_dispatch", + "status": "completed", + "conclusion": "success", + "repository": { + "full_name": adapter.DEPLOYMENT_REPOSITORY, + "id": int(adapter.trust.TARGET_CONTRACTS["cloud"]["repository_id"]), + "owner": {"id": int(adapter.PRODUCER_OWNER_ID)}, + }, + } + self.add("deployment_workflow_run", self.run) + self.readback = { + "schema_version": "openadapt.cloud-production-deployment-readback/v1", + "observed_at": "2026-09-11T00:05:00.123Z", + "source_commit": self.source, + "manifest_sha256": "sha256:" + manifest_hash, + "manifest_bytes_sha256": "sha256:" + manifest_hash, + "target_attestation_sha256": self.manifest["target"]["sha256"], + "admission_activated": False, + "provider_observation": { + "github": { + "run_id": "456", + "run_attempt": "2", + "source_commit": self.source, + "event": "workflow_dispatch", + }, + "host": { + **{ + key: host[key] + for key in ( + "deploy_id", + "immutable_url", + "created_at", + "published_at", + ) + }, + "site_id": self.uu(1), + }, + "modal": { + **{ + key: runner[key] + for key in ( + "environment", + "app_id", + "app_name", + "app_version", + "deployed_at", + ) + }, + "source_commit": self.source, + "source_dirty": False, + "sdk_version": runtime["modal_sdk"], + "endpoint_origin": "https://test-only.modal.run/", + "function_id": "fu-" + "a" * 22, + "function_definition_id": "test-only-definition", + }, + "environment_contract_sha256": host["environment_contract_sha256"], + }, + "observed_runtime_context": { + "deployment_manifest_sha256": manifest_hash, + "runtime_build_identity": self.build, + "control_image_id": "im-test-only-control", + "modal_image_id": "im-test-only-sandbox", + "runtime_environment_sha256": adapter.semantic_digest( + {"environment": "test-only-boundary"} + ), + }, + } + self.readback["function_readbacks"] = [ + { + "function_tag": tag, + "function_id": "test-only-" + tag, + "function_definition_id": "test-only-definition-" + tag, + } + for tag in ("enqueue", "run_flow", "run_teach") + ] + self.add("deployment_readback", self.readback) + commitments = {} + manifest_opening = self.add( + "evidence-manifest", {"test_only": "protected raw inventory"} + ) + for key in sorted(adapter.shared.FILE_COMMITMENTS): + if key in { + "bundle_sha256", + "admitted_runtime_sha256", + "evidence_manifest_sha256", + }: + commitments[key] = { + "bundle_sha256": self.bundle, + "admitted_runtime_sha256": self.wheel, + "evidence_manifest_sha256": manifest_opening, + }[key] + continue + opening = {"test_only_role": key, "admitted_subject": self.subject} + field = { + "organization_id_sha256": "tenant_id", + "workflow_id_sha256": "workflow_id", + "workflow_version_id_sha256": "workflow_version_id", + }.get(key) + if field: + opening["id"] = self.identity[field] + if key == "workflow_version_id_sha256": + opening["bundle_version"] = "1.35.1-hosted-reference.1" + if key == "decision_identity_sha256": + opening.update(id=self.uu(6), decision_revision=1) + if key in { + "decision_commitment_sha256", + "evidence_manifest_readback_sha256", + }: + opening["evidence_manifest_sha256"] = manifest_opening + if key == "decision_commitment_sha256": + opening.update(decision_id=self.uu(6), decision_revision=1) + if key == "campaign_artifact_sha256": + opening = { + "schema_version": "openadapt.qualification-campaign/v2", + "campaign_id": self.uu(2), + "evidence_identity_sha256": self.subject[ + "evidence_identity_sha256" + ], + } + commitments[key] = self.add("opening-" + key, opening) + groups = [] + conditions = [(name, name) for name in adapter.PHASES if name != "safe_halt"] + conditions += [("safe_halt", f"safe_halt_{n}") for n in range(5)] + for cell, (name, condition) in enumerate( + sorted(conditions, key=lambda x: x[1]) + ): + for ordinal in range(1, 4): + trial = cell * 3 + ordinal + run = self.digest("run" + str(trial)) + counts = dict.fromkeys(adapter.GROUP_COUNT_FIELDS, 0) + if name == "uncertain_delivery": + counts["reconciliation_required_count"] = 1 + if name == "declared_attended": + counts.update( + authenticated_bound_decision_count=1, + live_target_revalidation_count=1, + ) + if name == "governed_repair": + counts.update( + policy_approved_repair_count=1, + approved_repair_count=1, + retained_repair_evidence_count=1, + live_target_revalidation_count=1, + ) + group = { + "trial_id": self.uu(100 + trial), + "task": "test-only-task", + "condition": condition, + "campaign_class": name, + "ordinal": ordinal, + "counts": counts, + "phases": [], + "observer_before_sha256": self.add( + f"before-{trial}", {"test_only_before": trial} + ), + "observer_after_sha256": self.add( + f"after-{trial}", {"test_only_after": trial} + ), + } + phase_names, _principal = adapter.PHASES[name] + input_hash = self.add(f"input-{trial}", {"test_only_input": trial}) + for phase in sorted(phase_names): + phase_run = ( + run + if phase != "replay" and phase != "repair_prior" + else self.digest(f"{run}/{phase}") + ) + outcome = { + "safe_halt": "HALTED_BEFORE_EFFECT", + "uncertain_delivery": "RECONCILIATION_REQUIRED", + }.get(name, "VERIFIED") + report = { + "run_id_sha256": phase_run, + "bundle_content_digest": self.subject["bundle_content_digest"], + "qualification_evidence_only": phase != "replay", + "production_eligible": False, + "transaction_outcome": outcome, + "success": outcome == "VERIFIED", + "idempotency_key": "test-only-key-" + str(trial), + "phase": phase, + } + if phase == "repair_prior": + report["bundle_content_digest"] = self.digest("prior") + if phase == "replay": + report.update( + qualification_evidence_only=False, + success=False, + idempotent_replay=True, + ) + group["phases"].append( + { + "phase": phase, + "qualification_run_id_sha256": phase_run, + "run_report_sha256": self.add( + f"report-{trial}-{phase}", report + ), + "input_sha256": input_hash, + "auxiliary_artifacts": [], + "runner_receipt_sha256": self.add( + f"receipt-{trial}-{phase}", + {"test_only": [trial, phase]}, + ), + } + ) + groups.append(group) + self.derivative = { + "schema_version": "openadapt.hosted-qualification-derivative/v1", + "target": "cloud", + "scope": "hosted-synthetic-qualification", + "producer": producer(), + "validator": { + "repository": adapter.DEPLOYMENT_REPOSITORY, + "source_commit": self.source, + "path": "runner/qualification_issuer.py", + "sha256": validator_hash, + "source_inventory_sha256": source_inventory, + }, + "subject": self.subject, + "campaign_id": self.uu(2), + "receipt_commitments": commitments, + "raw_evidence_refs": sorted( + self.refs, key=lambda x: (x["role"], x["sha256"]) + ), + "groups": groups, + } + artifact = { + "name": "deployment.json", + "kind": "deployment-manifest", + "sha256": "sha256:" + manifest_hash, + "size_bytes": len(self.inventory[manifest_hash][1]), + "media_type": "application/vnd.openadapt.production-deployment-manifest+json;version=1", + "publish_destinations": ["deployment"], + } + self.artifacts = { + "schema_version": "openadapt.production-release-artifact-inventory/v1", + "target": "cloud", + "claim_scope": "production_cloud", + "artifacts": [artifact], + } + self.release = { + "schema_version": "openadapt.production-release-candidate/v1", + "kind": "deployment", + "source_repository": adapter.DEPLOYMENT_REPOSITORY, + "source_repository_id": adapter.trust.TARGET_CONTRACTS["cloud"][ + "repository_id" + ], + "source_commit": self.source, + "version": None, + "tag": None, + "deployment_id": "456", + "deployment_sha256": "sha256:" + manifest_hash, + "artifacts": [artifact], + } + self.staging = { + "schema_version": "openadapt.production-release-staging-evidence/v1", + "publication_mode": "already-published-deployment", + "repository": adapter.DEPLOYMENT_REPOSITORY, + "repository_id": self.release["source_repository_id"], + "target_commitish": self.source, + "draft": False, + "prerelease": False, + "pypi_files": None, + "deployment_id": "456", + "deployment_url": host["immutable_url"], + "tag": "v0.0.0-deployment.456", + "assets": [ + { + **artifact, + "asset_id": None, + "uploader_id": None, + "uploader_login": None, + } + ], + "observed_at": "2026-09-11T00:05:00Z", + } + + def add(self, role, value): + raw = value if isinstance(value, bytes) else adapter.canonical_json(value) + digest = adapter.shared.sha(raw)[7:] + path = self.directory / f"{len(self.refs)}.json" + path.write_bytes(raw) + self.inventory[digest] = (path, raw) + self.refs.append({"role": role, "sha256": digest}) + return digest + + def replace(self, digest, value): + old_path, _ = self.inventory.pop(digest) + raw = value if isinstance(value, bytes) else adapter.canonical_json(value) + new = adapter.shared.sha(raw)[7:] + old_path.write_bytes(raw) + self.inventory[new] = old_path, raw + + def update(node): + if isinstance(node, dict): + for key, item in node.items(): + node[key] = new if item == digest else update(item) + elif isinstance(node, list): + for index, item in enumerate(node): + node[index] = update(item) + return node + + update(self.derivative) + self.derivative["raw_evidence_refs"].sort( + key=lambda x: (x["role"], x["sha256"]) + ) + return new + + def gh(self, endpoint): + if "/actions/runs/" in endpoint: + return self.run + for path, raw in ( + (adapter.DEPLOYMENT_WORKFLOW, self.workflow_raw), + ("runner/runtime-version.json", adapter.canonical_json(self.runtime)), + ("runner/qualification_issuer.py", self.validator_raw), + ): + if f"/contents/{path}?ref={self.source}" in endpoint: + return { + "type": "file", + "path": path, + "encoding": "base64", + "content": base64.b64encode(raw).decode(), + } + raise AssertionError("unexpected GitHub interface: " + endpoint) + + +class CloudMeasuredInputsTests(unittest.TestCase): + def setUp(self): + temporary = tempfile.TemporaryDirectory() + self.addCleanup(temporary.cleanup) + self.fixture = MeasuredFixture(temporary.name) + + def test_counts_are_derived_for_ten_cells_without_fixed_total(self): + f = self.fixture + summary = adapter.verify_derivative(f.derivative, f.inventory) + self.assertEqual(sum(c["observed_trial_count"] for c in summary.values()), 30) + self.assertEqual(summary["safe_halt"]["task_condition_cell_count"], 5) + self.assertEqual(summary["safe_halt"]["observed_trial_count"], 15) + self.assertEqual( + adapter.verify_subject_openings(f.derivative, f.inventory), + ("1.35.1-hosted-reference.1", 1), + ) + self.assertIsNone( + adapter.verify_deployment( + f.derivative, f.inventory, f.release, f.staging, gh=f.gh + ) + ) + + def test_wrong_scope_source_target_counts_and_missing_phase_refuse(self): + original = copy.deepcopy(self.fixture.derivative) + mutations = [ + lambda d: d.update(target="flow"), + lambda d: d.update(scope="production_cloud"), + lambda d: d["validator"].update(source_commit="e" * 40), + lambda d: d["groups"][0]["counts"].update(model_call_count=True), + lambda d: d["groups"][0]["counts"].pop("model_call_count"), + lambda d: d["groups"][0]["phases"].pop(), + lambda d: d["groups"].append(copy.deepcopy(d["groups"][0])), + lambda d: d["receipt_commitments"].update(bundle_sha256="a" * 64), + ] + for mutation in mutations: + value = copy.deepcopy(original) + mutation(value) + with ( + self.subTest(mutation=mutation), + self.assertRaises((ValueError, KeyError)), + ): + adapter.verify_derivative(value, self.fixture.inventory) + + def test_native_outcome_bundle_scope_and_run_mismatch_refuse(self): + f = self.fixture + primary = next( + g for g in f.derivative["groups"] if g["campaign_class"] == "healthy" + )["phases"][0] + original = adapter.evidence_json(f.inventory, primary["run_report_sha256"]) + for key, value in ( + ("success", False), + ("transaction_outcome", "COMPLETED_UNVERIFIED"), + ("qualification_evidence_only", False), + ("production_eligible", True), + ("bundle_content_digest", "a" * 64), + ("run_id_sha256", "b" * 64), + ): + f.replace(primary["run_report_sha256"], {**original, key: value}) + with self.subTest(key=key), self.assertRaises(ValueError): + adapter.verify_derivative(f.derivative, f.inventory) + f.replace(primary["run_report_sha256"], original) + self.assertIsInstance( + adapter.verify_derivative(f.derivative, f.inventory), dict + ) + + def test_replay_keeps_native_early_refusal_but_requires_same_key(self): + f = self.fixture + group = next( + g + for g in f.derivative["groups"] + if g["campaign_class"] == "idempotency_replay" + ) + replay = next(p for p in group["phases"] if p["phase"] == "replay") + report = adapter.evidence_json(f.inventory, replay["run_report_sha256"]) + self.assertFalse(report["qualification_evidence_only"]) + for key, value in ( + ("idempotency_key", "other"), + ("idempotency_key", None), + ("idempotent_replay", False), + ("success", True), + ): + f.replace(replay["run_report_sha256"], {**report, key: value}) + with ( + self.subTest(key=key), + self.assertRaisesRegex(ValueError, "replay lacks"), + ): + adapter.verify_derivative(f.derivative, f.inventory) + f.replace(replay["run_report_sha256"], report) + self.assertIsInstance( + adapter.verify_derivative(f.derivative, f.inventory), dict + ) + + def test_per_group_attended_proof_cannot_hide_in_aggregate(self): + d = self.fixture.derivative + rows = [g for g in d["groups"] if g["campaign_class"] == "declared_attended"] + rows[0]["counts"]["authenticated_bound_decision_count"] = 0 + rows[1]["counts"]["authenticated_bound_decision_count"] = 2 + with self.assertRaisesRegex(ValueError, "each attended"): + adapter.verify_derivative(d, self.fixture.inventory) + + def test_archive_raw_hash_is_not_native_or_runtime_semantic_hash(self): + f = self.fixture + self.assertNotEqual( + f.subject["bundle_artifact_sha256"], f.subject["bundle_content_digest"] + ) + self.assertNotEqual( + f.subject["runtime_build_identity_sha256"], adapter.semantic_digest(f.build) + ) + self.assertNotEqual( + f.subject["runtime_manifest_sha256"], f.build["runtime_manifest_sha256"] + ) + for key in ( + "runtime_build_identity_sha256", + "evidence_identity_sha256", + "runtime_manifest_sha256", + ): + original = f.subject[key] + f.subject[key] = "a" * 64 + with ( + self.subTest(key=key), + self.assertRaisesRegex(ValueError, "semantic runtime"), + ): + adapter.verify_subject_openings(f.derivative, f.inventory) + f.subject[key] = original + + def test_deployment_source_attempt_readback_and_staging_mismatch_refuse(self): + f = self.fixture + for key, value in ( + ("deployment_id", "455"), + ("source_commit", "d" * 40), + ("deployment_sha256", "sha256:" + "a" * 64), + ): + with ( + self.subTest(key=key), + self.assertRaisesRegex(ValueError, "deployment release"), + ): + adapter.verify_deployment( + f.derivative, + f.inventory, + {**f.release, key: value}, + f.staging, + gh=f.gh, + ) + for key, value in ( + ("run_attempt", 1), + ("run_attempt", True), + ("status", "in_progress"), + ("conclusion", "failure"), + ("head_sha", "d" * 40), + ): + + def gh(endpoint, key=key, value=value): + return ( + {**f.run, key: value} + if "/actions/runs/" in endpoint + else f.gh(endpoint) + ) + + with ( + self.subTest(key=key), + self.assertRaisesRegex(ValueError, "successful protected"), + ): + adapter.verify_deployment( + f.derivative, f.inventory, f.release, f.staging, gh=gh + ) + changed = copy.deepcopy(f.staging) + changed["assets"][0]["sha256"] = "sha256:" + "b" * 64 + with self.assertRaisesRegex(ValueError, "publication staging differs"): + adapter.verify_deployment( + f.derivative, f.inventory, f.release, changed, gh=f.gh + ) + self.assertIsNone(f.release["tag"]) + self.assertIsNone(f.release["version"]) + self.assertEqual(f.staging["tag"], "v0.0.0-deployment.456") + + def test_provider_runtime_and_omitted_readback_refuse(self): + f = self.fixture + digest = next( + r["sha256"] + for r in f.derivative["raw_evidence_refs"] + if r["role"] == "deployment_readback" + ) + changed = copy.deepcopy(f.readback) + changed["provider_observation"]["modal"]["app_version"] = "v3" + f.replace(digest, changed) + with self.assertRaisesRegex(ValueError, "runner provider identity"): + adapter.verify_deployment( + f.derivative, f.inventory, f.release, f.staging, gh=f.gh + ) + f.derivative["raw_evidence_refs"] = [ + r + for r in f.derivative["raw_evidence_refs"] + if r["role"] != "deployment_readback" + ] + with self.assertRaisesRegex(ValueError, "exactly one deployment_readback"): + adapter.verify_deployment( + f.derivative, f.inventory, f.release, f.staging, gh=f.gh + ) + + def test_inventory_actual_bytes_and_scope_refuse_before_issuer(self): + f = self.fixture + refs = [ + {"path": path.name, "sha256": "sha256:" + digest, "size_bytes": len(raw)} + for digest, (path, raw) in f.inventory.items() + ] + self.assertEqual( + adapter.retained_inventory(f.directory / "mapping.json", refs), f.inventory + ) + path, raw = next(iter(f.inventory.values())) + path.write_bytes(raw + b" ") + with self.assertRaisesRegex(ValueError, "bytes differ"): + adapter.retained_inventory(f.directory / "mapping.json", refs) + for target, scope in ( + ("cloud", "all-seven-production"), + ("docs", "production_docs"), + ): + with self.subTest(target=target), self.assertRaises(ValueError): + adapter.shared.prepared_target( + { + "release": f.release, + "artifact_inventory": { + **f.artifacts, + "target": target, + "claim_scope": scope, + }, + } + ) + + def write_mapping(self): + f = self.fixture + + def write(name, value): + raw = adapter.canonical_json(value) + (f.directory / name).write_bytes(raw) + return { + "path": name, + "sha256": adapter.shared.sha(raw), + "size_bytes": len(raw), + } + + candidate = { + "schema_version": "openadapt.measured-cloud-release-candidate/v1", + "state": "ready-for-review", + "target": "cloud", + "release": f.release, + "artifact_inventory": f.artifacts, + "proposed_release_identity": { + "schema_version": "openadapt.monotonic-production-release/v1", + "channel": "production", + "sequence": 2, + "previous_admission_sha256": "sha256:" + "e" * 64, + }, + } + mapping = { + "schema_version": "openadapt.measured-cloud-admission-mapping/v1", + "candidate": write("candidate.json", candidate), + "derivative": write("derivative.json", f.derivative), + "publication_staging": write("staging.json", f.staging), + "publication_observation": None, + "provenance": {"test_only": "mocked after separate crypto tests"}, + "retained_files": [ + { + "path": path.name, + "sha256": "sha256:" + digest, + "size_bytes": len(raw), + } + for digest, (path, raw) in f.inventory.items() + ], + } + reference = write("mapping.json", mapping) + return f.directory / reference["path"], reference["sha256"] + + def test_prepare_composes_verified_inputs_without_issuer_or_signer(self): + f = self.fixture + path, digest = self.write_mapping() + with ( + mock.patch.object(adapter, "verify_derivative_provenance") as provenance, + mock.patch.object( + adapter.shared, + "current_context", + side_effect=AssertionError("no issuer in input preparation"), + ), + mock.patch.object( + adapter.shared.software, + "keychain_read", + side_effect=AssertionError("no signing"), + ), + ): + result = adapter.prepare_inputs(path, digest, gh=f.gh) + provenance.assert_called_once() + self.assertEqual( + provenance.call_args.args[1], (f.directory / "derivative.json").read_bytes() + ) + self.assertEqual(result["release"], f.release) + self.assertEqual(result["artifact_inventory"], f.artifacts) + self.assertEqual( + result["campaign_summary"], + adapter.verify_derivative(f.derivative, f.inventory), + ) + self.assertEqual(result["bundle_version"], "1.35.1-hosted-reference.1") + self.assertEqual(result["mapping_sha256"], digest) + self.assertEqual(set(result["commitments"]), adapter.shared.FILE_COMMITMENTS) + self.assertNotIn("raw_evidence_refs", result) + self.assertNotIn("admitted_subject", result) + + def test_prepare_requires_provenance_and_actual_candidate_bytes(self): + path, digest = self.write_mapping() + with ( + mock.patch.object( + adapter, + "verify_derivative_provenance", + side_effect=ValueError("crypto refused"), + ), + self.assertRaisesRegex(ValueError, "crypto refused"), + ): + adapter.prepare_inputs(path, digest, gh=self.fixture.gh) + (self.fixture.directory / "candidate.json").write_bytes(b"{}") + with ( + mock.patch.object(adapter, "verify_derivative_provenance") as provenance, + self.assertRaisesRegex(ValueError, "bytes differ"), + ): + adapter.prepare_inputs(path, digest, gh=self.fixture.gh) + provenance.assert_not_called() + + def test_flow_cli_cannot_consume_cloud_inputs_or_request_another_target(self): + from datetime import datetime, timezone + + path = self.fixture.directory / "request.json" + request = { + "phase": "receipt", + "issued_at": datetime.now(timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ"), + "request_handle": "qair_" + "A" * 43, + "issuer_source_commit": "a" * 40, + "expires_at": None, + } + path.write_bytes(adapter.canonical_json(request)) + output = self.fixture.directory / "unsigned.json" + args = [ + "--mapping", + str(path), + "--mapping-sha256", + adapter.shared.sha(path.read_bytes()), + "--phase-request", + str(path), + "--output", + str(output), + ] + inputs = { + "release": self.fixture.release, + "artifact_inventory": self.fixture.artifacts, + } + with ( + mock.patch.object( + adapter.shared, "prepare_inputs", return_value=inputs + ) as prepare, + mock.patch.object(adapter.shared, "current_context") as context, + mock.patch.object(adapter.shared.software, "keychain_read") as key, + mock.patch.object(adapter.shared, "persist_once") as state, + ): + self.assertEqual(adapter.shared.main(args), 1) + prepare.assert_called_once() + context.assert_not_called() + key.assert_not_called() + state.assert_not_called() + self.assertFalse(output.exists()) + with self.assertRaises(SystemExit): + adapter.shared.main(args + ["--target", "cloud"]) + + +class CloudPublicationObservationTests(unittest.TestCase): + def setUp(self): + temporary = tempfile.TemporaryDirectory() + self.addCleanup(temporary.cleanup) + self.fixture = f = MeasuredFixture(temporary.name) + self.owner = f.directory / "mapping.json" + self.derivative_raw = adapter.canonical_json(f.derivative) + prior = f.readback["provider_observation"] + context = f.readback["observed_runtime_context"] + runtime = f.runtime + self.observation = { + "observed_at": "2026-09-11T01:00:00.123Z", + "app_id": prior["modal"]["app_id"], + "app_version": prior["modal"]["app_version"], + "source_commit": f.source, + "netlify_deploy_id": prior["host"]["deploy_id"], + "runtime_boundary_id": "test-only-boundary", + "deployment_manifest_sha256": f.subject["deployment_manifest_sha256"], + "control_image_id": context["control_image_id"], + "sandbox_image_id": context["modal_image_id"], + "provider_observation": {key: prior[key] for key in ("host", "modal")}, + "health": { + "ready": True, + "service": "runner", + "mode": "live", + "boundary_id": "test-only-boundary", + "flow_version": runtime["openadapt_flow"], + "modal_sdk": runtime["modal_sdk"], + "fastapi": runtime["fastapi"], + "starlette": runtime["starlette"], + "runner_build": runtime["runner_build"], + "runner_artifact_sha256": runtime["runner_artifact_sha256"], + "sandbox_network_policy": runtime["sandbox_network_policy"], + "deployment_manifest_sha256": f.subject["deployment_manifest_sha256"], + "runtime_build_identity": f.build, + "runtime_environment_sha256": context["runtime_environment_sha256"], + "function_readbacks": f.readback["function_readbacks"], + "deployment_readback": { + "app_id": prior["modal"]["app_id"], + "app_version": prior["modal"]["app_version"], + "function_id": prior["modal"]["function_id"], + "function_definition_id": prior["modal"]["function_definition_id"], + "control_image_id": context["control_image_id"], + "sandbox_image_id": context["modal_image_id"], + }, + }, + } + self.run = { + "run_started_at": "2026-09-11T00:59:00Z", + "updated_at": "2026-09-11T01:01:00Z", + } + self.staging = {**f.staging, "observed_at": "2026-09-11T01:00:00Z"} + + def references(self, observation=None, **changes): + f = self.fixture + + def write(name, value): + raw = adapter.canonical_json(value) + (f.directory / name).write_bytes(raw) + return { + "path": name, + "sha256": adapter.shared.sha(raw), + "size_bytes": len(raw), + } + + observation = observation or self.observation + raw_ref = write("fresh-readback.json", observation) + proof = { + "schema_version": "openadapt.hosted-publication-observation/v1", + "producer": producer(), + "qualification_derivative_sha256": adapter.shared.sha(self.derivative_raw)[ + 7: + ], + "subject": f.subject, + "observed_at": observation["observed_at"], + "provider_readback_sha256": raw_ref["sha256"][7:], + **changes, + } + return { + "artifact": write("fresh-observation.json", proof), + "provider_readback": raw_ref, + "provenance": {"test_only": "separate fixed-workflow crypto tests"}, + } + + def verify(self, reference, staging=None): + f = self.fixture + return adapter.verify_publication_observation( + self.owner, + reference, + f.derivative, + self.derivative_raw, + f.inventory, + staging or self.staging, + gh=f.gh, + ) + + def test_original_time_cannot_be_refreshed_without_new_observation(self): + self.assertIsNone(self.verify(None, self.fixture.staging)) + with self.assertRaisesRegex(ValueError, "staging time differs"): + self.verify(None) + + def test_separate_attested_refresh_preserves_original_bytes(self): + f = self.fixture + originals = {digest: raw for digest, (_, raw) in f.inventory.items()} + reference = self.references() + with mock.patch.object( + adapter, "verify_derivative_provenance", return_value=self.run + ) as verifier: + self.assertIsNone(self.verify(reference)) + verifier.assert_called_once() + self.assertEqual( + verifier.call_args.args[1], + (f.directory / reference["artifact"]["path"]).read_bytes(), + ) + self.assertEqual(verifier.call_args.args[2], producer()) + self.assertEqual( + {digest: path.read_bytes() for digest, (path, _) in f.inventory.items()}, + originals, + ) + self.assertEqual(adapter.canonical_json(f.derivative), self.derivative_raw) + + def test_unverified_or_wrong_original_subject_and_bytes_refuse(self): + reference = self.references() + with ( + mock.patch.object( + adapter, + "verify_derivative_provenance", + side_effect=ValueError("crypto refused"), + ), + self.assertRaisesRegex(ValueError, "crypto refused"), + ): + self.verify(reference) + for changes in ( + {"qualification_derivative_sha256": "a" * 64}, + {"subject": {**self.fixture.subject, "source_commit": "b" * 40}}, + {"provider_readback_sha256": "a" * 64}, + {"observed_at": "2026-09-11T01:00:01.000Z"}, + {"extra": "not allowed"}, + ): + with self.subTest(changes=changes), self.assertRaises(ValueError): + self.verify(self.references(**changes)) + reference = self.references() + (self.fixture.directory / reference["provider_readback"]["path"]).write_bytes( + b"{}" + ) + with self.assertRaisesRegex(ValueError, "bytes differ"): + self.verify(reference) + + def test_staging_and_millisecond_time_must_fit_authenticated_attempt(self): + for observed_at in ("2026-09-11T00:58:59.999Z", "2026-09-11T01:01:01.000Z"): + changed = {**self.observation, "observed_at": observed_at} + with ( + self.subTest(observed_at=observed_at), + mock.patch.object( + adapter, "verify_derivative_provenance", return_value=self.run + ), + self.assertRaisesRegex(ValueError, "outside its authenticated"), + ): + self.verify(self.references(changed)) + for value in ( + "2026-09-11T01:00:00Z", + "2026-09-11T01:00:00.123000Z", + "2026-09-11T01:00:00.123+00:00", + "invalid", + ): + with ( + self.subTest(value=value), + self.assertRaisesRegex(ValueError, "UTC milliseconds"), + ): + self.verify(self.references({**self.observation, "observed_at": value})) + with ( + mock.patch.object( + adapter, "verify_derivative_provenance", return_value=self.run + ), + self.assertRaisesRegex(ValueError, "staging time differs"), + ): + self.verify( + self.references(), + {**self.staging, "observed_at": "2026-09-11T01:00:01Z"}, + ) + + def test_changed_provider_image_build_or_health_refuses(self): + for path, value in ( + (("source_commit",), "a" * 40), + (("app_version",), "v3"), + (("control_image_id",), "other"), + (("sandbox_image_id",), "other"), + (("netlify_deploy_id",), "b" * 24), + (("provider_observation", "modal", "function_id"), "other"), + (("health", "runtime_build_identity"), {}), + (("health", "runtime_environment_sha256"), "a" * 64), + (("health", "flow_version"), "1.34.0"), + (("health", "ready"), 1), + (("health", "function_readbacks", 1, "function_definition_id"), "other"), + (("health", "deployment_readback", "function_definition_id"), "other"), + ): + changed = copy.deepcopy(self.observation) + target = changed + for field in path[:-1]: + target = target[field] + target[path[-1]] = value + with ( + self.subTest(path=path), + mock.patch.object( + adapter, "verify_derivative_provenance", return_value=self.run + ), + self.assertRaises(ValueError), + ): + self.verify(self.references(changed)) + + +if __name__ == "__main__": + unittest.main()