diff --git a/CHANGELOG.md b/CHANGELOG.md index 70eb40e..4ee9b98 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -9,12 +9,19 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ### Added +- Live A2A JSON-RPC peer listener (`ca2a start` / `ca2a_runtime.server`): serves + `message/send`, runs `parse_peer_request` → `handle_peer_request` end to end + (verify chain, intersect local/Cedar policy, open sealed payload with a + configured key, emit provenance). Ordinary A2A (no cA2A keys) returns + `ca2a: null`. Fail-closed on malformed metadata and denials. Optional + `local_policy` / `enclave_private_key_hex` config fields; install with + `pip install 'ca2a-runtime[serve]'`. Does **not** add attestation handshake + or seal-to-verified-measurement binding. See issue #47. - A2A transport adapter (`ca2a_runtime.transport`): parse/attach cA2A extension metadata on A2A `SendMessage`-shaped messages into `PeerRequest` (and the reverse). Extension URI `https://agentrust.io/extensions/ca2a/v0.1`. Fail closed on malformed cA2A metadata; absence of all cA2A keys returns `None` (ordinary - A2A). Does not add HTTP serving, `ca2a start`, or seal-to-verified-measurement - binding. New error `TRANSPORT_ERROR`. See issue #47. + A2A). New error `TRANSPORT_ERROR`. See issue #47. ## [0.1.0a1] - 2026-07-09 diff --git a/LIMITATIONS.md b/LIMITATIONS.md index b4ecc08..d0c14d2 100644 --- a/LIMITATIONS.md +++ b/LIMITATIONS.md @@ -9,8 +9,8 @@ cA2A is a pre-release profile in active design. This document states plainly wha ## What is stubbed or not yet implemented -- **Live A2A transport serving.** An A2A metadata adapter (`ca2a_runtime.transport`) can parse/attach cA2A extension fields into a `PeerRequest`, and `handle_peer_request` enforces that request in-process. There is still no HTTP/JSON-RPC listener or `ca2a start`: nothing accepts a live inbound A2A socket call end to end. Parsing extension metadata is Tier 2 transport progress, not a claim that cA2A is attested across trust domains. -- **Sealed peer channel (live binding).** The channel is implemented: a payload is sealed to the peer's attested X25519 key (X25519 ECDH, HKDF-SHA256, ChaCha20-Poly1305), and only the holder of the peer's private key can open it. The remaining gap is the hardware property that the private key never leaves the peer's enclave (established by attestation), and wiring the seal to a verified report on a live inbound call. Until that end-to-end binding lands on hardware, do not assume a payload is confined to a specific attested measurement. Adapter-decoded `sealed_payload` bytes are opaque ciphertext only. +- **Live A2A transport serving.** An A2A metadata adapter (`ca2a_runtime.transport`) parses/attaches cA2A extension fields into a `PeerRequest`, and `ca2a start` serves JSON-RPC `message/send` end to end through `handle_peer_request` (verify chain → intersect policy → open sealed payload when a key is configured → emit a provenance record), fail-closed. There is still **no** attestation handshake on the live call, and the seal is **not** bound to a verified measurement: any enclave key is software-configured. Live serving is Tier 2 transport progress, not a claim that cA2A is attested across trust domains. +- **Sealed peer channel (live binding).** The channel is implemented: a payload is sealed to the peer's attested X25519 key (X25519 ECDH, HKDF-SHA256, ChaCha20-Poly1305), and only the holder of the peer's private key can open it. The live listener can open with a configured key. The remaining gap is the hardware property that the private key never leaves the peer's enclave (established by attestation), and wiring the seal to a verified report on a live inbound call. Until that end-to-end binding lands on hardware, do not assume a payload is confined to a specific attested measurement. Adapter-decoded `sealed_payload` bytes are opaque ciphertext only. - **Real hardware attestation.** The **SEV-SNP verifier is implemented**: report parsing, VCEK certificate chain verification, ECDSA-P384 report-signature verification, and measurement/report-data binding, all fail-closed. The chain path is validated against the genuine AMD Milan root chain; the report-signature path is validated with synthetic vectors, since a real report plus VCEK pair needs SEV-SNP hardware. Report generation (`SevSnpProvider.attest`) still requires a real SEV-SNP guest. The **Intel TDX verifier** (DCAP Quote v4: PCK chain to the genuine Intel SGX Root CA, QE report, attestation-key binding, quote signature, MRTD binding) and the **TPM 2.0 verifier** (AK chain to a caller-supplied vendor root, AK signature, magic/type, qualifying-data and PCR-digest binding) are also implemented and synthetic-vector validated; quote generation needs the respective hardware. Until a backend verifies a real quote end to end against a golden measurement on hardware, cA2A must not be described as fully attested across trust domains. ## Out of scope diff --git a/ROADMAP.md b/ROADMAP.md index af79966..8965426 100644 --- a/ROADMAP.md +++ b/ROADMAP.md @@ -22,7 +22,7 @@ Already implemented and tested elsewhere; cA2A depends on it rather than reimple ## v0.2: Runtime enforcement and sealed channel -- Runtime peer-delegation enforcement: **decision core landed** (`ca2a_runtime.peer.enforce_peer_call`: verify chain, intersect delegated scope with local policy, enforce, emit provenance record; claim C3 validated), now with a **real Cedar policy engine** option (`ca2a_runtime.cedar.CedarPolicy`) alongside the allow-set `LocalPolicy`. Remaining: wire the decision core to a live A2A transport (Tier 2) +- Runtime peer-delegation enforcement: **decision core landed** (`ca2a_runtime.peer.enforce_peer_call`: verify chain, intersect delegated scope with local policy, enforce, emit provenance record; claim C3 validated), now with a **real Cedar policy engine** option (`ca2a_runtime.cedar.CedarPolicy`) alongside the allow-set `LocalPolicy`. **Live A2A serving landed** (`ca2a start` / `ca2a_runtime.server`: JSON-RPC `message/send` → parse → `handle_peer_request`). Remaining: attestation handshake on the live call, and seal bound to a verified measurement (Tier 2/3) - Sealed peer channel: **landed** (`ca2a_runtime.channel`: HPKE-style X25519 -> HKDF-SHA256 -> ChaCha20-Poly1305 sealing to the peer's attested key; claim C4 validated). Remaining: bind the seal to a verified attestation report on a live call, and rely on the enclave to hold the private key (hardware property) - Linked runtime evidence: each hop's TRACE record references the parent record hash and delegation credential id, producing a verifiable delegation DAG (Tier 2) @@ -34,7 +34,7 @@ Real hardware attestation verification (SEV-SNP VCEK chain, Intel TDX quote via - **TDX verifier: landed.** DCAP Quote v4 parsing, PCK chain to the genuine Intel SGX Root CA, QE report signature, attestation-key binding, quote signature, and MRTD binding, all fail-closed. Quote generation requires a real TDX guest. See `ca2a_verify.tdx`. - **TPM 2.0 verifier: landed.** TPMS_ATTEST parsing, AK chain to a caller-supplied vendor root, AK signature (ECDSA or RSA), magic/type checks, and qualifying-data/PCR-digest binding, all fail-closed. Quote generation requires a real TPM. See `ca2a_verify.tpm`. - **Cross-operator attestation (C6): validated in software.** A two-operator harness (SEV-SNP verifier + measurement pinning + sealed channel) shows independent keys, mutual attestation, confidential cross-operator delegation, and binary-swap detection. All six claims (C1-C6) are now validated experiments. -- **Pending:** end-to-end validation of the SEV-SNP, TDX, and TPM signature paths against real hardware quotes on a confidential VM; and a transport that parses real A2A wire messages into a `PeerRequest`. +- **Pending:** end-to-end validation of the SEV-SNP, TDX, and TPM signature paths against real hardware quotes on a confidential VM; live attestation handshake and seal-to-verified-measurement binding on the inbound call. ## v1.0: Stable profile diff --git a/docs/configuration.md b/docs/configuration.md index 2d0ff18..3068512 100644 --- a/docs/configuration.md +++ b/docs/configuration.md @@ -1,6 +1,8 @@ # Configuration -The cA2A runtime reads a YAML config. The peer runtime path is under construction; this page documents the surface it consumes, validated today by `ca2a validate-config`. +The cA2A runtime reads a YAML config. Offline commands validate it with +`ca2a validate-config`. `ca2a start` consumes the same file for the live +JSON-RPC listener. ## Reference @@ -12,24 +14,31 @@ attestation: max_delegation_depth: 8 # reject chains deeper than this listen_addr: "0.0.0.0:8443" -policy_bundle_path: policy/ # optional: Cedar bundle for local scope intersection +local_policy: ["read", "write"] # allow-set for scope intersection (or use Cedar below) +# policy_bundle_path: policy.cedar +# enclave_private_key_hex: "<64 hex chars>" # optional; opens sealed_payload ``` ## Fields | Field | Default | Description | |---|---|---| -| `attestation.provider` | `auto` | TEE provider for peer attestation. `auto` selects the strongest available hardware; `software-only` requires no hardware. `opaque` is explicit opt-in and never auto-selected. | -| `attestation.enforcement_mode` | `enforcing` | `enforcing` denies peer calls that fail verification; `advisory` logs and proceeds; `silent` evaluates without logging or blocking. | +| `attestation.provider` | `auto` | TEE provider for peer attestation. `auto` selects the strongest available hardware; `software-only` requires no hardware. `opaque` is explicit opt-in and never auto-selected. Live attestation on the call is not wired yet. | +| `attestation.enforcement_mode` | `enforcing` | Intended mode. The live listener always fails closed on cA2A denials today; advisory/silent are accepted in config but not applied on the wire. | | `max_delegation_depth` | `8` | Chains deeper than this are rejected with `DELEGATION_DEPTH_EXCEEDED`. | -| `listen_addr` | `0.0.0.0:8443` | Address the peer runtime listens on. | -| `policy_bundle_path` | none | Directory of Cedar policy the runtime intersects with a delegated scope. | +| `listen_addr` | `0.0.0.0:8443` | Address `ca2a start` binds. | +| `local_policy` | none | Capability allow set for `LocalPolicy`. Required for `ca2a start` unless `policy_bundle_path` is set. | +| `policy_bundle_path` | none | Path to a Cedar policy file. When set, used instead of `local_policy`. | +| `enclave_private_key_hex` | none | Optional 32-byte X25519 private key (hex) for opening `sealed_payload`. Override with `--enclave-key-hex` or `CA2A_ENCLAVE_PRIVATE_KEY_HEX`. Software-configured only; not measurement-bound. | -## Validate +## Validate / start ```bash ca2a validate-config --config examples/minimal/ca2a-config.yaml -# ok: provider=auto enforcement=advisory +# ok: provider=software-only enforcement=enforcing + +pip install 'ca2a-runtime[serve]' +ca2a start --config examples/minimal/ca2a-config.yaml ``` Invalid values fail fast with a `CONFIG_ERROR` and a message naming the offending field. diff --git a/docs/spec/transport.md b/docs/spec/transport.md index e9b3f76..50faf05 100644 --- a/docs/spec/transport.md +++ b/docs/spec/transport.md @@ -7,13 +7,13 @@ cA2A is a profile on A2A, not a competing transport. A2A moves tasks and context | Piece | Status | |---|---| | Extension URI + namespaced metadata keys (below) | Specified | -| `ca2a_runtime.transport` adapter: A2A metadata ↔ `PeerRequest` | Implemented (parse/attach only) | -| Hand-off into `handle_peer_request` once a `PeerRequest` exists | Implemented in-process; callers invoke it after parsing | -| HTTP/JSON-RPC serving / `ca2a start` | Not yet — separate Tier 2 checkbox | +| `ca2a_runtime.transport` adapter: A2A metadata ↔ `PeerRequest` | Implemented (parse/attach) | +| Hand-off into `handle_peer_request` once a `PeerRequest` exists | Implemented | +| HTTP/JSON-RPC serving / `ca2a start` | Implemented (live serving only) | | Live attestation handshake on an inbound call | Not yet — separate Tier 2/3 checkbox | | Seal bound to a **verified** attestation measurement on a live call | Not yet — separate Tier 2/3 checkbox | -Parsing real A2A extension metadata is progress on Tier 2 transport wiring. It is **not** evidence that cA2A is attested across trust domains. See [LIMITATIONS.md](../../LIMITATIONS.md) and [ROADMAP.md](../../ROADMAP.md). +Live serving wires parse → `handle_peer_request` end to end on `message/send`. It is **not** evidence that cA2A is attested across trust domains: there is still no attestation handshake on the call, and sealed open uses a configured software/enclave key rather than a key bound to a verified measurement. See [LIMITATIONS.md](../../LIMITATIONS.md) and [ROADMAP.md](../../ROADMAP.md). ## Overlay, not fork @@ -88,26 +88,44 @@ Once a `PeerRequest` has been parsed, the intended inbound order is: 4. Open any sealed payload with the enclave-held key. Crypto implemented; live seal-to-verified-measurement binding is not. See [the sealed channel](sealed-channel.md). 5. Emit a linked provenance/TRACE record referencing the parent record hash and `credential_id`. See [the TRACE A2A profile](trace-a2a-profile.md). -Steps 1, 3, and 5 (and sealed open when a key is supplied) run today inside `handle_peer_request` after the adapter produces a `PeerRequest`. Steps 2 and the measurement-bound seal are remaining Tier 2/3 work. +Steps 1, 3, and 5 (and sealed open when a key is supplied) run inside `handle_peer_request` after the adapter produces a `PeerRequest`, including on the live `ca2a start` listener. Steps 2 and the measurement-bound seal are remaining Tier 2/3 work. + +## Live listener (`ca2a start`) + +```bash +pip install 'ca2a-runtime[serve]' +ca2a start --config examples/minimal/ca2a-config.yaml +``` + +The listener accepts JSON-RPC `message/send` (and `SendMessage`) on `POST /` and `POST /rpc`: + +1. `parse_peer_request` on the envelope. +2. If no cA2A keys: return `{"ca2a": null, ...}` (ordinary A2A; no invented trust state). +3. If cA2A keys present: `handle_peer_request` with the configured policy and optional enclave key; map `CA2AError` to a JSON-RPC error with `data.ca2a_code`. + +Config for the listener needs either `local_policy` (capability allow set) or `policy_bundle_path` (Cedar). Optional `enclave_private_key_hex` (or `--enclave-key-hex` / `CA2A_ENCLAVE_PRIVATE_KEY_HEX`) opens sealed payloads. That key is software-configured for this slice; it is not proven enclave-bound or measurement-linked. ## Enforcement is a peer decision -How strictly a cA2A peer acts on the extension fields is local configuration, not a transport-level flag. `Ca2aConfig.enforcement_mode` selects the behavior: +How strictly a cA2A peer acts on the extension fields is local configuration, not a transport-level flag. `Ca2aConfig.enforcement_mode` selects the intended behavior: -- `enforcing`: an unverifiable chain or a missing required credential denies the call. This is the default and the fail-closed posture the profile calls for. +- `enforcing`: an unverifiable chain or a missing required credential denies the call. This is the default and the fail-closed posture the profile calls for. The live listener always uses this posture today. - `advisory`: the failure is recorded but the call proceeds. - `silent`: the check runs without a visible signal. ```yaml # ca2a config attestation: - provider: auto + provider: software-only enforcement_mode: enforcing max_delegation_depth: 8 -listen_addr: "0.0.0.0:8443" +listen_addr: "127.0.0.1:8443" +local_policy: ["read", "write"] +# policy_bundle_path: policy.cedar # alternative to local_policy +# enclave_private_key_hex: "..." # optional; for sealed_payload open ``` -`max_delegation_depth` bounds the chain length a peer will accept and is passed through to `verify_chain`. `listen_addr` is reserved for the future HTTP listener (`ca2a start`); this adapter does not open a socket. See [failure modes](failure-modes.md). +`max_delegation_depth` bounds the chain length a peer will accept and is passed through to `verify_chain`. `listen_addr` is the HTTP bind address for `ca2a start`. See [failure modes](failure-modes.md). ## Transport stability diff --git a/examples/minimal/ca2a-config.yaml b/examples/minimal/ca2a-config.yaml index f15e4ce..e032c83 100644 --- a/examples/minimal/ca2a-config.yaml +++ b/examples/minimal/ca2a-config.yaml @@ -1,6 +1,7 @@ attestation: - provider: auto # auto | tpm | sev-snp | tdx | opaque | software-only - enforcement_mode: advisory # enforcing | advisory | silent + provider: software-only # auto | tpm | sev-snp | tdx | opaque | software-only + enforcement_mode: enforcing # live listener always fails closed today max_delegation_depth: 8 -listen_addr: "0.0.0.0:8443" +listen_addr: "127.0.0.1:8443" +local_policy: ["read", "write"] diff --git a/pyproject.toml b/pyproject.toml index d9d7135..86c5d74 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -30,6 +30,10 @@ dependencies = [ ] [project.optional-dependencies] +serve = [ + "starlette>=0.37", + "uvicorn>=0.29", +] dev = [ "pytest>=8.0", "pytest-asyncio>=0.23", @@ -39,6 +43,9 @@ dev = [ "types-pyyaml", "bandit[toml]>=1.7", "pip-audit>=2.6", + "starlette>=0.37", + "uvicorn>=0.29", + "httpx>=0.27", ] [project.scripts] diff --git a/src/ca2a_runtime/cli.py b/src/ca2a_runtime/cli.py index c7eee53..27b5ca6 100644 --- a/src/ca2a_runtime/cli.py +++ b/src/ca2a_runtime/cli.py @@ -4,6 +4,7 @@ import argparse import json +import os import sys from pathlib import Path from typing import Any @@ -11,7 +12,7 @@ from ca2a_runtime import __version__ from ca2a_runtime.config import Ca2aConfig from ca2a_runtime.delegation import DelegationCredential, verify_chain -from ca2a_runtime.errors import CA2AError, InvalidCredential, ProvenanceLinkBroken +from ca2a_runtime.errors import CA2AError, ConfigError, InvalidCredential, ProvenanceLinkBroken from ca2a_runtime.provenance import DelegationRecord, cross_check_chain, verify_dag from ca2a_verify import verify_chain_file @@ -107,6 +108,58 @@ def _cmd_verify_dag(args: argparse.Namespace) -> int: return 0 +def _cmd_start(args: argparse.Namespace) -> int: + """Start the live A2A JSON-RPC peer listener (``ca2a start``).""" + try: + import uvicorn + except ImportError: + print( + "ca2a start requires the 'serve' extra: pip install 'ca2a-runtime[serve]'", + file=sys.stderr, + ) + return 1 + + from ca2a_runtime.policy_loader import load_policy + from ca2a_runtime.server import ( + PeerRuntime, + create_app, + load_enclave_private_key, + parse_listen_addr, + ) + + try: + cfg = Ca2aConfig.load(args.config) + config_path = Path(args.config).resolve() + policy = load_policy(cfg, config_dir=config_path.parent) + key_hex = ( + args.enclave_key_hex + or os.environ.get("CA2A_ENCLAVE_PRIVATE_KEY_HEX") + or cfg.enclave_private_key_hex + ) + enclave_key = load_enclave_private_key(key_hex) + host, port = parse_listen_addr(cfg.listen_addr) + except ConfigError as exc: + print(f"invalid config: {exc}", file=sys.stderr) + return 1 + + if cfg.enforcement_mode != "enforcing": + print( + f"note: enforcement_mode={cfg.enforcement_mode!r} is accepted in config, " + "but the live listener always fails closed on cA2A denials", + file=sys.stderr, + ) + + runtime = PeerRuntime(config=cfg, policy=policy, enclave_private_key=enclave_key) + app = create_app(runtime) + print( + f"ca2a starting: listen={cfg.listen_addr} provider={cfg.provider} " + f"enforcement={cfg.enforcement_mode} " + f"(live serving only; no attestation handshake / seal-to-measurement binding)" + ) + uvicorn.run(app, host=host, port=port) + return 0 + + def build_parser() -> argparse.ArgumentParser: parser = argparse.ArgumentParser(prog="ca2a", description="Confidential agent-to-agent") parser.add_argument("--version", action="version", version=f"ca2a {__version__}") @@ -130,6 +183,18 @@ def build_parser() -> argparse.ArgumentParser: vd.add_argument("--max-depth", type=int, default=8) vd.set_defaults(func=_cmd_verify_dag) + st = sub.add_parser( + "start", + help="Start the live A2A JSON-RPC peer listener", + ) + st.add_argument("--config", required=True, help="Path to ca2a-config.yaml") + st.add_argument( + "--enclave-key-hex", + default=None, + help="X25519 private key as 32-byte hex (overrides config / CA2A_ENCLAVE_PRIVATE_KEY_HEX)", + ) + st.set_defaults(func=_cmd_start) + return parser diff --git a/src/ca2a_runtime/config.py b/src/ca2a_runtime/config.py index 84561c0..0a9702d 100644 --- a/src/ca2a_runtime/config.py +++ b/src/ca2a_runtime/config.py @@ -1,7 +1,7 @@ """Runtime configuration for the cA2A peer runtime. -The runtime peer path is under construction (see ROADMAP.md); this module -defines and validates the configuration surface it will consume. +Defines and validates the configuration surface consumed by ``ca2a start`` +and the offline CLI. See ROADMAP.md / LIMITATIONS.md for claim boundaries. """ from __future__ import annotations @@ -28,7 +28,9 @@ class Ca2aConfig: enforcement_mode: str = "enforcing" max_delegation_depth: int = 8 policy_bundle_path: str | None = None + local_policy: frozenset[str] | None = None listen_addr: str = "0.0.0.0:8443" + enclave_private_key_hex: str | None = None @classmethod def from_dict(cls, data: dict[str, Any]) -> Ca2aConfig: @@ -51,12 +53,33 @@ def from_dict(cls, data: dict[str, Any]) -> Ca2aConfig: if not isinstance(depth, int) or depth < 1: raise ConfigError("max_delegation_depth must be a positive integer") + raw_local = data.get("local_policy") + local_policy: frozenset[str] | None = None + if raw_local is not None: + if not isinstance(raw_local, list) or not all( + isinstance(item, str) and item for item in raw_local + ): + raise ConfigError( + "local_policy must be a list of non-empty capability strings" + ) + local_policy = frozenset(raw_local) + + bundle = data.get("policy_bundle_path") + if bundle is not None and not isinstance(bundle, str): + raise ConfigError("policy_bundle_path must be a string path") + + key_hex = data.get("enclave_private_key_hex") + if key_hex is not None and not isinstance(key_hex, str): + raise ConfigError("enclave_private_key_hex must be a string") + return cls( provider=provider, enforcement_mode=enforcement, max_delegation_depth=depth, - policy_bundle_path=data.get("policy_bundle_path"), + policy_bundle_path=bundle, + local_policy=local_policy, listen_addr=data.get("listen_addr", "0.0.0.0:8443"), + enclave_private_key_hex=key_hex, ) @classmethod diff --git a/src/ca2a_runtime/policy_loader.py b/src/ca2a_runtime/policy_loader.py new file mode 100644 index 0000000..51f4472 --- /dev/null +++ b/src/ca2a_runtime/policy_loader.py @@ -0,0 +1,39 @@ +"""Load a local Policy for the live peer listener from config.""" + +from __future__ import annotations + +from pathlib import Path + +from ca2a_runtime.cedar import CedarPolicy +from ca2a_runtime.config import Ca2aConfig +from ca2a_runtime.errors import ConfigError +from ca2a_runtime.policy import LocalPolicy, Policy + + +def load_policy(config: Ca2aConfig, *, config_dir: Path | None = None) -> Policy: + """Resolve the callee policy from ``local_policy`` or ``policy_bundle_path``. + + Cedar wins when ``policy_bundle_path`` is set. Otherwise ``local_policy`` + becomes a ``LocalPolicy`` allow set. At least one must be present for + ``ca2a start``. + """ + if config.policy_bundle_path: + path = Path(config.policy_bundle_path) + if not path.is_absolute() and config_dir is not None: + path = config_dir / path + if not path.is_file(): + raise ConfigError(f"policy_bundle_path not found: {path}") + try: + text = path.read_text(encoding="utf-8") + except OSError as exc: + raise ConfigError(f"cannot read policy_bundle_path: {path}", detail=str(exc)) from exc + if not text.strip(): + raise ConfigError(f"policy_bundle_path is empty: {path}") + return CedarPolicy(text) + + if config.local_policy is not None: + return LocalPolicy(allow=config.local_policy) + + raise ConfigError( + "ca2a start requires local_policy or policy_bundle_path in the config", + ) diff --git a/src/ca2a_runtime/server.py b/src/ca2a_runtime/server.py new file mode 100644 index 0000000..47b015e --- /dev/null +++ b/src/ca2a_runtime/server.py @@ -0,0 +1,231 @@ +"""Live A2A JSON-RPC peer listener. + +Wires ``parse_peer_request`` into ``handle_peer_request`` on an inbound +``message/send`` call. This is the Tier 2 live-serving slice: an HTTP listener +and ``ca2a start``. It does **not** perform an attestation handshake or bind +the sealed channel to a verified measurement; sealed open uses a configured +software/enclave key only. +""" + +from __future__ import annotations + +import base64 +from dataclasses import dataclass +from typing import Any + +from cryptography.hazmat.primitives.asymmetric.x25519 import X25519PrivateKey +from starlette.applications import Starlette +from starlette.requests import Request +from starlette.responses import JSONResponse, Response +from starlette.routing import Route + +from ca2a_runtime.config import Ca2aConfig +from ca2a_runtime.errors import CA2AError, ConfigError, TransportError +from ca2a_runtime.peer import PeerResult, handle_peer_request +from ca2a_runtime.policy import Policy +from ca2a_runtime.transport import EXTENSION_URI, parse_peer_request + +JSONRPC_PARSE_ERROR = -32700 +JSONRPC_INVALID_REQUEST = -32600 +JSONRPC_METHOD_NOT_FOUND = -32601 +JSONRPC_INVALID_PARAMS = -32602 +JSONRPC_APPLICATION_ERROR = -32000 + +SUPPORTED_METHODS = frozenset({"message/send", "SendMessage"}) + + +@dataclass(frozen=True) +class PeerRuntime: + """Runtime context for a live cA2A peer listener.""" + + config: Ca2aConfig + policy: Policy + enclave_private_key: X25519PrivateKey | None = None + + +def _b64url_encode(data: bytes) -> str: + return base64.urlsafe_b64encode(data).decode("ascii").rstrip("=") + + +def _result_payload(result: PeerResult) -> dict[str, Any]: + body: dict[str, Any] = { + "granted_capability": result.granted_capability, + "effective_scope": sorted(result.effective_scope), + "record": result.record.body(), + "record_hash": result.record.record_hash(), + } + if result.payload is not None: + body["payload"] = _b64url_encode(result.payload) + return {"ca2a": body} + + +def _jsonrpc_error( + request_id: Any, + *, + code: int, + message: str, + data: dict[str, Any] | None = None, + http_status: int = 200, +) -> JSONResponse: + error: dict[str, Any] = {"code": code, "message": message} + if data is not None: + error["data"] = data + return JSONResponse( + {"jsonrpc": "2.0", "id": request_id, "error": error}, + status_code=http_status, + ) + + +def _jsonrpc_result(request_id: Any, result: dict[str, Any]) -> JSONResponse: + return JSONResponse({"jsonrpc": "2.0", "id": request_id, "result": result}) + + +def handle_jsonrpc(runtime: PeerRuntime, body: dict[str, Any]) -> JSONResponse: + """Handle one JSON-RPC request body against the peer runtime.""" + request_id = body.get("id") + if body.get("jsonrpc") != "2.0": + return _jsonrpc_error( + request_id, + code=JSONRPC_INVALID_REQUEST, + message="jsonrpc must be '2.0'", + ) + + method = body.get("method") + if not isinstance(method, str): + return _jsonrpc_error( + request_id, + code=JSONRPC_INVALID_REQUEST, + message="method must be a string", + ) + if method not in SUPPORTED_METHODS: + return _jsonrpc_error( + request_id, + code=JSONRPC_METHOD_NOT_FOUND, + message=f"method not found: {method}", + ) + + params = body.get("params") + if params is None: + params = {} + if not isinstance(params, dict): + return _jsonrpc_error( + request_id, + code=JSONRPC_INVALID_PARAMS, + message="params must be an object", + ) + + try: + peer_request = parse_peer_request(body) + except TransportError as exc: + return _jsonrpc_error( + request_id, + code=JSONRPC_APPLICATION_ERROR, + message=str(exc), + data={"ca2a_code": exc.code, "detail": exc.detail}, + http_status=exc.http_status, + ) + + if peer_request is None: + # Ordinary A2A: no cA2A trust envelope. Do not invent a partial trust state. + return _jsonrpc_result( + request_id, + { + "ca2a": None, + "note": "ordinary A2A input; no cA2A extension metadata present", + }, + ) + + try: + result = handle_peer_request( + peer_request, + policy=runtime.policy, + enclave_private_key=runtime.enclave_private_key, + max_depth=runtime.config.max_delegation_depth, + ) + except CA2AError as exc: + return _jsonrpc_error( + request_id, + code=JSONRPC_APPLICATION_ERROR, + message=str(exc), + data={"ca2a_code": exc.code, "detail": exc.detail}, + http_status=exc.http_status, + ) + + return _jsonrpc_result(request_id, _result_payload(result)) + + +def create_app(runtime: PeerRuntime) -> Starlette: + """Build the Starlette ASGI app for a peer runtime.""" + + async def rpc(request: Request) -> Response: + try: + body = await request.json() + except Exception: + return _jsonrpc_error( + None, + code=JSONRPC_PARSE_ERROR, + message="parse error", + http_status=400, + ) + if not isinstance(body, dict): + return _jsonrpc_error( + None, + code=JSONRPC_INVALID_REQUEST, + message="request body must be a JSON object", + http_status=400, + ) + return handle_jsonrpc(runtime, body) + + async def health(_: Request) -> JSONResponse: + return JSONResponse( + { + "status": "ok", + "extension": EXTENSION_URI, + "provider": runtime.config.provider, + "enforcement_mode": runtime.config.enforcement_mode, + } + ) + + return Starlette( + routes=[ + Route("/", rpc, methods=["POST"]), + Route("/rpc", rpc, methods=["POST"]), + Route("/health", health, methods=["GET"]), + ] + ) + + +def parse_listen_addr(listen_addr: str) -> tuple[str, int]: + """Split ``host:port`` into a host string and port int.""" + host, sep, port_str = listen_addr.rpartition(":") + if not sep or not port_str: + raise ConfigError( + f"listen_addr must be host:port, got {listen_addr!r}", + ) + try: + port = int(port_str) + except ValueError as exc: + raise ConfigError( + f"listen_addr port must be an integer, got {port_str!r}", + ) from exc + if not (1 <= port <= 65535): + raise ConfigError(f"listen_addr port out of range: {port}") + if not host: + raise ConfigError( + f"listen_addr must include an explicit host (e.g. 127.0.0.1:{port}), " + f"got {listen_addr!r}", + ) + return host, port + + +def load_enclave_private_key(hex_key: str | None) -> X25519PrivateKey | None: + """Load an X25519 private key from raw hex, or return None.""" + if hex_key is None or hex_key == "": + return None + try: + raw = bytes.fromhex(hex_key.strip()) + if len(raw) != 32: + raise ValueError(f"expected 32 bytes, got {len(raw)}") + return X25519PrivateKey.from_private_bytes(raw) + except ValueError as exc: + raise ConfigError("invalid enclave_private_key_hex", detail=str(exc)) from exc diff --git a/tests/unit/test_config.py b/tests/unit/test_config.py index df40391..5def608 100644 --- a/tests/unit/test_config.py +++ b/tests/unit/test_config.py @@ -57,3 +57,20 @@ def test_load_invalid_yaml(tmp_path: Path) -> None: p.write_text("attestation: [unclosed\n") with pytest.raises(ConfigError): Ca2aConfig.load(p) + + +def test_local_policy_from_dict() -> None: + cfg = Ca2aConfig.from_dict({"local_policy": ["read", "write"]}) + assert cfg.local_policy == frozenset({"read", "write"}) + + +def test_bad_local_policy_rejected() -> None: + with pytest.raises(ConfigError): + Ca2aConfig.from_dict({"local_policy": "read"}) + with pytest.raises(ConfigError): + Ca2aConfig.from_dict({"local_policy": [""]}) + + +def test_enclave_key_hex_from_dict() -> None: + cfg = Ca2aConfig.from_dict({"enclave_private_key_hex": "ab" * 32}) + assert cfg.enclave_private_key_hex == "ab" * 32 diff --git a/tests/unit/test_policy_loader.py b/tests/unit/test_policy_loader.py new file mode 100644 index 0000000..ab27c3b --- /dev/null +++ b/tests/unit/test_policy_loader.py @@ -0,0 +1,35 @@ +"""Tests for policy loading used by ca2a start.""" + +from __future__ import annotations + +from pathlib import Path + +import pytest + +from ca2a_runtime.cedar import CedarPolicy +from ca2a_runtime.config import Ca2aConfig +from ca2a_runtime.errors import ConfigError +from ca2a_runtime.policy import LocalPolicy +from ca2a_runtime.policy_loader import load_policy + + +def test_load_local_policy() -> None: + cfg = Ca2aConfig(local_policy=frozenset({"read"})) + policy = load_policy(cfg) + assert isinstance(policy, LocalPolicy) + assert policy.allow == frozenset({"read"}) + + +def test_load_cedar_bundle(tmp_path: Path) -> None: + cedar = tmp_path / "policy.cedar" + cedar.write_text('permit(principal, action == Action::"read", resource);') + cfg = Ca2aConfig(policy_bundle_path=str(cedar.name)) + policy = load_policy(cfg, config_dir=tmp_path) + assert isinstance(policy, CedarPolicy) + assert policy.permits("read") + assert not policy.permits("write") + + +def test_missing_policy_rejected() -> None: + with pytest.raises(ConfigError): + load_policy(Ca2aConfig()) diff --git a/tests/unit/test_server.py b/tests/unit/test_server.py new file mode 100644 index 0000000..0c8608a --- /dev/null +++ b/tests/unit/test_server.py @@ -0,0 +1,200 @@ +"""Unit tests for the live A2A JSON-RPC peer listener.""" + +from __future__ import annotations + +import base64 +import json + +import httpx +import pytest +from starlette.testclient import TestClient + +from ca2a_runtime.channel import SealedChannel, generate_channel_keypair +from ca2a_runtime.config import Ca2aConfig +from ca2a_runtime.errors import ConfigError +from ca2a_runtime.policy import LocalPolicy +from ca2a_runtime.server import ( + PeerRuntime, + create_app, + handle_jsonrpc, + load_enclave_private_key, + parse_listen_addr, +) +from ca2a_runtime.transport import ( + KEY_DELEGATION_CHAIN, + KEY_PARENT_RECORD_HASH, + KEY_RECORD_ID, + KEY_REQUESTED_CAPABILITY, + KEY_SEALED_PAYLOAD, +) +from tests.unit.conftest import build_chain + + +def _cred_dicts(scopes: list[frozenset[str]]) -> list[dict]: + chain = build_chain(scopes) + return [{**c.body(), "signature": c.signature} for c in chain] + + +def _envelope( + *, + capability: str = "read", + record_id: str = "rec-live", + sealed_b64: str | None = None, + include_ca2a: bool = True, +) -> dict: + message: dict = { + "messageId": "msg-1", + "role": "user", + "parts": [{"text": "task"}], + "metadata": {}, + } + if include_ca2a: + meta = { + KEY_DELEGATION_CHAIN: _cred_dicts( + [frozenset({"read", "write"}), frozenset({"read"})] + ), + KEY_REQUESTED_CAPABILITY: capability, + KEY_RECORD_ID: record_id, + KEY_PARENT_RECORD_HASH: None, + } + if sealed_b64 is not None: + meta[KEY_SEALED_PAYLOAD] = sealed_b64 + message["metadata"] = meta + else: + message["metadata"] = {"https://example.com/ext/other/v1/note": "keep"} + + return { + "jsonrpc": "2.0", + "id": "1", + "method": "message/send", + "params": {"message": message}, + } + + +def _runtime( + *, + allow: list[str] | None = None, + enclave_key_hex: str | None = None, +) -> PeerRuntime: + cfg = Ca2aConfig( + provider="software-only", + enforcement_mode="enforcing", + local_policy=frozenset(allow or ["read", "write"]), + listen_addr="127.0.0.1:8443", + enclave_private_key_hex=enclave_key_hex, + ) + return PeerRuntime( + config=cfg, + policy=LocalPolicy(allow=cfg.local_policy or frozenset()), + enclave_private_key=load_enclave_private_key(enclave_key_hex), + ) + + +def test_parse_listen_addr() -> None: + assert parse_listen_addr("127.0.0.1:9443") == ("127.0.0.1", 9443) + assert parse_listen_addr("0.0.0.0:8443") == ("0.0.0.0", 8443) + + +def test_parse_listen_addr_rejects_bad() -> None: + with pytest.raises(ConfigError): + parse_listen_addr("no-port") + with pytest.raises(ConfigError): + parse_listen_addr("host:abc") + with pytest.raises(ConfigError): + parse_listen_addr(":8443") + + +def test_live_inbound_grants_and_emits_record() -> None: + runtime = _runtime() + resp = handle_jsonrpc(runtime, _envelope()) + assert resp.status_code == 200 + data = json.loads(resp.body) + assert "result" in data + ca2a = data["result"]["ca2a"] + assert ca2a["granted_capability"] == "read" + assert ca2a["effective_scope"] == ["read"] + assert ca2a["record"]["record_id"] == "rec-live" + assert isinstance(ca2a["record_hash"], str) and len(ca2a["record_hash"]) == 64 + + +def test_live_inbound_denies_out_of_scope() -> None: + runtime = _runtime(allow=["audit"]) + resp = handle_jsonrpc(runtime, _envelope(capability="read")) + data = json.loads(resp.body) + assert "error" in data + assert data["error"]["data"]["ca2a_code"] == "SCOPE_NOT_PERMITTED" + assert resp.status_code == 403 + + +def test_ordinary_a2a_returns_null_ca2a() -> None: + runtime = _runtime() + resp = handle_jsonrpc(runtime, _envelope(include_ca2a=False)) + data = json.loads(resp.body) + assert data["result"]["ca2a"] is None + + +def test_malformed_ca2a_metadata_fails_closed() -> None: + runtime = _runtime() + envelope = _envelope() + envelope["params"]["message"]["metadata"][KEY_DELEGATION_CHAIN] = "not-a-list" + resp = handle_jsonrpc(runtime, envelope) + data = json.loads(resp.body) + assert data["error"]["data"]["ca2a_code"] == "TRANSPORT_ERROR" + assert resp.status_code == 400 + + +def test_sealed_payload_opens_with_configured_key() -> None: + priv, pub_hex = generate_channel_keypair() + sealed = SealedChannel(pub_hex).seal(b"secret-task") + sealed_b64 = base64.urlsafe_b64encode(sealed).decode("ascii").rstrip("=") + key_hex = priv.private_bytes_raw().hex() + runtime = _runtime(enclave_key_hex=key_hex) + + resp = handle_jsonrpc(runtime, _envelope(sealed_b64=sealed_b64)) + data = json.loads(resp.body) + assert data["result"]["ca2a"]["payload"] == base64.urlsafe_b64encode(b"secret-task").decode( + "ascii" + ).rstrip("=") + + +def test_sealed_payload_without_key_fails_closed() -> None: + priv, pub_hex = generate_channel_keypair() + sealed = SealedChannel(pub_hex).seal(b"secret-task") + sealed_b64 = base64.urlsafe_b64encode(sealed).decode("ascii").rstrip("=") + runtime = _runtime() # no enclave key + + resp = handle_jsonrpc(runtime, _envelope(sealed_b64=sealed_b64)) + data = json.loads(resp.body) + assert data["error"]["data"]["ca2a_code"] == "SEALED_CHANNEL_ERROR" + + +def test_asgi_app_health_and_rpc() -> None: + app = create_app(_runtime()) + client = TestClient(app) + health = client.get("/health") + assert health.status_code == 200 + assert health.json()["status"] == "ok" + + rpc = client.post("/rpc", json=_envelope()) + assert rpc.status_code == 200 + assert rpc.json()["result"]["ca2a"]["granted_capability"] == "read" + + +@pytest.mark.asyncio +async def test_asgi_via_httpx() -> None: + app = create_app(_runtime()) + transport = httpx.ASGITransport(app=app) + async with httpx.AsyncClient(transport=transport, base_url="http://test") as client: + resp = await client.post("/", json=_envelope()) + assert resp.status_code == 200 + assert resp.json()["result"]["ca2a"]["granted_capability"] == "read" + + +def test_method_not_found() -> None: + runtime = _runtime() + resp = handle_jsonrpc( + runtime, + {"jsonrpc": "2.0", "id": "1", "method": "tasks/get", "params": {}}, + ) + data = json.loads(resp.body) + assert data["error"]["code"] == -32601