feat(seal): pin Cloud Seal keys (person-confirmed TOFU), verify Seals, seal Cloud-connected runs

README.md

@@ -360,13 +360,36 @@ It claims the job, polls openadapt.ai over outgoing HTTPS, and asks you in the t
 
 Report vulnerabilities as described in [SECURITY.md](SECURITY.md).
 
+## Seals
+
+A Seal is OpenAdapt Cloud's Ed25519 signature on an Execute receipt. The agent reports `sealed: true` only when all of this holds:
+
+1. The run is connected to a Cloud Execute execution: `FlowRunner.run(..., cloud_execution_id=...)`, or `openadapt-flow run --cloud-execution-id ID`, with `OPENADAPT_EXECUTE_API_KEY` set. After a VERIFIED outcome Flow fetches Cloud's receipt and writes it into `report.json` as `execute_receipt`. `openadapt-flow attach-seal RUN_DIR --execution-id ID` does the same later.
+2. The receipt's signature verifies against keys a person pinned on this computer. The agent never fetches keys while it checks a run.
+
+Pin the keys once. Compare the printed fingerprint with the one OpenAdapt publishes before you type the key id back:
+
+```bash
+openadapt-agent seal keys pin          # fetches https://app.openadapt.ai/api/public/seal-keys
+openadapt-agent seal keys show
+```
+
+Run `pin` again when OpenAdapt rotates its key. A new key needs the same confirmation. A key that Cloud revokes is marked revoked without asking. A key Cloud stops listing stays pinned.
+
+Anyone can check a Seal offline (a receipt, a run report, or a webhook body):
+
+```bash
+openadapt-agent seal verify runs/<run_id>/report.json
+openadapt-agent seal verify https://example.com/receipt.json --json
+```
+
 ## Limits
 
 - Your agent, or a program acting for it, has to run on your customer's computer.
 - Nobody is notified when a run stops. Someone has to check `list_needs_attention` or Flow's console.
 - The check that stops a second entry covers one runs folder. Pass the same absolute `--runs-dir` every time.
 - If the server exits during a run, the run reads as `not_sure_if_saved`, and Flow may still finish it.
-- This server issues no signed receipts, so `proof` is never `sealed`.
+- This server issues no signed receipts itself. `proof` is `sealed` only when a run is connected to OpenAdapt Cloud Execute, Cloud signed its receipt, and that signature verifies against keys you pinned (see [Seals](#seals)).
 - Each call carries one request. There's no batch call.
 - Accounts, separation between customers, fleet policy, and hosted runs belong to OpenAdapt Cloud.
 

src/openadapt_agent/cli.py

@@ -11,6 +11,8 @@
 - ``authoring connect`` — outbound mailbox client for hosted ChatGPT.com /
   Claude.ai (claim ``oab_``, poll wait=0, Allow-per-sub). Not an HTTP
   listener. Overlay chrome stays Desktop-only.
+- ``seal verify`` / ``seal keys pin`` — check a Seal offline against keys a
+  person pinned (trust on first use, explicit rotation).
 - ``emit-skill`` — write a PHI-safe Agent Skill for one workflow from its
   card: purpose, inputs, and how to read the four outcomes.
 
@@ -302,6 +304,10 @@ def build_parser() -> argparse.ArgumentParser:
     )
     p.set_defaults(func=_cmd_emit_skill)
 
+    from openadapt_agent.seal_cli import add_seal_parser
+
+    add_seal_parser(sub)
+
     return parser
 
 

src/openadapt_agent/runner.py

@@ -577,7 +577,12 @@ def __init__(self, config: RunnerConfig):
         self.config = config
 
     def _build_command(
-        self, bundle_dir: Path, run_dir: Path, params_file: Path, url: Optional[str]
+        self,
+        bundle_dir: Path,
+        run_dir: Path,
+        params_file: Path,
+        url: Optional[str],
+        cloud_execution_id: Optional[str] = None,
     ) -> list[str]:
         cmd = [
             *self.config.flow_cli,
@@ -596,6 +601,10 @@ def _build_command(
             cmd += ["--policy", self.config.policy]
         if self.config.approval_mode:
             cmd += ["--approval-mode", self.config.approval_mode]
+        if cloud_execution_id:
+            # Flow fetches Cloud's signed receipt after a VERIFIED outcome and
+            # writes it into report.json (needs OPENADAPT_EXECUTE_API_KEY).
+            cmd += ["--cloud-execution-id", cloud_execution_id]
         cmd += list(self.config.extra_run_args)
         return cmd
 
@@ -607,6 +616,7 @@ def run(
         params: dict[str, str],
         url_override: Optional[str] = None,
         run_id: Optional[str] = None,
+        cloud_execution_id: Optional[str] = None,
     ) -> RunOutcome:
         """Run the bundle once. Params travel via ``--params-file`` (never argv).
 
@@ -645,7 +655,9 @@ def run(
         try:
             with os.fdopen(params_fd, "w") as fh:
                 json.dump({k: str(v) for k, v in (params or {}).items()}, fh)
-            cmd = self._build_command(Path(bundle_dir), run_dir, params_file, url)
+            cmd = self._build_command(
+                Path(bundle_dir), run_dir, params_file, url, cloud_execution_id
+            )
             try:
                 # Decode with replacement: a strict decode error would surface
                 # after Flow already ran and read as a launch failure.

src/openadapt_agent/seal.py

@@ -8,9 +8,12 @@
 still done with ``proof: "local"``.
 
 The registry is the JSON from ``GET https://app.openadapt.ai/api/public/seal-keys``
-saved locally and named by ``OPENADAPT_SEAL_KEYS_PATH`` (or inline in
-``OPENADAPT_SEAL_KEYS_JSON``). It is never fetched at verification time, so a
-network attacker cannot substitute keys. Byte-compatible with
+pinned by a person with ``openadapt-agent seal keys pin`` (trust on first use,
+explicit confirmation) to ``~/.openadapt/seal-keys.json``, or named by
+``OPENADAPT_SEAL_KEYS_PATH`` (or inline in ``OPENADAPT_SEAL_KEYS_JSON``). It is
+never fetched at verification time, so a network attacker cannot substitute
+keys. A Cloud-connected run's receipt names the Cloud execution; Flow records
+that id as ``cloud_execution_id`` when it attaches the receipt. Byte-compatible with
 ``openadapt_types.seal`` and Cloud ``src/lib/sealSignature.ts``.
 """
 
@@ -67,28 +70,46 @@ def parse_registry(document: Any) -> list[dict]:
     return keys
 
 
+def default_keys_path() -> Path:
+    """Where ``openadapt-agent seal keys pin`` stores the pinned registry."""
+    return Path.home() / ".openadapt" / "seal-keys.json"
+
+
+def pinned_keys_path(env: Optional[Mapping[str, str]] = None) -> Path:
+    env = os.environ if env is None else env
+    named = env.get("OPENADAPT_SEAL_KEYS_PATH", "").strip()
+    return Path(named).expanduser() if named else default_keys_path()
+
+
 def load_pinned_registry(env: Optional[Mapping[str, str]] = None) -> list[dict]:
     """The pinned registry, or ``[]`` (which makes every Seal unverifiable)."""
     env = os.environ if env is None else env
     try:
         path = env.get("OPENADAPT_SEAL_KEYS_PATH", "").strip()
         if path:
-            return parse_registry(json.loads(Path(path).read_text(encoding="utf-8")))
+            return parse_registry(json.loads(Path(path).expanduser().read_text(encoding="utf-8")))
         inline = env.get("OPENADAPT_SEAL_KEYS_JSON", "").strip()
         if inline:
             return parse_registry(json.loads(inline))
-    except (OSError, ValueError):
+        default = default_keys_path()
+        if default.is_file():
+            return parse_registry(json.loads(default.read_text(encoding="utf-8")))
+    except (OSError, ValueError, TypeError, AttributeError):
         return []
     return []
 
 
-def verify_receipt_seal(receipt: Any, registry: Sequence[Mapping[str, Any]]) -> Optional[str]:
-    """Return the verifying key id, or ``None`` for any failure."""
+def explain_receipt_seal(
+    receipt: Any, registry: Sequence[Mapping[str, Any]]
+) -> tuple[Optional[str], str]:
+    """Return ``(key_id, "verified")`` or ``(None, closed refusal reason)``."""
     if not isinstance(receipt, Mapping):
-        return None
+        return None, "malformed"
     sig = receipt.get("seal_signature")
+    if sig is None:
+        return None, "unsigned"
     if not isinstance(sig, Mapping) or set(sig) != set(_FIELDS):
-        return None
+        return None, "malformed"
     if (
         sig["schema"] != SEAL_SIGNATURE_SCHEMA
         or sig["issuer"] != "openadapt-cloud"
@@ -98,25 +119,70 @@ def verify_receipt_seal(receipt: Any, registry: Sequence[Mapping[str, Any]]) ->
         or not isinstance(sig["issued_at"], str) or not _ISO.match(sig["issued_at"])
         or not isinstance(sig["signature"], str)
     ):
-        return None
+        return None, "malformed"
     if sig["receipt_sha256"] != receipt_sha256(receipt):
-        return None
+        return None, "receipt_mismatch"
     if sig["run_id"] != receipt.get("execution_id"):
-        return None
+        return None, "receipt_mismatch"
+    if not registry:
+        return None, "no_pinned_keys"
     key = next((k for k in registry if k.get("key_id") == sig["key_id"]), None)
     if key is None:
-        return None
+        return None, "unknown_key"
     if key["status"] == "revoked" and (not key.get("revoked_at") or sig["issued_at"] >= key["revoked_at"]):
-        return None
+        return None, "revoked_key"
     try:
         from cryptography.exceptions import InvalidSignature
         from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PublicKey
 
         payload = _DOMAIN + _canonical({k: v for k, v in sig.items() if k != "signature"}).encode("utf-8")
         Ed25519PublicKey.from_public_bytes(key["raw"]).verify(base64.b64decode(sig["signature"], validate=True), payload)
     except (ImportError, InvalidSignature, ValueError):
+        return None, "bad_signature"
+    return str(key["key_id"]), "verified"
+
+
+def verify_receipt_seal(receipt: Any, registry: Sequence[Mapping[str, Any]]) -> Optional[str]:
+    """Return the verifying key id, or ``None`` for any failure."""
+    return explain_receipt_seal(receipt, registry)[0]
+
+
+def receipt_from_document(document: Any) -> Optional[Mapping[str, Any]]:
+    """Find the Execute receipt in a receipt, run report, webhook, or wrapper."""
+    if not isinstance(document, Mapping):
         return None
-    return str(key["key_id"])
+    if "seal_signature" in document or document.get("schema_version") == "openadapt.execute-evidence-receipt/v1":
+        return document
+    for key in ("execute_receipt", "receipt"):
+        inner = document.get(key)
+        if isinstance(inner, Mapping):
+            return inner
+    return None
+
+
+def verify_document(document: Any, registry: Sequence[Mapping[str, Any]]) -> dict:
+    """Plain verdict for ``openadapt-agent seal verify``. Offline; pinned keys only."""
+    receipt = receipt_from_document(document)
+    if receipt is None:
+        return {"verified": False, "reason": "no_receipt"}
+    key_id, reason = explain_receipt_seal(receipt, registry)
+    verdict: dict = {
+        "verified": key_id is not None,
+        "reason": reason,
+        "execution_id": receipt.get("execution_id"),
+        "outcome": receipt.get("outcome"),
+        "receipt_sha256": receipt_sha256(receipt),
+    }
+    if key_id is not None:
+        verdict["key_id"] = key_id
+        sig = receipt.get("seal_signature") or {}
+        verdict["issued_at"] = sig.get("issued_at")
+        if receipt.get("outcome") not in _SEALABLE:
+            # A genuine signature over a non-verified outcome is not a Seal of success.
+            verdict["verified_outcome"] = False
+        else:
+            verdict["verified_outcome"] = True
+    return verdict
 
 
 def run_is_sealed(
@@ -132,7 +198,13 @@ def run_is_sealed(
     receipt = report.get("execute_receipt")
     if not isinstance(receipt, Mapping):
         return False
-    if receipt.get("execution_id") != run_id or receipt.get("outcome") not in _SEALABLE:
+    # The receipt names either this run or the Cloud execution Flow recorded
+    # when it attached the receipt to this run's report.
+    bound_ids = {run_id}
+    cloud_execution_id = report.get("cloud_execution_id")
+    if isinstance(cloud_execution_id, str) and cloud_execution_id:
+        bound_ids.add(cloud_execution_id)
+    if receipt.get("execution_id") not in bound_ids or receipt.get("outcome") not in _SEALABLE:
         return False
     registry = load_pinned_registry() if registry is None else registry
     return verify_receipt_seal(receipt, registry) is not None