diff --git a/packages/flow/docs/VERIFY_RUN.md b/packages/flow/docs/VERIFY_RUN.md new file mode 100644 index 000000000..ef25e33ce --- /dev/null +++ b/packages/flow/docs/VERIFY_RUN.md @@ -0,0 +1,141 @@ +# Check a write that something else made + +Preview, aligned with RFC-0002 (Open Effect Receipt). The receipt format may +still change before the RFC is final. + +An agent, a script, or a person says it saved a record. `verify-run` reads +the system of record and decides whether that is true. It uses the same +record checks a production `run` uses (`rest`, `sql`, `fhir`, `graphql`, and +the other kinds in [`EFFECT_KIT.md`](EFFECT_KIT.md)). It makes no AI model +calls. + +```bash +# Before the agent starts: read the store so new rows can be told from old ones. +openadapt-flow verify-run --verify verify.yaml --params task.json \ + --capture-baseline before.json + +# ... the agent does its work ... + +openadapt-flow verify-run --verify verify.yaml --params task.json \ + --run-record agent-run.json --authorization-ref TICKET-1234 \ + --baseline before.json --out share/ +``` + +`verify.yaml` has two parts, each in a shape Flow already reads: + +```yaml +schema_version: openadapt.verify/v1 +verifier: # the same keys as the effects: section of deployment.yaml + kind: rest + base_url: http://localhost:8080 + records_path: /api/db +effects: # the same list as effect_contract.yaml + - kind: record_written + match: + patient_id: {param: patient_id} + note: {param: note} + expected_count: 1 + count_new_only: true +``` + +It is the same file `openadapt-flow qualify --verify` reads (see +[`QUALIFY_A_WORKFLOW.md`](QUALIFY_A_WORKFLOW.md)). verify-run ignores the +qualify-only sections (`app`, `save_step`, `inputs`, `campaign`, and +`verifier.session`), expands `record` to the same effect list qualify uses, +and expands `${NAME}` environment references the same way. So the receipt's +`workflow_digest` equals the `verify_config_sha256` qualify reports for the +same file. `max_execution_window_s` (default 3600) is the one key only +verify-run reads. + +The three inputs come from different people on purpose: + +- `--params` holds the values the task named, such as which patient. They + come from whoever assigned the task, so the agent can't pick what it is + checked against. +- `--run-record` is what the agent reports. It must name `executor_kind` + (`agent`, `script`, `human`, `flow_replay`, or `unknown`). The receipt + keeps a salted digest of the file and reads nothing else from it except + `executor_id` and `executor_build`, which it also keeps only as digests. +- `--baseline` is the read taken before the agent started. Without it, a + contract that counts new rows can't be judged, and the verdict is + `INDETERMINATE`. A baseline works once: a second run with the same file + is refused. One older than `max_execution_window_s` (default 3600) counts + as no baseline. + +Optional flags: + +- `--approver approver.json` names who approved the write, as + `{kind, id, policy_ref}` with an opaque `id`. If the approver is the + executor itself, the outcome is `rejected_policy` even when the write + landed. +- `--remediates RECEIPT_ID` links this receipt to an earlier one it fixes. +- `--correlation-id ID` carries your own ticket or trace id. + +Both link ids must be opaque (8 to 128 of `A-Za-z0-9._:-`). They are absent +unless given, and when present they are signed with the rest of the receipt. + +## Verdicts + +| What the store shows | `effect_verdict` | `outcome` | +|---|---|---| +| The declared record, exactly once, through a system-of-record read | `CONFIRMED` | `verified` | +| The same, but the executor approved its own write | `CONFIRMED` | `rejected_policy` | +| Nothing that matches | `REFUTED` | `halted_before_effect` | +| A duplicate, a wrong value, or only some of the effects | `REFUTED` | `reconciliation_required` | +| The store could not be read, no baseline, or a separate-session read | `INDETERMINATE` or `CONFIRMED` | `reconciliation_required` | + +Only a read through the system of record produces `verified`. A screen +read-back (`onscreen`) is refused before anything runs. + +`verification_tier` on the receipt says how the effects were checked +(RFC-0002, weakest effect wins): + +- `independent_readback`: every effect was read from the system of record + without the UI (Flow evidence tier 1). +- `screen_readback`: at least one effect was read through a UI session, + including a separate signed-in session (Flow tier 2). +- `unchecked`: at least one effect was not read. + +Only `independent_readback` can yield `verified`. + +## The receipt + +`share/effect-receipt.json` holds `receipt` and `evidence`. `receipt` has +every field of the Execute receipt (`openadapt.execute-evidence-receipt/v1`) +plus `executor_kind`, `executor_ref_sha256`, `effect_verdict`, and +`verification_tier` (and `remediates` and `correlation_id` when given), +under the schema `openadapt.effect-receipt/v1`. `evidence.readback` lists +each effect's contract hash, verdict, and counts. `receipt.evidence_digest` +is the SHA-256 of `evidence`. Neither part holds record values, URLs, or the +agent's own text. + +Every digest whose input could hold customer data (the run record, the +authorization reference, the baseline, and each bound parameter inside an +effect's contract hash) is an HMAC-SHA256 keyed with a 256-bit salt drawn +for this run (`openadapt.run-salt/v1`). `evidence.salting.salt_commitment` +is the SHA-256 of the salt. The salt itself stays in a private file: the +baseline, or `share/effect-run.private.json` when there is no baseline. Each +REST read carries `X-OpenAdapt-Execution-Id` and a `traceparent` header so +the store's own logs can be matched to the run. + +`evidence.authorization.approver` keeps a placeholder shape until OpenAdapt +types defines approver identity. + +The file is unsigned. An issuer signs it with +`openadapt.effect-seal-signature/v1`, over its own domain, so a verifier +built for Execute Seals refuses an effect receipt instead of passing it. +Check a signed file offline against keys you pinned: + +```bash +openadapt-flow verify-run --check signed.json --keys seal-keys.json +``` + +The check reports the signature (`verified`) and the outcome +(`verified_outcome`) separately. It exits 0 only when both hold. A genuine +signature over `halted_before_effect` still exits 1 unless you pass +`--accept-outcome halted_before_effect` (repeatable). An unsigned file is +never a pass. + +The baseline and run files hold raw records or the salt. They are written +with mode 0600 and never go into the receipt. Delete them when you no longer +need them. diff --git a/packages/flow/openadapt_flow/__main__.py b/packages/flow/openadapt_flow/__main__.py index 3baf2e40e..4ba6d66f1 100644 --- a/packages/flow/openadapt_flow/__main__.py +++ b/packages/flow/openadapt_flow/__main__.py @@ -2415,6 +2415,80 @@ def _cmd_attach_seal(args: argparse.Namespace) -> int: return 1 +def _cmd_verify_run(args: argparse.Namespace) -> int: + """``verify-run``: judge an effect another executor claims, from the store.""" + from openadapt_flow.verify_run import ( + VerifyRunError, + capture_baseline, + check_effect_receipt, + load_verify_config, + verify_run, + write_result, + ) + + def _json(path: str) -> Any: + return json.loads(Path(path).read_text(encoding="utf-8")) + + if args.check: + if not args.keys: + print( + "verify-run --check needs --keys (a pinned Seal key registry)", + file=sys.stderr, + ) + return 2 + registry = _json(args.keys) + if isinstance(registry, dict): + registry = registry.get("keys", []) + result = check_effect_receipt( + _json(args.check), registry, accept_outcomes=args.accept_outcome or () + ) + print(json.dumps(result, indent=2, sort_keys=True)) + if result.get("verified") and not result.get("verified_outcome"): + print( + "The signature is genuine, but the outcome is not verified.", + file=sys.stderr, + ) + return 0 if result.get("ok") else 1 + if not args.verify: + print("verify-run needs --verify verify.yaml", file=sys.stderr) + return 2 + try: + cfg = load_verify_config(args.verify) + params = _json(args.params) if args.params else {} + if not isinstance(params, dict): + raise VerifyRunError("--params must be a JSON object") + if args.capture_baseline: + capture_baseline(cfg, params, Path(args.capture_baseline)) + print( + f"baseline written: {args.capture_baseline} (private; holds raw records)" + ) + return 0 + if not args.run_record or not args.authorization_ref or not args.out: + raise VerifyRunError( + "verify-run needs --run-record, --authorization-ref, and --out" + ) + result = verify_run( + cfg, + params=params, + run_record=_json(args.run_record), + authorization_ref=args.authorization_ref, + baseline_path=Path(args.baseline) if args.baseline else None, + private_dir=Path(args.out), + approver=_json(args.approver) if args.approver else None, + remediates=args.remediates, + correlation_id=args.correlation_id, + ) + except VerifyRunError as exc: + print(f"verify-run REFUSED: {exc}", file=sys.stderr) + return 2 + out = write_result(result, Path(args.out)) + receipt = result["receipt"] + print(f"effect verdict: {receipt['effect_verdict']}") + print(f"outcome: {receipt['outcome']} executor: {receipt['executor_kind']}") + print(f"receipt (unsigned): {out}") + return 0 if receipt["outcome"] == "verified" else 1 + + def _cmd_run(args: argparse.Namespace) -> int: """Execute a bundle under a named deployment profile -- FAIL-CLOSED. @@ -7448,6 +7522,100 @@ def build_parser() -> argparse.ArgumentParser: ) sa.set_defaults(func=_cmd_attach_seal) + p = sub.add_parser( + "verify-run", + help=( + "Read the system of record and rule on an effect that an agent, " + "script, or person claims it made; writes an unsigned " + "openadapt.effect-receipt/v1. The executor's own record is " + "hashed, never trusted" + ), + ) + p.add_argument( + "--verify", + default=None, + metavar="YAML", + help="verify.yaml: verifier wiring and effect contract", + ) + p.add_argument( + "--params", + default=None, + metavar="JSON", + help="The authorized task's values the contract binds", + ) + p.add_argument( + "--run-record", + default=None, + metavar="JSON", + help="The executor's run record (needs executor_kind)", + ) + p.add_argument( + "--authorization-ref", + default=None, + help="Opaque id of the authorization in your own system", + ) + p.add_argument( + "--baseline", + default=None, + metavar="JSON", + help="Baseline captured before the executor ran", + ) + p.add_argument( + "--capture-baseline", + default=None, + metavar="JSON", + help="Read the store now and write a private baseline, then exit", + ) + p.add_argument( + "--out", default=None, metavar="DIR", help="Directory for effect-receipt.json" + ) + p.add_argument( + "--check", + default=None, + metavar="JSON", + help="Check a signed effect receipt offline instead", + ) + p.add_argument( + "--keys", + default=None, + metavar="JSON", + help="Pinned Seal key registry for --check", + ) + p.add_argument( + "--accept-outcome", + action="append", + default=None, + choices=[ + "halted_before_effect", + "reconciliation_required", + "rejected_policy", + "failed_platform", + ], + help=( + "With --check: also exit 0 for this outcome when the signature is " + "genuine (repeatable; no value accepts every outcome)" + ), + ) + p.add_argument( + "--approver", + default=None, + metavar="JSON", + help="Who approved the write: {kind, id, policy_ref}; id is opaque", + ) + p.add_argument( + "--remediates", + default=None, + metavar="RECEIPT_ID", + help="receipt_id of an earlier effect receipt this run remediates", + ) + p.add_argument( + "--correlation-id", + default=None, + metavar="ID", + help="Your own opaque id linking this receipt to a ticket or trace", + ) + p.set_defaults(func=_cmd_verify_run) + p = sub.add_parser( "resume", help=( diff --git a/packages/flow/openadapt_flow/verify_run.py b/packages/flow/openadapt_flow/verify_run.py new file mode 100644 index 000000000..69b26dc1a --- /dev/null +++ b/packages/flow/openadapt_flow/verify_run.py @@ -0,0 +1,1071 @@ +"""Verify an effect that something other than Flow replay claims it made. + +``openadapt-flow verify-run`` reads the system of record through an existing +:class:`~openadapt_flow.runtime.effects.EffectVerifier` and rules on a declared +effect contract. The executor can be a computer-use agent, a script, a person, +or a Flow replay. Its run record is bound into the receipt by digest and is +never read as evidence: the verdict comes from the read-back alone. + +Three inputs, kept apart on purpose: + +* ``verify.yaml``: the verifier wiring (the same keys as the ``effects:`` + section of ``deployment.yaml``) and the effect contract (the same list as + ``effect_contract.yaml``). +* ``--params``: the values the authorized task named (which patient, which + note). They come from whoever authorized the work, not from the executor, + so the executor cannot choose what it is checked against. +* ``--run-record``: what the executor says it did. Hashed, never trusted. + +The output is an ``openadapt.effect-receipt/v1`` document as RFC-0002 +(openadapt-internal ``docs/RFC-0002_OPEN_EFFECT_RECEIPT.md``) specifies: + +* The receipt keeps every field of ``openadapt.execute-evidence-receipt/v1`` + and adds ``executor_kind``, ``executor_ref_sha256``, ``effect_verdict`` and + ``verification_tier``. Two optional fields, ``remediates`` and + ``correlation_id``, link a receipt to the receipt it remediates and to the + caller's own trace; they are absent unless given, and when present they + are inside ``receipt_sha256`` and so inside the signature. +* Every digest whose input can hold customer data is salted with a per-run + 256-bit salt (``openadapt.run-salt/v1``). The salt lives only in a private + file (mode 0600): the baseline, or a run file when there is no baseline. +* ``verification_tier`` is ``independent_readback`` only when every effect + was read from the system of record without the UI (Flow tier 1). A + separate signed-in UI session (tier 2) is ``screen_readback``. Only + ``independent_readback`` can yield ``verified``. +* A signature is ``openadapt.effect-seal-signature/v1``, over its own domain, + so a verifier that predates the RFC refuses an effect receipt instead of + passing it as an Execute Seal. verify-run never signs; it checks. + +Nothing here signs, uploads, or calls a model. +""" + +from __future__ import annotations + +import base64 +import hashlib +import hmac +import json +import os +import re +import secrets +from datetime import datetime, timezone +from pathlib import Path +from typing import Any, Literal, Mapping, Optional, Sequence, cast +from uuid import uuid4 + +from pydantic import BaseModel, ConfigDict, Field + +from openadapt_flow.deployment import EffectsConfig, build_effect_verifier +from openadapt_flow.qualify.result import verify_config_sha256 +from openadapt_flow.runtime.effects.adapter import verifier_identity +from openadapt_flow.runtime.effects.effect import ( + Effect, + EffectState, + EffectVerdict, + Verdict, +) +from openadapt_flow.verification import ( + EffectContractMutationError, + VerificationTier, + oracle_tier_from_verification_tier, + verifier_effect_tier, + verify_effect_without_mutation, +) + +EFFECT_RECEIPT_SCHEMA = "openadapt.effect-receipt/v1" +EFFECT_EVIDENCE_SCHEMA = "openadapt.effect-evidence/v1" +EFFECT_SEAL_SIGNATURE_SCHEMA = "openadapt.effect-seal-signature/v1" +EFFECT_SEAL_ISSUER = "openadapt-cloud:unattested-verifier" +VERIFY_CONFIG_SCHEMA = "openadapt.verify/v1" +BASELINE_SCHEMA = "openadapt.effect-baseline/v1" +RUN_FILE_SCHEMA = "openadapt.effect-run/v1" +SALT_SCHEME = "openadapt.run-salt/v1" + +_EFFECT_SEAL_DOMAIN = (EFFECT_SEAL_SIGNATURE_SCHEMA + "\0").encode("utf-8") + +#: The baseline file holds exactly these keys. The first five are bound by +#: ``baseline_sha256``; ``salt`` is the HMAC key and never part of a message. +BASELINE_KEYS: tuple[str, ...] = ( + "schema_version", + "execution_id", + "baseline_read_at", + "verifier_identity", + "state", + "salt", +) +_BASELINE_BOUND_KEYS = BASELINE_KEYS[:5] + +ExecutorKind = Literal["flow_replay", "agent", "script", "human", "unknown"] +EXECUTOR_KINDS: tuple[str, ...] = ("flow_replay", "agent", "script", "human", "unknown") +APPROVER_KINDS: tuple[str, ...] = ("person", "rule", "agent", "system") + +EffectVerdictLabel = Literal["CONFIRMED", "REFUTED", "INDETERMINATE"] + +#: How the effects were checked (RFC-0002 decision 6 and the tier table). +ReadbackClass = Literal["independent_readback", "screen_readback", "unchecked"] + +#: Verifier kinds that never read independently of the acting surface. +_NOT_INDEPENDENT = {"none", "onscreen"} + +_OPAQUE = re.compile(r"^[A-Za-z0-9][A-Za-z0-9._:-]{7,127}$") +_DIGEST = re.compile(r"^sha256:[0-9a-f]{64}$") +_SALTED = re.compile(r"^hmac-sha256:[0-9a-f]{64}$") +_ISO_SECONDS = re.compile(r"^\d{4}-\d{2}-\d{2}T\d{2}:\d{2}:\d{2}Z$") + +_STRENGTH_BY_TIER = { + VerificationTier.INDEPENDENT_SYSTEM: "independent_system_of_record", + VerificationTier.INDEPENDENT_SESSION: "independent_session", + VerificationTier.PERSISTED_STATE_REACQUISITION: "persisted_state_reacquisition", + VerificationTier.IMMEDIATE_SCREEN: "immediate_screen_confirmation", +} +_STRENGTH_RANK = { + "immediate_screen_confirmation": 1, + "persisted_state_reacquisition": 2, + "independent_session": 3, + "independent_system_of_record": 4, +} + +#: Receipt fields beyond Execute v1. The last two are optional. +_NEW_RECEIPT_FIELDS = ( + "executor_kind", + "executor_ref_sha256", + "effect_verdict", + "verification_tier", +) +_OPTIONAL_RECEIPT_FIELDS = ("remediates", "correlation_id") +_SIGNATURE_FIELDS = frozenset( + { + "schema", + "issuer", + "algorithm", + "key_id", + "receipt_schema", + "receipt_sha256", + "run_id", + "issued_at", + "signature", + } +) + + +class VerifyRunError(ValueError): + """verify-run refused its inputs. Nothing was read or written.""" + + +class VerifyConfig(BaseModel): + """``verify.yaml``: what to read, and what must be true there.""" + + model_config = ConfigDict(extra="forbid") + + schema_version: Literal["openadapt.verify/v1"] = "openadapt.verify/v1" + #: Opaque ids copied into the receipt. Defaults say plainly that nothing + #: qualified this contract. + qualification_id: str = Field( + default="unqualified-verify-run", + pattern=r"^[A-Za-z0-9][A-Za-z0-9._:-]{7,127}$", + ) + environment_id: str = Field( + default="unnamed-environment", + pattern=r"^[A-Za-z0-9][A-Za-z0-9._:-]{7,127}$", + ) + minimum_effect_strength: Literal[ + "independent_system_of_record", "independent_session" + ] = "independent_system_of_record" + #: Longest time from the baseline read to the read-back. Outside it the + #: baseline counts as absent, so change claims are INDETERMINATE. + max_execution_window_s: int = Field(default=3600, ge=1, le=7 * 24 * 3600) + #: Same shape as ``deployment.yaml`` ``effects:``. + verifier: EffectsConfig + #: Same shape as ``effect_contract.yaml`` ``effects:``. + effects: list[Effect] = Field(min_length=1) + + +def load_verify_config(path: str | Path) -> VerifyConfig: + import yaml + + source = Path(path) + if not source.is_file(): + raise VerifyRunError(f"verify config {source} is not a file") + try: + data = yaml.safe_load(source.read_text(encoding="utf-8")) or {} + except yaml.YAMLError as exc: + raise VerifyRunError(f"cannot parse {source}: {exc}") from exc + return parse_verify_config(data, source=f"verify config {source}") + + +#: ``openadapt.verify/v1`` sections only qualify reads (RFC-0002, Qualify-only +#: sections). verify-run accepts and ignores them, so one file serves both. +QUALIFY_ONLY_SECTIONS: tuple[str, ...] = ("app", "save_step", "inputs", "campaign") + + +def _record_effects(record: Any) -> list[Effect]: + """Expand qualify's ``record`` shorthand exactly as qualify does.""" + + from types import SimpleNamespace + + from openadapt_flow.qualify.campaign import record_effects + from openadapt_flow.qualify.config import RecordConfig + + try: + parsed = RecordConfig.model_validate(record) + except Exception as exc: + raise VerifyRunError(f"invalid record section: {exc}") from exc + if not parsed.fields or not parsed.match: + raise VerifyRunError("record must name fields and match") + return list(record_effects(cast(Any, SimpleNamespace(record=parsed)))) + + +def parse_verify_config(data: Any, *, source: str = "verify config") -> VerifyConfig: + """Validate a parsed ``verify.yaml`` the way verify-run reads it. + + The same file qualify reads: ``${NAME}`` environment references are + expanded the same way, qualify-only sections and ``verifier.session`` are + ignored, and ``record`` expands to the same effect list qualify uses, so + ``workflow_digest`` equals qualify's ``verify_config_sha256``. + """ + + from openadapt_flow.qualify.config import _expand + + if not isinstance(data, dict): + raise VerifyRunError(f"{source} must be a YAML mapping") + missing: list[str] = [] + data = _expand(dict(data), missing) + if missing: + raise VerifyRunError( + "environment variable(s) not set: " + ", ".join(sorted(set(missing))) + ) + for name in QUALIFY_ONLY_SECTIONS: + data.pop(name, None) + verifier = data.get("verifier") + if isinstance(verifier, dict) and "session" in verifier: + # A qualify hook for an authenticated client; not part of the + # verifier's identity, and verify-run does not import hooks. + data["verifier"] = {k: v for k, v in verifier.items() if k != "session"} + record = data.pop("record", None) + if record is not None: + if data.get("effects") is not None: + raise VerifyRunError( + "give the save's checks as either effects or record, not both" + ) + data["effects"] = [e.model_dump(mode="json") for e in _record_effects(record)] + try: + cfg = VerifyConfig.model_validate(data) + except Exception as exc: + raise VerifyRunError(f"invalid {source}: {exc}") from exc + if cfg.verifier.kind in _NOT_INDEPENDENT: + raise VerifyRunError( + f"verifier kind {cfg.verifier.kind!r} does not read independently of " + "the executor; verify-run needs a system-of-record or " + "separate-session verifier" + ) + if cfg.verifier.candidates: + raise VerifyRunError("verify-run does not support verifier candidates yet") + for effect in cfg.effects: + if effect.needs_operator_confirmation: + raise VerifyRunError( + "the effect contract still holds an unconfirmed placeholder" + ) + if effect.readback is not None: + raise VerifyRunError( + "an on-screen read-back effect cannot be verified after the fact" + ) + return cfg + + +# -- hashing ------------------------------------------------------------------ + + +def _canonical(value: Any) -> bytes: + """Sorted-key compact JSON; the Seal's canonical form.""" + + return json.dumps( + value, sort_keys=True, separators=(",", ":"), ensure_ascii=False + ).encode("utf-8") + + +def sha256_tagged(value: Any) -> str: + return "sha256:" + hashlib.sha256(_canonical(value)).hexdigest() + + +def _salted_hex(salt: bytes, label: str, value: Any) -> str: + message = label.encode("utf-8") + b"\0" + _canonical(value) + return hmac.new(salt, message, hashlib.sha256).hexdigest() + + +def salted_digest(salt: bytes, label: str, value: Any) -> str: + """``openadapt.run-salt/v1``: HMAC-SHA256 keyed with the run salt.""" + + return "hmac-sha256:" + _salted_hex(salt, label, value) + + +def salt_commitment(salt: bytes) -> str: + return "sha256:" + hashlib.sha256(salt).hexdigest() + + +def receipt_sha256(receipt: Mapping[str, Any]) -> str: + """64 hex over the receipt without ``seal_signature``.""" + + body = {k: v for k, v in receipt.items() if k != "seal_signature"} + return hashlib.sha256(_canonical(body)).hexdigest() + + +def effect_signing_payload(signature: Mapping[str, Any]) -> bytes: + """Bytes an ``openadapt.effect-seal-signature/v1`` signature covers.""" + + unsigned = {k: v for k, v in signature.items() if k != "signature"} + return _EFFECT_SEAL_DOMAIN + _canonical(unsigned) + + +def _now() -> str: + return ( + datetime.now(timezone.utc) + .replace(microsecond=0) + .isoformat() + .replace("+00:00", "Z") + ) + + +def _parse_time(value: str) -> datetime: + return datetime.strptime(value, "%Y-%m-%dT%H:%M:%SZ").replace(tzinfo=timezone.utc) + + +def _write_private(path: Path, payload: Mapping[str, Any]) -> None: + path.parent.mkdir(parents=True, exist_ok=True) + tmp = path.with_name(path.name + ".tmp") + fd = os.open(tmp, os.O_WRONLY | os.O_CREAT | os.O_TRUNC, 0o600) + with os.fdopen(fd, "w", encoding="utf-8") as fh: + fh.write(json.dumps(payload, indent=2, sort_keys=True) + "\n") + os.chmod(tmp, 0o600) + os.replace(tmp, path) + + +def _new_execution_id() -> str: + return f"verify-{uuid4().hex}" + + +# -- verifier ----------------------------------------------------------------- + + +def _build_verifier(cfg: VerifyConfig, params: Mapping[str, object]) -> Any: + try: + verifier = build_effect_verifier(cfg.verifier, params) + except ValueError as exc: + raise VerifyRunError(f"cannot build the verifier: {exc}") from exc + if verifier is None: + raise VerifyRunError("verify config built no verifier") + return verifier + + +def _tag_reads(verifier: Any, execution_id: str) -> None: + """Add the run's own headers to every REST read. + + Called after ``verifier_identity`` is taken, and on the built verifier, + never on the config, so the identity and ``workflow_digest`` stay equal + across runs of one ``verify.yaml``. + """ + + if not hasattr(verifier, "headers"): + return + trace = secrets.token_hex(16) + span = secrets.token_hex(8) + verifier.headers = { + **(verifier.headers or {}), + "X-OpenAdapt-Execution-Id": execution_id, + "traceparent": f"00-{trace}-{span}-01", + } + + +# -- baseline ----------------------------------------------------------------- + + +def capture_baseline( + cfg: VerifyConfig, params: Mapping[str, object], out: Path +) -> dict[str, Any]: + """Read the system of record before the executor starts. + + The baseline is what lets ``count_new_only`` and ``exact_new_set`` tell a + record the executor added from one that was already there. It holds raw + records and the run salt, so it stays private on this machine: mode 0600, + never inside a receipt. Only its salted digest is bound. It also starts + the run: ``execution_id`` is generated here and the read step reuses it. + """ + + verifier = _build_verifier(cfg, params) + identity = verifier_identity(verifier) + execution_id = _new_execution_id() + _tag_reads(verifier, execution_id) + state = verifier.capture_pre_state() + if not isinstance(state, EffectState): + raise VerifyRunError("verifier returned no readable baseline") + payload = { + "schema_version": BASELINE_SCHEMA, + "execution_id": execution_id, + "baseline_read_at": _now(), + "verifier_identity": identity, + "state": state.model_dump(mode="json"), + "salt": secrets.token_bytes(32).hex(), + } + _write_private(Path(out), payload) + return payload + + +def baseline_digest(baseline: Mapping[str, Any], salt: bytes) -> str: + """Salted digest (label ``baseline``) of the five bound baseline fields.""" + + return salted_digest( + salt, "baseline", {key: baseline[key] for key in _BASELINE_BOUND_KEYS} + ) + + +def _used_marker(path: Path) -> Path: + return path.with_name(path.name + ".used") + + +def _load_baseline(path: Path, identity: str) -> tuple[dict[str, Any], bytes]: + try: + raw = json.loads(Path(path).read_text(encoding="utf-8")) + except (OSError, ValueError) as exc: + raise VerifyRunError(f"cannot read baseline {path}: {exc}") from exc + if not isinstance(raw, dict) or raw.get("schema_version") != BASELINE_SCHEMA: + raise VerifyRunError(f"{path} is not an {BASELINE_SCHEMA} file") + if set(raw) != set(BASELINE_KEYS): + # Every field must be one the digest names; nothing rides along unbound. + raise VerifyRunError( + "baseline file must hold exactly " + ", ".join(BASELINE_KEYS) + ) + if raw["verifier_identity"] != identity: + raise VerifyRunError("baseline was captured by a different verifier") + execution_id = raw["execution_id"] + if not isinstance(execution_id, str) or not _OPAQUE.match(execution_id): + raise VerifyRunError("baseline execution_id is not an opaque id") + read_at = raw["baseline_read_at"] + if not isinstance(read_at, str) or not _ISO_SECONDS.match(read_at): + raise VerifyRunError("baseline_read_at is not whole-second UTC") + try: + salt = bytes.fromhex(str(raw["salt"])) + except ValueError as exc: + raise VerifyRunError("baseline salt is not hex") from exc + if len(salt) != 32: + raise VerifyRunError("baseline salt must be 32 bytes") + return raw, salt + + +def _consume_baseline(path: Path, execution_id: str) -> None: + """One use: record the consumed run beside the baseline, refuse a reuse.""" + + marker = _used_marker(Path(path)) + try: + fd = os.open(marker, os.O_WRONLY | os.O_CREAT | os.O_EXCL, 0o600) + except FileExistsError as exc: + raise VerifyRunError( + f"baseline {path} was already used for {execution_id}; capture a new one" + ) from exc + with os.fdopen(fd, "w", encoding="utf-8") as fh: + fh.write(execution_id + "\n") + + +# -- the run record and the approver ----------------------------------------- + + +def _executor(run_record: Mapping[str, Any]) -> tuple[str, str, Optional[str]]: + kind = run_record.get("executor_kind") + if kind not in EXECUTOR_KINDS: + raise VerifyRunError( + "run record must name executor_kind as one of " + ", ".join(EXECUTOR_KINDS) + ) + executor_id = run_record.get("executor_id") + ref = ( + executor_id if isinstance(executor_id, str) and executor_id else "unidentified" + ) + build = run_record.get("executor_build") + build_digest: Optional[str] = None + if build is not None: + if not isinstance(build, Mapping) or not all( + isinstance(k, str) and isinstance(v, str) for k, v in build.items() + ): + raise VerifyRunError("executor_build must be an object of strings") + build_digest = sha256_tagged(dict(build)) + return ( + str(kind), + "sha256:" + hashlib.sha256(ref.encode("utf-8")).hexdigest(), + build_digest, + ) + + +def _approver(approver: Optional[Mapping[str, Any]]) -> Optional[dict[str, str]]: + """Placeholder shape until the openadapt-types approver model lands.""" + + if approver is None: + return None + if not isinstance(approver, Mapping) or set(approver) != { + "kind", + "id", + "policy_ref", + }: + raise VerifyRunError("approver must be {kind, id, policy_ref}") + if approver["kind"] not in APPROVER_KINDS: + raise VerifyRunError( + "approver kind must be one of " + ", ".join(APPROVER_KINDS) + ) + if not isinstance(approver["id"], str) or not _OPAQUE.match(approver["id"]): + raise VerifyRunError("approver id must be an opaque id, never a name or email") + if not isinstance(approver["policy_ref"], str) or not _DIGEST.match( + approver["policy_ref"] + ): + raise VerifyRunError("approver policy_ref must be sha256:") + return {k: str(approver[k]) for k in ("kind", "id", "policy_ref")} + + +def _opaque_or_none(name: str, value: Optional[str]) -> Optional[str]: + if value is None: + return None + if not isinstance(value, str) or not _OPAQUE.match(value): + raise VerifyRunError(f"{name} must be an opaque id (8-128 of A-Za-z0-9._:-)") + return value + + +# -- verdicts ----------------------------------------------------------------- + + +def _label(verdict: Verdict) -> EffectVerdictLabel: + labels: dict[Verdict, EffectVerdictLabel] = { + Verdict.CONFIRMED: "CONFIRMED", + Verdict.REFUTED: "REFUTED", + Verdict.INDETERMINATE: "INDETERMINATE", + } + return labels[verdict] + + +def _aggregate(verdicts: Sequence[EffectVerdict]) -> EffectVerdictLabel: + if any(v.verdict is Verdict.INDETERMINATE for v in verdicts): + return "INDETERMINATE" + if any(v.verdict is Verdict.REFUTED for v in verdicts): + return "REFUTED" + return "CONFIRMED" + + +def _observed(verdicts: Sequence[EffectVerdict]) -> str: + seen = {v.observed_effect for v in verdicts} + if "unknown" in seen: + return "unknown" + if "conflicting" in seen or seen == {"absent", "present"}: + # A wrong record, or some effects landed and others did not. + return "conflicting" + return "absent" if "absent" in seen else "present" + + +def readback_class( + verdicts: Sequence[EffectVerdict], tiers: Sequence[Optional[VerificationTier]] +) -> ReadbackClass: + """RFC-0002 tier rule, weakest effect wins. + + ``unchecked`` when any effect was not read (no tier, store unreachable, + stale). ``independent_readback`` only when every effect was read from the + system of record without the UI (Flow tier 1). Anything else, including a + separate signed-in UI session (tier 2), is ``screen_readback``. + """ + + if ( + any(t is None for t in tiers) + or any(v.unavailable for v in verdicts) + or any(v.stale for v in verdicts) + ): + return "unchecked" + if all(t is not None and t.reads_system_of_record() for t in tiers): + return "independent_readback" + return "screen_readback" + + +def verify_run( + cfg: VerifyConfig, + *, + params: Mapping[str, object], + run_record: Mapping[str, Any], + authorization_ref: str, + baseline_path: Optional[Path] = None, + private_dir: Optional[Path] = None, + approver: Optional[Mapping[str, Any]] = None, + remediates: Optional[str] = None, + correlation_id: Optional[str] = None, +) -> dict[str, Any]: + """Read the system of record and return ``{"receipt", "evidence"}``. + + The run record contributes ``executor_kind``, a digest of + ``executor_id``, a digest of ``executor_build``, and a salted digest of + itself. No other field of it is read. ``verified`` requires every effect + CONFIRMED by a system-of-record read at or above + ``minimum_effect_strength``; anything weaker is never upgraded. + + With ``baseline_path`` the run continues the one the baseline started + (its ``execution_id`` and salt) and the baseline is consumed. Without one, + a new salt is drawn and written to ``private_dir/effect-run.private.json`` + (mode 0600), so ``private_dir`` is required. + """ + + from openadapt_flow.contracts.execute import ( + EffectStrengthV1, + ExecuteEvidenceContractV1, + ExecuteEvidenceReceiptV1, + ExecuteTerminalOutcomeV1, + ) + + if not isinstance(run_record, Mapping): + raise VerifyRunError("run record must be a JSON object") + if not isinstance(authorization_ref, str) or not authorization_ref: + raise VerifyRunError("an authorization reference is required") + remediates = _opaque_or_none("remediates", remediates) + correlation_id = _opaque_or_none("correlation_id", correlation_id) + approver_doc = _approver(approver) + executor_kind, executor_ref, build_digest = _executor(run_record) + + verifier = _build_verifier(cfg, params) + identity = verifier_identity(verifier) + + baseline: Optional[dict[str, Any]] = None + if baseline_path is not None: + baseline, salt = _load_baseline(Path(baseline_path), identity) + execution_id = str(baseline["execution_id"]) + _consume_baseline(Path(baseline_path), execution_id) + else: + if private_dir is None: + raise VerifyRunError( + "without a baseline, verify-run needs a private directory for " + "the run salt" + ) + salt = secrets.token_bytes(32) + execution_id = _new_execution_id() + _write_private( + Path(private_dir) / "effect-run.private.json", + { + "schema_version": RUN_FILE_SCHEMA, + "execution_id": execution_id, + "salt": salt.hex(), + }, + ) + _tag_reads(verifier, execution_id) + + read_at = _now() + before: EffectState + baseline_sha: Optional[str] = None + baseline_read_at: Optional[str] = None + if baseline is not None: + elapsed = ( + _parse_time(read_at) - _parse_time(str(baseline["baseline_read_at"])) + ).total_seconds() + if 0 <= elapsed <= cfg.max_execution_window_s: + before = EffectState.model_validate(baseline["state"]) + baseline_sha = baseline_digest(baseline, salt) + baseline_read_at = str(baseline["baseline_read_at"]) + else: + # Outside the window the baseline proves nothing about this run. + baseline = None + if baseline is None: + # No baseline: the judge treats the pre-state as unknown, so every + # claim about change is INDETERMINATE, never guessed. + before = EffectState( + substrate=str(getattr(verifier, "substrate", "")), reachable=False + ) + + # Same digest qualify reports as verify_config_sha256 for this file. + contract_digest = verify_config_sha256(identity, list(cfg.effects)) + verdicts: list[EffectVerdict] = [] + rows: list[dict[str, Any]] = [] + weakest: Optional[VerificationTier] = None + tiers: list[Optional[VerificationTier]] = [] + for effect in cfg.effects: + try: + bound = effect.resolve(params) + # Raw values withheld; each bound parameter enters as its salted + # digest in the opaque-parameter slot (RFC-0002, salted digests 5). + salted_params = { + name: _salted_hex(salt, f"effect_param:{name}", params[name]) + for name in sorted(effect.referenced_params()) + } + contract_hash = effect.resolved_contract_hash( + {}, opaque_param_sha256=salted_params + ) + except (KeyError, ValueError) as exc: + raise VerifyRunError(f"effect contract cannot bind params: {exc}") from exc + tier = verifier_effect_tier(verifier, bound) + try: + verdict = verify_effect_without_mutation(verifier, bound, before) + except EffectContractMutationError as exc: + raise VerifyRunError(str(exc)) from exc + if tier is None and verdict.verdict is Verdict.CONFIRMED: + # An untyped verifier cannot certify anything. + verdict = EffectVerdict( + verdict=Verdict.INDETERMINATE, + kind=bound.kind, + substrate=verdict.substrate, + reason="verifier declares no evidence tier", + ) + if tier is not None and (weakest is None or int(tier) > int(weakest)): + weakest = tier + verdicts.append(verdict) + tiers.append(tier) + # Closed fields only: no reason text, no observed value, no records. + rows.append( + { + "effect_contract_hash": contract_hash, + "kind": bound.kind.value, + "substrate": verdict.substrate, + "flow_verification_tier": int(tier) if tier is not None else None, + "verdict": _label(verdict.verdict), + "observed_effect": verdict.observed_effect, + "observed_count": verdict.observed_count, + "expected_count": verdict.expected_count, + "unavailable": verdict.unavailable, + "stale": verdict.stale, + } + ) + + effect_verdict = _aggregate(verdicts) + observed_effect = _observed(verdicts) + verification_tier = readback_class(verdicts, tiers) + unread = verification_tier == "unchecked" + strength = ( + _STRENGTH_BY_TIER.get(weakest) if weakest is not None and not unread else None + ) + oracle_tier = ( + oracle_tier_from_verification_tier(weakest) + if weakest is not None and not unread + else 0 + ) + strong_enough = strength is not None and ( + _STRENGTH_RANK[strength] >= _STRENGTH_RANK[cfg.minimum_effect_strength] + ) + self_approved: Optional[bool] = None + if approver_doc is not None: + approver_ref = ( + "sha256:" + hashlib.sha256(approver_doc["id"].encode("utf-8")).hexdigest() + ) + self_approved = approver_ref == executor_ref + + delivery_uncertain = False + read_proves_write = ( + effect_verdict == "CONFIRMED" + and verification_tier == "independent_readback" + and strong_enough + and oracle_tier in (2, 3) + ) + if read_proves_write and self_approved is True: + # The write landed, but the executor approved its own write: no + # oversight separate from itself. Never verified. + outcome = ExecuteTerminalOutcomeV1.REJECTED_POLICY + elif read_proves_write: + outcome = ExecuteTerminalOutcomeV1.VERIFIED + elif effect_verdict == "REFUTED" and observed_effect == "absent": + # The store shows nothing landed, whatever the executor said. + outcome = ExecuteTerminalOutcomeV1.HALTED_BEFORE_EFFECT + else: + # A wrong effect, an unreadable store, or a read too weak to certify. + outcome = ExecuteTerminalOutcomeV1.RECONCILIATION_REQUIRED + delivery_uncertain = True + verified = outcome is ExecuteTerminalOutcomeV1.VERIFIED + landed = read_proves_write + + evidence = { + "schema_version": EFFECT_EVIDENCE_SCHEMA, + "executor": { + "kind": executor_kind, + "ref_sha256": executor_ref, + "build_sha256": build_digest, + "run_record_sha256": salted_digest(salt, "run_record", dict(run_record)), + "claims_counted_as_evidence": False, + }, + "authorization": { + "reference_sha256": salted_digest( + salt, "authorization_reference", authorization_ref + ), + "evaluated_by_verifier": False, + # Placeholder shape until the openadapt-types approver identity + # model ships (RFC-0002, Who authorized). + "approver": approver_doc, + "self_approved": self_approved, + }, + "salting": { + "scheme": SALT_SCHEME, + "salt_commitment": salt_commitment(salt), + }, + "contract": { + "contract_sha256": contract_digest, + "minimum_effect_strength": cfg.minimum_effect_strength, + }, + "readback": { + "verifier_identity": identity, + "verification_tier": verification_tier, + "baseline_sha256": baseline_sha, + "baseline_read_at": baseline_read_at, + "read_at": read_at, + "max_execution_window_s": cfg.max_execution_window_s, + "effects": rows, + }, + "verifier": { + "name": "openadapt-flow verify-run", + "model_calls": 0, + }, + "effect_verdict": effect_verdict, + } + contracts = ExecuteEvidenceContractV1( + # A reference was bound and the approver is not the executor; + # verify-run did not evaluate a policy (evidence.authorization says so). + authorization_passed=verified, + # Identity is the contract's record selector, judged in the store. + identity_passed=landed, + postcondition_passed=landed, + effect_passed=landed, + minimum_effect_strength=EffectStrengthV1(cfg.minimum_effect_strength), + observed_effect_strength=( + EffectStrengthV1(strength) if strength is not None else None + ), + # Nothing says the executor used no model, so the receipt cannot. + model_used=executor_kind != "flow_replay", + external_network_used=executor_kind != "flow_replay", + ) + contract_hex = contract_digest.split(":", 1)[1] + issued_at = _now() + v1 = ExecuteEvidenceReceiptV1( + receipt_id=f"receipt-{uuid4().hex}", + execution_id=execution_id, + workflow_digest=contract_digest, + workflow_version=f"contract-{contract_hex[:16]}", + qualification_id=cfg.qualification_id, + environment_id=cfg.environment_id, + runner_id=f"executor-{executor_ref.split(':', 1)[1][:16]}", + nonce=f"nonce-{uuid4().hex}", + oracle_tier=cast(Literal[0, 1, 2, 3], oracle_tier), + outcome=outcome, + contracts=contracts, + delivery_uncertain=delivery_uncertain, + evidence_digest=sha256_tagged(evidence), + issued_at=issued_at, + ) + receipt = v1.model_dump(mode="json", exclude={"seal_signature"}) + receipt["schema_version"] = EFFECT_RECEIPT_SCHEMA + receipt["executor_kind"] = executor_kind + receipt["executor_ref_sha256"] = executor_ref + receipt["effect_verdict"] = effect_verdict + receipt["verification_tier"] = verification_tier + if remediates is not None: + receipt["remediates"] = remediates + if correlation_id is not None: + receipt["correlation_id"] = correlation_id + problem = receipt_problem(receipt) or evidence_problem(evidence, receipt) + if problem is not None: # pragma: no cover - a producer bug, never input + raise VerifyRunError(f"verify-run built an invalid receipt: {problem}") + return { + "schema_version": EFFECT_RECEIPT_SCHEMA, + "receipt": receipt, + "evidence": evidence, + } + + +def write_result(result: Mapping[str, Any], out_dir: Path) -> Path: + out = Path(out_dir) / "effect-receipt.json" + _write_private(out, result) + return out + + +# -- offline check ------------------------------------------------------------ + + +def receipt_problem(receipt: Mapping[str, Any]) -> Optional[str]: + """None when the receipt meets the schema and the cross-field rules.""" + + from openadapt_flow.contracts.execute import ( + EXECUTE_EVIDENCE_RECEIPT_SCHEMA, + ExecuteEvidenceReceiptV1, + ) + + if receipt.get("schema_version") != EFFECT_RECEIPT_SCHEMA: + return "schema_version" + v1_names = set(ExecuteEvidenceReceiptV1.model_fields) + allowed = v1_names | set(_NEW_RECEIPT_FIELDS) | set(_OPTIONAL_RECEIPT_FIELDS) + if set(receipt) - allowed: + return "unknown_field" + if any(name not in receipt for name in _NEW_RECEIPT_FIELDS): + return "missing_field" + v1 = {k: v for k, v in receipt.items() if k in v1_names and k != "seal_signature"} + v1["schema_version"] = EXECUTE_EVIDENCE_RECEIPT_SCHEMA + try: + ExecuteEvidenceReceiptV1.model_validate(v1) + except Exception: + return "execute_v1_rules" + if not _ISO_SECONDS.match(str(receipt.get("issued_at", ""))): + return "issued_at" + if receipt["executor_kind"] not in EXECUTOR_KINDS: + return "executor_kind" + if not _DIGEST.match(str(receipt["executor_ref_sha256"])): + return "executor_ref_sha256" + if receipt["effect_verdict"] not in ("CONFIRMED", "REFUTED", "INDETERMINATE"): + return "effect_verdict" + if receipt["verification_tier"] not in ( + "independent_readback", + "screen_readback", + "unchecked", + ): + return "verification_tier" + for name in _OPTIONAL_RECEIPT_FIELDS: + if name in receipt and not _OPAQUE.match(str(receipt[name])): + return name + outcome = receipt.get("outcome") + if outcome in ("verified", "rolled_back_verified") and ( + receipt["effect_verdict"] != "CONFIRMED" + or receipt["verification_tier"] != "independent_readback" + or receipt.get("oracle_tier") not in (2, 3) + ): + return "verified_without_independent_readback" + if ( + outcome == "reconciliation_required" + and receipt.get("delivery_uncertain") is not True + ): + return "delivery_uncertain" + return None + + +def evidence_problem( + evidence: Mapping[str, Any], receipt: Mapping[str, Any] +) -> Optional[str]: + """None when the manifest is bound to the receipt and meets its rules.""" + + if sha256_tagged(dict(evidence)) != receipt.get("evidence_digest"): + return "evidence_mismatch" + executor = evidence.get("executor") or {} + readback = evidence.get("readback") or {} + authorization = evidence.get("authorization") or {} + salting = evidence.get("salting") or {} + if ( + evidence.get("effect_verdict") != receipt.get("effect_verdict") + or executor.get("kind") != receipt.get("executor_kind") + or readback.get("verification_tier") != receipt.get("verification_tier") + ): + return "evidence_mismatch" + if salting.get("scheme") != SALT_SCHEME or not _DIGEST.match( + str(salting.get("salt_commitment", "")) + ): + return "unsalted" + salted = [ + executor.get("run_record_sha256"), + authorization.get("reference_sha256"), + ] + if readback.get("baseline_sha256") is not None: + salted.append(readback.get("baseline_sha256")) + if not all(isinstance(v, str) and _SALTED.match(v) for v in salted): + return "unsalted" + base_sha = readback.get("baseline_sha256") + base_at = readback.get("baseline_read_at") + if (base_sha is None) != (base_at is None): + return "baseline_window" + if base_at is not None: + try: + elapsed = ( + _parse_time(str(readback.get("read_at"))) - _parse_time(str(base_at)) + ).total_seconds() + except ValueError: + return "baseline_window" + window = readback.get("max_execution_window_s") + if not isinstance(window, int) or not 0 <= elapsed <= window: + return "baseline_window" + if authorization.get("self_approved") is True and ( + (receipt.get("contracts") or {}).get("authorization_passed") is not False + ): + return "self_approved" + read_at = readback.get("read_at") + if isinstance(read_at, str) and str(receipt.get("issued_at", "")) < read_at: + return "issued_before_read" + return None + + +def check_effect_receipt( + document: Mapping[str, Any], + registry: Sequence[Mapping[str, Any]], + *, + accept_outcomes: Sequence[str] = (), +) -> dict[str, Any]: + """RFC-0002 offline verification against a pinned key registry. + + Reports the signature (``verified``) and the outcome + (``verified_outcome``) separately; ``ok`` is true only when both hold, or + when the outcome is one the caller named in ``accept_outcomes``. A + missing signature is ``unsigned`` and is never a pass. + """ + + from openadapt_flow.contracts.seal import parse_key_registry + + receipt = document.get("receipt") if "receipt" in document else document + result: dict[str, Any] = {"verified": False, "verified_outcome": False, "ok": False} + if ( + not isinstance(receipt, Mapping) + or receipt.get("schema_version") != EFFECT_RECEIPT_SCHEMA + ): + return {**result, "reason": "no_effect_receipt"} + outcome = receipt.get("outcome") + result.update( + { + "schema_version": receipt.get("schema_version"), + "executor_kind": receipt.get("executor_kind"), + "effect_verdict": receipt.get("effect_verdict"), + "verification_tier": receipt.get("verification_tier"), + "outcome": outcome, + "verified_outcome": outcome in ("verified", "rolled_back_verified"), + } + ) + if receipt_problem(receipt) is not None: + return {**result, "verified_outcome": False, "reason": "malformed"} + evidence = document.get("evidence") + result["evidence_bound"] = isinstance(evidence, Mapping) + signature = receipt.get("seal_signature") + if signature is None: + return {**result, "reason": "unsigned"} + if ( + not isinstance(signature, Mapping) + or set(signature) != _SIGNATURE_FIELDS + or signature.get("schema") != EFFECT_SEAL_SIGNATURE_SCHEMA + or signature.get("issuer") != EFFECT_SEAL_ISSUER + or signature.get("algorithm") != "ed25519" + or signature.get("receipt_schema") != receipt.get("schema_version") + or not _ISO_SECONDS.match(str(signature.get("issued_at", ""))) + ): + return {**result, "reason": "malformed"} + result["issuer"] = signature.get("issuer") + if signature.get("receipt_sha256") != receipt_sha256(receipt) or signature.get( + "run_id" + ) != receipt.get("execution_id"): + return {**result, "reason": "receipt_mismatch"} + try: + keys = parse_key_registry(list(registry)) + except ValueError: + return {**result, "reason": "bad_registry"} + key = next((k for k in keys if k.key_id == signature.get("key_id")), None) + if key is None: + return {**result, "reason": "unknown_key"} + if key.status == "revoked" and ( + key.revoked_at is None or str(signature["issued_at"]) >= key.revoked_at + ): + return {**result, "reason": "revoked_key"} + try: + from cryptography.exceptions import InvalidSignature + from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PublicKey + except ImportError: # pragma: no cover - cryptography is a core dependency + return {**result, "reason": "verifier_unavailable"} + try: + Ed25519PublicKey.from_public_bytes(base64.b64decode(key.public_key)).verify( + base64.b64decode(str(signature["signature"]), validate=True), + effect_signing_payload(signature), + ) + except (InvalidSignature, ValueError): + return {**result, "reason": "bad_signature"} + result["key_id"] = key.key_id + if isinstance(evidence, Mapping): + problem = evidence_problem(evidence, receipt) + if problem is not None: + return {**result, "reason": problem} + result["verified"] = True + accepted = result["verified_outcome"] or ( + outcome in set(accept_outcomes) and outcome not in (None, "") + ) + result["ok"] = bool(accepted) + result["reason"] = ( + "verified" if result["verified_outcome"] else "outcome_not_verified" + ) + return result diff --git a/packages/flow/tests/test_verify_run.py b/packages/flow/tests/test_verify_run.py new file mode 100644 index 000000000..82ed1ab92 --- /dev/null +++ b/packages/flow/tests/test_verify_run.py @@ -0,0 +1,921 @@ +"""verify-run: an external actor writes to MockMed; the store decides. + +The "executor" here is test code posting straight to MockMed's encounter API, +the way a computer-use agent or a script would. Flow replay never runs. The +verdict comes from ``RestRecordVerifier`` reading ``GET /api/db``. No model +calls, no network beyond localhost. +""" + +from __future__ import annotations + +import base64 +import json +import subprocess +import sys +from pathlib import Path + +import pytest +import requests +import yaml +from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PrivateKey +from cryptography.hazmat.primitives.serialization import Encoding, PublicFormat + +import openadapt_flow.runtime.effects.adapter as adapter_mod +import openadapt_flow.verify_run as verify_run_mod +from openadapt_flow.contracts.execute import ExecuteEvidenceReceiptV1 +from openadapt_flow.contracts.seal import ( + parse_key_registry, + seal_key_id, + signing_payload, + verify_sealed_receipt, +) +from openadapt_flow.mockmed.fault_server import serve as fault_serve +from openadapt_flow.runtime.effects import ( + ConnectionProbe, + EffectState, + EffectVerdict, + Verdict, + register_verifier_factory, +) +from openadapt_flow.verification import VerificationTier +from openadapt_flow.verify_run import ( + EFFECT_RECEIPT_SCHEMA, + VerifyConfig, + VerifyRunError, + baseline_digest, + capture_baseline, + check_effect_receipt, + effect_signing_payload, + load_verify_config, + receipt_sha256, + salted_digest, + verify_run, +) + +PARAMS = {"patient_id": "p1", "note": "Follow-up in two weeks"} +AGENT_RECORD = { + "executor_kind": "agent", + "executor_id": "cua-test-agent", + # What the agent says. None of it may count as evidence. + "claims": [{"wrote": "encounter", "status": "saved"}], + "self_reported_verdict": "CONFIRMED", +} + +# The fixed TEST seed openadapt-cloud's sealSignature tests use. Not a real key. +TEST_SEED = bytes([7]) * 32 + + +def _verify_yaml(base_url: str, **extra: object) -> dict: + doc = { + "schema_version": "openadapt.verify/v1", + "verifier": {"kind": "rest", "base_url": base_url, "records_path": "/api/db"}, + "effects": [ + { + "kind": "record_written", + "match": { + "patient_id": {"param": "patient_id"}, + "type": "Triage", + "note": {"param": "note"}, + }, + "expected_count": 1, + "count_new_only": True, + # The verifier polls for a late write; keep absent cases short. + "timeout_s": 0.5, + } + ], + } + doc.update(extra) + return doc + + +@pytest.fixture +def sor(): + url, db, stop = fault_serve() + try: + yield url.rstrip("/"), db + finally: + stop() + + +@pytest.fixture +def cfg(sor): + base, _db = sor + return VerifyConfig.model_validate(_verify_yaml(base)) + + +def _actor_writes( + base: str, *, patient: str = "p1", note: str = PARAMS["note"], fault: str = "" +) -> None: + url = f"{base}/api/encounter" + (f"?fault={fault}" if fault else "") + requests.post( + url, json={"patient_id": patient, "type": "Triage", "note": note}, timeout=5 + ) + + +def _baseline(cfg: VerifyConfig, tmp_path: Path) -> Path: + path = tmp_path / "baseline.json" + capture_baseline(cfg, PARAMS, path) + return path + + +def _run(cfg, tmp_path, *, baseline=True, record=AGENT_RECORD, pre=None, **kw): + path = _baseline(cfg, tmp_path) if baseline else None + if pre is not None: + pre() + return verify_run( + cfg, + params=PARAMS, + run_record=record, + authorization_ref="ticket-0000-test", + baseline_path=path, + private_dir=tmp_path / "private", + **kw, + ) + + +def _registry() -> list[dict]: + raw = ( + Ed25519PrivateKey.from_private_bytes(TEST_SEED) + .public_key() + .public_bytes(Encoding.Raw, PublicFormat.Raw) + ) + return [ + { + "key_id": seal_key_id(raw), + "public_key": base64.b64encode(raw).decode(), + "status": "active", + } + ] + + +def _issuer_sign(receipt: dict) -> dict: + """Stand in for the issuer: openadapt.effect-seal-signature/v1 (RFC-0002).""" + + key = Ed25519PrivateKey.from_private_bytes(TEST_SEED) + unsigned = { + "schema": "openadapt.effect-seal-signature/v1", + "issuer": "openadapt-cloud:unattested-verifier", + "algorithm": "ed25519", + "key_id": _registry()[0]["key_id"], + "receipt_schema": "openadapt.effect-receipt/v1", + "receipt_sha256": receipt_sha256(receipt), + "run_id": receipt["execution_id"], + "issued_at": receipt["issued_at"], + } + signature = base64.b64encode(key.sign(effect_signing_payload(unsigned))).decode() + return {**receipt, "seal_signature": {**unsigned, "signature": signature}} + + +def _old_domain_sign(receipt: dict) -> dict: + """The pre-RFC Execute Seal signature, which an effect receipt must refuse.""" + + key = Ed25519PrivateKey.from_private_bytes(TEST_SEED) + unsigned = { + "schema": "openadapt.seal-signature/v1", + "issuer": "openadapt-cloud", + "algorithm": "ed25519", + "key_id": _registry()[0]["key_id"], + "receipt_sha256": receipt_sha256(receipt), + "run_id": receipt["execution_id"], + "issued_at": receipt["issued_at"], + } + signature = base64.b64encode(key.sign(signing_payload(unsigned))).decode() + return {**receipt, "seal_signature": {**unsigned, "signature": signature}} + + +# -- the three outcomes ------------------------------------------------------ + + +def test_correct_external_write_is_confirmed_and_verified(cfg, sor, tmp_path): + base, _db = sor + result = _run(cfg, tmp_path, pre=lambda: _actor_writes(base)) + receipt = result["receipt"] + assert receipt["effect_verdict"] == "CONFIRMED" + assert receipt["outcome"] == "verified" + assert receipt["oracle_tier"] == 2 + assert receipt["executor_kind"] == "agent" + assert ( + receipt["contracts"]["observed_effect_strength"] + == "independent_system_of_record" + ) + # The executor's model use is unknown, so the receipt cannot claim none. + assert receipt["contracts"]["model_used"] is True + assert receipt["verification_tier"] == "independent_readback" + assert result["evidence"]["verifier"]["model_calls"] == 0 + + +def test_incorrect_external_write_is_refuted(cfg, sor, tmp_path): + base, _db = sor + # The note the agent typed is not the note the task named. + result = _run(cfg, tmp_path, pre=lambda: _actor_writes(base, note="Wrong note")) + assert result["receipt"]["effect_verdict"] == "REFUTED" + assert result["receipt"]["outcome"] != "verified" + + +def test_wrong_patient_is_refuted(cfg, sor, tmp_path): + base, _db = sor + result = _run(cfg, tmp_path, pre=lambda: _actor_writes(base, fault="wrong_record")) + assert result["receipt"]["effect_verdict"] == "REFUTED" + assert result["receipt"]["outcome"] == "halted_before_effect" + + +def test_duplicate_write_needs_reconciliation(cfg, sor, tmp_path): + base, _db = sor + + def twice() -> None: + _actor_writes(base) + _actor_writes(base) + + result = _run(cfg, tmp_path, pre=twice) + receipt = result["receipt"] + assert receipt["effect_verdict"] == "REFUTED" + assert receipt["outcome"] == "reconciliation_required" + assert receipt["delivery_uncertain"] is True + assert ( + result["evidence"]["readback"]["effects"][0]["observed_effect"] == "conflicting" + ) + + +def test_no_record_with_baseline_is_refuted_absent(cfg, sor, tmp_path): + base, _db = sor + # The agent got a 2xx and reported success; nothing persisted. + result = _run(cfg, tmp_path, pre=lambda: _actor_writes(base, fault="no_persist")) + assert result["receipt"]["effect_verdict"] == "REFUTED" + assert result["receipt"]["outcome"] == "halted_before_effect" + assert result["evidence"]["readback"]["effects"][0]["observed_effect"] == "absent" + + +def test_no_baseline_means_indeterminate_never_confirmed(cfg, sor, tmp_path): + base, _db = sor + _actor_writes(base) # a correct write, but nobody read the store before it + result = _run(cfg, tmp_path, baseline=False) + receipt = result["receipt"] + assert receipt["effect_verdict"] == "INDETERMINATE" + assert receipt["outcome"] == "reconciliation_required" + assert result["evidence"]["readback"]["baseline_sha256"] is None + + +def test_unreachable_store_is_indeterminate(tmp_path): + cfg = VerifyConfig.model_validate(_verify_yaml("http://127.0.0.1:1")) + result = verify_run( + cfg, + params=PARAMS, + run_record=AGENT_RECORD, + authorization_ref="ticket-0000-test", + private_dir=tmp_path, + ) + assert result["receipt"]["effect_verdict"] == "INDETERMINATE" + assert result["receipt"]["outcome"] == "reconciliation_required" + assert result["receipt"]["verification_tier"] == "unchecked" + + +def test_screen_readback_never_becomes_verified(clean_registry, tmp_path): + register_verifier_factory( + "fake-screen", + lambda cfg, params: _FakeVerifier( + VerificationTier.PERSISTED_STATE_REACQUISITION + ), + ) + result = verify_run( + _fake_cfg("fake-screen"), + params=PARAMS, + run_record=AGENT_RECORD, + authorization_ref="ticket-0000-test", + private_dir=tmp_path, + ) + receipt = result["receipt"] + assert receipt["effect_verdict"] == "CONFIRMED" + assert receipt["verification_tier"] == "screen_readback" + assert receipt["outcome"] == "reconciliation_required" + + +def test_verified_without_independent_readback_fails_the_check(cfg, sor, tmp_path): + base, _db = sor + result = _run(cfg, tmp_path, pre=lambda: _actor_writes(base)) + forged = _issuer_sign({**result["receipt"], "verification_tier": "screen_readback"}) + verdict = check_effect_receipt({"receipt": forged}, _registry()) + assert verdict["verified"] is False + assert verdict["reason"] == "malformed" + + +# -- the executor's claims never count --------------------------------------- + + +def test_executor_claims_do_not_change_the_verdict(cfg, sor, tmp_path): + base, _db = sor + loud = {**AGENT_RECORD, "claims": [{"verified": True}] * 5, "outcome": "verified"} + result = _run( + cfg, tmp_path, record=loud, pre=lambda: _actor_writes(base, fault="no_persist") + ) + assert result["receipt"]["effect_verdict"] == "REFUTED" + evidence = result["evidence"] + assert evidence["executor"]["claims_counted_as_evidence"] is False + # The run record is bound by digest only; none of its text is copied. + blob = json.dumps(result) + assert "cua-test-agent" not in blob + assert "saved" not in blob + + +def test_params_come_from_the_task_not_the_run_record(cfg, sor, tmp_path): + base, _db = sor + # The agent wrote for p2 and its record says p2 was the task. The task + # said p1, so the store holds nothing that satisfies the contract. + record = {**AGENT_RECORD, "params": {"patient_id": "p2", "note": PARAMS["note"]}} + result = _run( + cfg, tmp_path, record=record, pre=lambda: _actor_writes(base, patient="p2") + ) + assert result["receipt"]["effect_verdict"] == "REFUTED" + + +def test_run_record_must_name_executor_kind(cfg): + with pytest.raises(VerifyRunError, match="executor_kind"): + verify_run( + cfg, + params=PARAMS, + run_record={}, + authorization_ref="ticket-0000-test", + private_dir=Path("/nonexistent"), + ) + + +# -- no VERIFIED without an independent read --------------------------------- + + +def test_screen_verifier_is_refused(tmp_path, sor): + base, _db = sor + doc = _verify_yaml(base) + doc["verifier"] = {"kind": "onscreen"} + path = tmp_path / "verify.yaml" + path.write_text(yaml.safe_dump(doc)) + with pytest.raises(VerifyRunError, match="independently"): + load_verify_config(path) + + +class _FakeVerifier: + """Confirms everything. Only its declared tier differs between tests.""" + + substrate = "fake" + + def __init__(self, tier): + if tier is not None: + self.verification_tier = tier + + def test_connection(self): + return ConnectionProbe(ok=True, substrate="fake") + + def capture_pre_state(self, context=None): + return EffectState(substrate="fake", reachable=True) + + capture_post_state = capture_pre_state + + def verify(self, expected, before, context=None): + return EffectVerdict( + verdict=Verdict.CONFIRMED, + kind=expected.kind, + substrate="fake", + observed_count=1, + expected_count=1, + ) + + +@pytest.fixture +def clean_registry(monkeypatch): + monkeypatch.setattr(adapter_mod, "_REGISTRY", {}) + monkeypatch.setattr(adapter_mod, "_ENTRY_POINTS_LOADED", True) + yield + + +def _fake_cfg(kind: str, **extra) -> VerifyConfig: + doc = _verify_yaml("http://unused") + doc["verifier"] = {"kind": kind} + doc["effects"][0].pop("count_new_only") + doc.update(extra) + return VerifyConfig.model_validate(doc) + + +def test_separate_session_read_is_screen_readback_and_never_verified( + clean_registry, tmp_path +): + register_verifier_factory( + "fake-session", + lambda cfg, params: _FakeVerifier(VerificationTier.INDEPENDENT_SESSION), + ) + cfg = _fake_cfg("fake-session", minimum_effect_strength="independent_session") + result = verify_run( + cfg, + params=PARAMS, + run_record=AGENT_RECORD, + authorization_ref="ticket-0000-test", + private_dir=tmp_path, + ) + receipt = result["receipt"] + assert receipt["effect_verdict"] == "CONFIRMED" + # RFC-0002 decision 6: a separate UI session (Flow tier 2) is a screen + # read, so it can never be independent_readback. + assert receipt["verification_tier"] == "screen_readback" + assert receipt["outcome"] == "reconciliation_required" + assert receipt["oracle_tier"] == 1 + assert result["evidence"]["readback"]["effects"][0]["flow_verification_tier"] == 2 + + +def test_untyped_verifier_is_refused(clean_registry, tmp_path): + register_verifier_factory("fake-untyped", lambda cfg, params: _FakeVerifier(None)) + with pytest.raises(VerifyRunError, match="verification_tier"): + verify_run( + _fake_cfg("fake-untyped"), + params=PARAMS, + run_record=AGENT_RECORD, + authorization_ref="ticket-0000-test", + private_dir=tmp_path, + ) + + +# -- the receipt: backward compatible, offline-verifiable --------------------- + + +def test_receipt_keeps_every_execute_v1_field(cfg, sor, tmp_path): + base, _db = sor + receipt = _run(cfg, tmp_path, pre=lambda: _actor_writes(base))["receipt"] + v1_fields = set(ExecuteEvidenceReceiptV1.model_fields) - {"seal_signature"} + assert v1_fields <= set(receipt) + assert set(receipt) - v1_fields == { + "executor_kind", + "executor_ref_sha256", + "effect_verdict", + "verification_tier", + } + assert receipt["schema_version"] == EFFECT_RECEIPT_SCHEMA + # A strict Execute consumer refuses it rather than mistaking it for a + # Flow-governed Execute receipt. + with pytest.raises(Exception): + ExecuteEvidenceReceiptV1.model_validate(receipt) + + +def test_signed_receipt_verifies_offline_under_the_effect_seal_domain( + cfg, sor, tmp_path +): + base, _db = sor + result = _run(cfg, tmp_path, pre=lambda: _actor_writes(base)) + signed = _issuer_sign(result["receipt"]) + registry = _registry() + # A pre-RFC Execute Seal verifier refuses it instead of passing it. + old = verify_sealed_receipt(signed, parse_key_registry(registry)) + assert not old.ok and old.reason == "malformed" + document = {**result, "receipt": signed} + verdict = check_effect_receipt(document, registry) + assert verdict["verified"] is True + assert verdict["verified_outcome"] is True + assert verdict["ok"] is True + assert verdict["issuer"] == "openadapt-cloud:unattested-verifier" + assert verdict["evidence_bound"] is True + assert verdict["executor_kind"] == "agent" + assert verdict["effect_verdict"] == "CONFIRMED" + + +def test_tampering_is_caught(cfg, sor, tmp_path): + base, _db = sor + result = _run(cfg, tmp_path, pre=lambda: _actor_writes(base)) + signed = _issuer_sign(result["receipt"]) + registry = _registry() + relabeled = {**result, "receipt": {**signed, "executor_kind": "flow_replay"}} + assert check_effect_receipt(relabeled, registry)["reason"] in { + "receipt_mismatch", + "evidence_mismatch", + } + evidence = json.loads(json.dumps(result["evidence"])) + evidence["readback"]["effects"][0]["verdict"] = "REFUTED" + assert ( + check_effect_receipt( + {**result, "receipt": signed, "evidence": evidence}, registry + )["reason"] + == "evidence_mismatch" + ) + assert check_effect_receipt(result, registry)["reason"] == "unsigned" + assert check_effect_receipt(result, [])["verified"] is False + + +def test_execute_seal_signature_on_an_effect_receipt_is_refused(cfg, sor, tmp_path): + base, _db = sor + result = _run(cfg, tmp_path, pre=lambda: _actor_writes(base)) + old = _old_domain_sign(result["receipt"]) + verdict = check_effect_receipt({**result, "receipt": old}, _registry()) + assert verdict["verified"] is False + assert verdict["reason"] == "malformed" + # Same fields under the new schema but signed over the old domain. + forged = _issuer_sign(result["receipt"]) + sig = dict(forged["seal_signature"]) + key = Ed25519PrivateKey.from_private_bytes(TEST_SEED) + sig["signature"] = base64.b64encode(key.sign(signing_payload(sig))).decode() + verdict = check_effect_receipt( + {**result, "receipt": {**forged, "seal_signature": sig}}, _registry() + ) + assert verdict["reason"] == "bad_signature" + + +def test_receipt_carries_no_record_values(cfg, sor, tmp_path): + base, _db = sor + result = _run(cfg, tmp_path, pre=lambda: _actor_writes(base)) + blob = json.dumps(result) + for value in (PARAMS["note"], "Triage", "patient_id"): + assert value not in blob + + +# -- CLI -------------------------------------------------------------------- + + +def test_cli_end_to_end(sor, tmp_path): + base, _db = sor + verify = tmp_path / "verify.yaml" + verify.write_text(yaml.safe_dump(_verify_yaml(base))) + params = tmp_path / "params.json" + params.write_text(json.dumps(PARAMS)) + record = tmp_path / "run.json" + record.write_text(json.dumps(AGENT_RECORD)) + baseline = tmp_path / "before.json" + cli = [ + sys.executable, + "-m", + "openadapt_flow", + "verify-run", + "--verify", + str(verify), + "--params", + str(params), + ] + + done = subprocess.run( + [*cli, "--capture-baseline", str(baseline)], capture_output=True, text=True + ) + assert done.returncode == 0, done.stderr + assert (baseline.stat().st_mode & 0o777) == 0o600 + _actor_writes(base) + out = tmp_path / "out" + done = subprocess.run( + [ + *cli, + "--run-record", + str(record), + "--authorization-ref", + "ticket-0000-test", + "--baseline", + str(baseline), + "--out", + str(out), + ], + capture_output=True, + text=True, + ) + assert done.returncode == 0, done.stderr + assert "effect verdict: CONFIRMED" in done.stdout + document = json.loads((out / "effect-receipt.json").read_text()) + signed = tmp_path / "signed.json" + signed.write_text( + json.dumps({**document, "receipt": _issuer_sign(document["receipt"])}) + ) + keys = tmp_path / "keys.json" + keys.write_text(json.dumps({"keys": _registry()})) + done = subprocess.run( + [ + sys.executable, + "-m", + "openadapt_flow", + "verify-run", + "--check", + str(signed), + "--keys", + str(keys), + ], + capture_output=True, + text=True, + ) + assert done.returncode == 0, done.stdout + done.stderr + assert json.loads(done.stdout)["verified"] is True + + +# -- RFC-0002: run salt, baseline binding, tiers, linking --------------------- + + +def _hmac(salt: bytes, label: str, value) -> str: + import hashlib + import hmac as hmac_mod + + message = ( + label.encode() + + b"\0" + + json.dumps( + value, sort_keys=True, separators=(",", ":"), ensure_ascii=False + ).encode("utf-8") + ) + return "hmac-sha256:" + hmac_mod.new(salt, message, hashlib.sha256).hexdigest() + + +def test_customer_digests_are_salted_with_the_run_salt(cfg, sor, tmp_path): + import hashlib + + base, _db = sor + path = _baseline(cfg, tmp_path) + baseline = json.loads(path.read_text()) + salt = bytes.fromhex(baseline["salt"]) + _actor_writes(base) + result = verify_run( + cfg, + params=PARAMS, + run_record=AGENT_RECORD, + authorization_ref="ticket-0000-test", + baseline_path=path, + ) + evidence = result["evidence"] + assert evidence["salting"] == { + "scheme": "openadapt.run-salt/v1", + "salt_commitment": "sha256:" + hashlib.sha256(salt).hexdigest(), + } + assert evidence["executor"]["run_record_sha256"] == _hmac( + salt, "run_record", AGENT_RECORD + ) + assert evidence["authorization"]["reference_sha256"] == _hmac( + salt, "authorization_reference", "ticket-0000-test" + ) + # The baseline digest covers the five named fields, never the salt. + five = { + k: baseline[k] + for k in ( + "schema_version", + "execution_id", + "baseline_read_at", + "verifier_identity", + "state", + ) + } + assert evidence["readback"]["baseline_sha256"] == _hmac(salt, "baseline", five) + assert evidence["readback"]["baseline_sha256"] == baseline_digest(baseline, salt) + # The receipt continues the run the baseline started. + assert result["receipt"]["execution_id"] == baseline["execution_id"] + # The salt itself never leaves the private file. + assert baseline["salt"] not in json.dumps(result) + # The effect hash binds the parameters only through salted digests. + effect = cfg.effects[0] + unsalted = effect.resolve(PARAMS).contract_hash() + row = evidence["readback"]["effects"][0] + assert row["effect_contract_hash"] != unsalted + salted = { + name: salted_digest(salt, f"effect_param:{name}", PARAMS[name]).split(":", 1)[1] + for name in effect.referenced_params() + } + assert row["effect_contract_hash"] == effect.resolved_contract_hash( + {}, opaque_param_sha256=salted + ) + + +def test_two_runs_differ_in_salt_and_execution_but_not_in_contract(cfg, sor, tmp_path): + seen: list = [] + original = verify_run_mod.build_effect_verifier + + def spy(config, params): + verifier = original(config, params) + seen.append(verifier) + return verifier + + base, _db = sor + _actor_writes(base) + with pytest.MonkeyPatch.context() as mp: + mp.setattr(verify_run_mod, "build_effect_verifier", spy) + one = verify_run( + cfg, + params=PARAMS, + run_record=AGENT_RECORD, + authorization_ref="ticket-0000-test", + private_dir=tmp_path / "a", + ) + two = verify_run( + cfg, + params=PARAMS, + run_record=AGENT_RECORD, + authorization_ref="ticket-0000-test", + private_dir=tmp_path / "b", + ) + r1, r2 = one["receipt"], two["receipt"] + assert r1["execution_id"] != r2["execution_id"] + assert r1["workflow_digest"] == r2["workflow_digest"] + assert ( + one["evidence"]["readback"]["verifier_identity"] + == two["evidence"]["readback"]["verifier_identity"] + ) + assert ( + one["evidence"]["salting"]["salt_commitment"] + != two["evidence"]["salting"]["salt_commitment"] + ) + assert ( + one["evidence"]["executor"]["run_record_sha256"] + != two["evidence"]["executor"]["run_record_sha256"] + ) + # Every REST read carries the run's own execution id and trace header. + for verifier, receipt in zip(seen, (r1, r2)): + assert verifier.headers["X-OpenAdapt-Execution-Id"] == receipt["execution_id"] + assert verifier.headers["traceparent"].startswith("00-") + # The run salt is kept privately, mode 0600. + private = tmp_path / "a" / "effect-run.private.json" + assert (private.stat().st_mode & 0o777) == 0o600 + assert json.loads(private.read_text())["execution_id"] == r1["execution_id"] + + +def test_a_baseline_with_an_extra_key_is_refused(cfg, sor, tmp_path): + path = _baseline(cfg, tmp_path) + doc = json.loads(path.read_text()) + doc["note"] = "unbound" + path.write_text(json.dumps(doc)) + with pytest.raises(VerifyRunError, match="exactly"): + verify_run( + cfg, + params=PARAMS, + run_record=AGENT_RECORD, + authorization_ref="ticket-0000-test", + baseline_path=path, + ) + + +def test_a_baseline_is_used_once(cfg, sor, tmp_path): + base, _db = sor + path = _baseline(cfg, tmp_path) + _actor_writes(base) + kw = dict( + params=PARAMS, + run_record=AGENT_RECORD, + authorization_ref="ticket-0000-test", + baseline_path=path, + ) + assert verify_run(cfg, **kw)["receipt"]["outcome"] == "verified" + with pytest.raises(VerifyRunError, match="already used"): + verify_run(cfg, **kw) + + +def test_a_baseline_outside_the_window_counts_as_absent(cfg, sor, tmp_path): + base, _db = sor + path = _baseline(cfg, tmp_path) + doc = json.loads(path.read_text()) + doc["baseline_read_at"] = "2020-01-01T00:00:00Z" + path.write_text(json.dumps(doc)) + _actor_writes(base) + result = verify_run( + cfg, + params=PARAMS, + run_record=AGENT_RECORD, + authorization_ref="ticket-0000-test", + baseline_path=path, + ) + assert result["receipt"]["effect_verdict"] == "INDETERMINATE" + assert result["receipt"]["outcome"] == "reconciliation_required" + readback = result["evidence"]["readback"] + assert readback["baseline_sha256"] is None + assert readback["baseline_read_at"] is None + + +def test_offline_check_refuses_a_broken_baseline_window(cfg, sor, tmp_path): + base, _db = sor + result = _run(cfg, tmp_path, pre=lambda: _actor_writes(base)) + evidence = json.loads(json.dumps(result["evidence"])) + evidence["readback"]["baseline_read_at"] = "2020-01-01T00:00:00Z" + receipt = dict(result["receipt"]) + receipt["evidence_digest"] = verify_run_mod.sha256_tagged(evidence) + signed = _issuer_sign(receipt) + verdict = check_effect_receipt( + {"receipt": signed, "evidence": evidence}, _registry() + ) + assert verdict["reason"] == "baseline_window" + assert verdict["verified"] is False + + +def test_offline_check_refuses_unsalted_evidence(cfg, sor, tmp_path): + base, _db = sor + result = _run(cfg, tmp_path, pre=lambda: _actor_writes(base)) + evidence = json.loads(json.dumps(result["evidence"])) + evidence["executor"]["run_record_sha256"] = "sha256:" + "0" * 64 + receipt = dict(result["receipt"]) + receipt["evidence_digest"] = verify_run_mod.sha256_tagged(evidence) + verdict = check_effect_receipt( + {"receipt": _issuer_sign(receipt), "evidence": evidence}, _registry() + ) + assert verdict["reason"] == "unsalted" + + +def test_remediates_and_correlation_id_are_signed_into_the_receipt(cfg, sor, tmp_path): + base, _db = sor + earlier = "receipt-" + "a" * 32 + result = _run( + cfg, + tmp_path, + pre=lambda: _actor_writes(base), + remediates=earlier, + correlation_id="ticket-INC-4242", + ) + receipt = result["receipt"] + assert receipt["remediates"] == earlier + assert receipt["correlation_id"] == "ticket-INC-4242" + signed = _issuer_sign(receipt) + assert check_effect_receipt({**result, "receipt": signed}, _registry())["ok"] + # Both are inside receipt_sha256, so editing either breaks the seal. + for field, value in ( + ("remediates", "receipt-" + "b" * 32), + ("correlation_id", "ticket-INC-9999"), + ): + edited = {**signed, field: value} + verdict = check_effect_receipt({**result, "receipt": edited}, _registry()) + assert verdict["reason"] == "receipt_mismatch" + # Absent unless given, so a plain receipt keeps the RFC field set. + plain = _run(cfg, tmp_path / "plain", pre=lambda: _actor_writes(base)) + assert "remediates" not in plain["receipt"] + assert "correlation_id" not in plain["receipt"] + + +def test_link_ids_must_be_opaque(cfg, tmp_path): + with pytest.raises(VerifyRunError, match="correlation_id"): + verify_run( + cfg, + params=PARAMS, + run_record=AGENT_RECORD, + authorization_ref="ticket-0000-test", + private_dir=tmp_path, + correlation_id="jane.doe@example.com x", + ) + + +def test_a_self_approved_write_is_rejected_policy(cfg, sor, tmp_path): + base, _db = sor + approver = { + "kind": "agent", + "id": AGENT_RECORD["executor_id"], + "policy_ref": "sha256:" + "1" * 64, + } + result = _run(cfg, tmp_path, pre=lambda: _actor_writes(base), approver=approver) + receipt = result["receipt"] + assert receipt["effect_verdict"] == "CONFIRMED" + assert receipt["outcome"] == "rejected_policy" + assert receipt["contracts"]["authorization_passed"] is False + assert receipt["contracts"]["effect_passed"] is True + assert result["evidence"]["authorization"]["self_approved"] is True + other = {**approver, "kind": "person", "id": "staff-00000042"} + ok = _run(cfg, tmp_path / "other", pre=lambda: _actor_writes(base), approver=other) + assert ok["evidence"]["authorization"]["self_approved"] is False + + +def test_a_genuine_signature_over_a_refutation_is_not_a_pass(cfg, sor, tmp_path): + base, _db = sor + result = _run(cfg, tmp_path, pre=lambda: _actor_writes(base, fault="no_persist")) + assert result["receipt"]["outcome"] == "halted_before_effect" + document = {**result, "receipt": _issuer_sign(result["receipt"])} + verdict = check_effect_receipt(document, _registry()) + assert verdict["verified"] is True + assert verdict["verified_outcome"] is False + assert verdict["ok"] is False + accepted = check_effect_receipt( + document, _registry(), accept_outcomes=["halted_before_effect"] + ) + assert accepted["ok"] is True + + +# -- one verify.yaml for qualify and verify-run -------------------------------- + + +def test_the_qualify_example_loads_and_hashes_like_qualify(monkeypatch): + from openadapt_flow.qualify.campaign import record_effects + from openadapt_flow.qualify.config import ( + load_verify_config as load_qualify_config, + ) + from openadapt_flow.qualify.result import verify_config_sha256 + + path = Path(__file__).resolve().parents[1] / "benchmark/openemr_local/verify.yaml" + for name in ("OPENEMR_UI_URL", "OPENEMR_API_URL"): + monkeypatch.delenv(name, raising=False) + ours = load_verify_config(path) + theirs = load_qualify_config(path) + # Qualify-only sections and verifier.session are ignored, not refused. + assert ours.verifier == theirs.verifier.verifier + assert ours.environment_id == theirs.environment_id + assert [e.model_dump(mode="json") for e in ours.effects] == [ + e.model_dump(mode="json") for e in record_effects(theirs) + ] + identity = "rest:example" + assert verify_config_sha256(identity, list(ours.effects)) == verify_config_sha256( + identity, record_effects(theirs) + ) + + +def test_record_and_effects_together_are_refused(): + from openadapt_flow.verify_run import parse_verify_config + + data = _verify_yaml("http://127.0.0.1:1") + data["record"] = { + "name": "note", + "match": ["patient_id"], + "fields": {"patient_id": "patient_id"}, + } + with pytest.raises(VerifyRunError, match="either effects or record"): + parse_verify_config(data) + + +def test_unset_environment_reference_is_refused(monkeypatch): + from openadapt_flow.verify_run import parse_verify_config + + monkeypatch.delenv("OA_TEST_UNSET_SOR_URL", raising=False) + data = _verify_yaml("${OA_TEST_UNSET_SOR_URL}") + with pytest.raises(VerifyRunError, match="OA_TEST_UNSET_SOR_URL"): + parse_verify_config(data)