From 48ee75f1dea97e093bbcf769f9ad4fcd74920817 Mon Sep 17 00:00:00 2001 From: Richard Abrich Date: Sat, 17 Oct 2026 00:23:45 -0400 Subject: [PATCH 1/4] feat(verify-run): judge an external executor's write from the system of record (prototype) openadapt-flow verify-run reads the store through an existing EffectVerifier (built from the deployment.yaml effects: shape) and rules on an effect_contract.yaml list. The executor's run record is bound by digest and never read as evidence; contract params come from --params, not from the executor. VERIFIED needs every effect CONFIRMED by a system-of-record read; screen read-backs are refused and separate-session reads stop at reconciliation_required. The output keeps every ExecuteEvidenceReceiptV1 field, adds executor_kind, executor_ref_sha256, and effect_verdict under schema openadapt.effect-receipt/v1, and binds an evidence manifest through evidence_digest. It is unsigned; a seal-signature/v1 over it verifies with openadapt_types.seal and openadapt-agent seal verify unchanged. Tests: MockMed fault server as the system of record, with test code as the external actor (correct, wrong note, wrong patient, duplicate, phantom write, no baseline, unreachable store). Ported from openadapt-flow feat/verify-any-executor (8f363086). Signed-off-by: Richard Abrich --- packages/flow/docs/VERIFY_RUN.md | 85 ++++ packages/flow/openadapt_flow/__main__.py | 123 +++++ packages/flow/openadapt_flow/verify_run.py | 488 ++++++++++++++++++++ packages/flow/tests/test_verify_run.py | 501 +++++++++++++++++++++ 4 files changed, 1197 insertions(+) create mode 100644 packages/flow/docs/VERIFY_RUN.md create mode 100644 packages/flow/openadapt_flow/verify_run.py create mode 100644 packages/flow/tests/test_verify_run.py diff --git a/packages/flow/docs/VERIFY_RUN.md b/packages/flow/docs/VERIFY_RUN.md new file mode 100644 index 000000000..8434a4c0b --- /dev/null +++ b/packages/flow/docs/VERIFY_RUN.md @@ -0,0 +1,85 @@ +# Check a write that something else made + +Prototype. The command and the receipt format may change. + +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 +``` + +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 SHA-256 digest of the file and reads nothing else from it. +- `--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`. + +## Verdicts + +| What the store shows | `effect_verdict` | `outcome` | +|---|---|---| +| The declared record, exactly once, through a system-of-record read | `CONFIRMED` | `verified` | +| 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. + +## 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`, and `effect_verdict`, under the +schema `openadapt.effect-receipt/v1`. `evidence.readback` lists each effect's +contract hash, verdict, and counts; `receipt.evidence_digest` is its +SHA-256. Neither part holds record values, URLs, or the agent's own text. + +The file is unsigned. An issuer adds `seal_signature` the same way OpenAdapt +Cloud signs Execute receipts. Check a signed file offline against keys you +pinned: + +```bash +openadapt-flow verify-run --check signed.json --keys seal-keys.json +``` + +`openadapt seal verify` also checks the signature, but it does not yet show +`executor_kind`. Read the receipt before you rely on it. + +The baseline file holds raw records. It is written with mode 0600 and never +goes into the receipt. Delete it when you no longer need it. diff --git a/packages/flow/openadapt_flow/__main__.py b/packages/flow/openadapt_flow/__main__.py index a654d5fc6..ebdf6cffb 100644 --- a/packages/flow/openadapt_flow/__main__.py +++ b/packages/flow/openadapt_flow/__main__.py @@ -2329,6 +2329,69 @@ def _cmd_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) + print(json.dumps(result, indent=2, sort_keys=True)) + return 0 if result.get("verified") 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, + ) + 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. @@ -7166,6 +7229,66 @@ def build_parser() -> argparse.ArgumentParser: ) sa.set_defaults(func=_cmd_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 effect " + "receipt. 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.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..1c31285ba --- /dev/null +++ b/packages/flow/openadapt_flow/verify_run.py @@ -0,0 +1,488 @@ +"""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. Its ``receipt`` +keeps every field of ``openadapt.execute-evidence-receipt/v1`` with the same +name, pattern, and meaning, adds ``executor_kind``, ``executor_ref_sha256``, +and ``effect_verdict``, and carries a new ``schema_version`` so a strict +Execute consumer refuses it instead of mistaking it for a governed Flow run. +``evidence_digest`` binds the evidence manifest written beside it. + +The receipt is unsigned. An issuer adds ``seal_signature`` +(``openadapt.seal-signature/v1``) over the same canonical hash Cloud uses for +Execute receipts, so the existing offline check verifies the signature +unchanged. Nothing here signs, uploads, or calls a model. +""" + +from __future__ import annotations + +import hashlib +import json +import os +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.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" +VERIFY_CONFIG_SCHEMA = "openadapt.verify/v1" +BASELINE_SCHEMA = "openadapt.effect-baseline/v1" + +ExecutorKind = Literal["flow_replay", "agent", "script", "human", "unknown"] +EXECUTOR_KINDS: tuple[str, ...] = ("flow_replay", "agent", "script", "human", "unknown") + +EffectVerdictLabel = Literal["CONFIRMED", "REFUTED", "INDETERMINATE"] + +#: Verifier kinds that never read independently of the acting surface. +_NOT_INDEPENDENT = {"none", "onscreen"} + +_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, +} + + +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" + #: 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") + data = yaml.safe_load(source.read_text(encoding="utf-8")) or {} + if not isinstance(data, dict): + raise VerifyRunError(f"verify config {source} must be a YAML mapping") + try: + cfg = VerifyConfig.model_validate(data) + except Exception as exc: + raise VerifyRunError(f"invalid verify config {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 + + +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 _now() -> str: + return ( + datetime.now(timezone.utc) + .replace(microsecond=0) + .isoformat() + .replace("+00:00", "Z") + ) + + +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") + tmp.write_text(json.dumps(payload, indent=2, sort_keys=True) + "\n", "utf-8") + os.chmod(tmp, 0o600) + os.replace(tmp, path) + + +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 capture_baseline( + cfg: VerifyConfig, params: Mapping[str, object], out: Path +) -> EffectState: + """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, so it stays private on this machine: mode 0600, never inside a + receipt. Only its digest is bound. + """ + + verifier = _build_verifier(cfg, params) + state = verifier.capture_pre_state() + if not isinstance(state, EffectState): + raise VerifyRunError("verifier returned no readable baseline") + _write_private( + out, + { + "schema_version": BASELINE_SCHEMA, + "verifier_identity": verifier_identity(verifier), + "state": state.model_dump(mode="json"), + }, + ) + return state + + +def _load_baseline( + path: Optional[Path], verifier: Any +) -> tuple[EffectState, Optional[str]]: + if path is None: + # No baseline: the judge treats the pre-state as unknown, so every + # claim about change is INDETERMINATE, never guessed. + return EffectState( + substrate=str(getattr(verifier, "substrate", "")), reachable=False + ), None + raw = json.loads(Path(path).read_text(encoding="utf-8")) + if not isinstance(raw, dict) or raw.get("schema_version") != BASELINE_SCHEMA: + raise VerifyRunError(f"{path} is not an {BASELINE_SCHEMA} file") + if raw.get("verifier_identity") != verifier_identity(verifier): + raise VerifyRunError("baseline was captured by a different verifier") + return EffectState.model_validate(raw["state"]), sha256_tagged(raw) + + +def _executor(run_record: Mapping[str, Any]) -> tuple[str, 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" + ) + return str(kind), "sha256:" + hashlib.sha256(ref.encode("utf-8")).hexdigest() + + +def _label(verdict: Verdict) -> EffectVerdictLabel: + return { + Verdict.CONFIRMED: "CONFIRMED", + Verdict.REFUTED: "REFUTED", + Verdict.INDETERMINATE: "INDETERMINATE", + }[verdict] # type: ignore[return-value] + + +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 verify_run( + cfg: VerifyConfig, + *, + params: Mapping[str, object], + run_record: Mapping[str, Any], + authorization_ref: str, + baseline_path: Optional[Path] = None, +) -> dict[str, Any]: + """Read the system of record and return ``{"receipt", "evidence"}``. + + The run record contributes ``executor_kind``, a digest of + ``executor_id``, and a 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. + """ + + from openadapt_types.execute import ( + EffectStrengthV1, + ExecuteEvidenceContractV1, + ExecuteEvidenceReceiptV1, + ExecuteTerminalOutcomeV1, + ) + + if not isinstance(run_record, Mapping): + raise VerifyRunError("run record must be a JSON object") + executor_kind, executor_ref = _executor(run_record) + run_record_digest = sha256_tagged(dict(run_record)) + verifier = _build_verifier(cfg, params) + before, baseline_digest = _load_baseline(baseline_path, verifier) + + contract_digest = sha256_tagged( + { + "verifier_identity": verifier_identity(verifier), + "effects": [e.model_dump(mode="json") for e in cfg.effects], + } + ) + verdicts: list[EffectVerdict] = [] + rows: list[dict[str, Any]] = [] + weakest: Optional[VerificationTier] = None + for effect in cfg.effects: + try: + bound = effect.resolve(params) + except 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) + # Closed fields only: no reason text, no observed value, no records. + rows.append( + { + "effect_contract_hash": bound.contract_hash(), + "kind": bound.kind.value, + "substrate": verdict.substrate, + "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) + strength = _STRENGTH_BY_TIER.get(weakest) if weakest is not None else None + strong_enough = strength is not None and ( + _STRENGTH_RANK[strength] >= _STRENGTH_RANK[cfg.minimum_effect_strength] + ) + sor_read = weakest is VerificationTier.INDEPENDENT_SYSTEM + + delivery_uncertain = False + if effect_verdict == "CONFIRMED" and strong_enough and sor_read: + 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 + + evidence = { + "schema_version": EFFECT_EVIDENCE_SCHEMA, + "executor": { + "kind": executor_kind, + "ref_sha256": executor_ref, + "run_record_sha256": run_record_digest, + "claims_counted_as_evidence": False, + }, + "authorization": { + "reference_sha256": "sha256:" + + hashlib.sha256(authorization_ref.encode("utf-8")).hexdigest(), + "evaluated_by_verifier": False, + }, + "contract": { + "contract_sha256": contract_digest, + "minimum_effect_strength": cfg.minimum_effect_strength, + }, + "readback": { + "verifier_identity": verifier_identity(verifier), + "baseline_sha256": baseline_digest, + "read_at": _now(), + "effects": rows, + }, + "verifier": { + "name": "openadapt-flow verify-run", + "model_calls": 0, + }, + "effect_verdict": effect_verdict, + } + contracts = ExecuteEvidenceContractV1( + # The caller bound an authorization reference; 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=verified, + postcondition_passed=verified, + effect_passed=verified, + 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] + v1 = ExecuteEvidenceReceiptV1( + receipt_id=f"receipt-{uuid4().hex}", + execution_id=f"verify-{uuid4().hex}", + 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_from_verification_tier(weakest) if weakest is not None else 0, + ), + outcome=outcome, + contracts=contracts, + delivery_uncertain=delivery_uncertain, + evidence_digest=sha256_tagged(evidence), + issued_at=_now(), + ) + 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 + 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 + + +def check_effect_receipt( + document: Mapping[str, Any], registry: Sequence[Mapping[str, Any]] +) -> dict[str, Any]: + """Offline check: pinned-key Seal signature, then the evidence binding. + + The signature check is ``openadapt_types.seal.verify_sealed_receipt``, + the same code path ``openadapt seal verify`` mirrors. A missing signature + is reported as ``unsigned`` and is never a pass. + """ + + from openadapt_types.seal import parse_key_registry, verify_sealed_receipt + + receipt = document.get("receipt") if "receipt" in document else document + if ( + not isinstance(receipt, Mapping) + or receipt.get("schema_version") != EFFECT_RECEIPT_SCHEMA + ): + return {"verified": False, "reason": "no_effect_receipt"} + result: dict[str, Any] = { + "verified": False, + "executor_kind": receipt.get("executor_kind"), + "effect_verdict": receipt.get("effect_verdict"), + "outcome": receipt.get("outcome"), + } + evidence = document.get("evidence") + if isinstance(evidence, Mapping): + if sha256_tagged(dict(evidence)) != receipt.get("evidence_digest"): + return {**result, "reason": "evidence_mismatch"} + if evidence.get("effect_verdict") != receipt.get("effect_verdict"): + return {**result, "reason": "evidence_mismatch"} + if (evidence.get("executor") or {}).get("kind") != receipt.get("executor_kind"): + return {**result, "reason": "evidence_mismatch"} + result["evidence_bound"] = True + else: + result["evidence_bound"] = False + seal = verify_sealed_receipt(receipt, parse_key_registry(list(registry))) + if not seal.ok: + return {**result, "reason": seal.reason} + signature = receipt.get("seal_signature") or {} + if signature.get("run_id") != receipt.get("execution_id"): + return {**result, "reason": "receipt_mismatch"} + return {**result, "verified": True, "reason": "verified", "key_id": seal.key_id} diff --git a/packages/flow/tests/test_verify_run.py b/packages/flow/tests/test_verify_run.py new file mode 100644 index 000000000..d1bbff27b --- /dev/null +++ b/packages/flow/tests/test_verify_run.py @@ -0,0 +1,501 @@ +"""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 +from openadapt_types.execute import ExecuteEvidenceReceiptV1 +from openadapt_types.seal import ( + parse_key_registry, + receipt_sha256, + seal_key_id, + signing_payload, + verify_sealed_receipt, +) + +import openadapt_flow.runtime.effects.adapter as adapter_mod +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, + capture_baseline, + check_effect_receipt, + load_verify_config, + 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): + 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, + ) + + +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: the same steps Cloud's signSeal takes.""" + + 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 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", + ) + assert result["receipt"]["effect_verdict"] == "INDETERMINATE" + assert result["receipt"]["outcome"] == "reconciliation_required" + + +# -- 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" + ) + + +# -- 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_never_becomes_verified(clean_registry): + 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", + ) + receipt = result["receipt"] + assert receipt["effect_verdict"] == "CONFIRMED" + assert receipt["outcome"] == "reconciliation_required" + assert receipt["oracle_tier"] == 1 + + +def test_untyped_verifier_is_refused(clean_registry): + 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", + ) + + +# -- 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", + } + 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_with_the_existing_seal_code( + cfg, sor, tmp_path +): + base, _db = sor + result = _run(cfg, tmp_path, pre=lambda: _actor_writes(base)) + signed = _issuer_sign(result["receipt"]) + registry = _registry() + # The existing Seal check in openadapt-types, unchanged. + assert verify_sealed_receipt(signed, parse_key_registry(registry)).ok + document = {**result, "receipt": signed} + verdict = check_effect_receipt(document, registry) + assert verdict["verified"] is True + 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_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 From d43f3f127b74c459bff1f088fd2020825df87763 Mon Sep 17 00:00:00 2001 From: Richard Abrich Date: Sat, 17 Oct 2026 01:03:30 -0400 Subject: [PATCH 2/4] feat(verify-run): record verification_tier and an approver placeholder verification_tier (independent_readback | screen_readback | unchecked) goes on the receipt and the evidence manifest; only independent_readback can yield verified, and check refuses a verified receipt without it. authorization.approver is null until the openadapt-types approver identity exists (RFC-0002). Ported from openadapt-flow feat/verify-any-executor (0139e1b8). Signed-off-by: Richard Abrich --- packages/flow/docs/VERIFY_RUN.md | 12 +++++- packages/flow/openadapt_flow/verify_run.py | 43 +++++++++++++++++++++- packages/flow/tests/test_verify_run.py | 30 +++++++++++++++ 3 files changed, 81 insertions(+), 4 deletions(-) diff --git a/packages/flow/docs/VERIFY_RUN.md b/packages/flow/docs/VERIFY_RUN.md index 8434a4c0b..af906d733 100644 --- a/packages/flow/docs/VERIFY_RUN.md +++ b/packages/flow/docs/VERIFY_RUN.md @@ -61,13 +61,21 @@ The three inputs come from different people on purpose: 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: +`independent_readback` (system of record or a separate session), +`screen_readback` (a plugin verifier that re-reads the acting session), or +`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`, and `effect_verdict`, under the +plus `executor_kind`, `executor_ref_sha256`, `effect_verdict`, and +`verification_tier`, under the schema `openadapt.effect-receipt/v1`. `evidence.readback` lists each effect's -contract hash, verdict, and counts; `receipt.evidence_digest` is its +contract hash, verdict, and counts. `evidence.authorization.approver` is +`null` until OpenAdapt types defines approver identity. `receipt.evidence_digest` is its SHA-256. Neither part holds record values, URLs, or the agent's own text. The file is unsigned. An issuer adds `seal_signature` the same way OpenAdapt diff --git a/packages/flow/openadapt_flow/verify_run.py b/packages/flow/openadapt_flow/verify_run.py index 1c31285ba..18ae4d129 100644 --- a/packages/flow/openadapt_flow/verify_run.py +++ b/packages/flow/openadapt_flow/verify_run.py @@ -19,7 +19,7 @@ The output is an ``openadapt.effect-receipt/v1`` document. Its ``receipt`` keeps every field of ``openadapt.execute-evidence-receipt/v1`` with the same name, pattern, and meaning, adds ``executor_kind``, ``executor_ref_sha256``, -and ``effect_verdict``, and carries a new ``schema_version`` so a strict +``effect_verdict``, and ``verification_tier``, and carries a new ``schema_version`` so a strict Execute consumer refuses it instead of mistaking it for a governed Flow run. ``evidence_digest`` binds the evidence manifest written beside it. @@ -67,6 +67,11 @@ EffectVerdictLabel = Literal["CONFIRMED", "REFUTED", "INDETERMINATE"] +#: How the effect was checked. Only ``independent_readback`` can yield +#: ``verified``, and only when the read also went through the system of +#: record (the v1 oracle-tier rule). +ReadbackClass = Literal["independent_readback", "screen_readback", "unchecked"] + #: Verifier kinds that never read independently of the acting surface. _NOT_INDEPENDENT = {"none", "onscreen"} @@ -265,6 +270,18 @@ def _observed(verdicts: Sequence[EffectVerdict]) -> str: return "absent" if "absent" in seen else "present" +def _readback_class( + verdicts: Sequence[EffectVerdict], tiers: Sequence[Optional[VerificationTier]] +) -> ReadbackClass: + """``unchecked`` unless every effect was actually read.""" + + if any(t is None for t in tiers) or any(v.unavailable for v in verdicts): + return "unchecked" + if all(t is not None and t.is_independent_of_actuation() for t in tiers): + return "independent_readback" + return "screen_readback" + + def verify_run( cfg: VerifyConfig, *, @@ -304,6 +321,7 @@ def verify_run( 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) @@ -325,6 +343,7 @@ def verify_run( 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( { @@ -348,9 +367,15 @@ def verify_run( _STRENGTH_RANK[strength] >= _STRENGTH_RANK[cfg.minimum_effect_strength] ) sor_read = weakest is VerificationTier.INDEPENDENT_SYSTEM + verification_tier = _readback_class(verdicts, tiers) delivery_uncertain = False - if effect_verdict == "CONFIRMED" and strong_enough and sor_read: + if ( + effect_verdict == "CONFIRMED" + and verification_tier == "independent_readback" + and strong_enough + and sor_read + ): outcome = ExecuteTerminalOutcomeV1.VERIFIED elif effect_verdict == "REFUTED" and observed_effect == "absent": # The store shows nothing landed, whatever the executor said. @@ -373,6 +398,9 @@ def verify_run( "reference_sha256": "sha256:" + hashlib.sha256(authorization_ref.encode("utf-8")).hexdigest(), "evaluated_by_verifier": False, + # approver{kind, id, policy_ref}: placeholder until the + # openadapt-types approver identity model exists (RFC-0002). + "approver": None, }, "contract": { "contract_sha256": contract_digest, @@ -380,6 +408,7 @@ def verify_run( }, "readback": { "verifier_identity": verifier_identity(verifier), + "verification_tier": verification_tier, "baseline_sha256": baseline_digest, "read_at": _now(), "effects": rows, @@ -431,6 +460,7 @@ def verify_run( receipt["executor_kind"] = executor_kind receipt["executor_ref_sha256"] = executor_ref receipt["effect_verdict"] = effect_verdict + receipt["verification_tier"] = verification_tier return { "schema_version": EFFECT_RECEIPT_SCHEMA, "receipt": receipt, @@ -462,6 +492,11 @@ def check_effect_receipt( or receipt.get("schema_version") != EFFECT_RECEIPT_SCHEMA ): return {"verified": False, "reason": "no_effect_receipt"} + if receipt.get("outcome") in {"verified", "rolled_back_verified"} and ( + receipt.get("verification_tier") != "independent_readback" + or receipt.get("effect_verdict") != "CONFIRMED" + ): + return {"verified": False, "reason": "malformed"} result: dict[str, Any] = { "verified": False, "executor_kind": receipt.get("executor_kind"), @@ -476,6 +511,10 @@ def check_effect_receipt( return {**result, "reason": "evidence_mismatch"} if (evidence.get("executor") or {}).get("kind") != receipt.get("executor_kind"): return {**result, "reason": "evidence_mismatch"} + if (evidence.get("readback") or {}).get("verification_tier") != receipt.get( + "verification_tier" + ): + return {**result, "reason": "evidence_mismatch"} result["evidence_bound"] = True else: result["evidence_bound"] = False diff --git a/packages/flow/tests/test_verify_run.py b/packages/flow/tests/test_verify_run.py index d1bbff27b..ddccc7791 100644 --- a/packages/flow/tests/test_verify_run.py +++ b/packages/flow/tests/test_verify_run.py @@ -176,6 +176,7 @@ def test_correct_external_write_is_confirmed_and_verified(cfg, sor, tmp_path): ) # 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 @@ -240,6 +241,34 @@ def test_unreachable_store_is_indeterminate(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): + 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", + ) + 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": False, "reason": "malformed"} # -- the executor's claims never count --------------------------------------- @@ -374,6 +403,7 @@ def test_receipt_keeps_every_execute_v1_field(cfg, sor, tmp_path): "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 From 3d16cb92f30daf659d6e35713127b03b442cbdd4 Mon Sep 17 00:00:00 2001 From: abrichr Date: Sat, 17 Oct 2026 16:43:45 -0400 Subject: [PATCH 3/4] feat(verify-run): align the effect receipt with RFC-0002 - Signatures are openadapt.effect-seal-signature/v1 over their own domain (issuer openadapt-cloud:unattested-verifier); an Execute Seal verifier refuses an effect receipt, and an Execute signature on one is refused here. - Per-run 256-bit salt (openadapt.run-salt/v1): the run record, authorization reference, baseline and every bound parameter are HMAC-SHA256 digests; evidence carries only the salt commitment. The salt stays in the 0600 baseline or effect-run.private.json. - Baselines start the run (execution_id), hold exactly six keys, bind five, work once, and count as absent outside max_execution_window_s. REST reads carry X-OpenAdapt-Execution-Id and traceparent. - verification_tier follows the RFC tier rule: independent_readback only for system-of-record reads (Flow tier 1); a separate UI session is screen_readback. - A self-approved write is rejected_policy. - New optional --remediates and --correlation-id, signed into the receipt body when given. - --check reports the signature and the outcome separately; exit 0 needs both, or an outcome named with --accept-outcome. Signed-off-by: Richard Abrich --- packages/flow/docs/VERIFY_RUN.md | 81 ++- packages/flow/openadapt_flow/__main__.py | 53 +- packages/flow/openadapt_flow/verify_run.py | 734 +++++++++++++++++---- packages/flow/tests/test_verify_run.py | 371 ++++++++++- 4 files changed, 1076 insertions(+), 163 deletions(-) diff --git a/packages/flow/docs/VERIFY_RUN.md b/packages/flow/docs/VERIFY_RUN.md index af906d733..c06c95d0e 100644 --- a/packages/flow/docs/VERIFY_RUN.md +++ b/packages/flow/docs/VERIFY_RUN.md @@ -1,6 +1,7 @@ # Check a write that something else made -Prototype. The command and the receipt format may change. +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 @@ -44,16 +45,32 @@ The three inputs come from different people on purpose: 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 SHA-256 digest of the file and reads nothing else from it. + 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`. + `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` | @@ -61,33 +78,55 @@ The three inputs come from different people on purpose: 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: -`independent_readback` (system of record or a separate session), -`screen_readback` (a plugin verifier that re-reads the acting session), or -`unchecked` (at least one effect was not read). Only `independent_readback` -can yield `verified`. +`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`, under the -schema `openadapt.effect-receipt/v1`. `evidence.readback` lists each effect's -contract hash, verdict, and counts. `evidence.authorization.approver` is -`null` until OpenAdapt types defines approver identity. `receipt.evidence_digest` is its -SHA-256. Neither part holds record values, URLs, or the agent's own text. - -The file is unsigned. An issuer adds `seal_signature` the same way OpenAdapt -Cloud signs Execute receipts. Check a signed file offline against keys you -pinned: +`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 ``` -`openadapt seal verify` also checks the signature, but it does not yet show -`executor_kind`. Read the receipt before you rely on it. +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 file holds raw records. It is written with mode 0600 and never -goes into the receipt. Delete it when you no longer need it. +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 ebdf6cffb..a1cf362ef 100644 --- a/packages/flow/openadapt_flow/__main__.py +++ b/packages/flow/openadapt_flow/__main__.py @@ -2353,9 +2353,16 @@ def _json(path: str) -> Any: registry = _json(args.keys) if isinstance(registry, dict): registry = registry.get("keys", []) - result = check_effect_receipt(_json(args.check), registry) + result = check_effect_receipt( + _json(args.check), registry, accept_outcomes=args.accept_outcome or () + ) print(json.dumps(result, indent=2, sort_keys=True)) - return 0 if result.get("verified") else 1 + 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 @@ -2380,6 +2387,10 @@ def _json(path: str) -> Any: 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) @@ -7233,8 +7244,9 @@ def build_parser() -> argparse.ArgumentParser: "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 effect " - "receipt. The executor's own record is hashed, never trusted" + "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( @@ -7287,6 +7299,39 @@ def build_parser() -> argparse.ArgumentParser: 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( diff --git a/packages/flow/openadapt_flow/verify_run.py b/packages/flow/openadapt_flow/verify_run.py index 18ae4d129..ef4698a8f 100644 --- a/packages/flow/openadapt_flow/verify_run.py +++ b/packages/flow/openadapt_flow/verify_run.py @@ -16,24 +16,38 @@ 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. Its ``receipt`` -keeps every field of ``openadapt.execute-evidence-receipt/v1`` with the same -name, pattern, and meaning, adds ``executor_kind``, ``executor_ref_sha256``, -``effect_verdict``, and ``verification_tier``, and carries a new ``schema_version`` so a strict -Execute consumer refuses it instead of mistaking it for a governed Flow run. -``evidence_digest`` binds the evidence manifest written beside it. - -The receipt is unsigned. An issuer adds ``seal_signature`` -(``openadapt.seal-signature/v1``) over the same canonical hash Cloud uses for -Execute receipts, so the existing offline check verifies the signature -unchanged. Nothing here signs, uploads, or calls a model. +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 @@ -59,22 +73,44 @@ 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 effect was checked. Only ``independent_readback`` can yield -#: ``verified``, and only when the read also went through the system of -#: record (the v1 oracle-tier rule). +#: 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", @@ -88,6 +124,28 @@ "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.""" @@ -102,14 +160,19 @@ class VerifyConfig(BaseModel): #: 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}$" + 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}$" + 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:``. @@ -149,6 +212,9 @@ def load_verify_config(path: str | Path) -> VerifyConfig: return cfg +# -- hashing ------------------------------------------------------------------ + + def _canonical(value: Any) -> bytes: """Sorted-key compact JSON; the Seal's canonical form.""" @@ -161,6 +227,35 @@ 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) @@ -170,14 +265,27 @@ def _now() -> str: ) +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") - tmp.write_text(json.dumps(payload, indent=2, sort_keys=True) + "\n", "utf-8") + 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) @@ -188,50 +296,118 @@ def _build_verifier(cfg: VerifyConfig, params: Mapping[str, object]) -> Any: 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 -) -> EffectState: +) -> 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, so it stays private on this machine: mode 0600, never inside a - receipt. Only its digest is bound. + 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") - _write_private( - out, - { - "schema_version": BASELINE_SCHEMA, - "verifier_identity": verifier_identity(verifier), - "state": state.model_dump(mode="json"), - }, + 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} ) - return state -def _load_baseline( - path: Optional[Path], verifier: Any -) -> tuple[EffectState, Optional[str]]: - if path is None: - # No baseline: the judge treats the pre-state as unknown, so every - # claim about change is INDETERMINATE, never guessed. - return EffectState( - substrate=str(getattr(verifier, "substrate", "")), reachable=False - ), None - raw = json.loads(Path(path).read_text(encoding="utf-8")) +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 raw.get("verifier_identity") != verifier_identity(verifier): + 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") - return EffectState.model_validate(raw["state"]), sha256_tagged(raw) + 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.""" -def _executor(run_record: Mapping[str, Any]) -> tuple[str, str]: + 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( @@ -241,15 +417,63 @@ def _executor(run_record: Mapping[str, Any]) -> tuple[str, str]: ref = ( executor_id if isinstance(executor_id, str) and executor_id else "unidentified" ) - return str(kind), "sha256:" + hashlib.sha256(ref.encode("utf-8")).hexdigest() + 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: - return { + labels: dict[Verdict, EffectVerdictLabel] = { Verdict.CONFIRMED: "CONFIRMED", Verdict.REFUTED: "REFUTED", Verdict.INDETERMINATE: "INDETERMINATE", - }[verdict] # type: ignore[return-value] + } + return labels[verdict] def _aggregate(verdicts: Sequence[EffectVerdict]) -> EffectVerdictLabel: @@ -270,14 +494,24 @@ def _observed(verdicts: Sequence[EffectVerdict]) -> str: return "absent" if "absent" in seen else "present" -def _readback_class( +def readback_class( verdicts: Sequence[EffectVerdict], tiers: Sequence[Optional[VerificationTier]] ) -> ReadbackClass: - """``unchecked`` unless every effect was actually read.""" + """RFC-0002 tier rule, weakest effect wins. - if any(t is None for t in tiers) or any(v.unavailable for v in verdicts): + ``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.is_independent_of_actuation() for t in tiers): + if all(t is not None and t.reads_system_of_record() for t in tiers): return "independent_readback" return "screen_readback" @@ -289,16 +523,26 @@ def verify_run( 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``, and a 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. + ``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_types.execute import ( + from openadapt_flow.contracts.execute import ( EffectStrengthV1, ExecuteEvidenceContractV1, ExecuteEvidenceReceiptV1, @@ -307,14 +551,64 @@ def verify_run( if not isinstance(run_record, Mapping): raise VerifyRunError("run record must be a JSON object") - executor_kind, executor_ref = _executor(run_record) - run_record_digest = sha256_tagged(dict(run_record)) + 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) - before, baseline_digest = _load_baseline(baseline_path, verifier) + 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 + ) contract_digest = sha256_tagged( { - "verifier_identity": verifier_identity(verifier), + "verifier_identity": identity, "effects": [e.model_dump(mode="json") for e in cfg.effects], } ) @@ -325,7 +619,16 @@ def verify_run( for effect in cfg.effects: try: bound = effect.resolve(params) - except ValueError as exc: + # 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: @@ -347,10 +650,10 @@ def verify_run( # Closed fields only: no reason text, no observed value, no records. rows.append( { - "effect_contract_hash": bound.contract_hash(), + "effect_contract_hash": contract_hash, "kind": bound.kind.value, "substrate": verdict.substrate, - "verification_tier": int(tier) if tier is not None else None, + "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, @@ -362,20 +665,38 @@ def verify_run( effect_verdict = _aggregate(verdicts) observed_effect = _observed(verdicts) - strength = _STRENGTH_BY_TIER.get(weakest) if weakest is not None else None + 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] ) - sor_read = weakest is VerificationTier.INDEPENDENT_SYSTEM - verification_tier = _readback_class(verdicts, tiers) + 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 - if ( + read_proves_write = ( effect_verdict == "CONFIRMED" and verification_tier == "independent_readback" and strong_enough - and sor_read - ): + 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. @@ -385,32 +706,42 @@ def verify_run( 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, - "run_record_sha256": run_record_digest, + "build_sha256": build_digest, + "run_record_sha256": salted_digest(salt, "run_record", dict(run_record)), "claims_counted_as_evidence": False, }, "authorization": { - "reference_sha256": "sha256:" - + hashlib.sha256(authorization_ref.encode("utf-8")).hexdigest(), + "reference_sha256": salted_digest( + salt, "authorization_reference", authorization_ref + ), "evaluated_by_verifier": False, - # approver{kind, id, policy_ref}: placeholder until the - # openadapt-types approver identity model exists (RFC-0002). - "approver": None, + # 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": verifier_identity(verifier), + "verifier_identity": identity, "verification_tier": verification_tier, - "baseline_sha256": baseline_digest, - "read_at": _now(), + "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": { @@ -420,13 +751,13 @@ def verify_run( "effect_verdict": effect_verdict, } contracts = ExecuteEvidenceContractV1( - # The caller bound an authorization reference; verify-run did not - # evaluate a policy. evidence.authorization says so. + # 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=verified, - postcondition_passed=verified, - effect_passed=verified, + 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 @@ -436,24 +767,22 @@ def verify_run( 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=f"verify-{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_from_verification_tier(weakest) if weakest is not None else 0, - ), + 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=_now(), + issued_at=issued_at, ) receipt = v1.model_dump(mode="json", exclude={"seal_signature"}) receipt["schema_version"] = EFFECT_RECEIPT_SCHEMA @@ -461,6 +790,13 @@ def verify_run( 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, @@ -474,54 +810,206 @@ def write_result(result: Mapping[str, Any], out_dir: Path) -> Path: 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]] + document: Mapping[str, Any], + registry: Sequence[Mapping[str, Any]], + *, + accept_outcomes: Sequence[str] = (), ) -> dict[str, Any]: - """Offline check: pinned-key Seal signature, then the evidence binding. + """RFC-0002 offline verification against a pinned key registry. - The signature check is ``openadapt_types.seal.verify_sealed_receipt``, - the same code path ``openadapt seal verify`` mirrors. A missing signature - is reported as ``unsigned`` and is never a pass. + 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_types.seal import parse_key_registry, verify_sealed_receipt + 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 {"verified": False, "reason": "no_effect_receipt"} - if receipt.get("outcome") in {"verified", "rolled_back_verified"} and ( - receipt.get("verification_tier") != "independent_readback" - or receipt.get("effect_verdict") != "CONFIRMED" - ): - return {"verified": False, "reason": "malformed"} - result: dict[str, Any] = { - "verified": False, - "executor_kind": receipt.get("executor_kind"), - "effect_verdict": receipt.get("effect_verdict"), - "outcome": receipt.get("outcome"), - } + 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") - if isinstance(evidence, Mapping): - if sha256_tagged(dict(evidence)) != receipt.get("evidence_digest"): - return {**result, "reason": "evidence_mismatch"} - if evidence.get("effect_verdict") != receipt.get("effect_verdict"): - return {**result, "reason": "evidence_mismatch"} - if (evidence.get("executor") or {}).get("kind") != receipt.get("executor_kind"): - return {**result, "reason": "evidence_mismatch"} - if (evidence.get("readback") or {}).get("verification_tier") != receipt.get( - "verification_tier" - ): - return {**result, "reason": "evidence_mismatch"} - result["evidence_bound"] = True - else: - result["evidence_bound"] = False - seal = verify_sealed_receipt(receipt, parse_key_registry(list(registry))) - if not seal.ok: - return {**result, "reason": seal.reason} - signature = receipt.get("seal_signature") or {} - if signature.get("run_id") != receipt.get("execution_id"): + 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"} - return {**result, "verified": True, "reason": "verified", "key_id": seal.key_id} + 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 index ddccc7791..a3d4323b4 100644 --- a/packages/flow/tests/test_verify_run.py +++ b/packages/flow/tests/test_verify_run.py @@ -19,16 +19,16 @@ import yaml from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PrivateKey from cryptography.hazmat.primitives.serialization import Encoding, PublicFormat -from openadapt_types.execute import ExecuteEvidenceReceiptV1 -from openadapt_types.seal import ( + +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, - receipt_sha256, seal_key_id, signing_payload, verify_sealed_receipt, ) - -import openadapt_flow.runtime.effects.adapter as adapter_mod from openadapt_flow.mockmed.fault_server import serve as fault_serve from openadapt_flow.runtime.effects import ( ConnectionProbe, @@ -42,9 +42,13 @@ EFFECT_RECEIPT_SCHEMA, VerifyConfig, VerifyRunError, + baseline_digest, capture_baseline, check_effect_receipt, + effect_signing_payload, load_verify_config, + receipt_sha256, + salted_digest, verify_run, ) @@ -114,7 +118,7 @@ def _baseline(cfg: VerifyConfig, tmp_path: Path) -> Path: return path -def _run(cfg, tmp_path, *, baseline=True, record=AGENT_RECORD, pre=None): +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() @@ -124,6 +128,8 @@ def _run(cfg, tmp_path, *, baseline=True, record=AGENT_RECORD, pre=None): run_record=record, authorization_ref="ticket-0000-test", baseline_path=path, + private_dir=tmp_path / "private", + **kw, ) @@ -143,7 +149,25 @@ def _registry() -> list[dict]: def _issuer_sign(receipt: dict) -> dict: - """Stand in for the issuer: the same steps Cloud's signSeal takes.""" + """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 = { @@ -238,13 +262,14 @@ def test_unreachable_store_is_indeterminate(tmp_path): 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): +def test_screen_readback_never_becomes_verified(clean_registry, tmp_path): register_verifier_factory( "fake-screen", lambda cfg, params: _FakeVerifier( @@ -256,6 +281,7 @@ def test_screen_readback_never_becomes_verified(clean_registry): params=PARAMS, run_record=AGENT_RECORD, authorization_ref="ticket-0000-test", + private_dir=tmp_path, ) receipt = result["receipt"] assert receipt["effect_verdict"] == "CONFIRMED" @@ -268,7 +294,8 @@ def test_verified_without_independent_readback_fails_the_check(cfg, sor, tmp_pat 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": False, "reason": "malformed"} + assert verdict["verified"] is False + assert verdict["reason"] == "malformed" # -- the executor's claims never count --------------------------------------- @@ -303,7 +330,11 @@ def test_params_come_from_the_task_not_the_run_record(cfg, sor, tmp_path): 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" + cfg, + params=PARAMS, + run_record={}, + authorization_ref="ticket-0000-test", + private_dir=Path("/nonexistent"), ) @@ -362,7 +393,9 @@ def _fake_cfg(kind: str, **extra) -> VerifyConfig: return VerifyConfig.model_validate(doc) -def test_separate_session_read_never_becomes_verified(clean_registry): +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), @@ -373,14 +406,19 @@ def test_separate_session_read_never_becomes_verified(clean_registry): 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): +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( @@ -388,6 +426,7 @@ def test_untyped_verifier_is_refused(clean_registry): params=PARAMS, run_record=AGENT_RECORD, authorization_ref="ticket-0000-test", + private_dir=tmp_path, ) @@ -412,18 +451,22 @@ def test_receipt_keeps_every_execute_v1_field(cfg, sor, tmp_path): ExecuteEvidenceReceiptV1.model_validate(receipt) -def test_signed_receipt_verifies_offline_with_the_existing_seal_code( +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() - # The existing Seal check in openadapt-types, unchanged. - assert verify_sealed_receipt(signed, parse_key_registry(registry)).ok + # 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" @@ -451,6 +494,24 @@ def test_tampering_is_caught(cfg, sor, tmp_path): 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)) @@ -529,3 +590,283 @@ def test_cli_end_to_end(sor, tmp_path): ) 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 From 3767053f2eafe953926d5528085d162122b36c41 Mon Sep 17 00:00:00 2001 From: abrichr Date: Sat, 17 Oct 2026 17:20:15 -0400 Subject: [PATCH 4/4] feat(verify-run): read the same verify.yaml qualify reads RFC-0002 makes openadapt.verify/v1 one file for both commands. verify-run now ignores the qualify-only sections (app, save_step, inputs, campaign, verifier.session), expands record with qualify's own record_effects, expands ${NAME} references with qualify's expander, and computes workflow_digest with qualify's verify_config_sha256, so a receipt and a qualification result over one file name the same contract. --- packages/flow/docs/VERIFY_RUN.md | 9 +++ packages/flow/openadapt_flow/verify_run.py | 74 +++++++++++++++++++--- packages/flow/tests/test_verify_run.py | 49 ++++++++++++++ 3 files changed, 123 insertions(+), 9 deletions(-) diff --git a/packages/flow/docs/VERIFY_RUN.md b/packages/flow/docs/VERIFY_RUN.md index c06c95d0e..ef25e33ce 100644 --- a/packages/flow/docs/VERIFY_RUN.md +++ b/packages/flow/docs/VERIFY_RUN.md @@ -38,6 +38,15 @@ effects: # the same list as effect_contract.yaml 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 diff --git a/packages/flow/openadapt_flow/verify_run.py b/packages/flow/openadapt_flow/verify_run.py index ef4698a8f..69b26dc1a 100644 --- a/packages/flow/openadapt_flow/verify_run.py +++ b/packages/flow/openadapt_flow/verify_run.py @@ -56,6 +56,7 @@ 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, @@ -185,13 +186,72 @@ def load_verify_config(path: str | Path) -> VerifyConfig: source = Path(path) if not source.is_file(): raise VerifyRunError(f"verify config {source} is not a file") - data = yaml.safe_load(source.read_text(encoding="utf-8")) or {} + 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"verify config {source} must be a YAML mapping") + 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 verify config {source}: {exc}") from 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 " @@ -606,12 +666,8 @@ def verify_run( substrate=str(getattr(verifier, "substrate", "")), reachable=False ) - contract_digest = sha256_tagged( - { - "verifier_identity": identity, - "effects": [e.model_dump(mode="json") for e in cfg.effects], - } - ) + # 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 diff --git a/packages/flow/tests/test_verify_run.py b/packages/flow/tests/test_verify_run.py index a3d4323b4..82ed1ab92 100644 --- a/packages/flow/tests/test_verify_run.py +++ b/packages/flow/tests/test_verify_run.py @@ -870,3 +870,52 @@ def test_a_genuine_signature_over_a_refutation_is_not_a_pass(cfg, sor, tmp_path) 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)