benchmark(lifecycle): the scripted person signs its approvals with a fixture-only key

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(),

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,
+        }

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,14 +536,17 @@ 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,
         "config": config,
         "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