diff --git a/CHANGELOG.md b/CHANGELOG.md index 1e6c59b..5789d78 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -5,7 +5,15 @@ All notable changes to this project will be documented in this file. The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.1.0/), and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.html). -## [Unreleased] +## [Unreleased] + +### Security + +- Runtime authorization now requires the delegation chain's root issuer to be + present in the callee's `trusted_root_issuers`. Previously, any party could + mint a self-consistent root chain granting itself a locally allowed capability; + every signature and attenuation check passed because no local trust anchor was + consulted. `ca2a start` now refuses to launch without at least one pinned root. ### Security diff --git a/docs/configuration.md b/docs/configuration.md index a73f782..70f4dc2 100644 --- a/docs/configuration.md +++ b/docs/configuration.md @@ -14,6 +14,8 @@ attestation: max_delegation_depth: 8 # reject chains deeper than this listen_addr: "127.0.0.1:8443" +trusted_root_issuers: + - "" local_policy: ["read", "write"] # allow-set for scope intersection (or use Cedar below) # policy_bundle_path: policy.cedar @@ -27,6 +29,7 @@ local_policy: ["read", "write"] # allow-set for scope intersection (or use Ced | `attestation.enforcement_mode` | `enforcing` | Intended mode. The peer path always fails closed on cA2A denials today; advisory and 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` | `127.0.0.1:8443` | Address `ca2a start` binds. The host is never defaulted, so serving on every interface has to be written out. | +| `trusted_root_issuers` | none | Ed25519 public keys allowed to originate delegation chains. At least one is required by `ca2a start`; an internally valid chain from any other root is denied before policy evaluation. | | `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, resolved relative to the config file. When set, used instead of `local_policy`. | diff --git a/docs/spec/delegation-chain.md b/docs/spec/delegation-chain.md index 310cf4d..5a23b95 100644 --- a/docs/spec/delegation-chain.md +++ b/docs/spec/delegation-chain.md @@ -33,6 +33,14 @@ The signed bytes are the RFC 8785 (JSON Canonicalization Scheme) encoding of the | Each hop's depth is previous + 1, and at most `max_depth` | `BROKEN_DELEGATION_LINK` / `DELEGATION_DEPTH_EXCEEDED` | | Each hop's scope is a subset of its parent's scope | `SCOPE_ESCALATION` | | No `credential_id` repeats | `CREDENTIAL_REPLAY` | +| The root issuer is pinned by the callee for runtime authorization | `UNTRUSTED_DELEGATION_ROOT` | + +Signature validity establishes who issued a chain; it does not establish that +the issuer is trusted. A live callee therefore supplies its local +`trusted_root_issuers` set when verifying a request and fails closed when the +root is absent. Offline tooling may omit that set when it only needs to check a +chain's internal structure, but structural verification alone does not authorize +work. ## Attenuation is the whole point diff --git a/docs/spec/error-codes.md b/docs/spec/error-codes.md index 7aaab4e..5039a5f 100644 --- a/docs/spec/error-codes.md +++ b/docs/spec/error-codes.md @@ -10,7 +10,8 @@ An error also carries a human-readable message and an optional `detail`. The mes |---|---|---|---| | `CA2AError` | `CA2A_ERROR` | 500 | Base class for all cA2A runtime and verifier errors. Not raised directly; caught to handle any cA2A failure generically. | | `ConfigError` | `CONFIG_ERROR` | 500 | `Ca2aConfig` construction or `verify_chain_file` config load failed: unknown field, `max_delegation_depth` not a positive integer, missing config file, invalid YAML, or a non-mapping config root. | -| `InvalidCredential` | `INVALID_CREDENTIAL` | 400 | A `DelegationCredential` is malformed or its Ed25519 signature does not verify: unsigned credential, bad signature, malformed fields, or a chain document that is not a list or `{"chain": [...]}`, a missing chain file, or invalid JSON. | +| `InvalidCredential` | `INVALID_CREDENTIAL` | 400 | A `DelegationCredential` is malformed or its Ed25519 signature does not verify: unsigned credential, bad signature, malformed fields, or a chain document that is not a list or `{"chain": [...]}`, a missing chain file, or invalid JSON. | +| `UntrustedDelegationRoot` | `UNTRUSTED_DELEGATION_ROOT` | 403 | A chain is internally valid, but its root issuer is not pinned in the callee's `trusted_root_issuers`. Runtime authorization checks this before policy evaluation. | | `ScopeEscalation` | `SCOPE_ESCALATION` | 403 | A child grant claims authority its parent did not hold. Raised by `verify_chain` when a hop's scope is not a subset of its parent's scope. | | `BrokenDelegationLink` | `BROKEN_DELEGATION_LINK` | 409 | A hop does not chain to its stated parent, or continuity is broken: empty chain, a root credential that names a parent or has nonzero depth, a hop whose parent link or subject does not match the previous hop, or a hop depth that is not previous + 1. | | `DelegationDepthExceeded` | `DELEGATION_DEPTH_EXCEEDED` | 403 | A chain is longer than the configured `max_delegation_depth`. Raised by `verify_chain`. | diff --git a/docs/spec/threat-model.md b/docs/spec/threat-model.md index 7a19712..7b49f68 100644 --- a/docs/spec/threat-model.md +++ b/docs/spec/threat-model.md @@ -32,6 +32,7 @@ Out of adversary scope: breaking the underlying cryptographic primitives (Ed2551 | Operator or network reads the task payload | Sealing to the peer's measurement; the path sees ciphertext | | Credential replayed into another workflow | Unique `credential_id` and parent-link checks in chain verification | | A copied chain presented by a party it was not issued to | Holder binding: the presenter must answer a callee-issued challenge with a signature under the leaf `subject` key (profile P-4a). Appraising the caller does not cover this: an attested runtime is not a claim to anyone's delegated authority | +| Attacker mints a self-consistent chain from its own root | Callee pins locally trusted root issuer keys before policy evaluation | | Reparented or forged provenance | Linked TRACE records; the DAG is verified offline against the chain | ## Residual risks in this release diff --git a/examples/minimal/ca2a-config.yaml b/examples/minimal/ca2a-config.yaml index 05bee10..8760996 100644 --- a/examples/minimal/ca2a-config.yaml +++ b/examples/minimal/ca2a-config.yaml @@ -5,3 +5,7 @@ attestation: max_delegation_depth: 8 listen_addr: "127.0.0.1:8443" local_policy: ["read", "write"] +trusted_root_issuers: + # Pin authorities allowed to originate delegation chains. This example key + # matches examples/minimal/chain.json; replace it in a real deployment. + - "eda38c446da3db3eba68852ca9869260c36badd48b2cab16ec8d8faf607eb162" diff --git a/examples/rejection-with-proof/demo.py b/examples/rejection-with-proof/demo.py index 7c9a476..17179da 100644 --- a/examples/rejection-with-proof/demo.py +++ b/examples/rejection-with-proof/demo.py @@ -122,6 +122,7 @@ def main() -> int: policy=CALLEE_POLICY, record_id="rec-check", parent_record_hash=parent_hash, + trusted_root_issuers={chain[0].issuer}, ) print(f"ALLOW tool:search effective scope {sorted(granted.effective_scope)}") @@ -133,6 +134,7 @@ def main() -> int: policy=CALLEE_POLICY, record_id="rec-denied-purchase", parent_record_hash=parent_hash, + trusted_root_issuers={chain[0].issuer}, ) except ScopeNotPermitted as exc: denial = exc.record diff --git a/src/ca2a_runtime/bootstrap.py b/src/ca2a_runtime/bootstrap.py index 4eb74db..66da9e3 100644 --- a/src/ca2a_runtime/bootstrap.py +++ b/src/ca2a_runtime/bootstrap.py @@ -94,8 +94,15 @@ def select_provider(config: Ca2aConfig) -> BaseProvider: def build_peer_node(config: Ca2aConfig, *, config_dir: Path | None = None) -> PeerNode: """Build the node ``ca2a start`` serves: policy, provider, and depth limit.""" + policy = load_policy(config, config_dir=config_dir) + if not config.trusted_root_issuers: + raise ConfigError( + "ca2a start requires at least one trusted_root_issuer", + detail="pin the Ed25519 public key of each authority allowed to originate delegation chains", + ) return PeerNode( - load_policy(config, config_dir=config_dir), + policy, provider=select_provider(config), max_depth=config.max_delegation_depth, + trusted_root_issuers=config.trusted_root_issuers, ) diff --git a/src/ca2a_runtime/config.py b/src/ca2a_runtime/config.py index 4740090..28b2ced 100644 --- a/src/ca2a_runtime/config.py +++ b/src/ca2a_runtime/config.py @@ -59,6 +59,7 @@ class Ca2aConfig: policy_bundle_path: str | None = None local_policy: frozenset[str] | None = None listen_addr: str = DEFAULT_LISTEN_ADDR + trusted_root_issuers: frozenset[str] = frozenset() def listen_host_port(self) -> tuple[str, int]: """Return ``listen_addr`` split into the host and port to bind.""" @@ -101,6 +102,12 @@ def from_dict(cls, data: dict[str, Any]) -> Ca2aConfig: listen_addr = data.get("listen_addr", DEFAULT_LISTEN_ADDR) split_listen_addr(listen_addr) + raw_roots = data.get("trusted_root_issuers", []) + if not isinstance(raw_roots, list) or not all( + isinstance(item, str) and item for item in raw_roots + ): + raise ConfigError("trusted_root_issuers must be a list of non-empty public-key strings") + return cls( provider=provider, enforcement_mode=enforcement, @@ -108,6 +115,7 @@ def from_dict(cls, data: dict[str, Any]) -> Ca2aConfig: policy_bundle_path=bundle, local_policy=local_policy, listen_addr=listen_addr, + trusted_root_issuers=frozenset(raw_roots), ) @classmethod diff --git a/src/ca2a_runtime/delegation/credential.py b/src/ca2a_runtime/delegation/credential.py index d221d4a..33faf30 100644 --- a/src/ca2a_runtime/delegation/credential.py +++ b/src/ca2a_runtime/delegation/credential.py @@ -18,6 +18,7 @@ from __future__ import annotations +from collections.abc import Collection from dataclasses import dataclass from typing import Any @@ -34,6 +35,7 @@ DelegationDepthExceeded, InvalidCredential, ScopeEscalation, + UntrustedDelegationRoot, ) @@ -124,7 +126,12 @@ def from_dict(cls, data: dict[str, Any]) -> DelegationCredential: raise InvalidCredential("malformed credential", detail=str(exc)) from exc -def verify_chain(chain: list[DelegationCredential], *, max_depth: int = 8) -> None: +def verify_chain( + chain: list[DelegationCredential], + *, + max_depth: int = 8, + trusted_root_issuers: Collection[str] | None = None, +) -> None: """Verify a root-to-leaf delegation chain, raising on the first violation. A well-formed chain of length N delegates from the root issuer down to the @@ -134,6 +141,15 @@ def verify_chain(chain: list[DelegationCredential], *, max_depth: int = 8) -> No if not chain: raise BrokenDelegationLink("empty delegation chain") + # ``None`` deliberately means structural/offline verification only. Runtime + # authorization always supplies its local trust set, including an empty set, + # so a self-consistent chain minted by an attacker cannot authorize a call. + if trusted_root_issuers is not None and chain[0].issuer not in trusted_root_issuers: + raise UntrustedDelegationRoot( + "delegation root issuer is not trusted by this peer", + detail=f"root_issuer={chain[0].issuer}", + ) + seen_ids: set[str] = set() prev: DelegationCredential | None = None diff --git a/src/ca2a_runtime/errors.py b/src/ca2a_runtime/errors.py index 5084ce8..6e4d0b4 100644 --- a/src/ca2a_runtime/errors.py +++ b/src/ca2a_runtime/errors.py @@ -26,6 +26,13 @@ class InvalidCredential(CA2AError): http_status = 400 +class UntrustedDelegationRoot(CA2AError): + """The chain is validly signed but its root issuer is not trusted locally.""" + + code = "UNTRUSTED_DELEGATION_ROOT" + http_status = 403 + + class ScopeEscalation(CA2AError): """A child grant claims authority its parent did not hold.""" diff --git a/src/ca2a_runtime/node.py b/src/ca2a_runtime/node.py index 24c4795..0359eaa 100644 --- a/src/ca2a_runtime/node.py +++ b/src/ca2a_runtime/node.py @@ -12,6 +12,7 @@ from __future__ import annotations +from collections.abc import Collection from typing import Any from ca2a_runtime.attestation import ChannelOffer, Verifier, attest_channel @@ -61,6 +62,7 @@ def __init__( challenge_ttl_seconds: int = DEFAULT_TTL_SECONDS, require_holder_proof: bool = True, seen_proofs: ProofReplayCache | None | _Unset = _UNSET, + trusted_root_issuers: Collection[str] = (), ) -> None: if require_caller_attestation not in REQUIREMENT_VALUES: raise ConfigError( @@ -95,6 +97,7 @@ def __init__( self.seen_proofs = ProofReplayCache(ttl_seconds=challenge_ttl_seconds) else: self.seen_proofs = seen_proofs + self.trusted_root_issuers = frozenset(trusted_root_issuers) self._private_key, self.channel_public_key = generate_channel_keypair() self._challenge_secret = generate_secret() @@ -124,4 +127,5 @@ def handle(self, message: dict[str, Any]) -> PeerResult: audience=self.channel_public_key, require_holder_proof=self.require_holder_proof, seen_proofs=self.seen_proofs, + trusted_root_issuers=self.trusted_root_issuers, ) diff --git a/src/ca2a_runtime/peer.py b/src/ca2a_runtime/peer.py index 29acdaa..60309ab 100644 --- a/src/ca2a_runtime/peer.py +++ b/src/ca2a_runtime/peer.py @@ -36,6 +36,7 @@ from __future__ import annotations +from collections.abc import Collection from dataclasses import dataclass from cryptography.hazmat.primitives.asymmetric.x25519 import X25519PrivateKey @@ -85,13 +86,17 @@ def effective_scope( - chain: list[DelegationCredential], policy: Policy, *, max_depth: int = 8 + chain: list[DelegationCredential], + policy: Policy, + *, + max_depth: int = 8, + trusted_root_issuers: Collection[str] = (), ) -> frozenset[str]: """Verify the chain and return the effective scope (delegated ∩ local policy). Raises the relevant CA2AError if the chain does not verify. """ - verify_chain(chain, max_depth=max_depth) + verify_chain(chain, max_depth=max_depth, trusted_root_issuers=trusted_root_issuers) return policy.intersect(chain[-1].scope) @@ -113,6 +118,7 @@ def enforce_peer_call( parent_record_hash: str | None = None, max_depth: int = 8, caller_attestation: str = CALLER_NOT_OFFERED, + trusted_root_issuers: Collection[str] = (), ) -> PeerDecision: """Verify, intersect with local policy, enforce, and emit a provenance record. @@ -125,7 +131,12 @@ def enforce_peer_call( :func:`handle_peer_request` does exactly that. The default is the honest value for a path that appraised nothing. """ - effective = effective_scope(chain, policy, max_depth=max_depth) + effective = effective_scope( + chain, + policy, + max_depth=max_depth, + trusted_root_issuers=trusted_root_issuers, + ) return decide_capability( chain, requested_capability, @@ -370,6 +381,7 @@ def handle_peer_request( audience: str | None = None, require_holder_proof: bool = True, seen_proofs: ProofReplayCache | None = None, + trusted_root_issuers: Collection[str] = (), ) -> PeerResult: """Run the full inbound pipeline for a parsed peer request. @@ -399,7 +411,14 @@ def handle_peer_request( there is no live caller to challenge, and must not be used on a live peer path. """ - verify_chain(request.chain, max_depth=max_depth) + # The trust set is supplied here as well as in the scope intersection below, + # so an untrusted root is refused before the caller is challenged for a proof + # about a credential this peer was never going to honour. + verify_chain( + request.chain, + max_depth=max_depth, + trusted_root_issuers=trusted_root_issuers, + ) if require_holder_proof: # Before the scope intersection, so an unauthenticated caller never # reaches authorization and never elicits a denial record. @@ -410,7 +429,12 @@ def handle_peer_request( seen_proofs=seen_proofs, ) - effective = effective_scope(request.chain, policy, max_depth=max_depth) + effective = effective_scope( + request.chain, + policy, + max_depth=max_depth, + trusted_root_issuers=trusted_root_issuers, + ) caller_attestation = appraise_caller_runtime( request, diff --git a/tests/conformance/test_profile_conformance.py b/tests/conformance/test_profile_conformance.py index 91c47cc..a83c8b0 100644 --- a/tests/conformance/test_profile_conformance.py +++ b/tests/conformance/test_profile_conformance.py @@ -32,7 +32,9 @@ ScopeNotPermitted, SealedChannelError, ) -from ca2a_runtime.peer import REQUIRE_ANY, PeerRequest, effective_scope, handle_peer_request +from ca2a_runtime.peer import REQUIRE_ANY, PeerRequest +from ca2a_runtime.peer import effective_scope as _effective_scope +from ca2a_runtime.peer import handle_peer_request as _handle_peer_request from ca2a_runtime.policy import LocalPolicy from ca2a_runtime.provenance import DelegationRecord, cross_check_chain, record_for, verify_dag from ca2a_runtime.tee.base import AttestationReport @@ -88,6 +90,14 @@ def _narrowing(): return build_chain([frozenset({"read", "write", "admin"}), frozenset({"read", "write"})]) +def effective_scope(chain, policy): + return _effective_scope(chain, policy, trusted_root_issuers={chain[0].issuer}) + + +def handle_peer_request(request, **kwargs): + return _handle_peer_request(request, trusted_root_issuers={request.chain[0].issuer}, **kwargs) + + def _deep3(): return build_chain([frozenset({"a", "b", "c"}), frozenset({"a", "b"}), frozenset({"a"})]) diff --git a/tests/unit/test_a2a_sdk_bridge.py b/tests/unit/test_a2a_sdk_bridge.py index 7507561..5ca952f 100644 --- a/tests/unit/test_a2a_sdk_bridge.py +++ b/tests/unit/test_a2a_sdk_bridge.py @@ -72,14 +72,22 @@ def _chain(hops: int = 1) -> list[DelegationCredential]: return _chain_with_keys(hops)[0] -def _request(*, node: PeerNode | None = None, **kwargs) -> PeerRequest: +def _request( + *, node: PeerNode | None = None, leaf_key: Ed25519PrivateKey | None = None, **kwargs +) -> PeerRequest: """A request for the bridge round trip. Pass ``node`` when the request goes on to :meth:`PeerNode.handle`, which requires holder binding as it ships: the chain is then built with its leaf key retained so a proof can be signed against a challenge that node issued. + + Pass ``chain`` and ``leaf_key`` together when the caller needs the chain + before the node exists, which it does whenever the node pins that chain's root + as its trusted issuer. The generated pair cannot serve both, since the trust + set has to be known at construction and the proof has to be signed afterwards. """ - chain, leaf_key = _chain_with_keys() + chain, generated_key = _chain_with_keys() + signing_key = leaf_key if leaf_key is not None else generated_key base: dict = { "chain": chain, "requested_capability": "read", @@ -89,7 +97,7 @@ def _request(*, node: PeerNode | None = None, **kwargs) -> PeerRequest: base.update(kwargs) if node is not None: base["holder_proof"] = build_holder_proof( - leaf_key, + signing_key, base["chain"][-1], audience=node.channel_public_key, challenge=node.issue_challenge(), @@ -270,8 +278,9 @@ def test_opted_in(values: list[str] | None, expected: bool) -> None: def test_an_sdk_message_drives_the_full_inbound_pipeline() -> None: """What an adopter actually gets: enforcement from an SDK message.""" - node = PeerNode(LocalPolicy.of({"read"})) - request = _request(node=node) + chain, leaf_key = _chain_with_keys() + node = PeerNode(LocalPolicy.of({"read"}), trusted_root_issuers={chain[0].issuer}) + request = _request(node=node, chain=chain, leaf_key=leaf_key) message = a2a_sdk.attach_to_sdk_message(Message(message_id="m1"), request) parsed = a2a_sdk.parse_sdk_message(message) @@ -283,7 +292,12 @@ def test_an_sdk_message_drives_the_full_inbound_pipeline() -> None: def test_mutual_attestation_works_over_the_sdk_bridge() -> None: """The newest part of the profile must reach SDK adopters too.""" - node = PeerNode(LocalPolicy.of({"read"}), require_caller_attestation=REQUIRE_ANY) + chain, leaf_key = _chain_with_keys() + node = PeerNode( + LocalPolicy.of({"read"}), + require_caller_attestation=REQUIRE_ANY, + trusted_root_issuers={chain[0].issuer}, + ) challenge = node.issue_challenge() offer = ChannelOffer( channel_public_key="k" * 43, @@ -295,7 +309,8 @@ def test_mutual_attestation_works_over_the_sdk_bridge() -> None: ), ) message = a2a_sdk.attach_to_sdk_message( - Message(message_id="m1"), _request(caller_offer=offer, node=node) + Message(message_id="m1"), + _request(caller_offer=offer, node=node, chain=chain, leaf_key=leaf_key), ) result = node.handle({"metadata": a2a_sdk.metadata_from_sdk_message(message)}) assert result.caller_attestation == "software-only" diff --git a/tests/unit/test_bootstrap.py b/tests/unit/test_bootstrap.py index d96ab41..6ac5263 100644 --- a/tests/unit/test_bootstrap.py +++ b/tests/unit/test_bootstrap.py @@ -99,7 +99,7 @@ def _delegation_chain() -> tuple[list[DelegationCredential], Ed25519PrivateKey]: return [cred], subject_priv -def _write_config(tmp_path: Path) -> Path: +def _write_config(tmp_path: Path, root_issuer: str = "test-root") -> Path: path = tmp_path / "ca2a-config.yaml" path.write_text( "attestation:\n" @@ -108,7 +108,8 @@ def _write_config(tmp_path: Path) -> Path: "max_delegation_depth: 3\n" "local_policy:\n" " - read\n" - "listen_addr: 127.0.0.1:8443\n", + "listen_addr: 127.0.0.1:8443\n" + f"trusted_root_issuers:\n - {root_issuer}\n", encoding="utf-8", ) return path @@ -121,10 +122,18 @@ def test_build_peer_node_carries_config(tmp_path: Path) -> None: assert node.policy.allow == frozenset({"read"}) assert isinstance(node.provider, SoftwareProvider) assert node.max_depth == 3 + assert node.trusted_root_issuers == frozenset({"test-root"}) + + +def test_build_peer_node_refuses_missing_trust_anchors() -> None: + cfg = Ca2aConfig(provider="software-only", local_policy=frozenset({"read"})) + with pytest.raises(ConfigError, match="trusted_root_issuer"): + build_peer_node(cfg) def test_config_built_node_serves_a_live_call(tmp_path: Path) -> None: - cfg = Ca2aConfig.load(_write_config(tmp_path)) + chain, leaf_key = _delegation_chain() + cfg = Ca2aConfig.load(_write_config(tmp_path, chain[0].issuer)) host, _ = cfg.listen_host_port() # Port 0 rather than the configured one: the test needs a free port, and the # host is what the config contributes here. @@ -133,8 +142,6 @@ def test_config_built_node_serves_a_live_call(tmp_path: Path) -> None: thread.start() try: base = f"http://{host}:{srv.server_address[1]}" - chain, leaf_key = _delegation_chain() - body = client.send_task( base, chain, "read", "r0", holder_key=leaf_key, payload=b"from a config file" ) diff --git a/tests/unit/test_cedar.py b/tests/unit/test_cedar.py index 44c07be..93966ee 100644 --- a/tests/unit/test_cedar.py +++ b/tests/unit/test_cedar.py @@ -8,7 +8,8 @@ from ca2a_runtime.cedar import CedarPolicy from ca2a_runtime.errors import ScopeNotPermitted -from ca2a_runtime.peer import effective_scope, enforce_peer_call +from ca2a_runtime.peer import effective_scope as _effective_scope +from ca2a_runtime.peer import enforce_peer_call as _enforce_peer_call from ca2a_runtime.policy import Policy from tests.unit.conftest import build_chain @@ -25,6 +26,14 @@ def _chain(): return build_chain([frozenset({"read", "write", "admin"}), frozenset({"read", "write"})]) +def effective_scope(chain, policy): + return _effective_scope(chain, policy, trusted_root_issuers={chain[0].issuer}) + + +def enforce_peer_call(chain, capability, **kwargs): + return _enforce_peer_call(chain, capability, trusted_root_issuers={chain[0].issuer}, **kwargs) + + def test_cedar_policy_satisfies_protocol() -> None: assert isinstance(CedarPolicy(POLICIES), Policy) diff --git a/tests/unit/test_claim3_scope_policy.py b/tests/unit/test_claim3_scope_policy.py index 850d109..7739b49 100644 --- a/tests/unit/test_claim3_scope_policy.py +++ b/tests/unit/test_claim3_scope_policy.py @@ -11,7 +11,8 @@ import pytest from ca2a_runtime.errors import ScopeNotPermitted -from ca2a_runtime.peer import effective_scope, enforce_peer_call +from ca2a_runtime.peer import effective_scope as _effective_scope +from ca2a_runtime.peer import enforce_peer_call as _enforce_peer_call from ca2a_runtime.policy import LocalPolicy from tests.unit.conftest import build_chain @@ -20,6 +21,14 @@ def _chain(): return build_chain([frozenset({"read", "write", "admin"}), frozenset({"read", "write"})]) +def effective_scope(chain, policy): + return _effective_scope(chain, policy, trusted_root_issuers={chain[0].issuer}) + + +def enforce_peer_call(chain, capability, **kwargs): + return _enforce_peer_call(chain, capability, trusted_root_issuers={chain[0].issuer}, **kwargs) + + def test_effective_scope_is_delegation_intersect_local_policy() -> None: policy = LocalPolicy.of(["read", "audit"]) # leaf delegated {read, write}; local allows {read, audit}; intersection {read}. diff --git a/tests/unit/test_cli_start.py b/tests/unit/test_cli_start.py index 31a907a..26fe11d 100644 --- a/tests/unit/test_cli_start.py +++ b/tests/unit/test_cli_start.py @@ -46,6 +46,7 @@ def fake_serve(node: PeerNode, host: str = "127.0.0.1", port: int = 8443) -> _Fa config = _config( tmp_path, "attestation:\n provider: software-only\nlocal_policy:\n - read\n" + "trusted_root_issuers:\n - test-root\n" "listen_addr: 127.0.0.1:9443\n", ) @@ -66,7 +67,11 @@ def test_start_refuses_a_config_with_no_policy( def test_start_refuses_auto_provider_off_hardware( tmp_path: Path, capsys: pytest.CaptureFixture[str] ) -> None: - config = _config(tmp_path, "attestation:\n provider: auto\nlocal_policy:\n - read\n") + config = _config( + tmp_path, + "attestation:\n provider: auto\nlocal_policy:\n - read\n" + "trusted_root_issuers:\n - test-root\n", + ) assert cli.main(["start", "--config", config]) == 1 assert "no hardware attestation provider" in capsys.readouterr().err @@ -81,6 +86,7 @@ def refuse(node: PeerNode, host: str, port: int) -> _FakeServer: config = _config( tmp_path, "attestation:\n provider: software-only\nlocal_policy:\n - read\n" + "trusted_root_issuers:\n - test-root\n" "listen_addr: 127.0.0.1:9443\n", ) assert cli.main(["start", "--config", config]) == 1 @@ -93,6 +99,10 @@ def test_start_warns_that_software_mode_has_no_guarantee( tmp_path: Path, monkeypatch: pytest.MonkeyPatch, capsys: pytest.CaptureFixture[str] ) -> None: monkeypatch.setattr(server, "serve", lambda node, host, port: _FakeServer(node)) - config = _config(tmp_path, "attestation:\n provider: software-only\nlocal_policy:\n - read\n") + config = _config( + tmp_path, + "attestation:\n provider: software-only\nlocal_policy:\n - read\n" + "trusted_root_issuers:\n - test-root\n", + ) assert cli.main(["start", "--config", config]) == 0 assert 'assurance="none"' in capsys.readouterr().err diff --git a/tests/unit/test_config.py b/tests/unit/test_config.py index 56f0cca..e7174ad 100644 --- a/tests/unit/test_config.py +++ b/tests/unit/test_config.py @@ -16,6 +16,18 @@ def test_defaults_from_empty_dict() -> None: assert cfg.enforcement_mode == "enforcing" assert cfg.max_delegation_depth == 8 assert cfg.listen_host_port() == ("127.0.0.1", 8443) + assert cfg.trusted_root_issuers == frozenset() + + +def test_trusted_root_issuers_are_loaded() -> None: + cfg = Ca2aConfig.from_dict({"trusted_root_issuers": ["root-a", "root-b"]}) + assert cfg.trusted_root_issuers == frozenset({"root-a", "root-b"}) + + +@pytest.mark.parametrize("value", ["root-a", [""], [1], {}]) +def test_malformed_trusted_root_issuers_are_rejected(value: object) -> None: + with pytest.raises(ConfigError, match="trusted_root_issuers"): + Ca2aConfig.from_dict({"trusted_root_issuers": value}) def test_unknown_provider_rejected() -> None: diff --git a/tests/unit/test_holder_binding.py b/tests/unit/test_holder_binding.py index 09b08c6..85b76e2 100644 --- a/tests/unit/test_holder_binding.py +++ b/tests/unit/test_holder_binding.py @@ -28,7 +28,8 @@ from ca2a_runtime.delegation.holder import HolderProof, ProofReplayCache, proof_body from ca2a_runtime.errors import HolderProofInvalid from ca2a_runtime.node import PeerNode -from ca2a_runtime.peer import REQUIRE_ANY, PeerRequest, handle_peer_request +from ca2a_runtime.peer import REQUIRE_ANY, PeerRequest +from ca2a_runtime.peer import handle_peer_request as _handle_peer_request from ca2a_runtime.policy import LocalPolicy from ca2a_runtime.tee.base import AttestationReport from ca2a_runtime.tee.software import SoftwareProvider @@ -47,6 +48,18 @@ def _chain(): return build_chain_with_keys([frozenset({"read", "write"})]) +def handle_peer_request(request, **kwargs): + """The handler with this request's own root trusted. + + Each test here mints a fresh chain, so there is no single root to pin at + module level. Trusting the presented root is right for these tests and only + these: root trust is what ``test_delegation_roots.py`` covers, and holding it + fixed here keeps a holder-binding failure from being reported as an + untrusted-root one. + """ + return _handle_peer_request(request, trusted_root_issuers={request.chain[0].issuer}, **kwargs) + + def _handle(req, **kwargs): return handle_peer_request( req, policy=POLICY, audience=TEST_AUDIENCE, challenge_secret=TEST_SECRET, **kwargs @@ -344,9 +357,9 @@ def test_cache_is_bounded_and_says_what_that_costs() -> None: def test_a_node_remembers_proofs_by_default() -> None: """The default posture, over the transport, not just the handler.""" - node = PeerNode(POLICY) - assert node.seen_proofs is not None chain, keys = _chain() + node = PeerNode(POLICY, trusted_root_issuers={chain[0].issuer}) + assert node.seen_proofs is not None message = a2a_adapter.attach_ca2a_metadata( {}, PeerRequest( @@ -437,13 +450,12 @@ def test_a_malformed_holder_proof_on_the_wire_fails_closed() -> None: def test_replay_over_http_is_refused() -> None: """Bob calls legitimately; every route Mallory has is refused.""" - node = PeerNode(POLICY) + chain, keys = _chain() + node = PeerNode(POLICY, trusted_root_issuers={chain[0].issuer}) srv = server.serve(node, host="127.0.0.1", port=0) threading.Thread(target=srv.serve_forever, daemon=True).start() base = f"http://127.0.0.1:{srv.server_address[1]}" try: - chain, keys = _chain() - # Bob, who holds the leaf key, gets through. ok = client.send_task(base, chain, "write", "r0", holder_key=keys[-1]) assert ok["accepted"] is True @@ -495,8 +507,9 @@ def test_replay_over_http_is_refused() -> None: def test_a_node_can_opt_out_for_offline_replay() -> None: - node = PeerNode(POLICY, require_holder_proof=False) + chain, _keys = _chain() + node = PeerNode(POLICY, require_holder_proof=False, trusted_root_issuers={chain[0].issuer}) message = a2a_adapter.attach_ca2a_metadata( - {}, PeerRequest(chain=_chain()[0], requested_capability="write", record_id="r0") + {}, PeerRequest(chain=chain, requested_capability="write", record_id="r0") ) assert node.handle(message).granted_capability == "write" diff --git a/tests/unit/test_live_call.py b/tests/unit/test_live_call.py index 3e53ccc..710d1ff 100644 --- a/tests/unit/test_live_call.py +++ b/tests/unit/test_live_call.py @@ -98,7 +98,7 @@ def _message( def test_live_inbound_flow_software_mode() -> None: chain, leaf_key = _chain() - node = PeerNode(LocalPolicy.of({"read"})) + node = PeerNode(LocalPolicy.of({"read"}), trusted_root_issuers={chain[0].issuer}) nonce = "nonce-abc" peer = verify_offer(node.offer(nonce), expected_nonce=nonce) @@ -116,18 +116,22 @@ def test_live_inbound_flow_software_mode() -> None: def test_over_scope_capability_is_denied() -> None: - node = PeerNode(LocalPolicy.of({"read"})) # policy does not allow "write" + chain, leaf_key = _chain() + node = PeerNode( + LocalPolicy.of({"read"}), trusted_root_issuers={chain[0].issuer} + ) # policy does not allow "write" with pytest.raises(ScopeNotPermitted): - node.handle(_message(node, *_chain(), "write", "r1")) + node.handle(_message(node, chain, leaf_key, "write", "r1")) def test_tampered_sealed_payload_fails_closed() -> None: - node = PeerNode(LocalPolicy.of({"read"})) + chain, leaf_key = _chain() + node = PeerNode(LocalPolicy.of({"read"}), trusted_root_issuers={chain[0].issuer}) peer = verify_offer(node.offer("n"), expected_nonce="n") sealed = bytearray(seal_to_peer(peer, b"payload")) sealed[-1] ^= 0x01 with pytest.raises(SealedChannelError): - node.handle(_message(node, *_chain(), "read", "r2", sealed=bytes(sealed))) + node.handle(_message(node, chain, leaf_key, "read", "r2", sealed=bytes(sealed))) def test_stale_offer_nonce_is_rejected() -> None: @@ -146,8 +150,9 @@ def test_channel_offer_wire_roundtrip() -> None: def test_serialize_result_shape() -> None: - node = PeerNode(LocalPolicy.of({"read"})) - result = node.handle(_message(node, *_chain(), "read", "r0")) + chain, leaf_key = _chain() + node = PeerNode(LocalPolicy.of({"read"}), trusted_root_issuers={chain[0].issuer}) + result = node.handle(_message(node, chain, leaf_key, "read", "r0")) body = wire.serialize_peer_result(result) assert body["accepted"] is True assert body["granted_capability"] == "read" @@ -160,15 +165,14 @@ def test_serialize_result_shape() -> None: def test_http_live_call_end_to_end() -> None: - node = PeerNode(LocalPolicy.of({"read"})) + chain, leaf_key = _chain() + node = PeerNode(LocalPolicy.of({"read"}), trusted_root_issuers={chain[0].issuer}) srv = server.serve(node, host="127.0.0.1", port=0) port = srv.server_address[1] thread = threading.Thread(target=srv.serve_forever, daemon=True) thread.start() try: base = f"http://127.0.0.1:{port}" - chain, leaf_key = _chain() - body = client.send_task( base, chain, "read", "r0", holder_key=leaf_key, payload=b"hello over the wire" ) diff --git a/tests/unit/test_mutual_attestation.py b/tests/unit/test_mutual_attestation.py index e251960..abda4a3 100644 --- a/tests/unit/test_mutual_attestation.py +++ b/tests/unit/test_mutual_attestation.py @@ -22,7 +22,7 @@ from ca2a_runtime.delegation.credential import DelegationCredential, new_keypair from ca2a_runtime.delegation.holder import build_holder_proof from ca2a_runtime.errors import AttestationFailed, CA2AError, ConfigError, TransportError -from ca2a_runtime.node import PeerNode +from ca2a_runtime.node import PeerNode as _PeerNode from ca2a_runtime.peer import ( REQUIRE_ANY, REQUIRE_HARDWARE, @@ -45,20 +45,6 @@ POLICY = LocalPolicy.of({"read"}) -def handle_peer_request(*args: object, **kwargs: object) -> object: - """``handle_peer_request`` with holder binding off, for this file only. - - Appraisal ("what is the caller running") and holder binding ("is the caller - the delegate") are independent, and holder binding is covered in - ``test_holder_binding.py``. Several tests here deliberately hand the callee a - challenge secret the offer was not issued under; since holder binding is - checked first and reads the same secret, leaving it on would surface as a - holder-proof failure and mask the appraisal outcome each test asserts. - """ - kwargs.setdefault("require_holder_proof", False) - return _handle_peer_request(*args, **kwargs) # type: ignore[arg-type] - - def _build_chain() -> tuple[list[DelegationCredential], Ed25519PrivateKey]: root_priv, root_pub = new_keypair() subject_priv, subject_pub = new_keypair() @@ -82,6 +68,24 @@ def _chain() -> list[DelegationCredential]: return _CHAIN +def handle_peer_request(request, **kwargs): + """``handle_peer_request`` with this file's trust set, and holder binding off. + + Appraisal ("what is the caller running") and holder binding ("is the caller + the delegate") are independent, and holder binding is covered in + ``test_holder_binding.py``. Several tests here deliberately hand the callee a + challenge secret the offer was not issued under; since holder binding is + checked first and reads the same secret, leaving it on would surface as a + holder-proof failure and mask the appraisal outcome each test asserts. + """ + kwargs.setdefault("require_holder_proof", False) + return _handle_peer_request(request, trusted_root_issuers={request.chain[0].issuer}, **kwargs) + + +def PeerNode(policy, **kwargs): + return _PeerNode(policy, trusted_root_issuers={_CHAIN[0].issuer}, **kwargs) + + def _caller_offer(challenge: str, *, platform: str = "software-only") -> ChannelOffer: """A caller's own attested channel key, bound to ``challenge``.""" public_key = SoftwareProvider().attest("x", "y").public_key # a well-formed key string diff --git a/tests/unit/test_peer.py b/tests/unit/test_peer.py index 8a1e699..6d513ec 100644 --- a/tests/unit/test_peer.py +++ b/tests/unit/test_peer.py @@ -4,8 +4,10 @@ import pytest -from ca2a_runtime.errors import ScopeEscalation, ScopeNotPermitted -from ca2a_runtime.peer import PeerDecision, effective_scope, enforce_peer_call +from ca2a_runtime.errors import ScopeEscalation, ScopeNotPermitted, UntrustedDelegationRoot +from ca2a_runtime.peer import PeerDecision +from ca2a_runtime.peer import effective_scope as _effective_scope +from ca2a_runtime.peer import enforce_peer_call as _enforce_peer_call from ca2a_runtime.policy import LocalPolicy from ca2a_runtime.provenance import verify_dag from tests.unit.conftest import build_chain @@ -16,6 +18,25 @@ def _chain(): return build_chain([frozenset({"read", "write", "admin"}), frozenset({"read", "write"})]) +def effective_scope(chain, policy): + return _effective_scope(chain, policy, trusted_root_issuers={chain[0].issuer}) + + +def enforce_peer_call(chain, capability, **kwargs): + return _enforce_peer_call(chain, capability, trusted_root_issuers={chain[0].issuer}, **kwargs) + + +def test_runtime_rejects_a_valid_chain_from_an_untrusted_root() -> None: + attacker_chain = _chain() + with pytest.raises(UntrustedDelegationRoot): + _enforce_peer_call( + attacker_chain, + "read", + policy=LocalPolicy.of(["read"]), + record_id="attack", + ) + + def test_local_policy_helpers() -> None: p = LocalPolicy.of(["read", "audit"]) assert p.permits("read") and not p.permits("write") diff --git a/tests/unit/test_peer_request.py b/tests/unit/test_peer_request.py index 18d25d2..ecfb8cd 100644 --- a/tests/unit/test_peer_request.py +++ b/tests/unit/test_peer_request.py @@ -11,7 +11,8 @@ from ca2a_runtime.channel import SealedChannel, generate_channel_keypair from ca2a_runtime.errors import ScopeEscalation, ScopeNotPermitted, SealedChannelError -from ca2a_runtime.peer import PeerResult, handle_peer_request +from ca2a_runtime.peer import PeerResult +from ca2a_runtime.peer import handle_peer_request as _handle_peer_request from ca2a_runtime.policy import LocalPolicy from ca2a_runtime.provenance import verify_dag from tests.unit.conftest import ( @@ -35,6 +36,10 @@ def _handle(req, policy, **kwargs): ) +def handle_peer_request(request, **kwargs): + return _handle_peer_request(request, trusted_root_issuers={request.chain[0].issuer}, **kwargs) + + def test_handles_request_without_payload() -> None: chain, keys = _chain() result = _handle(