diff --git a/.github/workflows/test.yml b/.github/workflows/test.yml index 8a23904..4c16892 100644 --- a/.github/workflows/test.yml +++ b/.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 diff --git a/openadapt_types/__init__.py b/openadapt_types/__init__.py index af70bb5..67c2759 100644 --- a/openadapt_types/__init__.py +++ b/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", diff --git a/openadapt_types/execute.py b/openadapt_types/execute.py index 26aaba9..02703bd 100644 --- a/openadapt_types/execute.py +++ b/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": diff --git a/openadapt_types/schemas/execute-v1-openapi.json b/openadapt_types/schemas/execute-v1-openapi.json index 1965e3e..a2aae46 100644 --- a/openadapt_types/schemas/execute-v1-openapi.json +++ b/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": { diff --git a/openadapt_types/seal.py b/openadapt_types/seal.py new file mode 100644 index 0000000..245a51f --- /dev/null +++ b/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 + ) diff --git a/tests/fixtures/cloud-seal-signature-vector.json b/tests/fixtures/cloud-seal-signature-vector.json new file mode 100644 index 0000000..065562c --- /dev/null +++ b/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==" + } +} diff --git a/tests/test_seal.py b/tests/test_seal.py new file mode 100644 index 0000000..a8e833b --- /dev/null +++ b/tests/test_seal.py @@ -0,0 +1,88 @@ +"""Seal issuer signatures: wire compatibility with Cloud and fail-closed checks.""" + +import base64 +import json +from pathlib import Path + +import pytest + +from openadapt_types import ExecuteEvidenceReceiptV1 +from openadapt_types.seal import ( + SealSignatureV1, + parse_key_registry, + receipt_sha256, + seal_key_id, + signing_payload, + verify_seal_signature, + verify_sealed_receipt, +) + +ed25519 = pytest.importorskip("cryptography.hazmat.primitives.asymmetric.ed25519") +from cryptography.hazmat.primitives import serialization # noqa: E402 + +VECTOR = json.loads((Path(__file__).parent / "fixtures" / "cloud-seal-signature-vector.json").read_text()) + + +def _key(): + private = ed25519.Ed25519PrivateKey.generate() + raw = private.public_key().public_bytes(serialization.Encoding.Raw, serialization.PublicFormat.Raw) + entry = {"key_id": seal_key_id(raw), "public_key": base64.b64encode(raw).decode(), "status": "active"} + return private, entry + + +def _sign(private, entry, sha, run_id="exec-0001-test", issued_at="2026-10-14T12:00:00Z"): + fields = { + "schema": "openadapt.seal-signature/v1", + "issuer": "openadapt-cloud", + "algorithm": "ed25519", + "key_id": entry["key_id"], + "receipt_sha256": sha, + "run_id": run_id, + "issued_at": issued_at, + } + fields["signature"] = base64.b64encode(private.sign(signing_payload(fields))).decode() + return fields + + +def test_cloud_typescript_vector_verifies_in_python() -> None: + registry = parse_key_registry(VECTOR["registry"]) + result = verify_seal_signature(VECTOR["signature"], "c" * 64, registry) + assert result.ok and result.key_id == VECTOR["registry"][0]["key_id"] + tampered = dict(VECTOR["signature"], run_id="exec-0002-test") + assert verify_seal_signature(tampered, "c" * 64, registry).reason == "bad_signature" + + +def test_refusals_are_closed() -> None: + private, entry = _key() + registry = parse_key_registry([entry]) + sig = _sign(private, entry, "a" * 64) + assert verify_seal_signature(None, "a" * 64, registry).reason == "unsigned" + assert verify_seal_signature(dict(sig, extra=1), "a" * 64, registry).reason == "malformed" + assert verify_seal_signature(sig, "b" * 64, registry).reason == "receipt_mismatch" + assert verify_seal_signature(sig, "a" * 64, []).reason == "unknown_key" + revoked = parse_key_registry([dict(entry, status="revoked", revoked_at="2026-01-01T00:00:00Z")]) + assert verify_seal_signature(sig, "a" * 64, revoked).reason == "revoked_key" + + +def test_receipt_field_is_optional_and_hash_excludes_it() -> None: + fields = ExecuteEvidenceReceiptV1.model_fields + assert "seal_signature" in fields and not fields["seal_signature"].is_required() + receipt = {"a": 1, "b": "é"} + assert receipt_sha256(receipt) == receipt_sha256({**receipt, "seal_signature": {"x": 1}}) + + +def test_sealed_receipt_round_trip() -> None: + private, entry = _key() + registry = parse_key_registry([entry]) + receipt = {"schema_version": "openadapt.execute-evidence-receipt/v1", "execution_id": "exec-0001-test"} + sealed = {**receipt, "seal_signature": _sign(private, entry, receipt_sha256(receipt))} + assert verify_sealed_receipt(sealed, registry).ok + assert not verify_sealed_receipt({**sealed, "execution_id": "exec-9999-test"}, registry).ok + assert verify_sealed_receipt(receipt, registry).reason == "unsigned" + assert SealSignatureV1.model_validate(sealed["seal_signature"]).wire() == sealed["seal_signature"] + + +def test_registry_rejects_mislabeled_keys() -> None: + _, entry = _key() + with pytest.raises(ValueError): + parse_key_registry([dict(entry, key_id="seal-ed25519-0000000000000000")])