From dcf0bb9c7473e464e6c5279276f08092f13c2cb0 Mon Sep 17 00:00:00 2001 From: Imran Siddique Date: Fri, 18 Sep 2026 11:42:40 -0700 Subject: [PATCH 1/2] feat(delegation): let an issuer revoke a grant before not_after A delegated grant could only lapse at not_after. Inside a still-valid window there was no way for the delegator to take it back. Add signed revocation statements. The issuer of a credential, or any issuer above it in the chain, signs a statement naming the credential by the SHA-256 of its canonical body, using the existing RFC 8785 helper. verify_chain takes an optional RevocationSnapshot and refuses a chain containing a revoked hop with CREDENTIAL_REVOKED, which also refuses every grant beneath it. A delegate cannot revoke upward, a statement from an unrelated key has no effect, nothing can un-revoke, and a snapshot containing a forged or unsigned statement is refused as a whole with INVALID_REVOCATION. Without a snapshot verification is unchanged and still offline, and verify_chain now returns a RevocationStatus that says revocation was not checked. ChainResult, PeerResult and the verify-chain / verify-dag output carry it. max_revocation_staleness fails closed with REVOCATION_STATUS_UNKNOWN when the snapshot is missing or older than the bound. PeerNode takes a revocation_source callable consulted on every call. Co-Authored-By: Claude Opus 5 (1M context) Signed-off-by: Imran Siddique --- README.md | 2 +- docs/quickstart.md | 2 +- docs/tutorials/verify-a-delegation-chain.md | 2 +- examples/cross-operator-delegation/README.md | 4 +- examples/rejection-with-proof/README.md | 2 +- src/ca2a_runtime/cli.py | 63 ++- src/ca2a_runtime/delegation/__init__.py | 12 + src/ca2a_runtime/delegation/credential.py | 95 +++- src/ca2a_runtime/delegation/revocation.py | 292 ++++++++++++ src/ca2a_runtime/errors.py | 37 ++ src/ca2a_runtime/node.py | 12 +- src/ca2a_runtime/peer.py | 36 +- src/ca2a_verify/__init__.py | 2 + src/ca2a_verify/verify.py | 46 +- tests/unit/test_docs_quickstart.py | 1 + tests/unit/test_revocation.py | 475 +++++++++++++++++++ 16 files changed, 1066 insertions(+), 17 deletions(-) create mode 100644 src/ca2a_runtime/delegation/revocation.py create mode 100644 tests/unit/test_revocation.py diff --git a/README.md b/README.md index 27cc18a..9d3c657 100644 --- a/README.md +++ b/README.md @@ -98,7 +98,7 @@ ca2a verify-dag --dag examples/rejection-with-proof/dag.json \ ```json {"verified": true, "records": 4, "outcome": "denied", "requested_capability": "tool:purchase", "effective_scope": ["tool:search"], - "cross_checked": true} + "cross_checked": true, "revocation": "not_checked"} ``` The callee's own policy permits `tool:purchase`. It is refused anyway, because diff --git a/docs/quickstart.md b/docs/quickstart.md index 31bcf6e..2c9c409 100644 --- a/docs/quickstart.md +++ b/docs/quickstart.md @@ -92,7 +92,7 @@ ca2a verify-chain --chain demo-chain.json --trusted-root-issuer "$(cat trusted-r Expected exit code: `0`. ```json -{"verified": true, "hops": 2, "leaf_scope": ["cap:read"]} +{"verified": true, "hops": 2, "leaf_scope": ["cap:read"], "revocation": "not_checked"} ``` In production, the relying party obtains trusted roots through its own approval process. Copying the issuer from an arbitrary incoming chain into the trust list would let that chain choose its own authority. diff --git a/docs/tutorials/verify-a-delegation-chain.md b/docs/tutorials/verify-a-delegation-chain.md index 4923e4a..ee811f9 100644 --- a/docs/tutorials/verify-a-delegation-chain.md +++ b/docs/tutorials/verify-a-delegation-chain.md @@ -55,7 +55,7 @@ In PowerShell: ca2a verify-chain --chain chain.json --trusted-root-issuer (Get-Content trusted-root.txt -Raw) ``` -Expect exit 0 and `{"verified": true, "hops": 3, "leaf_scope": ["cap:read"]}`. Substituting `tampered-chain.json` should exit nonzero with `INVALID_CREDENTIAL`. +Expect exit 0 and `{"verified": true, "hops": 3, "leaf_scope": ["cap:read"], "revocation": "not_checked"}`. Substituting `tampered-chain.json` should exit nonzero with `INVALID_CREDENTIAL`. ## What this establishes diff --git a/examples/cross-operator-delegation/README.md b/examples/cross-operator-delegation/README.md index eaf691b..05434c9 100644 --- a/examples/cross-operator-delegation/README.md +++ b/examples/cross-operator-delegation/README.md @@ -78,10 +78,10 @@ Cross-operator delegation example (offline; synthetic SEV-SNP vectors) wrote chain.json and dag.json; re-verifying via the CLI: $ ca2a verify-chain --chain .../chain.json --trusted-root-issuer - {"verified": true, "hops": 2, "leaf_scope": ["task:read", "task:write"]} + {"verified": true, "hops": 2, "leaf_scope": ["task:read", "task:write"], "revocation": "not_checked"} [11] ca2a verify-chain accepts chain.json: OK $ ca2a verify-dag --dag .../dag.json --chain .../chain.json --trusted-root-issuer - {"verified": true, "records": 2, "leaf_scope": ["task:read", "task:write"], "cross_checked": true} + {"verified": true, "records": 2, "leaf_scope": ["task:read", "task:write"], "cross_checked": true, "revocation": "not_checked"} [12] ca2a verify-dag accepts dag.json and cross-checks the chain: OK KEY RESULT: 12/12 ... diff --git a/examples/rejection-with-proof/README.md b/examples/rejection-with-proof/README.md index 8a0589b..5475088 100644 --- a/examples/rejection-with-proof/README.md +++ b/examples/rejection-with-proof/README.md @@ -39,7 +39,7 @@ ca2a verify-dag --dag examples/rejection-with-proof/dag.json \ "outcome": "denied", "requested_capability": "tool:purchase", "effective_scope": ["tool:search"], "denial_reason": "capability 'tool:purchase' is not in the effective scope", - "cross_checked": true} + "cross_checked": true, "revocation": "not_checked"} ``` ## Why the callee's policy is permissive here diff --git a/src/ca2a_runtime/cli.py b/src/ca2a_runtime/cli.py index 2a6edf3..e703b41 100644 --- a/src/ca2a_runtime/cli.py +++ b/src/ca2a_runtime/cli.py @@ -10,7 +10,12 @@ from ca2a_runtime import __version__ from ca2a_runtime.config import Ca2aConfig -from ca2a_runtime.delegation import DelegationCredential, verify_chain +from ca2a_runtime.delegation import ( + DelegationCredential, + RevocationSnapshot, + RevocationStatus, + verify_chain, +) from ca2a_runtime.errors import CA2AError, ConfigError, InvalidCredential, ProvenanceLinkBroken from ca2a_runtime.provenance import ( CALLER_NOT_OFFERED, @@ -18,7 +23,7 @@ cross_check_chain, verify_dag, ) -from ca2a_verify import verify_chain_file +from ca2a_verify import load_revocation_snapshot, verify_chain_file def _cmd_validate_config(args: argparse.Namespace) -> int: @@ -31,6 +36,20 @@ def _cmd_validate_config(args: argparse.Namespace) -> int: return 0 +def _revocation_fields(status: RevocationStatus) -> dict[str, Any]: + # Printed always, including "not_checked". A chain that verified offline + # with no revocation data may still have been revoked, and output that only + # mentioned revocation when it was checked would let a reader miss that. + out: dict[str, Any] = {"revocation": status.state} + if status.checked: + out["revocation_as_of"] = status.as_of + return out + + +def _load_revocations(args: argparse.Namespace) -> RevocationSnapshot | None: + return None if args.revocations is None else load_revocation_snapshot(args.revocations) + + def _cmd_verify_chain(args: argparse.Namespace) -> int: try: result = verify_chain_file( @@ -38,11 +57,17 @@ def _cmd_verify_chain(args: argparse.Namespace) -> int: trusted_root_issuers=args.trusted_root_issuer, max_depth=args.max_depth, at_time=args.at_time, + revocations=_load_revocations(args), + max_revocation_staleness=args.max_revocation_staleness, ) except CA2AError as exc: print(json.dumps({"verified": False, "code": exc.code, "error": str(exc)})) return 1 - print(json.dumps({"verified": True, "hops": result.hops, "leaf_scope": result.leaf_scope})) + out: dict[str, Any] = {"verified": True, "hops": result.hops, "leaf_scope": result.leaf_scope} + out.update( + _revocation_fields(RevocationStatus(result.revocation_checked, result.revocation_as_of)) + ) + print(json.dumps(out)) return 0 @@ -116,13 +141,16 @@ def _cmd_verify_dag(args: argparse.Namespace) -> int: try: records = verify_dag(_load_records(args.dag)) cross_checked = False + revocation: RevocationStatus | None = None if args.chain: chain = _load_chain(args.chain) - verify_chain( + revocation = verify_chain( chain, max_depth=args.max_depth, trusted_root_issuers=args.trusted_root_issuer, at_time=args.at_time, + revocations=_load_revocations(args), + max_revocation_staleness=args.max_revocation_staleness, ) cross_check_chain(records, chain) cross_checked = True @@ -147,6 +175,8 @@ def _cmd_verify_dag(args: argparse.Namespace) -> int: out["denial_reason"] = leaf.denial_reason if args.chain: out["cross_checked"] = cross_checked + if revocation is not None: + out.update(_revocation_fields(revocation)) print(json.dumps(out)) return 0 @@ -215,6 +245,21 @@ def _cmd_start(args: argparse.Namespace) -> int: return 0 +def _add_revocation_args(p: argparse.ArgumentParser, note: str = "") -> None: + p.add_argument( + "--revocations", + default=None, + help="Revocation snapshot JSON to check every hop against" + note, + ) + p.add_argument( + "--max-revocation-staleness", + type=int, + default=None, + help="Fail closed unless a revocation snapshot no older than this many " + "seconds (relative to the evaluation time) is supplied" + note, + ) + + 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__}") @@ -239,6 +284,7 @@ def build_parser() -> argparse.ArgumentParser: default=None, help="Unix time validity windows are evaluated at (default: now)", ) + _add_revocation_args(vch) vch.set_defaults(func=_cmd_verify_chain) vd = sub.add_parser("verify-dag", help="Verify a provenance DAG offline") @@ -260,6 +306,7 @@ def build_parser() -> argparse.ArgumentParser: default=None, help="Unix time validity windows are evaluated at (default: now)", ) + _add_revocation_args(vd, note=" (applies with --chain)") vd.set_defaults(func=_cmd_verify_dag) st = sub.add_parser( @@ -277,6 +324,14 @@ def main(argv: list[str] | None = None) -> int: args = parser.parse_args(argv) if args.command == "verify-dag" and args.chain and not args.trusted_root_issuer: parser.error("verify-dag with --chain requires --trusted-root-issuer") + if ( + args.command == "verify-dag" + and not args.chain + and (args.revocations is not None or args.max_revocation_staleness is not None) + ): + # Revocation applies to credentials, and without --chain none are checked. + # Accepting the flag silently would look like a check that never ran. + parser.error("verify-dag revocation options require --chain") result: int = args.func(args) return result diff --git a/src/ca2a_runtime/delegation/__init__.py b/src/ca2a_runtime/delegation/__init__.py index cd96da8..766e66f 100644 --- a/src/ca2a_runtime/delegation/__init__.py +++ b/src/ca2a_runtime/delegation/__init__.py @@ -16,13 +16,25 @@ build_holder_proof, verify_holder_proof, ) +from ca2a_runtime.delegation.revocation import ( + RevocationSnapshot, + RevocationStatement, + RevocationStatus, + credential_digest, + revoke, +) __all__ = [ "DelegationCredential", "HolderProof", + "RevocationSnapshot", + "RevocationStatement", + "RevocationStatus", "build_holder_proof", "canonical_bytes", + "credential_digest", "new_keypair", + "revoke", "verify_chain", "verify_holder_proof", ] diff --git a/src/ca2a_runtime/delegation/credential.py b/src/ca2a_runtime/delegation/credential.py index c3d29c0..54263b2 100644 --- a/src/ca2a_runtime/delegation/credential.py +++ b/src/ca2a_runtime/delegation/credential.py @@ -13,6 +13,10 @@ 5. Validity: each hop's validity window, when present, contains the evaluation time. +When the caller supplies a revocation snapshot, a sixth check refuses a chain +containing a hop revoked by its issuer or an issuer above it. See +ca2a_runtime.delegation.revocation. + Canonicalization uses RFC 8785 (JSON Canonicalization Scheme), so the signed byte string is identical across conforming implementations and cA2A signatures are cross-verifiable with agent-manifest. See ca2a_runtime.canonical. @@ -33,13 +37,21 @@ ) from ca2a_runtime.canonical import canonicalize +from ca2a_runtime.delegation.revocation import ( + REVOCATION_NOT_CHECKED, + RevocationSnapshot, + RevocationStatus, + credential_digest, +) from ca2a_runtime.errors import ( BrokenDelegationLink, CredentialExpired, CredentialNotYetValid, CredentialReplay, + CredentialRevoked, DelegationDepthExceeded, InvalidCredential, + RevocationStatusUnknown, ScopeEscalation, UntrustedDelegationRoot, ) @@ -227,7 +239,9 @@ def verify_chain( max_depth: int = 8, trusted_root_issuers: Collection[str] = (), at_time: int | None = None, -) -> None: + revocations: RevocationSnapshot | None = None, + max_revocation_staleness: int | None = None, +) -> RevocationStatus: """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 @@ -239,6 +253,20 @@ def verify_chain( replaying recorded evidence passes the time the action was decided, since a window that has lapsed by audit time says nothing about validity at decision time. + + ``revocations`` is an optional snapshot of signed revocation statements. + With it, a chain containing a hop revoked by that hop's issuer or by any + issuer above it raises ``CredentialRevoked``. Without it, verification is + exactly as before and stays fully offline, and the returned status has + ``checked=False``: the chain may have been revoked and this call would not + know. The return value exists so that no caller can read a successful + offline verification as "not revoked". + + ``max_revocation_staleness`` (seconds) makes revocation checking mandatory: + with it set, a missing snapshot, or one whose ``as_of`` is more than that + many seconds before the evaluation time, raises ``RevocationStatusUnknown``. + It defaults to ``None`` so that offline verification with no revocation data + keeps working. """ # Bounds on the wire are strict JSON integers; the evaluation time they are # compared against holds the same line, or True / 1.5 / -1 from a library @@ -247,6 +275,12 @@ def verify_chain( isinstance(at_time, bool) or not isinstance(at_time, int) or at_time < 0 ): raise ValueError("at_time must be a non-negative integer or None") + if max_revocation_staleness is not None and ( + isinstance(max_revocation_staleness, bool) + or not isinstance(max_revocation_staleness, int) + or max_revocation_staleness < 0 + ): + raise ValueError("max_revocation_staleness must be a non-negative integer or None") if not chain: raise BrokenDelegationLink("empty delegation chain") @@ -312,3 +346,62 @@ def verify_chain( ) prev = cred + + return _check_revocation( + chain, + revocations, + at_time=at_time, + now=now, + max_staleness=max_revocation_staleness, + ) + + +def _check_revocation( + chain: list[DelegationCredential], + revocations: RevocationSnapshot | None, + *, + at_time: int | None, + now: int, + max_staleness: int | None, +) -> RevocationStatus: + """Refuse a revoked hop, then enforce the staleness policy. + + Runs only on a chain whose structure has already verified, so the issuers + used as revocation authority are a real, continuous line of delegators. + """ + if revocations is None: + if max_staleness is not None: + raise RevocationStatusUnknown( + "revocation status is required but no revocation snapshot was supplied", + detail=f"max_revocation_staleness={max_staleness}", + ) + return REVOCATION_NOT_CHECKED + + # Revocation is looked up before staleness is judged. Revocation is + # monotonic, so a statement in an old snapshot is still true: a stale + # snapshot can prove a hop revoked, it just cannot prove one is not. + authorities: set[str] = set() + for i, cred in enumerate(chain): + # The issuer of hop i, plus every issuer above it. The subject of hop i + # is not added until it issues hop i + 1, so a delegate cannot revoke + # the grant it received, or anything above it. + authorities.add(cred.issuer) + digest = credential_digest(cred) + stmt = revocations.effective_revocation(digest, authorities, at_time=at_time) + if stmt is not None: + raise CredentialRevoked( + f"hop {i} credential has been revoked", + detail=( + f"credential_id={cred.credential_id} digest={digest} " + f"revoker={stmt.revoker} issued_at={stmt.issued_at}" + ), + ) + + if max_staleness is not None and now - revocations.as_of > max_staleness: + raise RevocationStatusUnknown( + "revocation snapshot is older than this verifier accepts", + detail=( + f"as_of={revocations.as_of} at_time={now} max_revocation_staleness={max_staleness}" + ), + ) + return RevocationStatus(checked=True, as_of=revocations.as_of) diff --git a/src/ca2a_runtime/delegation/revocation.py b/src/ca2a_runtime/delegation/revocation.py new file mode 100644 index 0000000..a28d9f1 --- /dev/null +++ b/src/ca2a_runtime/delegation/revocation.py @@ -0,0 +1,292 @@ +"""Revocation of delegation credentials before their validity window closes. + +A validity window (``not_before`` / ``not_after``) bounds how long a grant lasts, +but inside a still-valid window nothing let the delegator take it back. This +module adds that: a delegator signs a :class:`RevocationStatement` naming the +credential by digest, and a verifier handed a :class:`RevocationSnapshot` +refuses any chain that contains a revoked hop. + +**Who may revoke.** A statement is effective against hop ``i`` of a chain only +when its ``revoker`` is the issuer of hop ``i`` or of an ancestor hop (``j < i``). +The delegator can withdraw what it granted, and anyone above it can withdraw a +grant made below it. A delegate cannot revoke upward, because the subject of hop +``i`` is the issuer of hop ``i + 1``, not of ``i``. A validly signed statement from +any other key has no effect on that chain. Authority is evaluated against the +presented chain at verification time, because a statement names a credential +and not the chain it will be presented in. + +**Cascade.** Revoking hop ``i`` refuses every chain that contains it, so every +grant made beneath it falls with it: none of them can be presented without it. + +**Monotonic.** There is no un-revoke. A statement cannot express one (the wire +object is strict and has no such field), and a revocation is decided by the +presence of any effective statement, so no later statement can reverse it. + +**Integrity is fail closed.** Every statement's signature is checked when a +snapshot is built. A snapshot carrying an unsigned or forged statement is refused +as a whole with ``INVALID_REVOCATION`` rather than having the bad entry dropped: +a tampered snapshot means the revocation feed itself is untrustworthy, and +silently discarding entries would turn tampering into un-revocation. + +**Offline verification is unchanged.** Revocation data is an optional input, like +the trusted root set. A verifier with no snapshot still verifies offline under +P-4, and the result says revocation was not checked, so "verified offline" cannot +be read as "not revoked". A snapshot is local data: consulting it contacts +nobody. Getting current snapshots to verifiers is the deployment's job. + +Statements are signed over the RFC 8785 form of their body using the same +canonicalization helper as credentials (:func:`ca2a_runtime.canonical.canonicalize`). +The body carries a ``type`` field so a revocation signature cannot be mistaken +for a signature over any other object this key signs. +""" + +from __future__ import annotations + +import hashlib +import re +import time +from collections.abc import Iterable, Mapping +from dataclasses import dataclass, field, replace +from typing import TYPE_CHECKING, Any + +from cryptography.exceptions import InvalidSignature +from cryptography.hazmat.primitives.asymmetric.ed25519 import ( + Ed25519PrivateKey, + Ed25519PublicKey, +) + +from ca2a_runtime.canonical import canonicalize as canonical_bytes +from ca2a_runtime.errors import InvalidRevocation + +if TYPE_CHECKING: + from ca2a_runtime.delegation.credential import DelegationCredential + +#: Domain separation tag, part of every signed revocation body. +REVOCATION_TYPE = "ca2a.delegation-revocation.v1" + +_HEX_32_RE = re.compile(r"[0-9a-f]{64}") +_HEX_64_RE = re.compile(r"[0-9a-f]{128}") +_DIGEST_RE = re.compile(r"sha256:[0-9a-f]{64}") +_STATEMENT_FIELDS = frozenset({"type", "revoked_digest", "revoker", "issued_at", "signature"}) +_SNAPSHOT_FIELDS = frozenset({"as_of", "revocations"}) + + +def credential_digest(credential: DelegationCredential) -> str: + """Return the ``sha256:``-prefixed digest that identifies a credential. + + Taken over the canonical signed body, the same bytes the issuer signed, so it + names exactly one grant: the issuer, subject, scope, parent link, depth and + validity window are all inside it. + """ + return "sha256:" + hashlib.sha256(canonical_bytes(credential.body())).hexdigest() + + +def _non_negative_int(value: Any) -> bool: + return isinstance(value, int) and not isinstance(value, bool) and value >= 0 + + +@dataclass(frozen=True) +class RevocationStatement: + """A signed statement that ``revoker`` withdraws the credential ``revoked_digest``.""" + + revoked_digest: str # sha256: of the credential's canonical body + revoker: str # Ed25519 public key, raw hex + issued_at: int # Unix epoch seconds, as claimed by the revoker + signature: str = "" # Ed25519 over canonical_bytes(body()), hex + + def __post_init__(self) -> None: + if not isinstance(self.revoked_digest, str) or not _DIGEST_RE.fullmatch( + self.revoked_digest + ): + raise InvalidRevocation("revoked_digest must be sha256: followed by 64 lowercase hex") + if not isinstance(self.revoker, str) or not _HEX_32_RE.fullmatch(self.revoker): + raise InvalidRevocation("revoker must be a lowercase 32-byte Ed25519 key in hex") + if not _non_negative_int(self.issued_at): + raise InvalidRevocation("issued_at must be a non-negative integer") + + def body(self) -> dict[str, Any]: + """The signed portion of the statement (everything but the signature).""" + return { + "type": REVOCATION_TYPE, + "revoked_digest": self.revoked_digest, + "revoker": self.revoker, + "issued_at": self.issued_at, + } + + def sign(self, private_key: Ed25519PrivateKey) -> RevocationStatement: + """Return a copy signed by ``private_key`` (must match ``revoker``).""" + expected = private_key.public_key().public_bytes_raw().hex() + if expected != self.revoker: + raise InvalidRevocation( + "signing key does not match revoker", + detail=f"revoker={self.revoker} key={expected}", + ) + return replace(self, signature=private_key.sign(canonical_bytes(self.body())).hex()) + + def verify_signature(self) -> None: + """Raise InvalidRevocation if the signature is absent or does not verify.""" + if not self.signature: + raise InvalidRevocation("revocation statement is unsigned") + if not isinstance(self.signature, str) or not _HEX_64_RE.fullmatch(self.signature): + raise InvalidRevocation( + "signature must be a lowercase 64-byte Ed25519 signature in hex" + ) + try: + pub = Ed25519PublicKey.from_public_bytes(bytes.fromhex(self.revoker)) + pub.verify(bytes.fromhex(self.signature), canonical_bytes(self.body())) + except (InvalidSignature, ValueError) as exc: + raise InvalidRevocation( + "revocation signature failed to verify", detail=str(exc) + ) from exc + + def to_dict(self) -> dict[str, Any]: + return {**self.body(), "signature": self.signature} + + @classmethod + def from_dict(cls, data: Mapping[str, Any]) -> RevocationStatement: + """Parse the wire form strictly: unknown or missing fields are rejected.""" + if not isinstance(data, Mapping): + raise InvalidRevocation("revocation statement must be a JSON object") + unknown = set(data) - _STATEMENT_FIELDS + missing = _STATEMENT_FIELDS - set(data) + if unknown or missing: + raise InvalidRevocation( + "malformed revocation fields", + detail=f"missing={sorted(missing)} unknown={sorted(unknown)}", + ) + if data["type"] != REVOCATION_TYPE: + raise InvalidRevocation("unsupported revocation type", detail=f"type={data['type']!r}") + signature = data["signature"] + if not isinstance(signature, str): + raise InvalidRevocation("signature must be a string") + return cls( + revoked_digest=data["revoked_digest"], + revoker=data["revoker"], + issued_at=data["issued_at"], + signature=signature, + ) + + +def revoke( + credential: DelegationCredential, + private_key: Ed25519PrivateKey, + *, + issued_at: int | None = None, +) -> RevocationStatement: + """Sign a statement withdrawing ``credential``. + + ``private_key`` should belong to the credential's issuer or to an issuer + above it in the chain; a statement from anyone else is well formed but has + no effect when a chain is verified. + """ + return RevocationStatement( + revoked_digest=credential_digest(credential), + revoker=private_key.public_key().public_bytes_raw().hex(), + issued_at=int(time.time()) if issued_at is None else issued_at, + ).sign(private_key) + + +@dataclass(frozen=True) +class RevocationSnapshot: + """The set of revocation statements a verifier holds, current as of ``as_of``. + + ``as_of`` is when the supplier of this snapshot last brought it up to date, + in Unix epoch seconds. It is asserted by whoever hands the snapshot to the + verifier and is not signed by any revoker, so a staleness bound built on it + detects a feed that has stopped updating, not a supplier that lies about it. + Individual statements are signed, so a supplier can withhold a statement but + cannot forge one. + """ + + as_of: int + statements: tuple[RevocationStatement, ...] = () + _by_digest: dict[str, tuple[RevocationStatement, ...]] = field( + init=False, repr=False, compare=False + ) + + def __post_init__(self) -> None: + if not _non_negative_int(self.as_of): + raise InvalidRevocation("as_of must be a non-negative integer") + statements = tuple(self.statements) + index: dict[str, list[RevocationStatement]] = {} + for i, stmt in enumerate(statements): + if not isinstance(stmt, RevocationStatement): + raise InvalidRevocation(f"revocation {i} is not a RevocationStatement") + try: + stmt.verify_signature() + except InvalidRevocation as exc: + raise InvalidRevocation( + f"revocation {i} does not verify; the snapshot is refused as a whole", + detail=str(exc) if exc.detail is None else f"{exc}: {exc.detail}", + ) from exc + index.setdefault(stmt.revoked_digest, []).append(stmt) + object.__setattr__(self, "statements", statements) + object.__setattr__(self, "_by_digest", {d: tuple(s) for d, s in index.items()}) + + def effective_revocation( + self, + digest: str, + authorized_revokers: Iterable[str], + *, + at_time: int | None = None, + ) -> RevocationStatement | None: + """Return a statement that revokes ``digest`` under this authority, or None. + + With ``at_time`` (an audit of a past decision), only statements issued at + or before that time count, so a revocation issued afterwards does not + rewrite a decision that was correct when it was made. Without it (a live + decision), every statement the verifier holds is in force whatever the + revoker's clock said. + """ + authorized = frozenset(authorized_revokers) + for stmt in self._by_digest.get(digest, ()): + if stmt.revoker not in authorized: + continue + if at_time is not None and stmt.issued_at > at_time: + continue + return stmt + return None + + def to_dict(self) -> dict[str, Any]: + return {"as_of": self.as_of, "revocations": [s.to_dict() for s in self.statements]} + + @classmethod + def from_dict(cls, data: Mapping[str, Any]) -> RevocationSnapshot: + """Parse ``{"as_of": int, "revocations": [statement, ...]}`` strictly.""" + if not isinstance(data, Mapping): + raise InvalidRevocation("revocation snapshot must be a JSON object") + unknown = set(data) - _SNAPSHOT_FIELDS + missing = _SNAPSHOT_FIELDS - set(data) + if unknown or missing: + raise InvalidRevocation( + "malformed revocation snapshot fields", + detail=f"missing={sorted(missing)} unknown={sorted(unknown)}", + ) + items = data["revocations"] + if not isinstance(items, list): + raise InvalidRevocation("revocations must be an array") + return cls( + as_of=data["as_of"], + statements=tuple(RevocationStatement.from_dict(item) for item in items), + ) + + +@dataclass(frozen=True) +class RevocationStatus: + """What a chain verification established about revocation. + + ``checked`` is False when no snapshot was supplied. The chain may then be + revoked and the verifier would not know, which is why this is returned rather + than left for the caller to assume. When True, no hop was revoked by any + statement in a snapshot current as of ``as_of``. + """ + + checked: bool + as_of: int | None = None + + @property + def state(self) -> str: + return "not_revoked" if self.checked else "not_checked" + + +REVOCATION_NOT_CHECKED = RevocationStatus(checked=False) diff --git a/src/ca2a_runtime/errors.py b/src/ca2a_runtime/errors.py index 0801682..d7523e8 100644 --- a/src/ca2a_runtime/errors.py +++ b/src/ca2a_runtime/errors.py @@ -77,6 +77,43 @@ class CredentialExpired(CA2AError): http_status = 403 +class CredentialRevoked(CA2AError): + """A hop in the chain was revoked by its issuer or by an issuer above it. + + 403 like the validity failures: the chain is well formed and validly signed, + but a party with authority over the grant has withdrawn it. Descendants of a + revoked hop are refused with it, since they cannot be presented without it. + """ + + code = "CREDENTIAL_REVOKED" + http_status = 403 + + +class RevocationStatusUnknown(CA2AError): + """The verifier's policy requires current revocation data and it has none. + + Raised when ``max_revocation_staleness`` is set and no snapshot was supplied, + or the supplied snapshot is older than the bound. 503 because nothing is + wrong with the chain: the verifier cannot currently establish that it has not + been revoked, and fails closed until it can. + """ + + code = "REVOCATION_STATUS_UNKNOWN" + http_status = 503 + + +class InvalidRevocation(CA2AError): + """A revocation statement or snapshot is malformed, unsigned, or forged. + + A snapshot containing any such statement is refused as a whole rather than + having the bad entry dropped, since dropping it would turn tampering with the + revocation feed into an un-revocation. + """ + + code = "INVALID_REVOCATION" + http_status = 400 + + class HolderProofInvalid(CA2AError): """The presenter of a delegation chain did not prove it holds the leaf key. diff --git a/src/ca2a_runtime/node.py b/src/ca2a_runtime/node.py index 675e634..765e8ae 100644 --- a/src/ca2a_runtime/node.py +++ b/src/ca2a_runtime/node.py @@ -12,13 +12,14 @@ from __future__ import annotations -from collections.abc import Collection +from collections.abc import Callable, Collection from typing import Any from ca2a_runtime.agent_manifest import AgentManifestBinding from ca2a_runtime.attestation import ChannelOffer, Verifier, attest_channel from ca2a_runtime.challenge import DEFAULT_TTL_SECONDS, generate_secret, issue_challenge from ca2a_runtime.channel import generate_channel_keypair +from ca2a_runtime.delegation.revocation import RevocationSnapshot from ca2a_runtime.errors import ConfigError, TransportError from ca2a_runtime.peer import ( REQUIRE_HARDWARE, @@ -56,6 +57,8 @@ def __init__( require_holder_proof: bool = True, trusted_root_issuers: Collection[str] = (), agent_manifest: AgentManifestBinding | None = None, + revocation_source: Callable[[], RevocationSnapshot | None] | None = None, + max_revocation_staleness: int | None = None, ) -> None: if require_caller_attestation not in REQUIREMENT_VALUES: raise ConfigError( @@ -80,6 +83,11 @@ def __init__( self.require_holder_proof = require_holder_proof self.trusted_root_issuers = frozenset(trusted_root_issuers) self.agent_manifest = agent_manifest + # A callable rather than a snapshot, because a snapshot goes stale: the + # node asks for the current one on every call. Fetching and refreshing + # it is the deployment's job. + self.revocation_source = revocation_source + self.max_revocation_staleness = max_revocation_staleness self._private_key, self.channel_public_key = generate_channel_keypair() self._challenge_secret = generate_secret() @@ -109,4 +117,6 @@ def handle(self, message: dict[str, Any]) -> PeerResult: audience=self.channel_public_key, require_holder_proof=self.require_holder_proof, trusted_root_issuers=self.trusted_root_issuers, + revocations=None if self.revocation_source is None else self.revocation_source(), + max_revocation_staleness=self.max_revocation_staleness, ) diff --git a/src/ca2a_runtime/peer.py b/src/ca2a_runtime/peer.py index 2aa7fd9..7a3ebb1 100644 --- a/src/ca2a_runtime/peer.py +++ b/src/ca2a_runtime/peer.py @@ -45,6 +45,11 @@ from ca2a_runtime.channel import open_sealed from ca2a_runtime.delegation.credential import DelegationCredential, verify_chain from ca2a_runtime.delegation.holder import HolderProof, verify_holder_proof +from ca2a_runtime.delegation.revocation import ( + REVOCATION_NOT_CHECKED, + RevocationSnapshot, + RevocationStatus, +) from ca2a_runtime.errors import ( AttestationFailed, ConfigError, @@ -91,12 +96,20 @@ def effective_scope( *, max_depth: int = 8, trusted_root_issuers: Collection[str] = (), + revocations: RevocationSnapshot | None = None, + max_revocation_staleness: int | None = None, ) -> 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, trusted_root_issuers=trusted_root_issuers) + verify_chain( + chain, + max_depth=max_depth, + trusted_root_issuers=trusted_root_issuers, + revocations=revocations, + max_revocation_staleness=max_revocation_staleness, + ) return policy.intersect(chain[-1].scope) @@ -119,6 +132,8 @@ def enforce_peer_call( max_depth: int = 8, caller_attestation: str = CALLER_NOT_OFFERED, trusted_root_issuers: Collection[str] = (), + revocations: RevocationSnapshot | None = None, + max_revocation_staleness: int | None = None, ) -> PeerDecision: """Verify, intersect with local policy, enforce, and emit a provenance record. @@ -136,6 +151,8 @@ def enforce_peer_call( policy, max_depth=max_depth, trusted_root_issuers=trusted_root_issuers, + revocations=revocations, + max_revocation_staleness=max_revocation_staleness, ) return decide_capability( chain, @@ -235,6 +252,10 @@ class PeerResult: caller_attestation: str = CALLER_NOT_OFFERED """What the callee established about the caller's runtime. Also on ``record``, where it is part of the portable evidence rather than just this return value.""" + revocation: RevocationStatus = REVOCATION_NOT_CHECKED + """Whether the chain was checked against a revocation snapshot. ``checked`` is + False when the callee was given none, in which case a revoked chain would + have been accepted.""" def appraise_caller_runtime( @@ -380,6 +401,8 @@ def handle_peer_request( audience: str | None = None, require_holder_proof: bool = True, trusted_root_issuers: Collection[str] = (), + revocations: RevocationSnapshot | None = None, + max_revocation_staleness: int | None = None, ) -> PeerResult: """Run the full inbound pipeline for a parsed peer request. @@ -408,16 +431,24 @@ def handle_peer_request( leaf's authority; it exists for offline replay of recorded evidence, where there is no live caller to challenge, and must not be used on a live peer path. + + ``revocations`` is the callee's current revocation snapshot, if it has one; + a chain with a revoked hop is then refused with ``CREDENTIAL_REVOKED`` before + the caller is challenged. ``max_revocation_staleness`` makes a snapshot of at + most that age mandatory. Without either, revocation is not checked and the + returned ``PeerResult.revocation`` says so. """ # The chain, trust set included, is verified exactly once: here, before the # caller is challenged, so an untrusted or malformed chain is refused before # a proof is demanded about a credential this peer was never going to honour. # The scope intersection below reads the leaf of this verified chain, so it # does not verify it again. - verify_chain( + revocation = verify_chain( request.chain, max_depth=max_depth, trusted_root_issuers=trusted_root_issuers, + revocations=revocations, + max_revocation_staleness=max_revocation_staleness, ) if require_holder_proof: # Before the scope intersection, so an unauthenticated caller never @@ -459,4 +490,5 @@ def handle_peer_request( record=decision.record, payload=payload, caller_attestation=caller_attestation, + revocation=revocation, ) diff --git a/src/ca2a_verify/__init__.py b/src/ca2a_verify/__init__.py index 05ad6ad..3207cda 100644 --- a/src/ca2a_verify/__init__.py +++ b/src/ca2a_verify/__init__.py @@ -12,6 +12,7 @@ from ca2a_verify.verify import ( ChainResult, VerificationError, + load_revocation_snapshot, verify_chain_file, verify_delegation_chain, ) @@ -22,6 +23,7 @@ "TraceDagResult", "VerificationError", "cross_check_trace_dag", + "load_revocation_snapshot", "verify_chain_file", "verify_delegation_chain", "verify_trace_dag", diff --git a/src/ca2a_verify/verify.py b/src/ca2a_verify/verify.py index 1a235c3..853e3df 100644 --- a/src/ca2a_verify/verify.py +++ b/src/ca2a_verify/verify.py @@ -13,8 +13,8 @@ from pathlib import Path from typing import Any -from ca2a_runtime.delegation import DelegationCredential, verify_chain -from ca2a_runtime.errors import CA2AError, InvalidCredential +from ca2a_runtime.delegation import DelegationCredential, RevocationSnapshot, verify_chain +from ca2a_runtime.errors import CA2AError, InvalidCredential, InvalidRevocation # Re-exported so callers can catch a single verify-layer error type. VerificationError = CA2AError @@ -28,6 +28,16 @@ class ChainResult: root_issuer: str leaf_subject: str leaf_scope: list[str] + revocation_checked: bool = False + """False when no revocation snapshot was supplied. The chain verified, but + whether any hop has been revoked was not checked and is not known.""" + revocation_as_of: int | None = None + """When checked, the ``as_of`` time of the snapshot that found no revoked hop.""" + + @property + def revocation(self) -> str: + """``"not_revoked"`` when checked against a snapshot, else ``"not_checked"``.""" + return "not_revoked" if self.revocation_checked else "not_checked" def verify_delegation_chain( @@ -36,18 +46,26 @@ def verify_delegation_chain( trusted_root_issuers: Collection[str], max_depth: int = 8, at_time: int | None = None, + revocations: RevocationSnapshot | None = None, + max_revocation_staleness: int | None = None, ) -> ChainResult: """Verify a root-to-leaf chain and summarize it. Raises on any violation. ``at_time`` is the Unix time validity windows are evaluated at; ``None`` means the current time. An auditor replaying recorded evidence passes the time the action was decided, not its own. + + ``revocations`` and ``max_revocation_staleness`` are passed to + :func:`~ca2a_runtime.delegation.verify_chain`. Without a snapshot the result + has ``revocation_checked=False``. """ - verify_chain( + status = verify_chain( chain, max_depth=max_depth, trusted_root_issuers=trusted_root_issuers, at_time=at_time, + revocations=revocations, + max_revocation_staleness=max_revocation_staleness, ) root = chain[0] leaf = chain[-1] @@ -56,6 +74,8 @@ def verify_delegation_chain( root_issuer=root.issuer, leaf_subject=leaf.subject, leaf_scope=sorted(leaf.scope), + revocation_checked=status.checked, + revocation_as_of=status.as_of, ) @@ -73,6 +93,8 @@ def verify_chain_file( trusted_root_issuers: Collection[str], max_depth: int = 8, at_time: int | None = None, + revocations: RevocationSnapshot | None = None, + max_revocation_staleness: int | None = None, ) -> ChainResult: """Load a delegation chain from a JSON file and verify it.""" p = Path(path) @@ -87,4 +109,22 @@ def verify_chain_file( trusted_root_issuers=trusted_root_issuers, max_depth=max_depth, at_time=at_time, + revocations=revocations, + max_revocation_staleness=max_revocation_staleness, ) + + +def load_revocation_snapshot(path: str | Path) -> RevocationSnapshot: + """Load a revocation snapshot (``{"as_of": ..., "revocations": [...]}``). + + Every statement's signature is checked on load; a snapshot containing any + statement that does not verify raises ``InvalidRevocation``. + """ + p = Path(path) + if not p.is_file(): + raise InvalidRevocation(f"revocation file not found: {p}") + try: + data = json.loads(p.read_text(encoding="utf-8")) + except json.JSONDecodeError as exc: + raise InvalidRevocation(f"invalid JSON in {p}", detail=str(exc)) from exc + return RevocationSnapshot.from_dict(data) diff --git a/tests/unit/test_docs_quickstart.py b/tests/unit/test_docs_quickstart.py index c883b85..76e0a37 100644 --- a/tests/unit/test_docs_quickstart.py +++ b/tests/unit/test_docs_quickstart.py @@ -35,4 +35,5 @@ def test_first_chain(tmp_path, capsys): "verified": True, "hops": 2, "leaf_scope": ["cap:read"], + "revocation": "not_checked", } diff --git a/tests/unit/test_revocation.py b/tests/unit/test_revocation.py new file mode 100644 index 0000000..e714464 --- /dev/null +++ b/tests/unit/test_revocation.py @@ -0,0 +1,475 @@ +"""Revocation of delegation credentials inside their validity window. + +Covers who may revoke (the hop's issuer or an issuer above it, never the delegate +and never a stranger), the cascade to descendants, fail-closed handling of forged +or unsigned statements, the not-checked status of offline verification with no +snapshot, the staleness policy, and that nothing can un-revoke. +""" + +from __future__ import annotations + +import json +import time +from dataclasses import replace + +import pytest +from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PrivateKey + +from ca2a_runtime.cli import main as cli_main +from ca2a_runtime.delegation import ( + DelegationCredential, + RevocationSnapshot, + RevocationStatement, + credential_digest, + new_keypair, + revoke, + verify_chain, +) +from ca2a_runtime.delegation.holder import build_holder_proof +from ca2a_runtime.delegation.revocation import REVOCATION_TYPE +from ca2a_runtime.errors import ( + CredentialRevoked, + InvalidRevocation, + RevocationStatusUnknown, +) +from ca2a_runtime.node import PeerNode +from ca2a_runtime.peer import PeerRequest, handle_peer_request +from ca2a_runtime.policy import LocalPolicy +from ca2a_runtime.transport import a2a_adapter +from ca2a_verify import load_revocation_snapshot, verify_chain_file, verify_delegation_chain +from tests.unit.conftest import TEST_AUDIENCE, TEST_SECRET, proved_request + +NOW = 2_000_000_000 + + +def _chain(hops: int = 3) -> tuple[list[DelegationCredential], list[Ed25519PrivateKey]]: + """Return a chain and every key on it: keys[i] issues hop i, keys[i + 1] is its subject.""" + keys = [new_keypair() for _ in range(hops + 1)] + scope = ["cap:a", "cap:b", "cap:c", "cap:d"] + chain: list[DelegationCredential] = [] + parent_id: str | None = None + for i in range(hops): + cred = DelegationCredential( + credential_id=f"cred-{i}", + issuer=keys[i][1], + subject=keys[i + 1][1], + scope=frozenset(scope[: len(scope) - i]), + depth=i, + parent_id=parent_id, + ).sign(keys[i][0]) + chain.append(cred) + parent_id = cred.credential_id + return chain, [k[0] for k in keys] + + +def _verify(chain, snapshot=None, **kwargs): + return verify_chain( + chain, + trusted_root_issuers={chain[0].issuer}, + revocations=snapshot, + **kwargs, + ) + + +def _snapshot(*statements: RevocationStatement, as_of: int = NOW) -> RevocationSnapshot: + return RevocationSnapshot(as_of=as_of, statements=statements) + + +# --- offline verification without revocation data --------------------------- + + +def test_offline_verify_without_snapshot_reports_not_checked() -> None: + chain, _ = _chain() + status = _verify(chain) + assert status.checked is False + assert status.as_of is None + assert status.state == "not_checked" + + +def test_chain_result_says_revocation_was_not_checked() -> None: + chain, _ = _chain() + result = verify_delegation_chain(chain, trusted_root_issuers={chain[0].issuer}) + assert result.revocation_checked is False + assert result.revocation == "not_checked" + + +def test_empty_snapshot_reports_checked_as_of() -> None: + chain, _ = _chain() + status = _verify(chain, _snapshot(as_of=NOW - 5)) + assert status.checked is True + assert status.as_of == NOW - 5 + assert status.state == "not_revoked" + + +# --- who may revoke ----------------------------------------------------------- + + +def test_issuer_revokes_leaf() -> None: + chain, keys = _chain() + stmt = revoke(chain[2], keys[2], issued_at=NOW) + with pytest.raises(CredentialRevoked) as exc: + _verify(chain, _snapshot(stmt)) + assert exc.value.code == "CREDENTIAL_REVOKED" + assert exc.value.http_status == 403 + assert "hop 2" in str(exc.value) + + +def test_revoking_middle_link_cascades_to_descendants() -> None: + chain, keys = _chain(4) + snapshot = _snapshot(revoke(chain[1], keys[1], issued_at=NOW)) + # Every chain that runs through hop 1 is refused, whatever sits beneath it, + # and the refusal names hop 1 rather than the leaf. + for end in (2, 3, 4): + with pytest.raises(CredentialRevoked, match="hop 1"): + _verify(chain[:end], snapshot) + # The grant above the revoked link is untouched. + assert _verify(chain[:1], snapshot).checked is True + + +def test_ancestor_revokes_grant_made_below_it() -> None: + chain, keys = _chain() + # The root issuer withdraws a grant two levels down that it did not sign. + with pytest.raises(CredentialRevoked, match="hop 2"): + _verify(chain, _snapshot(revoke(chain[2], keys[0], issued_at=NOW))) + + +def test_delegate_cannot_revoke_its_delegator() -> None: + chain, keys = _chain() + # keys[2] is the subject of hop 1 and the issuer of hop 2. It may revoke + # hop 2, but it holds no authority over hop 1 or hop 0 above it. + upward = _snapshot( + revoke(chain[1], keys[2], issued_at=NOW), + revoke(chain[0], keys[2], issued_at=NOW), + # The leaf delegate trying to withdraw its own grant, or its parent's. + revoke(chain[2], keys[3], issued_at=NOW), + revoke(chain[1], keys[3], issued_at=NOW), + ) + status = _verify(chain, upward) + assert status.checked is True + assert status.state == "not_revoked" + + +def test_unrelated_key_cannot_revoke() -> None: + chain, _ = _chain() + stranger, _ = new_keypair() + snapshot = _snapshot(*(revoke(cred, stranger, issued_at=NOW) for cred in chain)) + assert _verify(chain, snapshot).checked is True + + +def test_revocation_names_one_credential_not_its_id() -> None: + chain, keys = _chain() + # A different grant that happens to reuse the credential_id is a different + # credential, and its revocation does not touch this one. + _, other_subject = new_keypair() + lookalike = replace(chain[2], subject=other_subject, signature="").sign(keys[2]) + assert credential_digest(lookalike) != credential_digest(chain[2]) + assert _verify(chain, _snapshot(revoke(lookalike, keys[2], issued_at=NOW))).checked + + +# --- integrity: forged or unsigned statements -------------------------------- + + +def test_unsigned_statement_refuses_the_snapshot() -> None: + chain, keys = _chain() + unsigned = replace(revoke(chain[2], keys[2], issued_at=NOW), signature="") + with pytest.raises(InvalidRevocation, match="refused as a whole"): + _snapshot(unsigned) + + +def test_forged_statement_refuses_the_snapshot() -> None: + chain, keys = _chain() + genuine = revoke(chain[1], keys[1], issued_at=NOW) + # A genuine statement re-pointed at a different credential: the signature + # no longer covers the body, so it is not accepted. + retargeted = replace(genuine, revoked_digest=credential_digest(chain[2])) + with pytest.raises(InvalidRevocation) as exc: + _snapshot(retargeted) + assert exc.value.code == "INVALID_REVOCATION" + # Claiming a different revoker under the original signature fails the same way. + with pytest.raises(InvalidRevocation): + _snapshot(replace(genuine, revoker=chain[0].issuer)) + + +def test_one_bad_statement_refuses_the_valid_ones_with_it() -> None: + chain, keys = _chain() + good = revoke(chain[2], keys[2], issued_at=NOW) + bad = replace(revoke(chain[1], keys[1], issued_at=NOW), issued_at=NOW + 1) + with pytest.raises(InvalidRevocation): + _snapshot(good, bad) + + +def test_signing_key_must_match_revoker() -> None: + chain, keys = _chain() + stmt = RevocationStatement(credential_digest(chain[0]), chain[0].issuer, NOW) + with pytest.raises(InvalidRevocation): + stmt.sign(keys[1]) + + +def test_credential_signature_is_not_a_revocation_signature() -> None: + chain, _ = _chain() + # Same key, same signature bytes, different object: the type tag in the + # revocation body keeps the two signing domains apart. + borrowed = RevocationStatement( + credential_digest(chain[0]), chain[0].issuer, NOW, signature=chain[0].signature + ) + with pytest.raises(InvalidRevocation): + _snapshot(borrowed) + + +# --- monotonic ---------------------------------------------------------------- + + +def test_later_statement_cannot_unrevoke() -> None: + chain, keys = _chain() + first = revoke(chain[2], keys[2], issued_at=NOW - 100) + later = revoke(chain[2], keys[2], issued_at=NOW) + for order in ((first, later), (later, first)): + with pytest.raises(CredentialRevoked): + _verify(chain, _snapshot(*order)) + + +@pytest.mark.parametrize( + "extra", + [{"revoked": False}, {"action": "unrevoke"}, {"reinstated_at": NOW}], +) +def test_wire_form_cannot_express_unrevoke(extra: dict[str, object]) -> None: + chain, keys = _chain() + wire = {**revoke(chain[2], keys[2], issued_at=NOW).to_dict(), **extra} + with pytest.raises(InvalidRevocation, match="malformed revocation fields"): + RevocationStatement.from_dict(wire) + + +def test_unknown_statement_type_is_rejected() -> None: + chain, keys = _chain() + wire = revoke(chain[2], keys[2], issued_at=NOW).to_dict() + wire["type"] = "ca2a.delegation-reinstatement.v1" + with pytest.raises(InvalidRevocation, match="unsupported revocation type"): + RevocationStatement.from_dict(wire) + + +def test_wire_round_trip() -> None: + chain, keys = _chain() + snapshot = _snapshot(revoke(chain[2], keys[2], issued_at=NOW)) + wire = json.loads(json.dumps(snapshot.to_dict())) + assert wire["revocations"][0]["type"] == REVOCATION_TYPE + assert RevocationSnapshot.from_dict(wire) == snapshot + + +# --- staleness policy --------------------------------------------------------- + + +def test_policy_without_snapshot_fails_closed() -> None: + chain, _ = _chain() + with pytest.raises(RevocationStatusUnknown) as exc: + _verify(chain, None, max_revocation_staleness=300) + assert exc.value.code == "REVOCATION_STATUS_UNKNOWN" + assert exc.value.http_status == 503 + + +def test_stale_snapshot_under_policy_fails_closed() -> None: + chain, _ = _chain() + with pytest.raises(RevocationStatusUnknown, match="older than"): + _verify(chain, _snapshot(as_of=NOW - 301), at_time=NOW, max_revocation_staleness=300) + + +def test_snapshot_at_the_bound_is_accepted() -> None: + chain, _ = _chain() + status = _verify(chain, _snapshot(as_of=NOW - 300), at_time=NOW, max_revocation_staleness=300) + assert status.checked is True + + +def test_stale_snapshot_still_proves_revocation() -> None: + chain, keys = _chain() + # Revocation is monotonic, so an old statement is still true. The refusal is + # the revocation, not the staleness. + old = _snapshot(revoke(chain[2], keys[2], issued_at=NOW - 1000), as_of=NOW - 900) + with pytest.raises(CredentialRevoked): + _verify(chain, old, at_time=NOW, max_revocation_staleness=300) + + +def test_stale_snapshot_without_policy_is_used_and_reported() -> None: + chain, _ = _chain() + status = _verify(chain, _snapshot(as_of=NOW - 10_000), at_time=NOW) + assert status.checked is True + assert status.as_of == NOW - 10_000 + + +@pytest.mark.parametrize("bad", [-1, True, 1.5]) +def test_staleness_bound_must_be_a_non_negative_int(bad: object) -> None: + chain, _ = _chain() + with pytest.raises(ValueError): + _verify(chain, _snapshot(), max_revocation_staleness=bad) + + +# --- audit at a past decision time --------------------------------------------- + + +def test_audit_ignores_revocation_issued_after_the_decision() -> None: + chain, keys = _chain() + snapshot = _snapshot(revoke(chain[2], keys[2], issued_at=NOW)) + assert _verify(chain, snapshot, at_time=NOW - 1).checked is True + with pytest.raises(CredentialRevoked): + _verify(chain, snapshot, at_time=NOW) + + +def test_live_verification_applies_every_held_statement() -> None: + chain, keys = _chain() + # A revoker whose clock runs ahead still revokes at once for a live verifier. + future = _snapshot(revoke(chain[2], keys[2], issued_at=4_000_000_000)) + with pytest.raises(CredentialRevoked): + _verify(chain, future) + + +# --- files and CLI ------------------------------------------------------------ + + +def _write(tmp_path, chain, snapshot): + chain_path = tmp_path / "chain.json" + chain_path.write_text( + json.dumps([{**c.body(), "signature": c.signature} for c in chain]), encoding="utf-8" + ) + rev_path = tmp_path / "revocations.json" + rev_path.write_text(json.dumps(snapshot.to_dict()), encoding="utf-8") + return chain_path, rev_path + + +def test_verify_chain_file_with_snapshot(tmp_path) -> None: + chain, keys = _chain() + chain_path, rev_path = _write(tmp_path, chain, _snapshot()) + result = verify_chain_file( + chain_path, + trusted_root_issuers={chain[0].issuer}, + revocations=load_revocation_snapshot(rev_path), + ) + assert result.revocation_checked is True + assert result.revocation_as_of == NOW + + _, rev_path = _write(tmp_path, chain, _snapshot(revoke(chain[1], keys[0], issued_at=NOW))) + with pytest.raises(CredentialRevoked): + verify_chain_file( + chain_path, + trusted_root_issuers={chain[0].issuer}, + revocations=load_revocation_snapshot(rev_path), + ) + + +def test_load_snapshot_with_tampered_statement_fails(tmp_path) -> None: + chain, keys = _chain() + _, rev_path = _write(tmp_path, chain, _snapshot(revoke(chain[2], keys[2], issued_at=NOW))) + data = json.loads(rev_path.read_text(encoding="utf-8")) + data["revocations"][0]["issued_at"] = NOW + 1 + rev_path.write_text(json.dumps(data), encoding="utf-8") + with pytest.raises(InvalidRevocation): + load_revocation_snapshot(rev_path) + + +def test_cli_reports_not_checked_without_snapshot(tmp_path, capsys) -> None: + chain, _ = _chain() + chain_path, _ = _write(tmp_path, chain, _snapshot()) + rc = cli_main( + ["verify-chain", "--chain", str(chain_path), "--trusted-root-issuer", chain[0].issuer] + ) + assert rc == 0 + out = json.loads(capsys.readouterr().out) + assert out["revocation"] == "not_checked" + assert "revocation_as_of" not in out + + +def test_cli_checks_and_refuses_revoked(tmp_path, capsys) -> None: + chain, keys = _chain() + chain_path, rev_path = _write(tmp_path, chain, _snapshot()) + base = ["verify-chain", "--chain", str(chain_path), "--trusted-root-issuer", chain[0].issuer] + assert cli_main([*base, "--revocations", str(rev_path)]) == 0 + out = json.loads(capsys.readouterr().out) + assert out["revocation"] == "not_revoked" + assert out["revocation_as_of"] == NOW + + _, rev_path = _write(tmp_path, chain, _snapshot(revoke(chain[2], keys[2], issued_at=NOW))) + assert cli_main([*base, "--revocations", str(rev_path)]) == 1 + assert json.loads(capsys.readouterr().out)["code"] == "CREDENTIAL_REVOKED" + + assert cli_main([*base, "--max-revocation-staleness", "60"]) == 1 + assert json.loads(capsys.readouterr().out)["code"] == "REVOCATION_STATUS_UNKNOWN" + + +def test_cli_verify_dag_rejects_revocation_flags_without_chain(tmp_path) -> None: + with pytest.raises(SystemExit) as exc: + cli_main(["verify-dag", "--dag", str(tmp_path / "dag.json"), "--revocations", "x.json"]) + assert exc.value.code == 2 + + +# --- live peer path ------------------------------------------------------------- + + +def test_peer_refuses_revoked_chain_before_the_holder_proof() -> None: + chain, keys = _chain(2) + request = proved_request(chain, keys[-1], "cap:a", "rec-0") + kwargs = { + "policy": LocalPolicy.of(["cap:a"]), + "audience": TEST_AUDIENCE, + "challenge_secret": TEST_SECRET, + "trusted_root_issuers": {chain[0].issuer}, + } + result = handle_peer_request(request, **kwargs) + assert result.revocation.checked is False + + result = handle_peer_request(request, revocations=_snapshot(), **kwargs) + assert result.revocation.checked is True + + with pytest.raises(CredentialRevoked): + handle_peer_request( + request, revocations=_snapshot(revoke(chain[1], keys[0], issued_at=NOW)), **kwargs + ) + # Refused before holder binding: a request with no proof at all gets the + # revocation error, not HOLDER_PROOF_INVALID. + with pytest.raises(CredentialRevoked): + handle_peer_request( + replace(request, holder_proof=None), + revocations=_snapshot(revoke(chain[1], keys[0], issued_at=NOW)), + **kwargs, + ) + + +def _node_message(node: PeerNode, chain, leaf_key, record_id: str) -> dict[str, object]: + request = PeerRequest( + chain=chain, + requested_capability="cap:a", + record_id=record_id, + holder_proof=build_holder_proof( + leaf_key, + chain[-1], + audience=node.channel_public_key, + challenge=node.issue_challenge(), + requested_capability="cap:a", + record_id=record_id, + ), + ) + return a2a_adapter.attach_ca2a_metadata({}, request) + + +def test_peer_node_consults_revocation_source_on_every_call() -> None: + chain, keys = _chain(2) + current: list[RevocationSnapshot | None] = [_snapshot(as_of=int(time.time()))] + node = PeerNode( + LocalPolicy.of(["cap:a"]), + trusted_root_issuers={chain[0].issuer}, + revocation_source=lambda: current[0], + max_revocation_staleness=3600, + ) + assert node.handle(_node_message(node, chain, keys[-1], "rec-0")).revocation.checked + + current[0] = _snapshot(revoke(chain[1], keys[1]), as_of=int(time.time())) + with pytest.raises(CredentialRevoked): + node.handle(_node_message(node, chain, keys[-1], "rec-1")) + + # The source going quiet is not a pass: the node's policy requires data. + current[0] = None + with pytest.raises(RevocationStatusUnknown): + node.handle(_node_message(node, chain, keys[-1], "rec-2")) + + +def test_peer_node_without_revocation_source_reports_not_checked() -> None: + chain, keys = _chain(2) + node = PeerNode(LocalPolicy.of(["cap:a"]), trusted_root_issuers={chain[0].issuer}) + result = node.handle(_node_message(node, chain, keys[-1], "rec-0")) + assert result.revocation.checked is False From 20c19367b78a00b19b13c9f23877f6703fb0c054 Mon Sep 17 00:00:00 2001 From: Imran Siddique Date: Fri, 18 Sep 2026 11:42:40 -0700 Subject: [PATCH 2/2] docs(spec): document delegation revocation and its offline limits threat-model.md said delegated authority could not be withdrawn early. Replace that with what now exists: who may revoke, the cascade, the not_checked status an offline verifier reports, the optional staleness bound, and what is still out of scope (distributing revocation data, and a verifier with no snapshot learning of a revocation). Add the statement format and snapshot semantics to delegation-chain.md, the three new error codes, the verification-library parameters, an informative note under profile P-4, and LIMITATIONS and CHANGELOG entries. Co-Authored-By: Claude Opus 5 (1M context) Signed-off-by: Imran Siddique --- CHANGELOG.md | 18 +++++++ LIMITATIONS.md | 1 + docs/spec/delegation-chain.md | 84 +++++++++++++++++++++++++++++++ docs/spec/error-codes.md | 7 ++- docs/spec/profile.md | 2 + docs/spec/threat-model.md | 5 +- docs/spec/verification-library.md | 14 +++++- 7 files changed, 127 insertions(+), 4 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 5b2f0aa..cdb4d96 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -9,6 +9,24 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ### Added +- **Delegation revocation.** Until now a delegated grant could not be withdrawn + inside its validity window (`docs/spec/threat-model.md`). The issuer of a + credential, or any issuer above it in the chain, can now sign a + `RevocationStatement` naming the credential by the SHA-256 of its canonical + body. `verify_chain`, `verify_delegation_chain`, `verify_chain_file`, + `handle_peer_request` and `ca2a verify-chain` / `verify-dag` take an optional + `RevocationSnapshot` and refuse a chain containing a revoked hop with + `CREDENTIAL_REVOKED`, which also refuses every grant beneath it. A delegate + cannot revoke upward, a statement from an unrelated key has no effect, nothing + can un-revoke, and a snapshot with a forged or unsigned statement is refused as + a whole with `INVALID_REVOCATION`. Without a snapshot, verification stays + offline and unchanged, and now reports revocation as `not_checked` + (`verify_chain` returns a `RevocationStatus`; `ChainResult`, `PeerResult` and + the CLI output carry it). An optional `max_revocation_staleness` fails closed + with `REVOCATION_STATUS_UNKNOWN` when the snapshot is missing or too old. + `PeerNode` takes a `revocation_source` callable consulted on every call. + Distributing revocation data remains the deployment's job. + - Add an explicit hardware floor for outbound peer appraisal, a pinned SNP verifier with platform/DEBUG/VMPL/guest-SVN checks, and a two-host acceptance harness. A same-operator hardware diagnostic completed both directions; diff --git a/LIMITATIONS.md b/LIMITATIONS.md index 29e45c9..c2850f2 100644 --- a/LIMITATIONS.md +++ b/LIMITATIONS.md @@ -5,6 +5,7 @@ cA2A 0.2 is a Developer Preview with a runnable, tested profile and runtime. Thi ## What is built - The delegation credential model and offline chain verifier: trusted-root checks, signatures, scope attenuation, depth and validity bounds, duplicate credential IDs within a chain, and cross-chain splice rejection. These checks do not maintain a global history of used credentials. +- Revocation of a delegation before its validity window closes. The issuer of a credential, or an issuer above it in the chain, signs a revocation statement naming the credential by digest; a verifier given a snapshot of such statements refuses a chain containing a revoked hop (`CREDENTIAL_REVOKED`). A delegate cannot revoke its delegator, statements cannot be un-revoked, and a snapshot carrying a forged or unsigned statement is refused as a whole. **What it does not do:** cA2A does not publish or fetch revocation data, so a verifier knows only what its supplied snapshot held at its `as_of` time, and a supplier can withhold a statement. A verifier with no snapshot still verifies offline and cannot learn of a revocation; the result says revocation was `not_checked`. `max_revocation_staleness` lets a verifier fail closed without a recent snapshot, but it is off by default, and `ca2a start` does not load a snapshot. - Configuration, error registry, and the CLI surface, including `ca2a start`, which builds a `PeerNode` from a config file and serves it over the reference transport. - A reference HTTP transport and the attestation handshake, in software mode. `ca2a_runtime.transport.server` and `ca2a_runtime.transport.client` (standard library only) run a live inbound A2A-profile call end to end: the caller fetches the callee's attested channel key, seals a payload to it, and sends a delegated task; the callee parses the A2A metadata with the adapter, runs verify + policy + enforce + open-sealed + provenance, and replies. `ca2a_runtime.attestation` gates the seal on a verified channel key. This is a **reference** transport, not part of the profile: the profile mandates no wire protocol (see Out of scope), and in software mode the peer key is accepted at `assurance="none"`. diff --git a/docs/spec/delegation-chain.md b/docs/spec/delegation-chain.md index 2757535..73334e4 100644 --- a/docs/spec/delegation-chain.md +++ b/docs/spec/delegation-chain.md @@ -55,6 +55,8 @@ without invalidating the signature. | No `credential_id` repeats | `CREDENTIAL_REPLAY` | | Each hop's validity window, when present, contains the evaluation time | `CREDENTIAL_NOT_YET_VALID` / `CREDENTIAL_EXPIRED` | | The root issuer is pinned by the callee for runtime authorization | `UNTRUSTED_DELEGATION_ROOT` | +| No hop is revoked by its issuer or by an issuer above it (checked only when a revocation snapshot is supplied) | `CREDENTIAL_REVOKED` | +| A snapshot no older than `max_revocation_staleness` was supplied (only when that bound is set) | `REVOCATION_STATUS_UNKNOWN` | Signature validity establishes who issued a chain; it does not establish that the issuer is trusted. A live callee therefore supplies its local @@ -83,6 +85,88 @@ Windows are not required to nest across hops. A chain is usable only at times inside every hop's window, so the effective window is already the intersection of the hops'; requiring structural nesting would add no authority bound. + + +## Revocation + +A validity window bounds how long a grant lasts. Revocation lets a party with +authority over a grant withdraw it before `not_after`. See +`ca2a_runtime.delegation.revocation`. + +### Revocation statement + +A `RevocationStatement` is a signed body plus a detached Ed25519 signature, +canonicalized with the same RFC 8785 helper as credentials: + +| Field | Type | Meaning | +|---|---|---| +| `type` | string | Always `ca2a.delegation-revocation.v1`. Signed, so a revocation signature cannot be taken for a signature over any other object the same key signs | +| `revoked_digest` | string | `sha256:` followed by the lowercase hex SHA-256 of the revoked credential's canonical body (the bytes its issuer signed) | +| `revoker` | hex | Ed25519 public key of the party withdrawing the grant | +| `issued_at` | int | Unix epoch seconds, as claimed by the revoker | +| `signature` | hex | Ed25519 over the canonical body, by `revoker` | + +The wire object is strict in the same way a credential is: unknown or missing +fields are rejected, so there is no field that could express an un-revoke. + +### Who may revoke + +A statement is effective against hop `i` of a chain only when `revoker` is the +issuer of hop `i` or of an earlier hop. The delegator can withdraw what it +granted, and any issuer above it can withdraw a grant made below it. The subject +of hop `i` is not in that set, so a delegate cannot revoke the grant it received +or anything above it. A validly signed statement from any other key has no effect +on the chain. Authority is judged against the presented chain at verification +time, after its structure has verified, because a statement names a credential +rather than a chain. + +Revoking hop `i` refuses every chain that contains it, which is every chain +through which a grant beneath it could be exercised. The error names the first +revoked hop. + +Revocation is monotonic. A hop is revoked when any effective statement for it is +present, so no statement, earlier or later, can reverse it. + +### Snapshots and what verification reports + +`verify_chain` takes an optional `RevocationSnapshot`: a set of statements plus +`as_of`, the time its supplier last brought it up to date. Every statement's +signature is checked when the snapshot is built, and a snapshot containing an +unsigned or forged statement is refused as a whole with `INVALID_REVOCATION` +rather than having the bad entry dropped, since dropping it would let tampering +with the feed act as an un-revocation. `as_of` is asserted by the supplier and is +not signed by any revoker. + +`verify_chain` returns a `RevocationStatus`. With no snapshot, verification is +exactly as before, still offline under P-4, and the status has `checked=False` +(`not_checked`): the chain may have been revoked and the verifier would not know. +With a snapshot and no revoked hop, the status has `checked=True` and carries the +snapshot's `as_of` (`not_revoked`). `ChainResult`, `PeerResult` and the +`ca2a verify-chain` output carry the same status, and the CLI prints it on every +successful verification. + +`max_revocation_staleness` (seconds, default unset) makes revocation checking +mandatory. With it set, a missing snapshot, or one whose `as_of` is more than +that many seconds before the evaluation time, raises +`REVOCATION_STATUS_UNKNOWN`. It is unset by default so that offline verification +with no revocation data keeps working. A stale snapshot can still prove a hop +revoked, since revocation is monotonic, so a revoked hop is reported as +`CREDENTIAL_REVOKED` before staleness is judged. + +When `at_time` is supplied, only statements with `issued_at` at or before it +count, so an audit of a past decision is not rewritten by a revocation issued +afterwards. Without `at_time` (a live decision), every statement the verifier +holds applies, whatever the revoker's clock said. + +### Out of scope + +cA2A defines no protocol for publishing or fetching revocation statements. +Getting current snapshots to verifiers is the deployment's job. A verifier with +no snapshot cannot learn of a revocation; it is told that it did not check. A +snapshot supplier can withhold statements. `PeerNode` accepts a +`revocation_source` callable that it consults on every call, but the +config-driven `ca2a start` does not load one. + ## Attenuation is the whole point diff --git a/docs/spec/error-codes.md b/docs/spec/error-codes.md index 6fe732e..f6627d7 100644 --- a/docs/spec/error-codes.md +++ b/docs/spec/error-codes.md @@ -18,6 +18,9 @@ An error also carries a human-readable message and an optional `detail`. The mes | `CredentialReplay` | `CREDENTIAL_REPLAY` | 409 | A `credential_id` appears more than once in a single chain. Raised by `verify_chain`. | | `CredentialNotYetValid` | `CREDENTIAL_NOT_YET_VALID` | 403 | A hop's `not_before` bound is after the evaluation time. The chain is well formed and validly signed, but the grant is not yet in force. Raised by `verify_chain`. | | `CredentialExpired` | `CREDENTIAL_EXPIRED` | 403 | A hop's `not_after` bound is before the evaluation time. Raised by `verify_chain`. | +| `CredentialRevoked` | `CREDENTIAL_REVOKED` | 403 | A hop's credential was revoked by its issuer or by an issuer above it in the chain, according to the revocation snapshot the verifier supplied. Every chain containing that hop is refused. Raised by `verify_chain` only when a snapshot is supplied. See [delegation chain](delegation-chain.md#ca2a-delegation-revocation). | +| `RevocationStatusUnknown` | `REVOCATION_STATUS_UNKNOWN` | 503 | The verifier set `max_revocation_staleness` and has no revocation snapshot, or its snapshot's `as_of` is older than that bound at the evaluation time. Nothing is wrong with the chain; the verifier cannot currently establish that it has not been revoked, and fails closed. Raised by `verify_chain`. | +| `InvalidRevocation` | `INVALID_REVOCATION` | 400 | A revocation statement or snapshot is malformed, unsigned, or carries a signature that does not verify, or a snapshot file is missing or not valid JSON. A snapshot with any such statement is refused as a whole. Raised by `RevocationStatement`, `RevocationSnapshot` and `load_revocation_snapshot`. | | `HolderProofInvalid` | `HOLDER_PROOF_INVALID` | 401 | The presenter of a delegation chain did not prove it controls the leaf `subject`: no proof was presented, the proof was malformed, it answered a challenge this callee did not issue or which has expired, or its signature did not verify over the exact request being made. 401 rather than 403 because the chain may well carry the authority requested while the caller has not shown it is the party that authority was delegated to. Distinct from `ATTESTATION_FAILED`, which is about what the caller is *running*: a caller can appraise perfectly and still fail this. Raised by `verify_holder_proof`, `handle_peer_request`, and the A2A adapter on a malformed proof. See [profile](profile.md) P-4a. | | `AttestationUnsupported` | `ATTESTATION_UNSUPPORTED` | 500 | An attestation provider was requested that the host cannot supply. Raised by any provider's `attest` when the host lacks what its collector needs, and by `OpaqueProvider`, which has no collector. The `detail` names the missing piece. See [Peer Attestation](attestation.md). | | `AttestationFailed` | `ATTESTATION_FAILED` | 412 | Attestation evidence was present but did not verify. Raised by the SEV-SNP verifier on a malformed report, an untrusted or broken certificate chain, a bad report signature, or a measurement / report-data mismatch. See [Peer Attestation](attestation.md). | @@ -29,7 +32,7 @@ An error also carries a human-readable message and an optional `detail`. The mes ## Which errors are live today -`ConfigError`, `InvalidCredential`, `ScopeEscalation`, `BrokenDelegationLink`, `DelegationDepthExceeded`, `CredentialReplay`, `CredentialNotYetValid`, `CredentialExpired`, , `ProvenanceLinkBroken`, and `TraceDigestUnsupported` are raised by shipping code paths: attenuated delegation, offline chain verification, and the provenance DAG. `ScopeNotPermitted` is raised by the peer-call enforcement decision core (`enforce_peer_call`), and `SealedChannelError` by the sealed channel (`SealedChannel.seal`, `open_sealed`), both of which are implemented. `TransportError` is raised by the A2A metadata adapter when cA2A keys are present but cannot be parsed into a `PeerRequest`. +`ConfigError`, `InvalidCredential`, `ScopeEscalation`, `BrokenDelegationLink`, `DelegationDepthExceeded`, `CredentialReplay`, `CredentialNotYetValid`, `CredentialExpired`, `CredentialRevoked`, `RevocationStatusUnknown`, `InvalidRevocation`, `ProvenanceLinkBroken`, and `TraceDigestUnsupported` are raised by shipping code paths: attenuated delegation, offline chain verification, and the provenance DAG. `ScopeNotPermitted` is raised by the peer-call enforcement decision core (`enforce_peer_call`), and `SealedChannelError` by the sealed channel (`SealedChannel.seal`, `open_sealed`), both of which are implemented. `TransportError` is raised by the A2A metadata adapter when cA2A keys are present but cannot be parsed into a `PeerRequest`. `AttestationFailed` is raised by the SEV-SNP verifier (chain, report signature, and measurement binding), and by a collector whose hardware returned evidence that does not commit the key and nonce it asked for. `AttestationUnsupported` is raised where a host cannot collect at all: no TPM or tpm2-pytss for `tpm`, no configfs-TSM or guest device for `sev-snp` and `tdx`, and on Azure confidential VMs, where SEV-SNP runs behind a paravisor that owns `REPORT_DATA`. See [Peer Attestation](attestation.md) and [ROADMAP.md](../../ROADMAP.md). @@ -57,7 +60,7 @@ Verification fails closed. `verify_chain`, `verify_dag`, and `cross_check_chain` ## See also -- [Delegation Chain](delegation-chain.md) for the checks behind `ScopeEscalation`, `BrokenDelegationLink`, `DelegationDepthExceeded`, `CredentialReplay`, `CredentialNotYetValid`, and `CredentialExpired`. +- [Delegation Chain](delegation-chain.md) for the checks behind `ScopeEscalation`, `BrokenDelegationLink`, `DelegationDepthExceeded`, `CredentialReplay`, `CredentialNotYetValid`, `CredentialExpired`, `CredentialRevoked`, `RevocationStatusUnknown`, and `InvalidRevocation`. - [Provenance DAG](provenance-dag.md) for the checks behind `ProvenanceLinkBroken`. - [Verification Library](verification-library.md) for `verify_chain`, `verify_chain_file`, `verify_dag`, and `cross_check_chain`. - [Failure Modes](failure-modes.md) for how these errors map to observable runtime behavior. diff --git a/docs/spec/profile.md b/docs/spec/profile.md index dc86536..94fc071 100644 --- a/docs/spec/profile.md +++ b/docs/spec/profile.md @@ -54,6 +54,8 @@ A peer that does not implement this profile MUST ignore the cA2A extension field A callee MUST verify the presented delegation chain before acting: every credential's signature, the continuity of each parent link, and the attenuation rule that a child's scope is a subset of its parent's. A callee MUST reject a chain whose depth exceeds its configured maximum, and MUST reject a credential replayed from a different chain. Verification MUST be possible offline, without contacting the issuer. +> **Revocation is an optional input, not a network dependency.** A verifier can also be given a snapshot of signed revocation statements and will then refuse a chain containing a hop revoked by its issuer or an issuer above it. Consulting a snapshot contacts nobody, so this does not conflict with offline verification. Without one, the reference verifier reports revocation as `not_checked` rather than implying the chain is unrevoked. See [delegation chain](delegation-chain.md#ca2a-delegation-revocation). + ### P-4a Holder binding A callee MUST NOT act on a delegation chain until the presenter has proved it controls the private key of the leaf credential's `subject`. The callee MUST issue the challenge the proof answers, and MUST reject a proof that does not commit to the callee's own identity, that challenge, the leaf `credential_id` and `subject`, the requested capability, the `record_id`, the `parent_record_hash`, the sealed payload if one is present, and the caller's own offered channel key if one is present. A chain presented without such a proof MUST be refused with `HOLDER_PROOF_INVALID`. diff --git a/docs/spec/threat-model.md b/docs/spec/threat-model.md index f69b1b2..e472689 100644 --- a/docs/spec/threat-model.md +++ b/docs/spec/threat-model.md @@ -31,6 +31,7 @@ Out of adversary scope: breaking the underlying cryptographic primitives (Ed2551 | Tampered peer wearing a valid Agent Card | Attestation: measurement must match an expected value before a task is accepted | | 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 compromised delegate keeps using a grant inside its validity window | Revocation by the grant's issuer or an issuer above it, checked against a revocation snapshot the verifier supplies. Only as current as that snapshot; see residual risks | | 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. A hop cannot be reparented in flight either: the holder proof commits to `parent_record_hash`, so altering it invalidates the proof before a record is emitted | @@ -43,4 +44,6 @@ Because attestation and sealing are not yet implemented (Tier 2/3), this release Closing the window entirely needs state, and the place for it is the challenge rather than the proof, so that the profile carries one such decision instead of two. A deployment that requires exactly-once should supply a stateful challenge and accept the shared-store or sticky-routing cost that comes with it. -**Delegated authority cannot be actively withdrawn.** A credential can carry a validity window (`not_before` / `not_after`, see [delegation chain](delegation-chain.md)), which bounds how long a compromised delegate keeps what it was granted — but there is no revocation path, so inside a still-valid window the grant cannot be withdrawn early. This interacts with P-4's requirement that verification work offline, since an offline verifier cannot learn that a credential was revoked. +**Revocation is only as current as the verifier's data.** The issuer of a credential, or any issuer above it in the chain, can withdraw it before `not_after` by signing a revocation statement that names the credential by digest (see [delegation chain](delegation-chain.md#ca2a-delegation-revocation)). A verifier given a revocation snapshot refuses a chain containing a revoked hop with `CREDENTIAL_REVOKED`, and every grant beneath that hop falls with it. A delegate cannot revoke its delegator, and a statement from any key outside that line of issuers has no effect. + +P-4 still requires verification to work offline, so a verifier that has no snapshot still verifies, and it still cannot learn that a credential was revoked. What changed is that it is told so: the verification result reports revocation as `not_checked` rather than leaving the caller to assume. A verifier that has a snapshot knows what the snapshot held at its `as_of` time and nothing later. Whoever supplies snapshots can withhold a statement, though not forge one, since each statement is signed by its revoker. A verifier can set `max_revocation_staleness` to refuse to decide without a snapshot of bounded age; that bounds the window between a revocation being issued and a verifier acting on it, but does not close it. Getting statements to verifiers is the deployment's job, and cA2A defines no distribution protocol. The validity window remains the backstop for a verifier that never receives the statement. diff --git a/docs/spec/verification-library.md b/docs/spec/verification-library.md index fc8f8ef..b6f8be7 100644 --- a/docs/spec/verification-library.md +++ b/docs/spec/verification-library.md @@ -10,16 +10,28 @@ from ca2a_verify import verify_delegation_chain, verify_chain_file, ChainResult result: ChainResult = verify_chain_file( "chain.json", trusted_root_issuers={""} ) -# result.hops, result.root_issuer, result.leaf_subject, result.leaf_scope +# result.hops, result.root_issuer, result.leaf_subject, result.leaf_scope, +# result.revocation ("not_checked" here, since no snapshot was supplied) ``` - `verify_delegation_chain(chain, trusted_root_issuers=..., max_depth=8, at_time=None)` verifies a list of `DelegationCredential` against an explicit local root trust set and returns a `ChainResult` summary, or raises a `CA2AError` subtype. - `verify_chain_file(path, trusted_root_issuers=..., max_depth=8, at_time=None)` loads a chain from JSON (a bare list, or `{"chain": [...]}`) and verifies it against that trust set. +- Both also take `revocations=None` and `max_revocation_staleness=None`. `load_revocation_snapshot(path)` loads a snapshot (`{"as_of": ..., "revocations": [...]}`) and checks every statement's signature. Root trust is mandatory. A self-consistent chain from an unknown root is cryptographically well formed but is not authorized and therefore does not produce a successful verification result. `at_time` is the Unix time validity windows are evaluated at; `None` means the current time. An auditor replaying recorded evidence passes the time the action was decided, not its own. See [delegation chain](delegation-chain.md). +## Revocation status + +`ChainResult.revocation_checked` is False, and `ChainResult.revocation` is `"not_checked"`, whenever no revocation snapshot was supplied. The chain verified, but whether any hop was revoked is not known. With a snapshot and no revoked hop they are True and `"not_revoked"`, and `revocation_as_of` is the snapshot's `as_of`. The CLI prints the same field: + +```bash +ca2a verify-chain --chain chain.json --trusted-root-issuer --revocations revocations.json --max-revocation-staleness 300 +``` + +Consulting a snapshot contacts nobody, so verification stays offline. Fetching current snapshots is the deployment's job. See [delegation chain](delegation-chain.md#ca2a-delegation-revocation). + ## Errors All verification failures are subtypes of `CA2AError`, re-exported as `VerificationError`. Each carries a stable `code` and an HTTP status. The specific codes and the invariants they map to are in [delegation chain](delegation-chain.md).