diff --git a/openadapt_flow/__main__.py b/openadapt_flow/__main__.py index f81a019a..6f91cc94 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)), @@ -1695,11 +1718,13 @@ def _cmd_run(args: argparse.Namespace) -> int: 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 +1851,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 +1869,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 +1905,86 @@ 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 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 + ) + authority_file = getattr(args, "qualification_authority_file", None) + if authority_file: + 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" @@ -1916,6 +2001,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 +2014,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 +2107,89 @@ 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.production_qualification_admission_id is not 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 + 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_campaign_permit_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 + 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 # durable resume entrypoint, which re-binds params from the run manifest. @@ -2076,6 +2239,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 +5318,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; when " + "supplied it is verified and re-read at every input edge" + ), + ) p.add_argument( "--dry-run", "--explain", @@ -5181,6 +5355,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 +5859,15 @@ 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", + default=None, + metavar="PATH", + help=( + "Private owner-only signed non-production authority for this exact " + "trial; required once the run actuates under a campaign permit" + ), + ) 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..91091e0b --- /dev/null +++ b/openadapt_flow/production_qualification.py @@ -0,0 +1,393 @@ +"""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 Final, 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: Final[Literal["openadapt.production-qualification-authority/v1"]] = ( + "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..a233e07b --- /dev/null +++ b/openadapt_flow/qualification_campaign_authority.py @@ -0,0 +1,299 @@ +"""Fail-closed authority for one signed non-production qualification trial.""" + +from __future__ import annotations + +from pathlib import Path +from typing import Final, 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: Final[Literal["openadapt.qualification-campaign-authority/v1"]] = ( + "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..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: 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/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 diff --git a/openadapt_flow/runtime/authorization.py b/openadapt_flow/runtime/authorization.py index e156dab2..5e814c85 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,18 @@ 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_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..aa2479c8 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,35 @@ 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") + 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 + 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 +1084,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 +9278,60 @@ 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: + # 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 + 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: + 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) + 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 +9344,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..99fe8411 --- /dev/null +++ b/tests/test_qualification_campaign_authority.py @@ -0,0 +1,206 @@ +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="requires the signed non-production campaign authority" + ): + 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..d459316a 100644 --- a/tests/test_qualification_project.py +++ b/tests/test_qualification_project.py @@ -2145,6 +2145,12 @@ 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 +2851,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: