fix: require trusted delegation roots

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
 

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:
+  - "<root Ed25519 public key as raw hex>"
 
 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`. |
 

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
 

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`. |

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

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"

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

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,
     )

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,13 +102,20 @@ 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,
             max_delegation_depth=depth,
             policy_bundle_path=bundle,
             local_policy=local_policy,
             listen_addr=listen_addr,
+            trusted_root_issuers=frozenset(raw_roots),
         )
 
     @classmethod

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
 

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."""
 

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,
         )

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,

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"})])