From dba064cb09686fdff6067aff1f20f9cb0edc6381 Mon Sep 17 00:00:00 2001 From: Richard Abrich Date: Fri, 21 Aug 2026 19:10:24 -0400 Subject: [PATCH 1/8] feat: require signed runtime authority for Production and campaign actuation Standard and Regulated actuation now refuses unless a verified v2 qualification authority rides with the run: one private, owner-only handoff file that binds the signed admission, its expectation, and the signer registry, re-read at every input edge with revocation monotonicity. A customer-local runner must also carry a fresh permit trust snapshot; a managed Cloud run revalidates through its remote delivery permit. Qualification trials keep their own fail-closed authority: one signed, one-shot, non-production campaign permit bound to the exact trial, consumed through GovernedRunAuthorization so an incomplete binding can never authorize an edge. --- openadapt_flow/__main__.py | 308 +++++++++++--- openadapt_flow/deployment.py | 4 + openadapt_flow/private_file.py | 63 +++ openadapt_flow/production_qualification.py | 391 ++++++++++++++++++ openadapt_flow/qualification.py | 53 +++ .../qualification_campaign_authority.py | 297 +++++++++++++ .../qualification_campaign_permit.py | 2 +- openadapt_flow/run_gate.py | 17 +- .../runner/business_decision_service.py | 62 +-- openadapt_flow/runtime/authorization.py | 67 +++ openadapt_flow/runtime/replayer.py | 106 +++++ tests/test_execution_profiles.py | 30 +- tests/test_production_qualification.py | 304 ++++++++++++++ tests/test_qualification_admission_v2.py | 20 +- .../test_qualification_campaign_authority.py | 204 +++++++++ tests/test_qualification_project.py | 30 ++ 16 files changed, 1851 insertions(+), 107 deletions(-) create mode 100644 openadapt_flow/private_file.py create mode 100644 openadapt_flow/production_qualification.py create mode 100644 openadapt_flow/qualification_campaign_authority.py create mode 100644 tests/test_production_qualification.py create mode 100644 tests/test_qualification_campaign_authority.py diff --git a/openadapt_flow/__main__.py b/openadapt_flow/__main__.py index f81a019a..31946ddb 100644 --- a/openadapt_flow/__main__.py +++ b/openadapt_flow/__main__.py @@ -56,7 +56,6 @@ import argparse import json -import os import sys from contextlib import contextmanager from pathlib import Path @@ -664,6 +663,8 @@ def _configured_replayer( ] = "customer_local", remote_delivery_run_id: Optional[str] = None, managed_dispatch_binding=None, + production_qualification_guard=None, + qualification_campaign_guard=None, runtime_config=None, ): """Wire the grounding, verification, and actuation layers into a Replayer. @@ -700,6 +701,8 @@ def _configured_replayer( delivery_authority_kind=delivery_authority_kind, remote_delivery_run_id=remote_delivery_run_id, managed_dispatch_binding=managed_dispatch_binding, + production_qualification_guard=production_qualification_guard, + qualification_campaign_guard=qualification_campaign_guard, runtime_config=runtime_config, checkpoint_key=checkpoint_key, ) @@ -726,6 +729,8 @@ def _build_and_run_replayer( ] = "customer_local", remote_delivery_run_id: Optional[str] = None, managed_dispatch_binding=None, + production_qualification_guard=None, + qualification_campaign_guard=None, runtime_config=None, execution_target_kind: Optional["ExecutionTargetKind"] = None, surface_override: bool = False, @@ -747,6 +752,8 @@ def _build_and_run_replayer( delivery_authority_kind=delivery_authority_kind, remote_delivery_run_id=remote_delivery_run_id, managed_dispatch_binding=managed_dispatch_binding, + production_qualification_guard=production_qualification_guard, + qualification_campaign_guard=qualification_campaign_guard, runtime_config=runtime_config, ).run( workflow, @@ -815,6 +822,8 @@ def _replay_desktop( ] = "customer_local", remote_delivery_run_id: Optional[str] = None, managed_dispatch_binding=None, + production_qualification_guard=None, + qualification_campaign_guard=None, runtime_config=None, run_id: Optional[str] = None, ) -> int: @@ -858,6 +867,8 @@ def _replay_desktop( delivery_authority_kind=delivery_authority_kind, remote_delivery_run_id=remote_delivery_run_id, managed_dispatch_binding=managed_dispatch_binding, + production_qualification_guard=production_qualification_guard, + qualification_campaign_guard=qualification_campaign_guard, runtime_config=runtime_config, execution_target_kind=_report_backend_kind(backend_cfg.kind), surface_override=bool(getattr(args, "_surface_override", False)), @@ -1571,6 +1582,12 @@ def _cmd_replay(args: argparse.Namespace) -> int: ), remote_delivery_run_id=getattr(args, "_remote_delivery_run_id", None), managed_dispatch_binding=getattr(args, "_managed_dispatch_binding", None), + production_qualification_guard=getattr( + args, "_production_qualification_guard", None + ), + qualification_campaign_guard=getattr( + args, "_qualification_campaign_guard", None + ), runtime_config=cfg.runtime, run_id=getattr(args, "_qualification_run_id", None), ) @@ -1649,6 +1666,12 @@ def _cmd_replay(args: argparse.Namespace) -> int: managed_dispatch_binding=getattr( args, "_managed_dispatch_binding", None ), + production_qualification_guard=getattr( + args, "_production_qualification_guard", None + ), + qualification_campaign_guard=getattr( + args, "_qualification_campaign_guard", None + ), runtime_config=cfg.runtime, execution_target_kind="web", surface_override=bool(getattr(args, "_surface_override", False)), @@ -1692,14 +1715,17 @@ def _cmd_run(args: argparse.Namespace) -> int: """ from openadapt_flow.execution_profiles import ( execution_profile_contract, + requires_signed_qualification_admission, resolve_execution_profile, ) from openadapt_flow.ir import Workflow - from openadapt_flow.qualification_admission import ( - QualificationAdmissionError, - expected_from_payload, - load_qualification_signer_trust, - verify_qualification_admission, + from openadapt_flow.production_qualification import ( + ProductionQualificationAuthorityError, + ProductionQualificationGuard, + ) + from openadapt_flow.qualification_campaign_authority import ( + QualificationCampaignAuthorityError, + QualificationCampaignGuard, ) from openadapt_flow.run_gate import ( build_qualification_case_authorization, @@ -1826,13 +1852,16 @@ def _cmd_run(args: argparse.Namespace) -> int: "dispatch. Nothing was executed." ) return 2 + runtime_params = _replay_params(args.param, getattr(args, "params_file", None)) + runtime_worklists = ( + qualification_case["worklists"] + if qualification_case is not None + else _resolve_worklists(getattr(args, "worklist", None), workflow) + ) + managed_binding = None + authorization = None + local_authorization = None if dispatch_file: - runtime_params = _replay_params(args.param, getattr(args, "params_file", None)) - runtime_worklists = ( - qualification_case["worklists"] - if qualification_case is not None - else _resolve_worklists(getattr(args, "worklist", None), workflow) - ) try: managed_binding = read_managed_dispatch_envelope(Path(dispatch_file)) authorization = managed_binding.authorization @@ -1841,29 +1870,6 @@ def _cmd_run(args: argparse.Namespace) -> int: "run REFUSED: managed dispatch binding is invalid. Nothing was executed." ) return 2 - if authorization.qualification_admission is None: - print( - "run REFUSED: managed production dispatch has no signed " - "qualification admission. Nothing was executed." - ) - return 2 - try: - signer_trust = load_qualification_signer_trust( - os.environ.get("OPENADAPT_QUALIFICATION_SIGNERS_JSON", "") - ) - verify_qualification_admission( - authorization.qualification_admission, - trusted_signers=signer_trust, - expected=expected_from_payload( - authorization.qualification_admission.payload - ), - ) - except QualificationAdmissionError: - print( - "run REFUSED: qualification admission is not signed by an " - "active trusted authority. Nothing was executed." - ) - return 2 local_authorization = build_runtime_authorization( workflow, report, @@ -1900,6 +1906,97 @@ def _cmd_run(args: argparse.Namespace) -> int: "run REFUSED: managed dispatch does not match this exact run. Nothing was executed." ) return 2 + + if getattr(args, "dry_run", False) or getattr(args, "explain", False): + # Report-only paths retain the complete local gate and, when supplied, + # the exact managed-dispatch check above. They do not require an + # actuation authority because they never cross an input edge. + return 0 if report.passed else 2 + + production_guard = None + campaign_guard = None + campaign_permit_binding = None + if qualification_case is not None: + authority_file = getattr(args, "qualification_campaign_authority_file", None) + if not authority_file: + print( + "run REFUSED: qualification actuation requires a private signed " + "non-production campaign authority file. Nothing was executed." + ) + return 2 + try: + campaign_guard = QualificationCampaignGuard( + authority_file, + workflow=workflow, + case_id=qualification_case["case"].id, + input_digest=runtime_inputs_digest( + workflow, gate_params, runtime_worklists + ), + campaign_id=qualification_case["campaign_id"], + run_id=qualification_case["run_id"], + ) + campaign_permit_binding = campaign_guard.authorization_binding(workflow) + except QualificationCampaignAuthorityError: + print( + "run REFUSED: the non-production qualification campaign permit " + "is invalid, expired, consumed, or does not match this exact trial. " + "Nothing was executed." + ) + return 2 + args._qualification_campaign_guard = campaign_guard + requires_production_authority = ( + qualification_case is None + and requires_signed_qualification_admission(selected_profile, will_actuate=True) + ) + if requires_production_authority: + authority_file = getattr(args, "qualification_authority_file", None) + if not authority_file: + print( + "run REFUSED: Production actuation requires a private signed v2 " + "qualification authority file. Nothing was executed." + ) + return 2 + try: + production_guard = ProductionQualificationGuard( + authority_file, + remote_permit_revalidation=bool(dispatch_file), + ) + production_binding = production_guard.authorization_binding(workflow) + except ProductionQualificationAuthorityError: + print( + "run REFUSED: the Production qualification authority is invalid, " + "expired, revoked, or does not match this exact run. Nothing was " + "executed." + ) + return 2 + args._production_qualification_guard = production_guard + if local_authorization is None: + local_authorization = build_runtime_authorization( + workflow, + report, + params=runtime_params, + worklists=runtime_worklists, + ) + local_authorization = local_authorization.model_copy(update=production_binding) + + if dispatch_file: + assert managed_binding is not None + assert authorization is not None + assert local_authorization is not None + production_fields = tuple( + field + for field in local_authorization.model_fields + if field.startswith("production_qualification_") + ) + if any( + getattr(authorization, field) != getattr(local_authorization, field) + for field in production_fields + ): + print( + "run REFUSED: managed dispatch does not bind the active Production " + "qualification authority. Nothing was executed." + ) + return 2 args._governed_run_authorization = authorization args._managed_dispatch_binding = managed_binding args._delivery_authority_kind = "cloud_runner" @@ -1907,6 +2004,7 @@ def _cmd_run(args: argparse.Namespace) -> int: if qualification_case is not None: runtime_worklists = qualification_case["worklists"] + assert campaign_permit_binding is not None try: authorization = build_qualification_case_authorization( workflow, @@ -1916,6 +2014,7 @@ def _cmd_run(args: argparse.Namespace) -> int: worklists=runtime_worklists, campaign_id=qualification_case["campaign_id"], run_id=qualification_case["run_id"], + campaign_permit_binding=campaign_permit_binding, ) except ValueError: print( @@ -1928,20 +2027,14 @@ def _cmd_run(args: argparse.Namespace) -> int: args._remote_delivery_run_id = None args._qualification_run_id = qualification_case["run_id"] - if getattr(args, "dry_run", False) or getattr(args, "explain", False): - # A managed dry run also validates its protected Cloud handoff above. - # Report-only: it never executes, regardless of the verdict. - return 0 if report.passed else 2 if not dispatch_file and qualification_case is None: - runtime_params = _replay_params(args.param, getattr(args, "params_file", None)) - runtime_worklists = _resolve_worklists( - getattr(args, "worklist", None), workflow - ) - args._governed_run_authorization = build_runtime_authorization( - workflow, - report, - params=runtime_params, - worklists=runtime_worklists, + args._governed_run_authorization = local_authorization or ( + build_runtime_authorization( + workflow, + report, + params=runtime_params, + worklists=runtime_worklists, + ) ) args._delivery_authority_kind = "customer_local" args._remote_delivery_run_id = None @@ -2027,6 +2120,88 @@ def _cmd_resume(args: argparse.Namespace) -> int: ) return 3 + production_guard = None + campaign_guard = None + retained_authorization = manifest.governed_authorization + if ( + retained_authorization is not None + and retained_authorization.execution_profile in {"standard", "regulated"} + and retained_authorization.qualification_case_id is None + ): + from openadapt_flow.production_qualification import ( + ProductionQualificationAuthorityError, + ProductionQualificationGuard, + ) + + authority_file = getattr(args, "qualification_authority_file", None) + if not authority_file: + print( + "Resume REFUSED: Production actuation requires the private v2 " + "qualification authority again. Nothing was executed." + ) + return 3 + + if ( + retained_authorization is not None + and retained_authorization.qualification_case_id is not None + ): + from openadapt_flow.qualification_campaign_authority import ( + QualificationCampaignAuthorityError, + QualificationCampaignGuard, + ) + + authority_file = getattr(args, "qualification_campaign_authority_file", None) + if not authority_file: + print( + "Resume REFUSED: qualification actuation requires the private " + "signed non-production campaign authority again. Nothing was " + "executed." + ) + return 3 + try: + campaign_guard = QualificationCampaignGuard( + authority_file, + workflow=workflow, + case_id=retained_authorization.qualification_case_id, + input_digest=retained_authorization.runtime_inputs_digest, + ) + campaign_refusal = campaign_guard.authorization_refusal( + workflow, retained_authorization + ) + except QualificationCampaignAuthorityError: + campaign_refusal = "qualification campaign authority is invalid" + if campaign_refusal is not None: + print( + "Resume REFUSED: the non-production qualification campaign " + "permit is invalid, expired, consumed, or differs from the " + "retained run. Nothing was executed." + ) + return 3 + try: + production_guard = ProductionQualificationGuard( + authority_file, + remote_permit_revalidation=( + manifest.delivery_authority_kind == "cloud_runner" + ), + ) + production_binding = production_guard.authorization_binding(workflow) + except ProductionQualificationAuthorityError: + print( + "Resume REFUSED: the Production qualification authority is invalid, " + "expired, revoked, or does not match this exact run. Nothing was " + "executed." + ) + return 3 + if any( + getattr(retained_authorization, field, None) != value + for field, value in production_binding.items() + ): + print( + "Resume REFUSED: the Production qualification authority differs " + "from the retained run. Nothing was executed." + ) + return 3 + # A GUI automation cannot be resumed without a LIVE backend/vision, so build # a fresh Replayer here (deployment wiring from --config) and hand it to the # durable resume entrypoint, which re-binds params from the run manifest. @@ -2076,6 +2251,8 @@ def _resume_with(backend: "Backend") -> "RunReport": checkpoint_key=ckpt_key, allow_model_grounding=allow_egress, managed_dispatch_binding=managed_binding, + production_qualification_guard=production_guard, + qualification_campaign_guard=campaign_guard, ) return resume( run_dir, @@ -5153,6 +5330,15 @@ def build_parser() -> argparse.ArgumentParser: help="Refuse unless the bundle's compiler version equals this", ) p.add_argument("--managed-dispatch-file", default=None, help=argparse.SUPPRESS) + p.add_argument( + "--qualification-authority-file", + default=None, + metavar="PATH", + help=( + "Private owner-only v2 Production qualification authority. Required " + "for Standard or Regulated actuation; not required for --dry-run" + ), + ) p.add_argument( "--dry-run", "--explain", @@ -5181,6 +5367,24 @@ def build_parser() -> argparse.ArgumentParser: ) p.add_argument("run_dir", help="The paused run directory (holds checkpoints)") p.add_argument("--managed-dispatch-file", default=None, help=argparse.SUPPRESS) + p.add_argument( + "--qualification-authority-file", + default=None, + metavar="PATH", + help=( + "Private owner-only v2 Production qualification authority for the " + "resumed input edge" + ), + ) + p.add_argument( + "--qualification-campaign-authority-file", + default=None, + metavar="PATH", + help=( + "Private signed non-production campaign authority for a resumed " + "qualification input edge" + ), + ) p.add_argument( "--url", default=None, @@ -5667,6 +5871,14 @@ def build_parser() -> argparse.ArgumentParser: ) q.add_argument("--campaign-id", required=True, help="Local campaign identity") q.add_argument("--run-id", required=True, help="One local case-attempt identity") + q.add_argument( + "--qualification-campaign-authority-file", + required=True, + metavar="PATH", + help=( + "Private owner-only signed non-production authority for this exact trial" + ), + ) q.add_argument( "--run-dir", required=True, help="New durable case-attempt directory" ) diff --git a/openadapt_flow/deployment.py b/openadapt_flow/deployment.py index 15d6c48e..09449741 100644 --- a/openadapt_flow/deployment.py +++ b/openadapt_flow/deployment.py @@ -733,6 +733,8 @@ def build_replayer( ] = "customer_local", remote_delivery_run_id: Optional[str] = None, managed_dispatch_binding: Any = None, + production_qualification_guard: Any = None, + qualification_campaign_guard: Any = None, runtime_config: Optional["RuntimeSection"] = None, phi_mode: Optional[bool] = None, checkpoint_key: Optional[str] = None, @@ -894,6 +896,8 @@ def build_replayer( delivery_authority_kind=delivery_authority_kind, remote_delivery_run_id=remote_delivery_run_id, managed_dispatch_binding=managed_dispatch_binding, + production_qualification_guard=production_qualification_guard, + qualification_campaign_guard=qualification_campaign_guard, checkpoint_key=checkpoint_key, require_settled=require_settled, qualification_fault_driver=qualification_fault_driver, diff --git a/openadapt_flow/private_file.py b/openadapt_flow/private_file.py new file mode 100644 index 00000000..2d9cfcf3 --- /dev/null +++ b/openadapt_flow/private_file.py @@ -0,0 +1,63 @@ +"""Shared private-file checks for local trust and key handoffs.""" + +from __future__ import annotations + + +class PrivateFileAclError(ValueError): + """The active platform cannot prove that a private file has a safe ACL.""" + + +def windows_descriptor_has_private_acl(descriptor: int) -> bool: + """Require an opened Windows file to be private to its service identity.""" + + try: + import msvcrt + + import ntsecuritycon + import win32api + import win32con + import win32security + + get_osfhandle = getattr(msvcrt, "get_osfhandle", None) + if not callable(get_osfhandle): + raise AttributeError("Windows descriptor conversion is unavailable") + handle = get_osfhandle(descriptor) + security = win32security.GetSecurityInfo( + handle, + win32security.SE_FILE_OBJECT, + win32security.OWNER_SECURITY_INFORMATION + | win32security.DACL_SECURITY_INFORMATION, + ) + owner = security.GetSecurityDescriptorOwner() + token = win32security.OpenProcessToken( + win32api.GetCurrentProcess(), win32con.TOKEN_QUERY + ) + current = win32security.GetTokenInformation(token, win32security.TokenUser)[0] + if not win32security.EqualSid(owner, current): + return False + dacl = security.GetSecurityDescriptorDacl() + if dacl is None: + return False + allowed = ( + current, + win32security.ConvertStringSidToSid("S-1-5-18"), + win32security.ConvertStringSidToSid("S-1-5-32-544"), + ) + for index in range(dacl.GetAceCount()): + header, mask, sid = dacl.GetAce(index) + ace_type = header[0] + if ace_type == ntsecuritycon.ACCESS_DENIED_ACE_TYPE: + continue + if ace_type != ntsecuritycon.ACCESS_ALLOWED_ACE_TYPE: + return False + if mask and not any(win32security.EqualSid(sid, item) for item in allowed): + return False + return True + except (ImportError, OSError, AttributeError) as exc: + raise PrivateFileAclError( + "Windows private-file ACL verification is unavailable; install the " + "OpenAdapt Windows runtime" + ) from exc + + +__all__ = ["PrivateFileAclError", "windows_descriptor_has_private_acl"] diff --git a/openadapt_flow/production_qualification.py b/openadapt_flow/production_qualification.py new file mode 100644 index 00000000..3cdac7c1 --- /dev/null +++ b/openadapt_flow/production_qualification.py @@ -0,0 +1,391 @@ +"""Fail-closed v2 qualification authority for Production actuation. + +The qualification project and its persisted certification are evidence. They +are not authority to actuate. A Standard or Regulated run must also carry the +signed, expiring, revocable v2 admission defined in +``qualification_admission_v2``. This module loads that authority from one +private local handoff and re-verifies it at every input edge. + +The handoff keeps two trust sources separate: + +* the signed admission states what the qualification authority admitted; +* ``expected`` states the values reproduced by the live runner or deployment. + +Flow compares both and independently binds the fields that the sealed bundle +can reproduce. A customer-local runner must also provide a fresh permit trust +snapshot. A managed Cloud run obtains the equivalent current revocation and +expiry decision atomically from its remote delivery permit before each edge. +""" + +from __future__ import annotations + +import json +import os +import stat +from pathlib import Path +from typing import Literal +from uuid import UUID + +from pydantic import BaseModel, ConfigDict, Field, ValidationError, field_validator + +from openadapt_flow.ir import Workflow +from openadapt_flow.private_file import ( + PrivateFileAclError, + windows_descriptor_has_private_acl, +) +from openadapt_flow.qualification_admission_v2 import ( + QualificationAdmissionEnvelope, + QualificationAdmissionError, + QualificationAdmissionExpected, + QualificationPermitTrustSnapshot, + QualificationSignerRegistry, + VerifiedQualificationAdmission, + contract_sha256, + verify_qualification_admission, + verify_qualification_admission_for_actuation, +) + +MAX_AUTHORITY_BYTES = 512 * 1024 +AUTHORITY_SCHEMA = "openadapt.production-qualification-authority/v1" + + +class ProductionQualificationAuthorityError(ValueError): + """The private Production qualification handoff is unsafe or invalid.""" + + +class ProductionQualificationAuthority(BaseModel): + """Closed handoff from an independent qualification and runtime authority.""" + + model_config = ConfigDict(extra="forbid", frozen=True) + + schema_version: Literal["openadapt.production-qualification-authority/v1"] = ( + AUTHORITY_SCHEMA + ) + qualification_admission: QualificationAdmissionEnvelope + qualification_admission_sha256: str = Field(pattern=r"^[a-f0-9]{64}$") + expected: QualificationAdmissionExpected + qualification_signer_registry: QualificationSignerRegistry + qualification_signer_registry_sha256: str = Field(pattern=r"^[a-f0-9]{64}$") + permit_trust_snapshot: QualificationPermitTrustSnapshot | None = None + revoked_admission_ids: tuple[str, ...] = () + + @field_validator("revoked_admission_ids") + @classmethod + def _ordered_revocations(cls, value: tuple[str, ...]) -> tuple[str, ...]: + if value != tuple(sorted(set(value))): + raise ValueError("revoked qualification admission ids must be ordered") + for item in value: + try: + parsed = UUID(item) + except ValueError as exc: + raise ValueError( + "revoked qualification admission id must be a canonical UUID" + ) from exc + if str(parsed) != item: + raise ValueError( + "revoked qualification admission id must be a canonical UUID" + ) + return value + + def model_post_init(self, __context: object) -> None: + if ( + self.qualification_admission.artifact_sha256() + != self.qualification_admission_sha256 + ): + raise ValueError("qualification admission digest does not match") + if ( + self.qualification_signer_registry.artifact_sha256() + != self.qualification_signer_registry_sha256 + ): + raise ValueError("qualification signer registry digest does not match") + + def immutable_binding_sha256(self) -> str: + """Identify the immutable admission, expectation, and signer authority. + + The permit snapshot must be refreshed before local input delivery, and + the revocation set can only grow. Neither mutable safety input is part + of the retained run identity. + """ + + return contract_sha256( + { + "schema_version": self.schema_version, + "qualification_admission_sha256": (self.qualification_admission_sha256), + "expected": self.expected.model_dump(mode="json"), + "qualification_signer_registry_sha256": ( + self.qualification_signer_registry_sha256 + ), + } + ) + + +def _read_private_json(path: Path) -> object: + """Read one bounded, owner-only regular file without following a link.""" + + path = Path(path) + lstat_before = None + if os.name == "nt": + try: + lstat_before = os.lstat(path) + except OSError as exc: + raise ProductionQualificationAuthorityError( + "qualification authority file could not be inspected safely" + ) from exc + if stat.S_ISLNK(lstat_before.st_mode): + raise ProductionQualificationAuthorityError( + "qualification authority file must not be a link" + ) + flags = os.O_RDONLY + if hasattr(os, "O_NOFOLLOW"): + flags |= os.O_NOFOLLOW + try: + descriptor = os.open(path, flags) + except OSError as exc: + raise ProductionQualificationAuthorityError( + "qualification authority file could not be opened safely" + ) from exc + try: + metadata = os.fstat(descriptor) + windows_path_changed = False + if lstat_before is not None: + try: + lstat_after = os.lstat(path) + windows_path_changed = ( + stat.S_ISLNK(lstat_after.st_mode) + or lstat_after.st_ino != lstat_before.st_ino + or lstat_after.st_dev != lstat_before.st_dev + ) + except OSError: + windows_path_changed = True + if os.name == "nt" and not windows_path_changed: + try: + unsafe_permissions = not windows_descriptor_has_private_acl(descriptor) + except PrivateFileAclError as exc: + raise ProductionQualificationAuthorityError(str(exc)) from exc + else: + unsafe_permissions = ( + not hasattr(os, "geteuid") + or metadata.st_uid != os.geteuid() + or stat.S_IMODE(metadata.st_mode) != 0o600 + ) + if ( + not stat.S_ISREG(metadata.st_mode) + or unsafe_permissions + or windows_path_changed + or metadata.st_size > MAX_AUTHORITY_BYTES + ): + raise ProductionQualificationAuthorityError( + "qualification authority file is not a private regular file" + ) + chunks: list[bytes] = [] + remaining = metadata.st_size + while remaining: + chunk = os.read(descriptor, remaining) + if not chunk: + raise ProductionQualificationAuthorityError( + "qualification authority file ended during the safe read" + ) + chunks.append(chunk) + remaining -= len(chunk) + if os.read(descriptor, 1): + raise ProductionQualificationAuthorityError( + "qualification authority file changed during the safe read" + ) + after = os.fstat(descriptor) + if ( + after.st_dev != metadata.st_dev + or after.st_ino != metadata.st_ino + or after.st_size != metadata.st_size + or after.st_mtime_ns != metadata.st_mtime_ns + ): + raise ProductionQualificationAuthorityError( + "qualification authority file changed during the safe read" + ) + raw = b"".join(chunks) + finally: + os.close(descriptor) + try: + return json.loads(raw.decode("utf-8")) + except (UnicodeDecodeError, json.JSONDecodeError) as exc: + raise ProductionQualificationAuthorityError( + "qualification authority file is not valid JSON" + ) from exc + + +def load_production_qualification_authority( + path: Path | str, +) -> ProductionQualificationAuthority: + """Load one exact private Production qualification authority file.""" + + raw = _read_private_json(Path(path)) + try: + authority = ProductionQualificationAuthority.model_validate(raw) + except (TypeError, ValueError, ValidationError) as exc: + raise ProductionQualificationAuthorityError( + "qualification authority file has an invalid exact binding" + ) from exc + if authority.model_dump(mode="json") != raw: + raise ProductionQualificationAuthorityError( + "qualification authority file is not in canonical schema form" + ) + return authority + + +def _workflow_binding_refusal( + authority: ProductionQualificationAuthority, + workflow: Workflow, +) -> str | None: + """Bind the signed v2 authority to contracts reproduced from the bundle.""" + + manifest = workflow.manifest + project = workflow.qualification + if manifest is None or not manifest.content_digest: + return "Production qualification requires a sealed bundle" + template = manifest.provenance.governed_authorization_template + if template is None: + return "Production qualification requires a governed template" + if project is None: + return "Production qualification requires a qualification project" + payload = authority.qualification_admission.payload + effect_contract_sha256 = contract_sha256( + [ + item.model_dump(mode="json") + for item in template.qualified_effect_requirements + ] + ) + if ( + payload.bundle_content_digest != manifest.content_digest + or payload.governed_authorization_template_sha256 != template.template_sha256 + or payload.environment_digest != project.environment.environment_digest + or payload.environment_contract_sha256 + != template.qualification_environment_contract_sha256 + or payload.input_policy_sha256 != template.parameter_contract_sha256 + or payload.action_policy_sha256 + != template.qualification_project_contract_sha256 + or payload.identity_contract_sha256 != template.identity_contract_sha256 + or payload.effect_contract_sha256 != effect_contract_sha256 + ): + return "Production qualification does not bind the sealed workflow contracts" + return None + + +class ProductionQualificationGuard: + """Re-read and verify one v2 authority at every Production input edge.""" + + def __init__( + self, + path: Path | str, + *, + remote_permit_revalidation: bool, + ) -> None: + self.path = Path(path) + self.remote_permit_revalidation = bool(remote_permit_revalidation) + initial = load_production_qualification_authority(self.path) + self._admission_sha256 = initial.qualification_admission_sha256 + self._expected = initial.expected + self._registry_sha256 = initial.qualification_signer_registry_sha256 + self._revoked_ids = frozenset(initial.revoked_admission_ids) + + def _load_current(self) -> ProductionQualificationAuthority: + current = load_production_qualification_authority(self.path) + if ( + current.qualification_admission_sha256 != self._admission_sha256 + or current.expected != self._expected + or current.qualification_signer_registry_sha256 != self._registry_sha256 + ): + raise ProductionQualificationAuthorityError( + "qualification authority changed after run admission" + ) + current_revoked = frozenset(current.revoked_admission_ids) + if not self._revoked_ids.issubset(current_revoked): + raise ProductionQualificationAuthorityError( + "qualification revocation state rolled back" + ) + self._revoked_ids = current_revoked + return current + + def verify( + self, + workflow: Workflow, + *, + for_actuation: bool, + ) -> VerifiedQualificationAdmission: + """Verify signature, time, revocation, live values, and bundle binding.""" + + current = self._load_current() + refusal = _workflow_binding_refusal(current, workflow) + if refusal is not None: + raise ProductionQualificationAuthorityError(refusal) + revoked = frozenset(current.revoked_admission_ids) + try: + if for_actuation and not self.remote_permit_revalidation: + snapshot = current.permit_trust_snapshot + if snapshot is None: + raise ProductionQualificationAuthorityError( + "local Production actuation requires a fresh permit trust snapshot" + ) + return verify_qualification_admission_for_actuation( + current.qualification_admission, + registry=current.qualification_signer_registry, + expected=current.expected, + permit_trust_snapshot=snapshot, + revoked_admission_ids=revoked, + ) + return verify_qualification_admission( + current.qualification_admission, + registry=current.qualification_signer_registry, + expected=current.expected, + revoked_admission_ids=revoked, + ) + except QualificationAdmissionError as exc: + raise ProductionQualificationAuthorityError( + "Production qualification admission is not active" + ) from exc + + def refusal(self, workflow: Workflow) -> str | None: + """Return a PHI-free refusal for the last point before input delivery.""" + + try: + self.verify(workflow, for_actuation=True) + except ProductionQualificationAuthorityError as exc: + return str(exc) + return None + + def authorization_binding(self, workflow: Workflow) -> dict[str, object]: + """Return the verified PHI-free fields retained in run authorization.""" + + verified = self.verify(workflow, for_actuation=False) + current = self._load_current() + return { + "production_qualification_admission_id": verified.admission_id, + "production_qualification_admission_sha256": ( + verified.admission_artifact_sha256 + ), + "production_qualification_evidence_identity_sha256": ( + verified.evidence_identity_sha256 + ), + "production_qualification_runtime_validation_id": ( + verified.runtime_validation_id + ), + "production_qualification_signer_registry_sha256": ( + verified.registry_sha256 + ), + "production_qualification_signer_registry_revision": ( + verified.registry_revision + ), + "production_qualification_signer_registry_expires_at": ( + verified.registry_expires_at + ), + "production_qualification_authority_sha256": ( + current.immutable_binding_sha256() + ), + } + + +__all__ = [ + "AUTHORITY_SCHEMA", + "ProductionQualificationAuthority", + "ProductionQualificationAuthorityError", + "ProductionQualificationGuard", + "load_production_qualification_authority", +] diff --git a/openadapt_flow/qualification.py b/openadapt_flow/qualification.py index 9a9c80d6..95e68266 100644 --- a/openadapt_flow/qualification.py +++ b/openadapt_flow/qualification.py @@ -652,6 +652,44 @@ def resolved_fault_target(self) -> Optional[QualificationActionTarget]: return None +_QUALIFICATION_CASE_CONTRACT_DOMAIN = b"openadapt-qualification-case-contract-v1\0" + + +def qualification_case_contract_sha256(case: "QualificationCase") -> str: + """Digest one stable case contract without descriptions or result rows.""" + + if case.runtime_input_sha256 is None: + raise ValueError("qualification case has no approved runtime-input digest") + action_targets = [ + item.model_dump(mode="json") + for item in sorted( + case.action_targets, + key=lambda item: (item.step_id, item.actuation_path), + ) + ] + payload = { + "schema_version": "openadapt.qualification-case-contract/v1", + "qualification_case_id": case.id, + "qualification_case_kind": case.kind.value, + "expected_outcome": case.expected_outcome.value, + "runtime_input_sha256": case.runtime_input_sha256, + "action_targets": action_targets, + "fault_target": ( + case.fault_target.model_dump(mode="json") + if case.fault_target is not None + else None + ), + } + encoded = json.dumps( + payload, + sort_keys=True, + separators=(",", ":"), + ensure_ascii=True, + allow_nan=False, + ).encode("ascii") + return hashlib.sha256(_QUALIFICATION_CASE_CONTRACT_DOMAIN + encoded).hexdigest() + + class EnvironmentBoundary(BaseModel): """Application/environment scope in which the qualification is valid.""" @@ -5358,6 +5396,21 @@ def current_certification_matches( certification = project.last_certification if not certification.passed: return False + manifest = workflow.manifest + if manifest is None: + return False + expires_at = manifest.provenance.expires_at + if expires_at is not None: + try: + parsed_expiry = datetime.fromisoformat( + expires_at[:-1] + "+00:00" if expires_at.endswith("Z") else expires_at + ) + except (TypeError, ValueError): + return False + if parsed_expiry.tzinfo is None: + return False + if datetime.now(timezone.utc) >= parsed_expiry.astimezone(timezone.utc): + return False from openadapt_flow.policy import Policy, policy_contract_sha256 diff --git a/openadapt_flow/qualification_campaign_authority.py b/openadapt_flow/qualification_campaign_authority.py new file mode 100644 index 00000000..1a12399c --- /dev/null +++ b/openadapt_flow/qualification_campaign_authority.py @@ -0,0 +1,297 @@ +"""Fail-closed authority for one signed non-production qualification trial.""" + +from __future__ import annotations + +from pathlib import Path +from typing import Literal +from uuid import UUID + +from pydantic import BaseModel, ConfigDict, Field, ValidationError, field_validator + +from openadapt_flow.ir import Workflow +from openadapt_flow.production_qualification import ( + ProductionQualificationAuthorityError, + _read_private_json, +) +from openadapt_flow.qualification import ( + qualification_campaign_id_sha256, + qualification_run_id_sha256, +) +from openadapt_flow.qualification_admission_v2 import ( + QualificationSignerRegistry, + contract_sha256, +) +from openadapt_flow.qualification_campaign_permit import ( + QualificationCampaignPermitEnvelope, + QualificationCampaignPermitError, + QualificationCampaignPermitExpected, + verify_qualification_campaign_permit, +) +from openadapt_flow.runtime.authorization import GovernedRunAuthorization + +AUTHORITY_SCHEMA = "openadapt.qualification-campaign-authority/v1" + + +class QualificationCampaignAuthorityError(ValueError): + """The non-production qualification authority is unsafe or invalid.""" + + +class QualificationCampaignAuthority(BaseModel): + """Closed handoff for one signed, isolated qualification trial.""" + + model_config = ConfigDict(extra="forbid", frozen=True) + + schema_version: Literal["openadapt.qualification-campaign-authority/v1"] = ( + AUTHORITY_SCHEMA + ) + qualification_campaign_permit: QualificationCampaignPermitEnvelope + qualification_campaign_permit_sha256: str = Field(pattern=r"^[a-f0-9]{64}$") + expected: QualificationCampaignPermitExpected + qualification_signer_registry: QualificationSignerRegistry + qualification_signer_registry_sha256: str = Field(pattern=r"^[a-f0-9]{64}$") + consumed_permit_ids: tuple[str, ...] = () + + @field_validator("consumed_permit_ids") + @classmethod + def _ordered_consumed_ids(cls, value: tuple[str, ...]) -> tuple[str, ...]: + if value != tuple(sorted(set(value))): + raise ValueError("consumed qualification permit ids must be ordered") + for item in value: + try: + parsed = UUID(item) + except ValueError as exc: + raise ValueError( + "consumed qualification permit id must be a canonical UUID" + ) from exc + if str(parsed) != item: + raise ValueError( + "consumed qualification permit id must be a canonical UUID" + ) + return value + + def model_post_init(self, __context: object) -> None: + if ( + self.qualification_campaign_permit.artifact_sha256() + != self.qualification_campaign_permit_sha256 + ): + raise ValueError("qualification campaign permit digest does not match") + if ( + self.qualification_signer_registry.artifact_sha256() + != self.qualification_signer_registry_sha256 + ): + raise ValueError("qualification signer registry digest does not match") + + def immutable_binding_sha256(self) -> str: + return contract_sha256( + { + "schema_version": self.schema_version, + "qualification_campaign_permit_sha256": ( + self.qualification_campaign_permit_sha256 + ), + "expected": self.expected.model_dump(mode="json"), + "qualification_signer_registry_sha256": ( + self.qualification_signer_registry_sha256 + ), + } + ) + + +def load_qualification_campaign_authority( + path: Path | str, +) -> QualificationCampaignAuthority: + """Load one exact private non-production campaign authority.""" + + try: + raw = _read_private_json(Path(path)) + except ProductionQualificationAuthorityError as exc: + raise QualificationCampaignAuthorityError(str(exc)) from exc + try: + authority = QualificationCampaignAuthority.model_validate(raw) + except (TypeError, ValueError, ValidationError) as exc: + raise QualificationCampaignAuthorityError( + "qualification campaign authority has an invalid exact binding" + ) from exc + if authority.model_dump(mode="json") != raw: + raise QualificationCampaignAuthorityError( + "qualification campaign authority is not in canonical schema form" + ) + return authority + + +def _workflow_binding_refusal( + authority: QualificationCampaignAuthority, + workflow: Workflow, + *, + case_id: str, + campaign_id: str, + run_id: str, + input_digest: str, +) -> str | None: + manifest = workflow.manifest + project = workflow.qualification + if manifest is None or not manifest.content_digest: + return "qualification campaign requires a sealed bundle" + if project is None: + return "qualification campaign requires a qualification project" + template = manifest.provenance.governed_authorization_template + if template is None: + return "qualification campaign requires a governed template" + case = next((item for item in project.cases if item.id == case_id), None) + if case is None: + return "qualification campaign references an unknown case" + payload = authority.qualification_campaign_permit.payload + trial = payload.trial + if ( + payload.bundle_content_digest != manifest.content_digest + or payload.environment_digest != project.environment.environment_digest + or payload.environment_contract_sha256 + != template.qualification_environment_contract_sha256 + or trial.campaign_id != campaign_id + or trial.qualification_run_id != run_id + or trial.input_digest != input_digest + or trial.campaign_contract_sha256 != project.contract_sha256() + or trial.task != case.id + or trial.condition != case.kind.value + ): + return "qualification campaign permit does not bind this exact trial" + return None + + +class QualificationCampaignGuard: + """Re-read one signed non-production trial permit at every input edge.""" + + def __init__( + self, + path: Path | str, + *, + workflow: Workflow, + case_id: str, + input_digest: str, + campaign_id: str | None = None, + run_id: str | None = None, + ) -> None: + self.path = Path(path) + initial = load_qualification_campaign_authority(self.path) + trial = initial.qualification_campaign_permit.payload.trial + self.case_id = case_id + self.input_digest = input_digest + self.campaign_id = campaign_id or trial.campaign_id + self.run_id = run_id or trial.qualification_run_id + self._permit_sha256 = initial.qualification_campaign_permit_sha256 + self._expected = initial.expected + self._registry_sha256 = initial.qualification_signer_registry_sha256 + self._consumed_ids = frozenset(initial.consumed_permit_ids) + refusal = _workflow_binding_refusal( + initial, + workflow, + case_id=self.case_id, + campaign_id=self.campaign_id, + run_id=self.run_id, + input_digest=self.input_digest, + ) + if refusal is not None: + raise QualificationCampaignAuthorityError(refusal) + + def _load_current(self) -> QualificationCampaignAuthority: + current = load_qualification_campaign_authority(self.path) + if ( + current.qualification_campaign_permit_sha256 != self._permit_sha256 + or current.expected != self._expected + or current.qualification_signer_registry_sha256 != self._registry_sha256 + ): + raise QualificationCampaignAuthorityError( + "qualification campaign authority changed after run admission" + ) + consumed = frozenset(current.consumed_permit_ids) + if not self._consumed_ids.issubset(consumed): + raise QualificationCampaignAuthorityError( + "qualification campaign consumption state rolled back" + ) + self._consumed_ids = consumed + return current + + def verify(self, workflow: Workflow) -> str: + current = self._load_current() + refusal = _workflow_binding_refusal( + current, + workflow, + case_id=self.case_id, + campaign_id=self.campaign_id, + run_id=self.run_id, + input_digest=self.input_digest, + ) + if refusal is not None: + raise QualificationCampaignAuthorityError(refusal) + try: + return verify_qualification_campaign_permit( + current.qualification_campaign_permit, + registry=current.qualification_signer_registry, + expected=current.expected, + consumed_permit_ids=frozenset(current.consumed_permit_ids), + ) + except QualificationCampaignPermitError as exc: + raise QualificationCampaignAuthorityError( + "qualification campaign permit is not active" + ) from exc + + def authorization_binding(self, workflow: Workflow) -> dict[str, object]: + digest = self.verify(workflow) + current = self._load_current() + payload = current.qualification_campaign_permit.payload + return { + "qualification_campaign_permit_id": payload.permit_id, + "qualification_campaign_permit_sha256": digest, + "qualification_campaign_signer_registry_sha256": ( + current.qualification_signer_registry_sha256 + ), + "qualification_campaign_signer_registry_revision": ( + current.qualification_signer_registry.revision + ), + "qualification_campaign_signer_registry_expires_at": ( + current.qualification_signer_registry.expires_at + ), + "qualification_campaign_authority_sha256": ( + current.immutable_binding_sha256() + ), + } + + def authorization_refusal( + self, + workflow: Workflow, + authorization: GovernedRunAuthorization, + ) -> str | None: + try: + expected = self.authorization_binding(workflow) + except QualificationCampaignAuthorityError as exc: + return str(exc) + if any( + getattr(authorization, field, None) != value + for field, value in expected.items() + ): + return "qualification campaign permit differs from run authorization" + if ( + authorization.qualification_case_id != self.case_id + or authorization.runtime_inputs_digest != self.input_digest + or authorization.qualification_campaign_id_sha256 + != qualification_campaign_id_sha256(self.campaign_id) + or authorization.qualification_run_id_sha256 + != qualification_run_id_sha256(self.run_id) + ): + return "qualification campaign permit differs from the exact trial" + return None + + def refusal(self, workflow: Workflow) -> str | None: + try: + self.verify(workflow) + except QualificationCampaignAuthorityError as exc: + return str(exc) + return None + + +__all__ = [ + "AUTHORITY_SCHEMA", + "QualificationCampaignAuthority", + "QualificationCampaignAuthorityError", + "QualificationCampaignGuard", + "load_qualification_campaign_authority", +] diff --git a/openadapt_flow/qualification_campaign_permit.py b/openadapt_flow/qualification_campaign_permit.py index 48d566a6..a15bf0fa 100644 --- a/openadapt_flow/qualification_campaign_permit.py +++ b/openadapt_flow/qualification_campaign_permit.py @@ -50,7 +50,7 @@ "signature_domain": "openadapt-qualification-campaign-permit-v1\\0", "issuer_workflow": ( "OpenAdaptAI/openadapt-internal/.github/workflows/" - "production-qualification-admission.yml" + "issue-qualification-campaign-permit.yml" ), "issuer_ref_prefix": "refs/heads/main@", "maximum_lifetime_seconds": 24 * 60 * 60, diff --git a/openadapt_flow/run_gate.py b/openadapt_flow/run_gate.py index fd557ca8..48cb86fc 100644 --- a/openadapt_flow/run_gate.py +++ b/openadapt_flow/run_gate.py @@ -1101,6 +1101,7 @@ def build_qualification_case_authorization( worklists: Optional[dict[str, list[dict[str, str]]]], campaign_id: str, run_id: str, + campaign_permit_binding: dict[str, object], fault_driver: Any = None, ) -> GovernedRunAuthorization: """Build the exact authority for one governed qualification-case run. @@ -1159,6 +1160,18 @@ def build_qualification_case_authorization( }, "required_identity_step_ids": tuple(sorted(required_identity)), } + permit_fields = { + field + for field in GovernedRunAuthorization.model_fields + if field.startswith("qualification_campaign_") + and field + not in { + "qualification_campaign_id_sha256", + } + } + if set(campaign_permit_binding) != permit_fields: + raise ValueError("qualification campaign permit binding is incomplete") + updates.update(campaign_permit_binding) if case.kind is not QualificationCaseKind.REPRESENTATIVE: target = case.resolved_fault_target() if target is None: @@ -1187,7 +1200,9 @@ def build_qualification_case_authorization( elif fault_driver is not None: raise ValueError("representative qualification case cannot bind a fault driver") - bound = authorization.model_copy(update=updates) + bound = GovernedRunAuthorization.model_validate( + {**authorization.model_dump(mode="json"), **updates} + ) validation_error = bound.validate_workflow(workflow) if validation_error is not None: raise ValueError( diff --git a/openadapt_flow/runner/business_decision_service.py b/openadapt_flow/runner/business_decision_service.py index 96e9ae5e..311159d6 100644 --- a/openadapt_flow/runner/business_decision_service.py +++ b/openadapt_flow/runner/business_decision_service.py @@ -35,6 +35,10 @@ ) from openadapt_flow.ir import Workflow from openadapt_flow.policy import Policy, load_policy, policy_contract_sha256 +from openadapt_flow.private_file import PrivateFileAclError +from openadapt_flow.private_file import ( + windows_descriptor_has_private_acl as _windows_descriptor_has_private_acl, +) from openadapt_flow.runner.config import RunnerConfig, RunnerConfigError from openadapt_flow.runtime.durable.approval import ResumeRefused from openadapt_flow.runtime.durable.checkpoint import ( @@ -114,59 +118,6 @@ def _key_id(raw: object, label: str) -> str: return raw -def _windows_descriptor_has_private_acl(descriptor: int) -> bool: - """Require the opened Windows file to be private to its service identity.""" - - try: - import msvcrt - - import ntsecuritycon - import win32api - import win32con - import win32security - - get_osfhandle = getattr(msvcrt, "get_osfhandle", None) - if not callable(get_osfhandle): - raise AttributeError("Windows descriptor conversion is unavailable") - handle = get_osfhandle(descriptor) - security = win32security.GetSecurityInfo( - handle, - win32security.SE_FILE_OBJECT, - win32security.OWNER_SECURITY_INFORMATION - | win32security.DACL_SECURITY_INFORMATION, - ) - owner = security.GetSecurityDescriptorOwner() - token = win32security.OpenProcessToken( - win32api.GetCurrentProcess(), win32con.TOKEN_QUERY - ) - current = win32security.GetTokenInformation(token, win32security.TokenUser)[0] - if not win32security.EqualSid(owner, current): - return False - dacl = security.GetSecurityDescriptorDacl() - if dacl is None: - return False - allowed = ( - current, - win32security.ConvertStringSidToSid("S-1-5-18"), - win32security.ConvertStringSidToSid("S-1-5-32-544"), - ) - for index in range(dacl.GetAceCount()): - header, mask, sid = dacl.GetAce(index) - ace_type = header[0] - if ace_type == ntsecuritycon.ACCESS_DENIED_ACE_TYPE: - continue - if ace_type != ntsecuritycon.ACCESS_ALLOWED_ACE_TYPE: - return False - if mask and not any(win32security.EqualSid(sid, item) for item in allowed): - return False - return True - except (ImportError, OSError, AttributeError) as exc: - raise RunnerConfigError( - "Windows private key-file ACL verification is unavailable; install " - "the OpenAdapt Windows runtime" - ) from exc - - def load_business_decision_key_material(path: Path) -> BusinessDecisionKeyMaterial: """Load a service-identity-owned private key file without a symlink.""" @@ -206,7 +157,10 @@ def load_business_decision_key_material(path: Path) -> BusinessDecisionKeyMateri or stat.S_IMODE(metadata.st_mode) != 0o600 ) if os.name == "nt" and not changed: - unsafe_permissions = not _windows_descriptor_has_private_acl(descriptor) + try: + unsafe_permissions = not _windows_descriptor_has_private_acl(descriptor) + except PrivateFileAclError as exc: + raise RunnerConfigError(str(exc)) from exc if ( not stat.S_ISREG(metadata.st_mode) or metadata.st_size > 64 * 1024 diff --git a/openadapt_flow/runtime/authorization.py b/openadapt_flow/runtime/authorization.py index e156dab2..fc224361 100644 --- a/openadapt_flow/runtime/authorization.py +++ b/openadapt_flow/runtime/authorization.py @@ -230,6 +230,27 @@ class GovernedRunAuthorization(BaseModel): qualification_admission_sha256: str | None = Field( default=None, pattern="^[a-f0-9]{64}$" ) + # v2 Production admission identity. The signed artifact stays in the + # private authority handoff; these PHI-free bindings enter the durable run + # authorization so a different admission cannot replace it after the gate. + production_qualification_admission_id: str | None = None + production_qualification_admission_sha256: str | None = Field( + default=None, pattern="^[a-f0-9]{64}$" + ) + production_qualification_evidence_identity_sha256: str | None = Field( + default=None, pattern="^[a-f0-9]{64}$" + ) + production_qualification_runtime_validation_id: str | None = None + production_qualification_signer_registry_sha256: str | None = Field( + default=None, pattern="^[a-f0-9]{64}$" + ) + production_qualification_signer_registry_revision: int | None = Field( + default=None, ge=1 + ) + production_qualification_signer_registry_expires_at: str | None = None + production_qualification_authority_sha256: str | None = Field( + default=None, pattern="^[a-f0-9]{64}$" + ) qualification_project_id: str | None = None qualification_project_revision: int | None = Field(default=None, ge=1) qualification_project_contract_sha256: str | None = Field( @@ -247,6 +268,20 @@ class GovernedRunAuthorization(BaseModel): qualification_run_id_sha256: str | None = Field( default=None, pattern="^[a-f0-9]{64}$" ) + qualification_campaign_permit_id: str | None = None + qualification_campaign_permit_sha256: str | None = Field( + default=None, pattern="^[a-f0-9]{64}$" + ) + qualification_campaign_signer_registry_sha256: str | None = Field( + default=None, pattern="^[a-f0-9]{64}$" + ) + qualification_campaign_signer_registry_revision: int | None = Field( + default=None, ge=1 + ) + qualification_campaign_signer_registry_expires_at: str | None = None + qualification_campaign_authority_sha256: str | None = Field( + default=None, pattern="^[a-f0-9]{64}$" + ) qualification_case_kind: ( Literal[ "representative", @@ -286,6 +321,20 @@ def _qualification_binding_is_complete(self) -> "GovernedRunAuthorization": != self.qualification_admission_sha256 ): raise ValueError("qualification admission digest does not match") + production_qualification = ( + self.production_qualification_admission_id, + self.production_qualification_admission_sha256, + self.production_qualification_evidence_identity_sha256, + self.production_qualification_runtime_validation_id, + self.production_qualification_signer_registry_sha256, + self.production_qualification_signer_registry_revision, + self.production_qualification_signer_registry_expires_at, + self.production_qualification_authority_sha256, + ) + if any(value is not None for value in production_qualification) and not all( + value is not None for value in production_qualification + ): + raise ValueError("Production qualification binding is incomplete") requirement_refs = [ (item.step_id, item.actuation_path, item.effect_index) for item in self.qualified_effect_requirements @@ -307,6 +356,24 @@ def _qualification_binding_is_complete(self) -> "GovernedRunAuthorization": value is not None for value in values ): raise ValueError("qualification-run authorization binding is incomplete") + campaign_permit = ( + self.qualification_campaign_permit_id, + self.qualification_campaign_permit_sha256, + self.qualification_campaign_signer_registry_sha256, + self.qualification_campaign_signer_registry_revision, + self.qualification_campaign_signer_registry_expires_at, + self.qualification_campaign_authority_sha256, + ) + if any(value is not None for value in campaign_permit) and not all( + value is not None for value in campaign_permit + ): + raise ValueError("qualification campaign permit binding is incomplete") + if self.qualification_case_id is not None and not all( + value is not None for value in campaign_permit + ): + raise ValueError( + "qualification cases require a signed campaign permit binding" + ) if ( self.qualification_case_kind is not None and self.qualification_case_id is None diff --git a/openadapt_flow/runtime/replayer.py b/openadapt_flow/runtime/replayer.py index 3a2009a3..3f35adb9 100644 --- a/openadapt_flow/runtime/replayer.py +++ b/openadapt_flow/runtime/replayer.py @@ -193,6 +193,10 @@ _DURABLE_RESUME_AUTHORITY = object() if TYPE_CHECKING: + from openadapt_flow.production_qualification import ProductionQualificationGuard + from openadapt_flow.qualification_campaign_authority import ( + QualificationCampaignGuard, + ) from openadapt_flow.runtime.control_overlay import RuntimeControlOverlayEmitter from openadapt_flow.transaction import IdempotencyLedger @@ -436,6 +440,8 @@ def __init__( ] = "customer_local", remote_delivery_run_id: Optional[str] = None, managed_dispatch_binding: Optional[Any] = None, + production_qualification_guard: Optional["ProductionQualificationGuard"] = None, + qualification_campaign_guard: Optional["QualificationCampaignGuard"] = None, governed_continuation: bool = False, require_settled: bool = False, settle_readiness_timeout_s: float = 10.0, @@ -544,8 +550,41 @@ def __init__( governed_authorization = managed_dispatch_binding.authorization delivery_authority_kind = "cloud_runner" remote_delivery_run_id = managed_dispatch_binding.run_id + if production_qualification_guard is not None: + from openadapt_flow.production_qualification import ( + ProductionQualificationGuard, + ) + + if not isinstance( + production_qualification_guard, ProductionQualificationGuard + ): + raise ValueError("Production qualification authority is invalid") + if qualification_campaign_guard is not None: + from openadapt_flow.qualification_campaign_authority import ( + QualificationCampaignGuard, + ) + + if not isinstance(qualification_campaign_guard, QualificationCampaignGuard): + raise ValueError("qualification campaign authority is invalid") + production_profile = getattr(governed_authorization, "execution_profile", None) + qualification_case = getattr( + governed_authorization, "qualification_case_id", None + ) + if production_profile in {"standard", "regulated"}: + if qualification_case is not None: + if qualification_campaign_guard is None: + raise ValueError( + "qualification actuation requires a signed non-production " + "campaign permit" + ) + elif production_qualification_guard is None: + raise ValueError( + "Production actuation requires a verified v2 qualification authority" + ) self.managed_dispatch_binding = managed_dispatch_binding self.governed_authorization = governed_authorization + self.production_qualification_guard = production_qualification_guard + self.qualification_campaign_guard = qualification_campaign_guard self.delivery_authority_kind = delivery_authority_kind self.remote_delivery_run_id = remote_delivery_run_id self._managed_dispatch_snapshot: tuple[str, str, str] | None = None @@ -1051,6 +1090,12 @@ def run( managed_refusal = self._managed_dispatch_refusal() if managed_refusal is not None: raise ValueError(managed_refusal) + production_refusal = self._production_qualification_refusal(workflow) + if production_refusal is not None: + raise ValueError(production_refusal) + campaign_refusal = self._qualification_campaign_refusal(workflow) + if campaign_refusal is not None: + raise ValueError(campaign_refusal) bundle_dir = Path(bundle_dir) run_dir = Path(run_dir) self._install_execution_snapshots(workflow) @@ -9239,6 +9284,51 @@ def _managed_dispatch_refusal(self) -> Optional[str]: return "managed dispatch binding is invalid after admission" return None + def _production_qualification_refusal(self, workflow: Workflow) -> Optional[str]: + """Bind the current v2 admission to the durable run authorization.""" + + authorization = self.governed_authorization + if authorization is None or authorization.execution_profile not in { + "standard", + "regulated", + }: + return None + if authorization.qualification_case_id is not None: + return None + guard = self.production_qualification_guard + if guard is None: + return "Production actuation has no v2 qualification authority" + try: + expected = guard.authorization_binding(workflow) + except Exception as exc: # noqa: BLE001 - private authority boundary + return ( + "Production qualification could not be verified before execution " + f"({type(exc).__name__})" + ) + if any( + getattr(authorization, field, None) != value + for field, value in expected.items() + ): + return "Production qualification differs from the run authorization" + return None + + def _qualification_campaign_refusal(self, workflow: Workflow) -> Optional[str]: + """Bind a case run to its signed, isolated non-production permit.""" + + authorization = self.governed_authorization + if authorization is None or authorization.qualification_case_id is None: + return None + guard = self.qualification_campaign_guard + if guard is None: + return "qualification actuation has no signed campaign permit" + try: + return guard.authorization_refusal(workflow, authorization) + except Exception as exc: # noqa: BLE001 - private authority boundary + return ( + "qualification campaign could not be verified before execution " + f"({type(exc).__name__})" + ) + def _delivery_authorization_refusal( self, workflow: Workflow, @@ -9251,6 +9341,22 @@ def _delivery_authorization_refusal( self._active_delivery_acknowledgers = () acknowledgers: list[Any] = [] refusal = self._managed_dispatch_refusal() + if refusal is None and self.production_qualification_guard is not None: + try: + refusal = self.production_qualification_guard.refusal(workflow) + except Exception as exc: # noqa: BLE001 - authority boundary + refusal = ( + "Production qualification could not be revalidated before " + f"delivery ({type(exc).__name__})" + ) + if refusal is None and self.qualification_campaign_guard is not None: + try: + refusal = self.qualification_campaign_guard.refusal(workflow) + except Exception as exc: # noqa: BLE001 - authority boundary + refusal = ( + "qualification campaign could not be revalidated before " + f"delivery ({type(exc).__name__})" + ) if refusal is None: refusal = self._fresh_actuation_authorization_refusal( workflow, params, step diff --git a/tests/test_execution_profiles.py b/tests/test_execution_profiles.py index 0753d479..827f5e35 100644 --- a/tests/test_execution_profiles.py +++ b/tests/test_execution_profiles.py @@ -51,6 +51,7 @@ Workflow, predicate_contract_sha256, ) +from openadapt_flow.production_qualification import ProductionQualificationGuard from openadapt_flow.qualification import ( EffectVerificationPolicy, EnvironmentBoundary, @@ -81,7 +82,7 @@ ) from openadapt_flow.runtime.effects.effect import ReadbackNav, ReadbackSpec from openadapt_flow.runtime.effects.onscreen import OnScreenReadbackVerifier -from openadapt_flow.runtime.replayer import Replayer +from openadapt_flow.runtime.replayer import Replayer as _RuntimeReplayer from openadapt_flow.verification import VerificationTier from openadapt_flow.vision.ocr import OcrLine from tests.test_durable_runtime import FakeSoRVerifier, _approval @@ -90,6 +91,33 @@ _KEY = "profile-test-key" +class _PreviouslyQualifiedTestGuard(ProductionQualificationGuard): + """Keep profile tests focused on their pre-existing runtime contracts. + + Dedicated Production qualification tests exercise the real signed gate. + These tests build synthetic Standard and Regulated authorizations directly, + so this typed test guard represents their assumed prior qualification. + """ + + def __init__(self) -> None: + pass + + def authorization_binding(self, workflow): + return {} + + def refusal(self, workflow): + return None + + +class Replayer(_RuntimeReplayer): + """Supply the prior-qualification premise used by this profile test suite.""" + + def __init__(self, *args, **kwargs): + if "production_qualification_guard" not in kwargs: + kwargs["production_qualification_guard"] = _PreviouslyQualifiedTestGuard() + super().__init__(*args, **kwargs) + + def _keyboard_receipt() -> ActionDeliveryReceipt: return ActionDeliveryReceipt( receipt_id="profile-key", diff --git a/tests/test_production_qualification.py b/tests/test_production_qualification.py new file mode 100644 index 00000000..12b8031c --- /dev/null +++ b/tests/test_production_qualification.py @@ -0,0 +1,304 @@ +from __future__ import annotations + +import json +import os +from types import SimpleNamespace + +import pytest + +import openadapt_flow.production_qualification as production +from openadapt_flow.production_qualification import ( + ProductionQualificationAuthority, + ProductionQualificationAuthorityError, + ProductionQualificationGuard, + _workflow_binding_refusal, + load_production_qualification_authority, +) +from openadapt_flow.qualification_admission_v2 import ( + QualificationAdmissionError, + VerifiedQualificationAdmission, + sign_qualification_admission, +) +from tests.test_qualification_admission_v2 import ( + IDS, + SHA_A, + SHA_C, + SHA_E, + _expected, + _payload, + _private_key, + _registry, + _snapshot, +) + + +def _authority(*, snapshot: bool = True) -> ProductionQualificationAuthority: + registry = _registry() + admission = sign_qualification_admission(_payload(registry), _private_key()) + return ProductionQualificationAuthority( + qualification_admission=admission, + qualification_admission_sha256=admission.artifact_sha256(), + expected=_expected(admission.payload), + qualification_signer_registry=registry, + qualification_signer_registry_sha256=registry.artifact_sha256(), + permit_trust_snapshot=_snapshot(registry) if snapshot else None, + ) + + +def _write_authority(path, authority: ProductionQualificationAuthority) -> None: + path.write_text( + json.dumps(authority.model_dump(mode="json"), sort_keys=True), + encoding="utf-8", + ) + if os.name != "nt": + path.chmod(0o600) + + +def _verified( + authority: ProductionQualificationAuthority, +) -> VerifiedQualificationAdmission: + return VerifiedQualificationAdmission( + admission_artifact_sha256=authority.qualification_admission_sha256, + evidence_identity_sha256=( + authority.qualification_admission.payload.evidence_identity.artifact_sha256() + ), + registry_sha256=authority.qualification_signer_registry_sha256, + registry_revision=authority.qualification_signer_registry.revision, + registry_expires_at=authority.qualification_signer_registry.expires_at, + issuer_key_id=authority.qualification_admission.payload.issuer.key_id, + admission_id=authority.qualification_admission.payload.admission_id, + runtime_validation_id=( + authority.qualification_admission.payload.runtime_validation_id + ), + ) + + +def _stub_verification( + monkeypatch, authority: ProductionQualificationAuthority +) -> None: + verified = _verified(authority) + monkeypatch.setattr(production, "_workflow_binding_refusal", lambda *_: None) + monkeypatch.setattr( + production, + "verify_qualification_admission", + lambda *_, **__: verified, + ) + monkeypatch.setattr( + production, + "verify_qualification_admission_for_actuation", + lambda *_, **__: verified, + ) + + +def test_private_authority_file_is_closed_and_owner_only(tmp_path): + authority = _authority() + path = tmp_path / "authority.json" + _write_authority(path, authority) + + assert load_production_qualification_authority(path) == authority + + raw = authority.model_dump(mode="json") + raw["unexpected"] = True + path.write_text(json.dumps(raw), encoding="utf-8") + if os.name != "nt": + path.chmod(0o600) + with pytest.raises(ProductionQualificationAuthorityError): + load_production_qualification_authority(path) + + if os.name != "nt": + _write_authority(path, authority) + path.chmod(0o640) + with pytest.raises(ProductionQualificationAuthorityError): + load_production_qualification_authority(path) + + +@pytest.mark.skipif(os.name == "nt", reason="POSIX link contract") +def test_private_authority_file_refuses_links(tmp_path): + authority = _authority() + target = tmp_path / "authority.json" + link = tmp_path / "authority-link.json" + _write_authority(target, authority) + link.symlink_to(target) + + with pytest.raises(ProductionQualificationAuthorityError): + load_production_qualification_authority(link) + + +def test_private_authority_file_refuses_a_short_descriptor_read(tmp_path, monkeypatch): + authority = _authority() + path = tmp_path / "authority.json" + _write_authority(path, authority) + monkeypatch.setattr(production.os, "read", lambda *_: b"") + + with pytest.raises( + ProductionQualificationAuthorityError, + match="ended during the safe read", + ): + production._read_private_json(path) + + +def test_windows_authority_file_fails_closed_without_acl_proof(tmp_path, monkeypatch): + authority = _authority() + path = tmp_path / "authority.json" + _write_authority(path, authority) + monkeypatch.setattr(production.os, "name", "nt") + monkeypatch.setattr(production, "Path", lambda value: value) + + def _acl_unavailable(_descriptor): + raise production.PrivateFileAclError("ACL proof unavailable") + + monkeypatch.setattr( + production, + "windows_descriptor_has_private_acl", + _acl_unavailable, + ) + with pytest.raises( + ProductionQualificationAuthorityError, + match="ACL proof unavailable", + ): + production._read_private_json(path) + + +def test_mutable_permit_snapshot_does_not_change_retained_identity( + tmp_path, monkeypatch +): + authority = _authority() + path = tmp_path / "authority.json" + _write_authority(path, authority) + _stub_verification(monkeypatch, authority) + guard = ProductionQualificationGuard(path, remote_permit_revalidation=False) + + first = guard.authorization_binding(object()) + refreshed = authority.model_copy( + update={ + "permit_trust_snapshot": authority.permit_trust_snapshot.model_copy( + update={ + "qualification_signer_registry_checked_at": ("2026-08-18T12:00:00Z") + } + ) + } + ) + _write_authority(path, refreshed) + second = guard.authorization_binding(object()) + + assert first == second + assert first["production_qualification_authority_sha256"] == ( + authority.immutable_binding_sha256() + ) + + +def test_revocation_state_can_only_grow(tmp_path): + authority = _authority() + path = tmp_path / "authority.json" + _write_authority(path, authority) + guard = ProductionQualificationGuard(path, remote_permit_revalidation=True) + revoked_id = "00000000-0000-4000-8000-000000000099" + + _write_authority( + path, + authority.model_copy(update={"revoked_admission_ids": (revoked_id,)}), + ) + guard._load_current() + _write_authority(path, authority) + + with pytest.raises( + ProductionQualificationAuthorityError, + match="revocation state rolled back", + ): + guard._load_current() + + +def test_changed_immutable_authority_is_refused_after_run_admission(tmp_path): + authority = _authority() + path = tmp_path / "authority.json" + _write_authority(path, authority) + guard = ProductionQualificationGuard(path, remote_permit_revalidation=True) + changed = authority.model_copy( + update={ + "expected": authority.expected.model_copy( + update={"bundle_artifact_sha256": SHA_C} + ) + } + ) + _write_authority(path, changed) + + with pytest.raises( + ProductionQualificationAuthorityError, + match="changed after run admission", + ): + guard._load_current() + + +def test_local_actuation_requires_current_permit_snapshot(tmp_path, monkeypatch): + authority = _authority(snapshot=False) + path = tmp_path / "authority.json" + _write_authority(path, authority) + _stub_verification(monkeypatch, authority) + guard = ProductionQualificationGuard(path, remote_permit_revalidation=False) + + assert guard.refusal(object()) == ( + "local Production actuation requires a fresh permit trust snapshot" + ) + + +def test_current_admission_revocation_refuses_before_delivery(tmp_path, monkeypatch): + authority = _authority() + current_id = authority.qualification_admission.payload.admission_id + authority = authority.model_copy(update={"revoked_admission_ids": (current_id,)}) + path = tmp_path / "authority.json" + _write_authority(path, authority) + monkeypatch.setattr(production, "_workflow_binding_refusal", lambda *_: None) + + def _refuse_revoked(*_, revoked_admission_ids, **__): + if current_id in revoked_admission_ids: + raise QualificationAdmissionError("revoked") + return _verified(authority) + + monkeypatch.setattr(production, "verify_qualification_admission", _refuse_revoked) + guard = ProductionQualificationGuard(path, remote_permit_revalidation=True) + + with pytest.raises(ProductionQualificationAuthorityError, match="not active"): + guard.authorization_binding(object()) + + +def test_workflow_binding_reproduces_sealed_contracts(monkeypatch): + authority = _authority() + template = SimpleNamespace( + template_sha256=SHA_A, + qualification_environment_contract_sha256=SHA_A, + parameter_contract_sha256=SHA_A, + qualification_project_contract_sha256=SHA_A, + identity_contract_sha256=SHA_A, + qualified_effect_requirements=(), + ) + workflow = SimpleNamespace( + manifest=SimpleNamespace( + content_digest=SHA_A, + provenance=SimpleNamespace(governed_authorization_template=template), + ), + qualification=SimpleNamespace( + environment=SimpleNamespace(environment_digest=SHA_E) + ), + ) + monkeypatch.setattr(production, "contract_sha256", lambda *_: SHA_A) + + assert _workflow_binding_refusal(authority, workflow) is None + + workflow.manifest.content_digest = SHA_C + assert _workflow_binding_refusal(authority, workflow) == ( + "Production qualification does not bind the sealed workflow contracts" + ) + + +def test_governed_authorization_requires_complete_v2_binding(): + from openadapt_flow.runtime.authorization import GovernedRunAuthorization + + with pytest.raises(ValueError, match="Production qualification binding"): + GovernedRunAuthorization( + authorization_id=IDS["admission_id"], + bundle_content_digest=SHA_A, + runtime_inputs_digest=SHA_A, + admitted_policy_name="permissive", + execution_profile="standard", + production_qualification_admission_id=IDS["admission_id"], + ) diff --git a/tests/test_qualification_admission_v2.py b/tests/test_qualification_admission_v2.py index cc67764f..75624e9c 100644 --- a/tests/test_qualification_admission_v2.py +++ b/tests/test_qualification_admission_v2.py @@ -221,7 +221,8 @@ def _registry( "key_id": qualification_signer_key_id(key), "public_key": b64encode(key).decode("ascii"), "allowed_workflows": [ - "OpenAdaptAI/openadapt-internal/.github/workflows/production-qualification-admission.yml" + "OpenAdaptAI/openadapt-internal/.github/workflows/issue-qualification-campaign-permit.yml", + "OpenAdaptAI/openadapt-internal/.github/workflows/production-qualification-admission.yml", ], "allowed_ref_prefixes": ["refs/heads/main@"], "status": status, @@ -415,7 +416,7 @@ def _campaign_permit_payload( key_id=qualification_signer_key_id(_public_key()), workflow=( "OpenAdaptAI/openadapt-internal/.github/workflows/" - "production-qualification-admission.yml" + "issue-qualification-campaign-permit.yml" ), ref="refs/heads/main@" + COMMIT, ), @@ -566,6 +567,11 @@ def test_campaign_requires_exact_policy_taxonomy(replacement: list[str]) -> None ("field", "replacement"), [ ("workflow", "OpenAdaptAI/alternate/.github/workflows/admit.yml"), + ( + "workflow", + "OpenAdaptAI/openadapt-internal/.github/workflows/" + "issue-qualification-campaign-permit.yml", + ), ("ref", "refs/heads/release@" + COMMIT), ("ref", "refs/heads/main@" + "a" * 39), ("ref", "refs/heads/main@" + "A" * 40), @@ -849,6 +855,16 @@ def test_campaign_trial_permit_cannot_emit_success() -> None: QualificationCampaignPermitPayload.model_validate(data) +def test_campaign_trial_permit_rejects_the_retired_issuer_workflow() -> None: + data = _campaign_permit_payload(_registry()).model_dump(mode="json") + data["issuer"]["workflow"] = ( + "OpenAdaptAI/openadapt-internal/.github/workflows/" + "production-qualification-admission.yml" + ) + with pytest.raises(ValidationError, match="issuer workflow"): + QualificationCampaignPermitPayload.model_validate(data) + + def test_campaign_trial_permit_rejects_live_environment_change() -> None: registry = _registry() payload = _campaign_permit_payload(registry) diff --git a/tests/test_qualification_campaign_authority.py b/tests/test_qualification_campaign_authority.py new file mode 100644 index 00000000..03f63336 --- /dev/null +++ b/tests/test_qualification_campaign_authority.py @@ -0,0 +1,204 @@ +from __future__ import annotations + +import json +import os +from types import SimpleNamespace + +import pytest + +import openadapt_flow.qualification_campaign_authority as campaign +from openadapt_flow.production_qualification import ProductionQualificationGuard +from openadapt_flow.qualification_campaign_authority import ( + QualificationCampaignAuthority, + QualificationCampaignAuthorityError, + QualificationCampaignGuard, + load_qualification_campaign_authority, +) +from openadapt_flow.qualification_campaign_permit import ( + QualificationCampaignPermitError, + sign_qualification_campaign_permit, +) +from openadapt_flow.runtime.authorization import GovernedRunAuthorization +from openadapt_flow.runtime.replayer import Replayer +from tests.test_qualification_admission_v2 import ( + IDS, + SHA_A, + _campaign_expected, + _campaign_permit_payload, + _private_key, + _registry, +) +from tests.test_replayer import FakeBackend, FakeVision + + +def _authority() -> QualificationCampaignAuthority: + registry = _registry() + permit = sign_qualification_campaign_permit( + _campaign_permit_payload(registry), _private_key() + ) + return QualificationCampaignAuthority( + qualification_campaign_permit=permit, + qualification_campaign_permit_sha256=permit.artifact_sha256(), + expected=_campaign_expected(permit.payload), + qualification_signer_registry=registry, + qualification_signer_registry_sha256=registry.artifact_sha256(), + ) + + +def _write(path, authority: QualificationCampaignAuthority) -> None: + path.write_text( + json.dumps(authority.model_dump(mode="json"), sort_keys=True), + encoding="utf-8", + ) + if os.name != "nt": + path.chmod(0o600) + + +def _stub_guard(monkeypatch, authority: QualificationCampaignAuthority) -> None: + monkeypatch.setattr( + campaign, "_workflow_binding_refusal", lambda *_args, **_kw: None + ) + monkeypatch.setattr( + campaign, + "verify_qualification_campaign_permit", + lambda *_, **__: authority.qualification_campaign_permit_sha256, + ) + + +def test_campaign_authority_is_private_closed_and_content_bound(tmp_path): + authority = _authority() + path = tmp_path / "campaign.json" + _write(path, authority) + + assert load_qualification_campaign_authority(path) == authority + + raw = authority.model_dump(mode="json") + raw["unexpected"] = True + path.write_text(json.dumps(raw), encoding="utf-8") + if os.name != "nt": + path.chmod(0o600) + with pytest.raises(QualificationCampaignAuthorityError): + load_qualification_campaign_authority(path) + + +def test_campaign_guard_binds_the_exact_trial_and_retained_authorization( + tmp_path, monkeypatch +): + authority = _authority() + trial = authority.qualification_campaign_permit.payload.trial + path = tmp_path / "campaign.json" + _write(path, authority) + _stub_guard(monkeypatch, authority) + guard = QualificationCampaignGuard( + path, + workflow=object(), + case_id=trial.task, + input_digest=trial.input_digest, + campaign_id=trial.campaign_id, + run_id=trial.qualification_run_id, + ) + binding = guard.authorization_binding(object()) + authorization = SimpleNamespace( + **binding, + qualification_case_id=trial.task, + runtime_inputs_digest=trial.input_digest, + qualification_campaign_id_sha256=campaign.qualification_campaign_id_sha256( + trial.campaign_id + ), + qualification_run_id_sha256=campaign.qualification_run_id_sha256( + trial.qualification_run_id + ), + ) + + assert guard.authorization_refusal(object(), authorization) is None + authorization.qualification_case_id = "different-case" + assert guard.authorization_refusal(object(), authorization) == ( + "qualification campaign permit differs from the exact trial" + ) + + +def test_campaign_consumption_state_can_only_grow(tmp_path, monkeypatch): + authority = _authority() + trial = authority.qualification_campaign_permit.payload.trial + path = tmp_path / "campaign.json" + _write(path, authority) + _stub_guard(monkeypatch, authority) + guard = QualificationCampaignGuard( + path, + workflow=object(), + case_id=trial.task, + input_digest=trial.input_digest, + ) + consumed_id = "00000000-0000-4000-8000-000000000099" + _write(path, authority.model_copy(update={"consumed_permit_ids": (consumed_id,)})) + guard._load_current() + _write(path, authority) + + with pytest.raises( + QualificationCampaignAuthorityError, + match="consumption state rolled back", + ): + guard._load_current() + + +def test_consumed_campaign_permit_refuses_before_input(tmp_path, monkeypatch): + authority = _authority() + permit_id = authority.qualification_campaign_permit.payload.permit_id + authority = authority.model_copy(update={"consumed_permit_ids": (permit_id,)}) + trial = authority.qualification_campaign_permit.payload.trial + path = tmp_path / "campaign.json" + _write(path, authority) + monkeypatch.setattr( + campaign, "_workflow_binding_refusal", lambda *_args, **_kw: None + ) + + def _refuse(*_, consumed_permit_ids, **__): + if permit_id in consumed_permit_ids: + raise QualificationCampaignPermitError("consumed") + return authority.qualification_campaign_permit_sha256 + + monkeypatch.setattr(campaign, "verify_qualification_campaign_permit", _refuse) + guard = QualificationCampaignGuard( + path, + workflow=object(), + case_id=trial.task, + input_digest=trial.input_digest, + ) + + assert guard.refusal(object()) == "qualification campaign permit is not active" + + +def test_forged_case_id_cannot_bypass_production_admission(): + authorization = GovernedRunAuthorization( + bundle_content_digest=SHA_A, + runtime_inputs_digest=SHA_A, + admitted_policy_name="permissive", + execution_profile="standard", + approval_source="qualification-campaign", + qualification_project_id="project-1", + qualification_project_revision=1, + qualification_project_contract_sha256=SHA_A, + qualification_case_id="forged-case", + qualification_campaign_id_sha256=SHA_A, + qualification_case_input_sha256=SHA_A, + qualification_run_id_sha256=SHA_A, + qualification_case_kind="representative", + qualification_case_action_paths={"write": "gui"}, + qualification_campaign_permit_id=IDS["admission_id"], + qualification_campaign_permit_sha256=SHA_A, + qualification_campaign_signer_registry_sha256=SHA_A, + qualification_campaign_signer_registry_revision=1, + qualification_campaign_signer_registry_expires_at="2099-01-01T00:00:00Z", + qualification_campaign_authority_sha256=SHA_A, + ) + production_guard = ProductionQualificationGuard.__new__( + ProductionQualificationGuard + ) + + with pytest.raises(ValueError, match="signed non-production campaign permit"): + Replayer( + FakeBackend(), + vision=FakeVision(), + governed_authorization=authorization, + production_qualification_guard=production_guard, + ) diff --git a/tests/test_qualification_project.py b/tests/test_qualification_project.py index d00e7d55..1f0b08ad 100644 --- a/tests/test_qualification_project.py +++ b/tests/test_qualification_project.py @@ -2145,6 +2145,14 @@ def test_qualification_authorization_cannot_omit_project_identity_scope( qualification_run_id_sha256="f" * 64, qualification_case_kind="representative", qualification_case_action_paths={"save": "gui"}, + qualification_campaign_permit_id="00000000-0000-4000-8000-000000000001", + qualification_campaign_permit_sha256="1" * 64, + qualification_campaign_signer_registry_sha256="2" * 64, + qualification_campaign_signer_registry_revision=1, + qualification_campaign_signer_registry_expires_at=( + "2099-01-01T00:00:00Z" + ), + qualification_campaign_authority_sha256="3" * 64, ) assert authorization.validate_workflow(workflow) == ( @@ -2845,6 +2853,28 @@ def test_persisted_certification_is_recomputed_and_policy_digest_bound( assert not current_certification_matches(workflow, policy=policy) +def test_persisted_certification_refuses_expired_or_invalid_bundle_expiry( + tmp_path: Path, +) -> None: + workflow = _workflow() + _configure(workflow, tier=VerificationTier.INDEPENDENT_SYSTEM) + evidence_root = tmp_path / "evidence" + _record_passing_campaign(workflow, evidence_root) + policy = load_policy("clinical-write") + assert certify_project(workflow, policy=policy, evidence_root=evidence_root).passed + assert workflow.manifest is not None + assert current_certification_matches(workflow, policy=policy) + + workflow.manifest.provenance.expires_at = "2000-01-01T00:00:00Z" + assert not current_certification_matches(workflow, policy=policy) + + workflow.manifest.provenance.expires_at = "invalid" + assert not current_certification_matches(workflow, policy=policy) + + workflow.manifest.provenance.expires_at = "2099-01-01T00:00:00Z" + assert current_certification_matches(workflow, policy=policy) + + def test_judgment_evidence_certification_reproduces_and_saves( tmp_path: Path, ) -> None: From d94309584a872c83feaefb92c858ca4d555081be Mon Sep 17 00:00:00 2001 From: Richard Abrich Date: Fri, 21 Aug 2026 19:12:50 -0400 Subject: [PATCH 2/8] style: ruff-format the campaign permit fixture test --- tests/test_qualification_project.py | 4 +--- 1 file changed, 1 insertion(+), 3 deletions(-) diff --git a/tests/test_qualification_project.py b/tests/test_qualification_project.py index 1f0b08ad..d459316a 100644 --- a/tests/test_qualification_project.py +++ b/tests/test_qualification_project.py @@ -2149,9 +2149,7 @@ def test_qualification_authorization_cannot_omit_project_identity_scope( qualification_campaign_permit_sha256="1" * 64, qualification_campaign_signer_registry_sha256="2" * 64, qualification_campaign_signer_registry_revision=1, - qualification_campaign_signer_registry_expires_at=( - "2099-01-01T00:00:00Z" - ), + qualification_campaign_signer_registry_expires_at=("2099-01-01T00:00:00Z"), qualification_campaign_authority_sha256="3" * 64, ) From 85a2e5c14576af30ef747464fd723b6222ae277e Mon Sep 17 00:00:00 2001 From: Richard Abrich Date: Fri, 21 Aug 2026 19:31:47 -0400 Subject: [PATCH 3/8] fix: scope the runtime authority gate to governed deliveries The v2 authority and campaign permit guards now enforce where an authority chain can exist: runs that carry the binding, and per-edge re-verification whenever a guard is supplied. A local run without an admission issuer keeps today's run-gate contracts instead of refusing with no way to obtain the missing artifact, and the qualification-case authorization builder keeps its released signature so existing campaign harnesses keep working. --- openadapt_flow/__main__.py | 64 ++++++++----------- openadapt_flow/production_qualification.py | 6 +- .../qualification_campaign_authority.py | 6 +- openadapt_flow/run_gate.py | 17 +---- openadapt_flow/runtime/replayer.py | 35 +++++----- .../test_qualification_campaign_authority.py | 2 +- 6 files changed, 55 insertions(+), 75 deletions(-) diff --git a/openadapt_flow/__main__.py b/openadapt_flow/__main__.py index 31946ddb..36936a1a 100644 --- a/openadapt_flow/__main__.py +++ b/openadapt_flow/__main__.py @@ -1715,7 +1715,6 @@ def _cmd_run(args: argparse.Namespace) -> int: """ from openadapt_flow.execution_profiles import ( execution_profile_contract, - requires_signed_qualification_admission, resolve_execution_profile, ) from openadapt_flow.ir import Workflow @@ -1918,44 +1917,33 @@ def _cmd_run(args: argparse.Namespace) -> int: campaign_permit_binding = None if qualification_case is not None: authority_file = getattr(args, "qualification_campaign_authority_file", None) - if not authority_file: - print( - "run REFUSED: qualification actuation requires a private signed " - "non-production campaign authority file. Nothing was executed." - ) - return 2 - try: - campaign_guard = QualificationCampaignGuard( - authority_file, - workflow=workflow, - case_id=qualification_case["case"].id, - input_digest=runtime_inputs_digest( - workflow, gate_params, runtime_worklists - ), - campaign_id=qualification_case["campaign_id"], - run_id=qualification_case["run_id"], - ) - campaign_permit_binding = campaign_guard.authorization_binding(workflow) - except QualificationCampaignAuthorityError: - print( - "run REFUSED: the non-production qualification campaign permit " - "is invalid, expired, consumed, or does not match this exact trial. " - "Nothing was executed." - ) - return 2 - args._qualification_campaign_guard = campaign_guard - requires_production_authority = ( - qualification_case is None - and requires_signed_qualification_admission(selected_profile, will_actuate=True) - ) - if requires_production_authority: - authority_file = getattr(args, "qualification_authority_file", None) - if not authority_file: - print( - "run REFUSED: Production actuation requires a private signed v2 " - "qualification authority file. Nothing was executed." + if authority_file: + try: + campaign_guard = QualificationCampaignGuard( + authority_file, + workflow=workflow, + case_id=qualification_case["case"].id, + input_digest=runtime_inputs_digest( + workflow, gate_params, runtime_worklists + ), + campaign_id=qualification_case["campaign_id"], + run_id=qualification_case["run_id"], + ) + campaign_permit_binding = campaign_guard.authorization_binding(workflow) + except QualificationCampaignAuthorityError: + print( + "run REFUSED: the non-production qualification campaign permit " + "is invalid, expired, consumed, or does not match this exact " + "trial. Nothing was executed." + ) + return 2 + args._qualification_campaign_guard = campaign_guard + if campaign_permit_binding is not None and local_authorization is not None: + local_authorization = local_authorization.model_copy( + update=campaign_permit_binding ) - return 2 + authority_file = getattr(args, "qualification_authority_file", None) + if authority_file: try: production_guard = ProductionQualificationGuard( authority_file, diff --git a/openadapt_flow/production_qualification.py b/openadapt_flow/production_qualification.py index 3cdac7c1..91091e0b 100644 --- a/openadapt_flow/production_qualification.py +++ b/openadapt_flow/production_qualification.py @@ -23,7 +23,7 @@ import os import stat from pathlib import Path -from typing import Literal +from typing import Final, Literal from uuid import UUID from pydantic import BaseModel, ConfigDict, Field, ValidationError, field_validator @@ -46,7 +46,9 @@ ) MAX_AUTHORITY_BYTES = 512 * 1024 -AUTHORITY_SCHEMA = "openadapt.production-qualification-authority/v1" +AUTHORITY_SCHEMA: Final[Literal["openadapt.production-qualification-authority/v1"]] = ( + "openadapt.production-qualification-authority/v1" +) class ProductionQualificationAuthorityError(ValueError): diff --git a/openadapt_flow/qualification_campaign_authority.py b/openadapt_flow/qualification_campaign_authority.py index 1a12399c..a233e07b 100644 --- a/openadapt_flow/qualification_campaign_authority.py +++ b/openadapt_flow/qualification_campaign_authority.py @@ -3,7 +3,7 @@ from __future__ import annotations from pathlib import Path -from typing import Literal +from typing import Final, Literal from uuid import UUID from pydantic import BaseModel, ConfigDict, Field, ValidationError, field_validator @@ -29,7 +29,9 @@ ) from openadapt_flow.runtime.authorization import GovernedRunAuthorization -AUTHORITY_SCHEMA = "openadapt.qualification-campaign-authority/v1" +AUTHORITY_SCHEMA: Final[Literal["openadapt.qualification-campaign-authority/v1"]] = ( + "openadapt.qualification-campaign-authority/v1" +) class QualificationCampaignAuthorityError(ValueError): diff --git a/openadapt_flow/run_gate.py b/openadapt_flow/run_gate.py index 48cb86fc..fd557ca8 100644 --- a/openadapt_flow/run_gate.py +++ b/openadapt_flow/run_gate.py @@ -1101,7 +1101,6 @@ def build_qualification_case_authorization( worklists: Optional[dict[str, list[dict[str, str]]]], campaign_id: str, run_id: str, - campaign_permit_binding: dict[str, object], fault_driver: Any = None, ) -> GovernedRunAuthorization: """Build the exact authority for one governed qualification-case run. @@ -1160,18 +1159,6 @@ def build_qualification_case_authorization( }, "required_identity_step_ids": tuple(sorted(required_identity)), } - permit_fields = { - field - for field in GovernedRunAuthorization.model_fields - if field.startswith("qualification_campaign_") - and field - not in { - "qualification_campaign_id_sha256", - } - } - if set(campaign_permit_binding) != permit_fields: - raise ValueError("qualification campaign permit binding is incomplete") - updates.update(campaign_permit_binding) if case.kind is not QualificationCaseKind.REPRESENTATIVE: target = case.resolved_fault_target() if target is None: @@ -1200,9 +1187,7 @@ def build_qualification_case_authorization( elif fault_driver is not None: raise ValueError("representative qualification case cannot bind a fault driver") - bound = GovernedRunAuthorization.model_validate( - {**authorization.model_dump(mode="json"), **updates} - ) + bound = authorization.model_copy(update=updates) validation_error = bound.validate_workflow(workflow) if validation_error is not None: raise ValueError( diff --git a/openadapt_flow/runtime/replayer.py b/openadapt_flow/runtime/replayer.py index 3f35adb9..aa2479c8 100644 --- a/openadapt_flow/runtime/replayer.py +++ b/openadapt_flow/runtime/replayer.py @@ -566,21 +566,15 @@ def __init__( if not isinstance(qualification_campaign_guard, QualificationCampaignGuard): raise ValueError("qualification campaign authority is invalid") - production_profile = getattr(governed_authorization, "execution_profile", None) - qualification_case = getattr( - governed_authorization, "qualification_case_id", None - ) - if production_profile in {"standard", "regulated"}: - if qualification_case is not None: - if qualification_campaign_guard is None: - raise ValueError( - "qualification actuation requires a signed non-production " - "campaign permit" - ) - elif production_qualification_guard is None: - raise ValueError( - "Production actuation requires a verified v2 qualification authority" - ) + carries_campaign_permit = ( + getattr(governed_authorization, "qualification_campaign_permit_id", None) + is not None + ) + if carries_campaign_permit and qualification_campaign_guard is None: + raise ValueError( + "qualification actuation bound to a campaign permit requires the " + "signed non-production campaign authority" + ) self.managed_dispatch_binding = managed_dispatch_binding self.governed_authorization = governed_authorization self.production_qualification_guard = production_qualification_guard @@ -9297,7 +9291,10 @@ def _production_qualification_refusal(self, workflow: Workflow) -> Optional[str] return None guard = self.production_qualification_guard if guard is None: - return "Production actuation has no v2 qualification authority" + # A local Standard/Regulated run keeps the run-gate contracts it + # always had; a v2 authority becomes mandatory once an admission + # issuer provisions one for the runner. + return None try: expected = guard.authorization_binding(workflow) except Exception as exc: # noqa: BLE001 - private authority boundary @@ -9320,6 +9317,12 @@ def _qualification_campaign_refusal(self, workflow: Workflow) -> Optional[str]: return None guard = self.qualification_campaign_guard if guard is None: + carries_permit = ( + getattr(authorization, "qualification_campaign_permit_id", None) + is not None + ) + if not carries_permit: + return None return "qualification actuation has no signed campaign permit" try: return guard.authorization_refusal(workflow, authorization) diff --git a/tests/test_qualification_campaign_authority.py b/tests/test_qualification_campaign_authority.py index 03f63336..bff98e87 100644 --- a/tests/test_qualification_campaign_authority.py +++ b/tests/test_qualification_campaign_authority.py @@ -195,7 +195,7 @@ def test_forged_case_id_cannot_bypass_production_admission(): ProductionQualificationGuard ) - with pytest.raises(ValueError, match="signed non-production campaign permit"): + with pytest.raises(ValueError, match="requires the signed non-production campaign authority"): Replayer( FakeBackend(), vision=FakeVision(), From 603c44c9d9f7a08735bd2f98de572b9eed7f05ae Mon Sep 17 00:00:00 2001 From: Richard Abrich Date: Fri, 21 Aug 2026 19:35:24 -0400 Subject: [PATCH 4/8] style: wrap the campaign authority refusal matcher --- tests/test_qualification_campaign_authority.py | 4 +++- 1 file changed, 3 insertions(+), 1 deletion(-) diff --git a/tests/test_qualification_campaign_authority.py b/tests/test_qualification_campaign_authority.py index bff98e87..99fe8411 100644 --- a/tests/test_qualification_campaign_authority.py +++ b/tests/test_qualification_campaign_authority.py @@ -195,7 +195,9 @@ def test_forged_case_id_cannot_bypass_production_admission(): ProductionQualificationGuard ) - with pytest.raises(ValueError, match="requires the signed non-production campaign authority"): + with pytest.raises( + ValueError, match="requires the signed non-production campaign authority" + ): Replayer( FakeBackend(), vision=FakeVision(), From 0509008c995ea0f3b175505884f6039d3c00a12f Mon Sep 17 00:00:00 2001 From: Richard Abrich Date: Fri, 21 Aug 2026 20:09:13 -0400 Subject: [PATCH 5/8] fix: keep the qualification-case CLI contract working without a permit file The campaign authority flag is opt-in again: existing run-case and resume invocations parse as before, a retained campaign or admission binding still requires its authority file on resume, and the resume path no longer reuses the campaign file as a Production authority. --- openadapt_flow/__main__.py | 64 ++++++++++++++++++++------------------ 1 file changed, 33 insertions(+), 31 deletions(-) diff --git a/openadapt_flow/__main__.py b/openadapt_flow/__main__.py index 36936a1a..cabd471c 100644 --- a/openadapt_flow/__main__.py +++ b/openadapt_flow/__main__.py @@ -2113,8 +2113,7 @@ def _cmd_resume(args: argparse.Namespace) -> int: retained_authorization = manifest.governed_authorization if ( retained_authorization is not None - and retained_authorization.execution_profile in {"standard", "regulated"} - and retained_authorization.qualification_case_id is None + and retained_authorization.production_qualification_admission_id is not None ): from openadapt_flow.production_qualification import ( ProductionQualificationAuthorityError, @@ -2128,10 +2127,35 @@ def _cmd_resume(args: argparse.Namespace) -> int: "qualification authority again. Nothing was executed." ) return 3 + try: + production_guard = ProductionQualificationGuard( + authority_file, + remote_permit_revalidation=( + manifest.delivery_authority_kind == "cloud_runner" + ), + ) + production_binding = production_guard.authorization_binding(workflow) + except ProductionQualificationAuthorityError: + print( + "Resume REFUSED: the Production qualification authority is invalid, " + "expired, revoked, or does not match this exact run. Nothing was " + "executed." + ) + return 3 + if any( + getattr(retained_authorization, field, None) != value + for field, value in production_binding.items() + ): + print( + "Resume REFUSED: the Production qualification authority differs " + "from the retained run. Nothing was executed." + ) + return 3 + args._production_qualification_guard = production_guard if ( retained_authorization is not None - and retained_authorization.qualification_case_id is not None + and retained_authorization.qualification_campaign_permit_id is not None ): from openadapt_flow.qualification_campaign_authority import ( QualificationCampaignAuthorityError, @@ -2165,30 +2189,7 @@ def _cmd_resume(args: argparse.Namespace) -> int: "retained run. Nothing was executed." ) return 3 - try: - production_guard = ProductionQualificationGuard( - authority_file, - remote_permit_revalidation=( - manifest.delivery_authority_kind == "cloud_runner" - ), - ) - production_binding = production_guard.authorization_binding(workflow) - except ProductionQualificationAuthorityError: - print( - "Resume REFUSED: the Production qualification authority is invalid, " - "expired, revoked, or does not match this exact run. Nothing was " - "executed." - ) - return 3 - if any( - getattr(retained_authorization, field, None) != value - for field, value in production_binding.items() - ): - print( - "Resume REFUSED: the Production qualification authority differs " - "from the retained run. Nothing was executed." - ) - return 3 + args._qualification_campaign_guard = campaign_guard # A GUI automation cannot be resumed without a LIVE backend/vision, so build # a fresh Replayer here (deployment wiring from --config) and hand it to the @@ -5323,8 +5324,8 @@ def build_parser() -> argparse.ArgumentParser: default=None, metavar="PATH", help=( - "Private owner-only v2 Production qualification authority. Required " - "for Standard or Regulated actuation; not required for --dry-run" + "Private owner-only v2 Production qualification authority; when " + "supplied it is verified and re-read at every input edge" ), ) p.add_argument( @@ -5861,10 +5862,11 @@ def build_parser() -> argparse.ArgumentParser: q.add_argument("--run-id", required=True, help="One local case-attempt identity") q.add_argument( "--qualification-campaign-authority-file", - required=True, + default=None, metavar="PATH", help=( - "Private owner-only signed non-production authority for this exact trial" + "Private owner-only signed non-production authority for this exact " + "trial; required once the run actuates under a campaign permit" ), ) q.add_argument( From eb742ecab7ea5b6a102d3d759d703260c75a230f Mon Sep 17 00:00:00 2001 From: Richard Abrich Date: Fri, 21 Aug 2026 20:44:19 -0400 Subject: [PATCH 6/8] fix: let case authorizations stay valid before a permit is bound A qualification-case authorization without campaign permit fields remains the released contract; the all-or-nothing validator still refuses any partial permit binding, and the replayer requires the signed authority once a binding is present. --- openadapt_flow/runtime/authorization.py | 6 ------ 1 file changed, 6 deletions(-) diff --git a/openadapt_flow/runtime/authorization.py b/openadapt_flow/runtime/authorization.py index fc224361..5e814c85 100644 --- a/openadapt_flow/runtime/authorization.py +++ b/openadapt_flow/runtime/authorization.py @@ -368,12 +368,6 @@ def _qualification_binding_is_complete(self) -> "GovernedRunAuthorization": value is not None for value in campaign_permit ): raise ValueError("qualification campaign permit binding is incomplete") - if self.qualification_case_id is not None and not all( - value is not None for value in campaign_permit - ): - raise ValueError( - "qualification cases require a signed campaign permit binding" - ) if ( self.qualification_case_kind is not None and self.qualification_case_id is None From 8abc1d049767be39e950ac35a508293190f93f61 Mon Sep 17 00:00:00 2001 From: Richard Abrich Date: Fri, 21 Aug 2026 20:48:42 -0400 Subject: [PATCH 7/8] fix: bind the campaign permit only when the authority is supplied The case authorization accepts an optional permit binding again: the CLI passes it when a campaign authority file is provided, the binding completeness check still refuses any partial set, and existing callers without one keep working. --- openadapt_flow/__main__.py | 1 - openadapt_flow/run_gate.py | 11 +++++++++++ 2 files changed, 11 insertions(+), 1 deletion(-) diff --git a/openadapt_flow/__main__.py b/openadapt_flow/__main__.py index cabd471c..6f91cc94 100644 --- a/openadapt_flow/__main__.py +++ b/openadapt_flow/__main__.py @@ -1992,7 +1992,6 @@ def _cmd_run(args: argparse.Namespace) -> int: if qualification_case is not None: runtime_worklists = qualification_case["worklists"] - assert campaign_permit_binding is not None try: authorization = build_qualification_case_authorization( workflow, diff --git a/openadapt_flow/run_gate.py b/openadapt_flow/run_gate.py index fd557ca8..632810a6 100644 --- a/openadapt_flow/run_gate.py +++ b/openadapt_flow/run_gate.py @@ -1101,6 +1101,7 @@ def build_qualification_case_authorization( worklists: Optional[dict[str, list[dict[str, str]]]], campaign_id: str, run_id: str, + campaign_permit_binding: Optional[dict[str, object]] = None, fault_driver: Any = None, ) -> GovernedRunAuthorization: """Build the exact authority for one governed qualification-case run. @@ -1187,6 +1188,16 @@ def build_qualification_case_authorization( elif fault_driver is not None: raise ValueError("representative qualification case cannot bind a fault driver") + if campaign_permit_binding is not None: + permit_fields = { + field + for field in GovernedRunAuthorization.model_fields + if field.startswith("qualification_campaign_") + } + if set(campaign_permit_binding) != permit_fields: + raise ValueError("qualification campaign permit binding is incomplete") + updates.update(campaign_permit_binding) + bound = authorization.model_copy(update=updates) validation_error = bound.validate_workflow(workflow) if validation_error is not None: From 445acfe3bd162da82e92014c5f910059771ffa79 Mon Sep 17 00:00:00 2001 From: Richard Abrich Date: Fri, 21 Aug 2026 21:26:50 -0400 Subject: [PATCH 8/8] fix: keep the dispatch binding digest grammar stable The runtime-local v2 authority bindings stay outside the cross-repo dispatch digest so Cloud and Flow keep hashing identical payloads. They remain enforced through direct field comparison on the dispatch path and through their own digests inside the v2 permit. --- openadapt_flow/runner/protocol.py | 32 ++++++++++++++++++++++++++++++- 1 file changed, 31 insertions(+), 1 deletion(-) diff --git a/openadapt_flow/runner/protocol.py b/openadapt_flow/runner/protocol.py index c4f67509..f469b603 100644 --- a/openadapt_flow/runner/protocol.py +++ b/openadapt_flow/runner/protocol.py @@ -34,6 +34,32 @@ #: Cloud long-poll ceiling (runners.ts POLL_MAX_WAIT_S). POLL_MAX_WAIT_S = 25 +#: Runtime-local v2 authority bindings are deliberately OUTSIDE the dispatch +#: binding digest. The digest grammar is a cross-repo contract: Cloud and Flow +#: must hash byte-identical payloads, so fields Cloud does not emit may not +#: move the digest. These bindings are still enforced directly — the dispatch +#: path compares every production_qualification_* field between the envelope +#: authorization and the local authorization, and the terminal-verification +#: v2 permit carries its own admission and authority digests. +DISPATCH_BINDING_LOCAL_FIELDS = frozenset( + { + "production_qualification_admission_id", + "production_qualification_admission_sha256", + "production_qualification_evidence_identity_sha256", + "production_qualification_runtime_validation_id", + "production_qualification_signer_registry_sha256", + "production_qualification_signer_registry_revision", + "production_qualification_signer_registry_expires_at", + "production_qualification_authority_sha256", + "qualification_campaign_permit_id", + "qualification_campaign_permit_sha256", + "qualification_campaign_signer_registry_sha256", + "qualification_campaign_signer_registry_revision", + "qualification_campaign_signer_registry_expires_at", + "qualification_campaign_authority_sha256", + } +) + class DispatchParseError(ValueError): """A dispatch payload could not be strictly parsed (contract drift).""" @@ -50,7 +76,11 @@ def dispatch_binding_sha256( payload = { "run_id": run_id, - "authorization": authorization.model_dump(mode="json"), + "authorization": { + key: value + for key, value in authorization.model_dump(mode="json").items() + if key not in DISPATCH_BINDING_LOCAL_FIELDS + }, } encoded = json.dumps( payload, sort_keys=True, separators=(",", ":"), ensure_ascii=False