diff --git a/packages/flow/benchmark/lifecycle/production_arm.py b/packages/flow/benchmark/lifecycle/production_arm.py index 874522f3f..ead8cb355 100644 --- a/packages/flow/benchmark/lifecycle/production_arm.py +++ b/packages/flow/benchmark/lifecycle/production_arm.py @@ -85,13 +85,15 @@ def __init__(self, ledger: _Ledger) -> None: self.item: Optional[WorkItem] = None self.operator: Any = None self.kind = "approval" + #: The run's fixture approver (benchmark.lifecycle_proof.approver). + self.approver: Any = None def watch(self, store: Any) -> None: self.stores.append(store) def sleep(self, seconds: float) -> None: time.sleep(min(seconds, 0.5)) - from openadapt_flow.approval_mode.item_approval import make_answer + from benchmark.lifecycle_proof import run as proof for store in list(self.stores): for request in store.requests(): @@ -115,15 +117,11 @@ def sleep(self, seconds: float) -> None: ) ) approve = decision.action in ("approve", "accept_repair") + # Signed by the run's fixture approver; an unsigned answer is + # refused, and the refusal is logged before it stops the gate. store.write_answer( - make_answer( - request, - decision="approve" if approve else "decline", - decline_reason=None if approve else "wrong_record", - decided_by="human", - route="desktop_run_it", - approver_id=APPROVER_ID, - auth_method="cloud_aal2", + proof.answer_signed( + self.ledger, self.approver, store, request, approve ) ) self.touches += 1 @@ -209,6 +207,7 @@ def setup(self, env: Env, operator: Any) -> list[dict[str, Any]]: ledger = _Ledger(out / "arm-ledger.jsonl", root=out.parent.parent) ctx = proof._context(out, self.keys_dir, ledger) # type: ignore[arg-type] person = OperatorPerson(ledger) + person.approver = ctx["approver"] person.operator = operator ctx["person"] = person self.ctx = ctx @@ -560,14 +559,13 @@ def _self_repair( configuration's ``repair.agent`` does. It re-runs the item itself.""" from benchmark.lifecycle_proof import run as proof - from openadapt_flow.approval_mode.item_approval import ItemApprovalStore from openadapt_flow.repair.self_repair import ScreenAccess, SelfRepair person: OperatorPerson = self.ctx["person"] run_dir = self._run_dir(item, "repair") handle = proof.Item(self.ctx, item.item_id, run_dir, params=item.params) person.kind = "repair_review" - person.watch(ItemApprovalStore(run_dir)) + person.watch(self.ctx["approver"].store(run_dir)) try: repair = SelfRepair( proof.LabelAgent(), diff --git a/packages/flow/benchmark/lifecycle_proof/approver.py b/packages/flow/benchmark/lifecycle_proof/approver.py new file mode 100644 index 000000000..ef580baab --- /dev/null +++ b/packages/flow/benchmark/lifecycle_proof/approver.py @@ -0,0 +1,165 @@ +"""A fixture-only approver for the scripted person in the proof and the soak. + +A person's approval counts only when a key on the signed approver roster +signs it, and the run pins that roster's root +(:mod:`openadapt_flow.approval_mode.approver_roster`). The scripted person +plays a front-desk approver, so each run gets its own: + +- a throwaway Ed25519 key, generated under the run's own directory and never + a real person's key; +- a one-entry roster that key roots; +- a deployment config that pins the root (``approvers.root_public_key``) and + the key directory, and whose bytes are pinned by SHA-256, as + ``--config-sha256`` pins them for the CLI. + +The approval stores take their pin only from that config +(``config_only``), never from ``OPENADAPT_APPROVER_ROOT``. The key, its +passphrase and the roster stay in the run directory, which is never +committed. The ledger records the public root and the config digest. +""" + +from __future__ import annotations + +import hashlib +import os +import secrets +from dataclasses import dataclass +from pathlib import Path +from typing import Any, Optional + +#: The scripted approver's name on the fixture roster (a person's role name). +PRINCIPAL = "soak-front-desk" +KEY_DIRNAME = "approver-keys" +CONFIG_FILENAME = "deployment.yaml" +_PASSPHRASE_FILENAME = ".passphrase" +ROUTE = "desktop_run_it" + + +@dataclass(frozen=True) +class FixtureApprover: + """The run's own approver key, roster and pinned deployment config.""" + + key_dir: Path + root_public_key: str + config_path: Path + config_sha256: str + signer: Any + + @classmethod + def for_run( + cls, run_root: Path, *, policy: Optional[str] = None + ) -> "FixtureApprover": + """Open the run's approver, creating it on the run's first phase.""" + + from openadapt_flow.approval_mode.approver_roster import ( + file_key_signer, + init_root, + ) + from openadapt_flow.approval_mode.human_attestation import load_signer + from openadapt_flow.jit.authority import ( + KEY_PASSPHRASE_ENV, + generate_encrypted_key, + load_encrypted_key, + public_hex, + ) + + key_dir = run_root / KEY_DIRNAME + key_path = key_dir / f"{PRINCIPAL}.approver.pem" + secret = key_dir / _PASSPHRASE_FILENAME + config_path = run_root / CONFIG_FILENAME + if not key_path.exists(): + key_dir.mkdir(parents=True, exist_ok=True, mode=0o700) + fd = os.open(secret, os.O_WRONLY | os.O_CREAT | os.O_EXCL, 0o600) + with os.fdopen(fd, "w") as handle: + handle.write(secrets.token_hex(24)) + env = {KEY_PASSPHRASE_ENV: secret.read_text().strip()} + if not key_path.exists(): + public = generate_encrypted_key(key_path, env=env) + init_root( + key_dir, + principal=PRINCIPAL, + public_key=public, + sign=file_key_signer(load_encrypted_key(key_path, env=env)), + ) + lines = ["name: lifecycle-soak-fixture"] + if policy: + lines += ["policy:", f" policy: {policy}"] + lines += [ + "approvers:", + f" root_public_key: {public}", + f" key_dir: {key_dir.resolve()}", + ] + config_path.write_text("\n".join(lines) + "\n") + public = public_hex(load_encrypted_key(key_path, env=env)) + digest = hashlib.sha256(config_path.read_bytes()).hexdigest() + return cls( + key_dir=key_dir.resolve(), + root_public_key=public, + config_path=config_path, + config_sha256=digest, + signer=load_signer( + key_path, key_dir, method="desktop_key", env=env, root=public + ), + ) + + def deployment(self) -> Any: + """The run's deployment config, read only if its bytes match the pin.""" + + from openadapt_flow.deployment import load_deployment, pin_deployment_bytes + + pin_deployment_bytes(self.config_path, self.config_sha256) + config = load_deployment(self.config_path) + approvers = config.approvers + if approvers is None or approvers.root_public_key != self.root_public_key: + raise ValueError( + "the run's deployment config doesn't pin its approver root" + ) + return config + + def store(self, run_dir: Path) -> Any: + """An item-approval store pinned only by the run's deployment config.""" + + from openadapt_flow.approval_mode.item_approval import ItemApprovalStore + + approvers = self.deployment().approvers + return ItemApprovalStore( + run_dir, + key_dir=Path(approvers.key_dir) if approvers.key_dir else self.key_dir, + config_root=approvers.root_public_key, + config_only=True, + ) + + def answer( + self, + store: Any, + request: Any, + *, + approve: bool, + decline_reason: Optional[str] = None, + ) -> Any: + """A signed answer from the fixture approver (``make_answer`` checks it).""" + + from openadapt_flow.approval_mode.item_approval import make_answer + + return make_answer( + request, + decision="approve" if approve else "decline", + decline_reason=None if approve else (decline_reason or "wrong_record"), + decided_by="human", + route=ROUTE, + # The signature names the approver; the older sign-in fields stay + # empty, as for any key-signed answer. + approver_id=None, + auth_method="none", + signer=self.signer, + policy=store.context(), + ) + + def ledger_fields(self) -> dict[str, str]: + """What the ledger records: public values only, never the key.""" + + return { + "principal": PRINCIPAL, + "root_public_key": self.root_public_key, + "config_sha256": self.config_sha256, + } diff --git a/packages/flow/benchmark/lifecycle_proof/run.py b/packages/flow/benchmark/lifecycle_proof/run.py index 142348081..4a6a17350 100644 --- a/packages/flow/benchmark/lifecycle_proof/run.py +++ b/packages/flow/benchmark/lifecycle_proof/run.py @@ -60,6 +60,7 @@ from typing import Any, Optional from benchmark.lifecycle.paths import relative_paths +from benchmark.lifecycle_proof.approver import FixtureApprover HERE = Path(__file__).resolve().parent FLOW = HERE.parents[1] @@ -186,18 +187,15 @@ def _config_and_hooks() -> tuple[Any, Any, dict[str, Any]]: class ScriptedPerson: - """Answers every waiting approval request, and counts each answer.""" - - APPROVE = dict( - decision="approve", - decided_by="human", - route="desktop_run_it", - approver_id="lifecycle-proof-operator", - auth_method="cloud_aal2", - ) + """Answers every waiting approval request, and counts each answer. + + Every answer is signed by the run's fixture approver + (:mod:`benchmark.lifecycle_proof.approver`); an unsigned one is refused. + """ - def __init__(self, ledger: Ledger) -> None: + def __init__(self, ledger: Ledger, approver: Any = None) -> None: self.ledger = ledger + self.approver = approver self.stores: list[Any] = [] self.seen: set[str] = set() self.touches = 0 @@ -207,14 +205,14 @@ def watch(self, store: Any) -> None: def sleep(self, seconds: float) -> None: time.sleep(min(seconds, 0.5)) - from openadapt_flow.approval_mode.item_approval import make_answer - for store in list(self.stores): for request in store.requests(): if request.request_id in self.seen: continue self.seen.add(request.request_id) - store.write_answer(make_answer(request, **self.APPROVE)) + store.write_answer( + answer_signed(self.ledger, self.approver, store, request, True) + ) self.touches += 1 self.ledger.add( "person", @@ -224,6 +222,26 @@ def sleep(self, seconds: float) -> None: ) +def answer_signed( + ledger: Any, approver: Any, store: Any, request: Any, approve: bool +) -> Any: + """The fixture approver's signed answer; a refusal is logged, then raised.""" + + from openadapt_flow.approval_mode.item_approval import ItemApprovalRefused + + try: + if approver is None: + raise ItemApprovalRefused( + "human_answer_unsigned", "no fixture approver for this run" + ) + return approver.answer(store, request, approve=approve) + except ItemApprovalRefused as exc: + ledger.add( + "person_answer_refused", request_id=request.request_id, code=exc.code + ) + raise + + class LocalStatusAuthority: """Stands in for Cloud's status route: signs ``on`` for the open switch.""" @@ -352,14 +370,13 @@ def __init__( def approval_gate(self, bundle: Path, page: Any) -> Any: from openadapt_flow.approval_mode.gate import ApprovalGate - from openadapt_flow.approval_mode.item_approval import ItemApprovalStore from openadapt_flow.approval_mode.model import ( DESKTOP_LIST_WITH_HEADER, ItemContext, ) ctx = self.ctx - store = ItemApprovalStore(self.run_dir) + store = ctx["approver"].store(self.run_dir) ctx["person"].watch(store) return ApprovalGate( @@ -392,7 +409,6 @@ def approval_gate(self, bundle: Path, page: Any) -> Any: ) def authorization(self, workflow: Any, bundle: Path, verifier: Any) -> Any: - from openadapt_flow.deployment import DeploymentConfig, PolicySection from openadapt_flow.execution_profiles import ( ExecutionProfile, execution_profile_contract, @@ -406,7 +422,7 @@ def authorization(self, workflow: Any, bundle: Path, verifier: Any) -> Any: gate = evaluate_run_gate( workflow, bundle_dir=bundle, - deployment=DeploymentConfig(policy=PolicySection(policy=POLICY)), + deployment=self.ctx["approver"].deployment(), effect_verifier=verifier, profile_contract=execution_profile_contract(ExecutionProfile.REGULATED), effective_durable=True, @@ -520,6 +536,8 @@ def _context(out: Path, keys_dir: Path, ledger: Ledger) -> dict[str, Any]: keys = _keys(keys_dir) _env(app, keys) config, hooks, lookup = _config_and_hooks() + approver = FixtureApprover.for_run(out, policy=POLICY) + ledger.add("fixture_approver", **approver.ledger_fields()) ctx: dict[str, Any] = { "app": app, "keys": keys, @@ -527,7 +545,8 @@ def _context(out: Path, keys_dir: Path, ledger: Ledger) -> dict[str, Any]: "hooks": hooks, "lookup": lookup, "ledger": ledger, - "person": ScriptedPerson(ledger), + "approver": approver, + "person": ScriptedPerson(ledger, approver), "store": out / "repair-store", "out": out, } @@ -1018,9 +1037,8 @@ def phase_drift(ctx: dict[str, Any], trials: int, agent: Any) -> dict[str, Any]: admission_key_dir=ctx["keys"]["admission_key_dir"], sleep=ctx["person"].sleep, ) - from openadapt_flow.approval_mode.item_approval import ItemApprovalStore - ctx["person"].watch(ItemApprovalStore(run_dir)) + ctx["person"].watch(ctx["approver"].store(run_dir)) t0 = time.time() result = repair.run(ctx["bundle"], run_dir) evidence = result.evidence diff --git a/packages/flow/tests/test_lifecycle_fixture_approver.py b/packages/flow/tests/test_lifecycle_fixture_approver.py new file mode 100644 index 000000000..1025ce57f --- /dev/null +++ b/packages/flow/tests/test_lifecycle_fixture_approver.py @@ -0,0 +1,152 @@ +"""The soak's scripted person signs its approvals with a fixture-only key. + +Since a person's approval needs a registered approver key's signature, the +harness's old unsigned answer is refused. The fixture approver gives each run +its own key, roster and pinned deployment config, so the scripted person's +answers count only the way a real approver's do. +""" + +from __future__ import annotations + +import json +from pathlib import Path + +import pytest + +from benchmark.lifecycle_proof.approver import FixtureApprover +from benchmark.lifecycle_proof.run import ScriptedPerson +from openadapt_flow.approval_mode.approver_roster import ROOT_ENV +from openadapt_flow.approval_mode.item_approval import ( + ItemApprovalRefused, + answer_refusal, + make_answer, +) +from openadapt_flow.approval_mode.model import ApprovalMode + +FRAME = b"\x89PNG\r\n\x1a\nmade-up filled form" + + +class _Ledger: + def __init__(self) -> None: + self.rows: list[dict] = [] + + def add(self, kind: str, **fields) -> None: + self.rows.append({"kind": kind, **fields}) + + +def _request(store): + return store.open_request( + run_id="run-0001", + workflow_name="referral", + step_id="step_040", + step_index=28, + step_intent="click 'Save new patient'", + mode=ApprovalMode.PROBLEMS_ONLY, + bundle_content_digest="a" * 64, + runtime_inputs_digest="b" * 64, + frame_png=FRAME, + form_state=("dom", "sha256:" + "1" * 64), + ) + + +@pytest.fixture +def run(tmp_path, monkeypatch): + # The environment pin is never used: a stray one must not help or hurt. + monkeypatch.delenv(ROOT_ENV, raising=False) + approver = FixtureApprover.for_run(tmp_path / "arm", policy="clinical-write") + store = approver.store(tmp_path / "arm" / "items" / "ref-00026") + store.write_approver_root() + return approver, store + + +def test_the_old_unsigned_scripted_answer_is_refused(run): + _approver, store = run + request = _request(store) + + with pytest.raises(ItemApprovalRefused) as refused: + make_answer( + request, + decision="approve", + decided_by="human", + route="desktop_run_it", + approver_id="lifecycle-soak-operator", + auth_method="cloud_aal2", + policy=store.context(), + ) + assert refused.value.code == "auth_method_unverified" + + with pytest.raises(ItemApprovalRefused): + make_answer( + request, + decision="approve", + decided_by="human", + route="desktop_run_it", + approver_id="soak-front-desk", + auth_method="none", + policy=store.context(), + ) + + +def test_the_fixture_approver_signs_an_answer_the_run_accepts(run): + approver, store = run + request = _request(store) + + answer = approver.answer(store, request, approve=True) + assert answer.attestation is not None + assert answer_refusal(request, answer, policy=store.context()) is None + store.write_answer(answer) + assert store.status(request) == "approved" + + +def test_the_scripted_person_signs_and_a_missing_approver_is_logged(run): + approver, store = run + request = _request(store) + + ledger = _Ledger() + person = ScriptedPerson(ledger, approver) + person.watch(store) + person.sleep(0) + assert store.status(request) == "approved" + assert [r["kind"] for r in ledger.rows] == ["person"] + + other = approver.store(store.run_dir.parent / "ref-00027") + other.write_approver_root() + unsigned = _request(other) + ledger = _Ledger() + person = ScriptedPerson(ledger, None) + person.watch(other) + with pytest.raises(ItemApprovalRefused): + person.sleep(0) + assert ledger.rows == [ + { + "kind": "person_answer_refused", + "request_id": unsigned.request_id, + "code": "human_answer_unsigned", + } + ] + assert other.status(unsigned) == "waiting" + + +def test_a_changed_deployment_config_is_refused(run): + approver, _store = run + approver.config_path.write_text( + approver.config_path.read_text().replace("lifecycle-soak-fixture", "edited") + ) + with pytest.raises(ValueError): + approver.deployment() + + +def test_the_run_reopens_its_own_approver_and_the_ledger_has_no_secret(run, tmp_path): + approver, _store = run + again = FixtureApprover.for_run(tmp_path / "arm", policy="clinical-write") + assert again.root_public_key == approver.root_public_key + assert again.config_sha256 == approver.config_sha256 + + fields = approver.ledger_fields() + assert set(fields) == {"principal", "root_public_key", "config_sha256"} + key_text = next(Path(approver.key_dir).glob("*.pem")).read_text() + passphrase = (Path(approver.key_dir) / ".passphrase").read_text() + blob = json.dumps(fields) + assert passphrase not in blob and key_text not in blob + # The key and its passphrase live in the run directory, owner-only. + assert (Path(approver.key_dir) / ".passphrase").stat().st_mode & 0o077 == 0