feat(execute): optional Cloud issuer seal_signature on ExecuteEvidenceReceiptV1

.github/workflows/test.yml

@@ -19,6 +19,8 @@ jobs:
           version: "0.11.29"
       - run: uv sync --locked --extra dev
       - run: uv run pytest
+      - name: Seal signature verification (needs cryptography)
+        run: uv run --with 'cryptography>=42' pytest tests/test_seal.py
       - name: Build and verify the public distribution boundary
         run: |
           uv build

openadapt_types/__init__.py

@@ -251,6 +251,13 @@
     ExecuteWebhookV1,
     sign_execute_webhook_hmac,
 )
+from openadapt_types.seal import (
+    SealPublicKey,
+    SealSignatureV1,
+    SealVerification,
+    verify_seal_signature,
+    verify_sealed_receipt,
+)
 from openadapt_types.execute_client import ExecuteApiError, ExecuteClient
 from openadapt_types.execute_openapi import execute_openapi_document
 from openadapt_types.oracle import (
@@ -624,6 +631,11 @@
     "ExecuteDecisionTaskV1",
     "ExecuteEvidenceContractV1",
     "ExecuteEvidenceReceiptV1",
+    "SealSignatureV1",
+    "SealPublicKey",
+    "SealVerification",
+    "verify_seal_signature",
+    "verify_sealed_receipt",
     "ExecuteLifecycleStateV1",
     "ExecuteRequestV1",
     "ExecuteStateChangedWebhookV1",

openadapt_types/execute.py

@@ -31,6 +31,7 @@
 
 from openadapt_types.business_decision import BusinessDecisionTaskV1
 from openadapt_types.human_decision import HumanDecisionTaskV1
+from openadapt_types.seal import SealSignatureV1
 from openadapt_types.oracle import (
     oracle_tier_from_effect_strength,
     production_seal_allowed,
@@ -187,6 +188,9 @@ class ExecuteEvidenceReceiptV1(_StrictContract):
     compensation_effect_verified: StrictBool = False
     evidence_digest: StrictStr = Field(pattern=_SHA256_PATTERN)
     issued_at: StrictStr = Field(pattern=_TIMESTAMP_PATTERN)
+    # Optional, additive: Cloud issuer signature over this receipt's canonical
+    # hash (openadapt_types.seal). Absent means unsigned, never sealed.
+    seal_signature: SealSignatureV1 | None = None
 
     @model_validator(mode="after")
     def _validate_proof(self) -> "ExecuteEvidenceReceiptV1":

openadapt_types/schemas/execute-v1-openapi.json

@@ -583,6 +583,17 @@
             "title": "Schema Version",
             "type": "string"
           },
+          "seal_signature": {
+            "anyOf": [
+              {
+                "$ref": "#/components/schemas/SealSignatureV1"
+              },
+              {
+                "type": "null"
+              }
+            ],
+            "default": null
+          },
           "workflow_digest": {
             "pattern": "^sha256:[0-9a-f]{64}$",
             "title": "Workflow Digest",
@@ -1275,7 +1286,66 @@
         "title": "HumanDecisionTaskV1",
         "type": "object"
       },
-      "JsonValue": {}
+      "JsonValue": {},
+      "SealSignatureV1": {
+        "additionalProperties": false,
+        "description": "Issuer signature over one receipt hash.",
+        "properties": {
+          "algorithm": {
+            "const": "ed25519",
+            "default": "ed25519",
+            "title": "Algorithm",
+            "type": "string"
+          },
+          "issued_at": {
+            "pattern": "^\\d{4}-\\d{2}-\\d{2}T\\d{2}:\\d{2}:\\d{2}Z$",
+            "title": "Issued At",
+            "type": "string"
+          },
+          "issuer": {
+            "const": "openadapt-cloud",
+            "default": "openadapt-cloud",
+            "title": "Issuer",
+            "type": "string"
+          },
+          "key_id": {
+            "pattern": "^seal-ed25519-[a-f0-9]{16}$",
+            "title": "Key Id",
+            "type": "string"
+          },
+          "receipt_sha256": {
+            "pattern": "^[a-f0-9]{64}$",
+            "title": "Receipt Sha256",
+            "type": "string"
+          },
+          "run_id": {
+            "maxLength": 128,
+            "minLength": 1,
+            "title": "Run Id",
+            "type": "string"
+          },
+          "schema": {
+            "const": "openadapt.seal-signature/v1",
+            "default": "openadapt.seal-signature/v1",
+            "title": "Schema",
+            "type": "string"
+          },
+          "signature": {
+            "pattern": "^[A-Za-z0-9+/]+={0,2}$",
+            "title": "Signature",
+            "type": "string"
+          }
+        },
+        "required": [
+          "key_id",
+          "receipt_sha256",
+          "run_id",
+          "issued_at",
+          "signature"
+        ],
+        "title": "SealSignatureV1",
+        "type": "object"
+      }
     },
     "securitySchemes": {
       "bearerAuth": {

openadapt_types/seal.py

@@ -0,0 +1,170 @@
+"""Cloud issuer signatures on Seals (``openadapt.seal-signature/v1``).
+
+A Seal is an :class:`~openadapt_types.execute.ExecuteEvidenceReceiptV1`. Cloud
+may attach ``seal_signature``: an Ed25519 signature by a published OpenAdapt
+Cloud issuer key over the receipt's canonical sha256 (the receipt without its
+``seal_signature``). Byte-compatible with Cloud ``src/lib/sealSignature.ts``.
+
+Verification is fail closed: every problem is a refusal with a closed reason,
+and an unsigned receipt is never "sealed". Verifying needs ``cryptography``
+(install ``cryptography`` alongside); without it verification refuses.
+"""
+
+from __future__ import annotations
+
+import base64
+import hashlib
+import json
+import re
+from dataclasses import dataclass
+from typing import Any, Literal, Mapping, Sequence
+
+from pydantic import BaseModel, ConfigDict, Field, StrictStr
+
+SEAL_SIGNATURE_SCHEMA = "openadapt.seal-signature/v1"
+SEAL_ISSUER = "openadapt-cloud"
+_DOMAIN = (SEAL_SIGNATURE_SCHEMA + "\0").encode("utf-8")
+_SHA256_HEX = r"^[a-f0-9]{64}$"
+_KEY_ID = r"^seal-ed25519-[a-f0-9]{16}$"
+_ISO_SECONDS = r"^\d{4}-\d{2}-\d{2}T\d{2}:\d{2}:\d{2}Z$"
+_B64 = r"^[A-Za-z0-9+/]+={0,2}$"
+
+SealRefusal = Literal[
+    "unsigned",
+    "malformed",
+    "receipt_mismatch",
+    "unknown_key",
+    "revoked_key",
+    "bad_signature",
+    "verifier_unavailable",
+]
+
+
+class SealSignatureV1(BaseModel):
+    """Issuer signature over one receipt hash."""
+
+    model_config = ConfigDict(extra="forbid", frozen=True)
+
+    schema_: Literal["openadapt.seal-signature/v1"] = Field(
+        default=SEAL_SIGNATURE_SCHEMA, alias="schema"
+    )
+    issuer: Literal["openadapt-cloud"] = SEAL_ISSUER
+    algorithm: Literal["ed25519"] = "ed25519"
+    key_id: StrictStr = Field(pattern=_KEY_ID)
+    receipt_sha256: StrictStr = Field(pattern=_SHA256_HEX)
+    run_id: StrictStr = Field(min_length=1, max_length=128)
+    issued_at: StrictStr = Field(pattern=_ISO_SECONDS)
+    signature: StrictStr = Field(pattern=_B64)
+
+    def wire(self) -> dict[str, str]:
+        return self.model_dump(mode="json", by_alias=True)
+
+
+@dataclass(frozen=True)
+class SealPublicKey:
+    key_id: str
+    public_key: str  # base64 raw 32 bytes
+    status: Literal["active", "revoked"]
+    revoked_at: str | None = None
+
+
+@dataclass(frozen=True)
+class SealVerification:
+    ok: bool
+    key_id: str | None = None
+    reason: SealRefusal | None = None
+
+
+def canonical_json(value: Any) -> str:
+    """Sorted-key compact JSON, matching Cloud's canonical form."""
+    return json.dumps(value, sort_keys=True, separators=(",", ":"), ensure_ascii=False)
+
+
+def receipt_sha256(receipt: Mapping[str, Any]) -> str:
+    """Hash of a receipt's JSON without its ``seal_signature``."""
+    body = {k: v for k, v in receipt.items() if k != "seal_signature"}
+    return hashlib.sha256(canonical_json(body).encode("utf-8")).hexdigest()
+
+
+def seal_key_id(public_key_raw: bytes) -> str:
+    return "seal-ed25519-" + hashlib.sha256(public_key_raw).hexdigest()[:16]
+
+
+def signing_payload(signature: Mapping[str, Any]) -> bytes:
+    unsigned = {k: v for k, v in signature.items() if k != "signature"}
+    return _DOMAIN + canonical_json(unsigned).encode("utf-8")
+
+
+def parse_key_registry(entries: Sequence[Mapping[str, Any]]) -> list[SealPublicKey]:
+    keys: list[SealPublicKey] = []
+    for entry in entries:
+        raw = base64.b64decode(str(entry.get("public_key", "")), validate=True)
+        if len(raw) != 32:
+            raise ValueError("seal public key must be 32 raw bytes")
+        if entry.get("key_id") != seal_key_id(raw):
+            raise ValueError("seal key_id does not match its public key")
+        status = entry.get("status")
+        if status not in ("active", "revoked"):
+            raise ValueError("seal key status must be active or revoked")
+        revoked_at = entry.get("revoked_at")
+        if status == "revoked" and not (
+            isinstance(revoked_at, str) and re.match(_ISO_SECONDS, revoked_at)
+        ):
+            raise ValueError("a revoked seal key names revoked_at")
+        keys.append(
+            SealPublicKey(
+                key_id=str(entry["key_id"]),
+                public_key=base64.b64encode(raw).decode(),
+                status=status,
+                revoked_at=revoked_at if status == "revoked" else None,
+            )
+        )
+    return keys
+
+
+def verify_seal_signature(
+    signature: Any,
+    expected_receipt_sha256: str,
+    registry: Sequence[SealPublicKey],
+) -> SealVerification:
+    """Verify a Seal signature against a pinned issuer key registry."""
+    if signature is None:
+        return SealVerification(False, reason="unsigned")
+    try:
+        parsed = (
+            signature
+            if isinstance(signature, SealSignatureV1)
+            else SealSignatureV1.model_validate(signature)
+        )
+    except Exception:
+        return SealVerification(False, reason="malformed")
+    if parsed.receipt_sha256 != expected_receipt_sha256:
+        return SealVerification(False, reason="receipt_mismatch")
+    key = next((k for k in registry if k.key_id == parsed.key_id), None)
+    if key is None:
+        return SealVerification(False, reason="unknown_key")
+    if key.status == "revoked" and (
+        key.revoked_at is None or parsed.issued_at >= key.revoked_at
+    ):
+        return SealVerification(False, reason="revoked_key")
+    try:
+        from cryptography.exceptions import InvalidSignature
+        from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PublicKey
+    except ImportError:
+        return SealVerification(False, reason="verifier_unavailable")
+    try:
+        Ed25519PublicKey.from_public_bytes(base64.b64decode(key.public_key)).verify(
+            base64.b64decode(parsed.signature), signing_payload(parsed.wire())
+        )
+    except (InvalidSignature, ValueError):
+        return SealVerification(False, reason="bad_signature")
+    return SealVerification(True, key_id=key.key_id)
+
+
+def verify_sealed_receipt(
+    receipt: Mapping[str, Any], registry: Sequence[SealPublicKey]
+) -> SealVerification:
+    """Verify an Execute receipt (wire JSON) carries a valid issuer Seal."""
+    return verify_seal_signature(
+        receipt.get("seal_signature"), receipt_sha256(receipt), registry
+    )

tests/fixtures/cloud-seal-signature-vector.json

@@ -0,0 +1,20 @@
+{
+  "note": "Generated by openadapt-cloud src/lib/sealSignature.ts with a fixed TEST seed (32 bytes of 0x07). Not a production key.",
+  "registry": [
+    {
+      "key_id": "seal-ed25519-fe812c12f3ab4ce6",
+      "public_key": "6kpsY+KcUgq+9VB7Ey7F+ZVHdq6+vnuSQh7qaRRG0iw=",
+      "status": "active"
+    }
+  ],
+  "signature": {
+    "schema": "openadapt.seal-signature/v1",
+    "issuer": "openadapt-cloud",
+    "algorithm": "ed25519",
+    "key_id": "seal-ed25519-fe812c12f3ab4ce6",
+    "receipt_sha256": "cccccccccccccccccccccccccccccccccccccccccccccccccccccccccccccccc",
+    "run_id": "exec-0001-test",
+    "issued_at": "2026-10-14T12:00:00Z",
+    "signature": "YWdCNlLtcwDpBTAPbziOYWLjLcI264mQ/dcGtTI/eFybvZThhhBVvQogcdU5dyzD8jJgCYmDSE95cL/NCTX0Dg=="
+  }
+}