diff --git a/docs/rfcs/a2a-delegation-profile.md b/docs/rfcs/a2a-delegation-profile.md index d3cab573..00313a60 100644 --- a/docs/rfcs/a2a-delegation-profile.md +++ b/docs/rfcs/a2a-delegation-profile.md @@ -1,11 +1,11 @@ # RFC Proposal: A2A delegation-link verification profile -When one AI agent hands a task to another, each can write a TRACE record, and the later record points back to the earlier one (a delegation link). This proposal sets out how a checker follows those links and decides whether each hand-off was allowed. It is for people building verifiers, or agents that pass work to each other over A2A (the agent-to-agent protocol). It is a draft: nothing in it binds an implementation yet, and it ships with 24 test vectors so independent implementations can be compared. +When one AI agent hands a task to another, each can write a TRACE record, and the later record points back to the earlier one (a delegation link). This proposal sets out how a checker follows those links and decides whether each hand-off was allowed. It is for people building verifiers, or agents that pass work to each other over A2A (the agent-to-agent protocol). It is a draft: nothing in it binds an implementation yet, and it ships with 26 test vectors so independent implementations can be compared. **Status:** Draft proposal. Binds nothing. **Scope:** Verification rules for the existing `delegation` block. No schema change. **Target:** `spec/trace-v0.2.md` §3.1 surface, for the v0.3 A2A profile named in `ROADMAP.md`. -**Conformance material: ** `examples/delegation-link/`: 24 vectors, generator, published key. +**Conformance material: ** `examples/delegation-link/`: 26 vectors, generator, published key. Requirement keywords are lowercase throughout this document, deliberately. `CONTRIBUTING.md` draws the line that normative text lives in the specifications and informative text binds no @@ -88,7 +88,7 @@ an answer that looks like verification and is not. Naming them is half the profi ## 3. The rules -Ten rules, each with a stable code, a classification, and two independent vectors. +Eleven rules, each with a stable code, a classification, and two independent vectors. Classification is the three-way distinction the conformance corpus in `agentrust-io/ca2a` already uses for its action cases: a structural failure, an authority failure, and a chain that could not be read are three different findings, and a verifier that collapses any two @@ -106,6 +106,7 @@ of them is reporting something other than what it found. | D-8 | `credential_window` | authorization | This record's `iat` falls inside the credential's validity window. | 16, 17 | | D-9 | `data_class_widened` | authorization | A hop's `data_class` is no more sensitive than its parent's, per the supplied lattice. | 18, 19 | | D-10 | `digest_algorithm_unsupported` | unverifiable | A link naming an algorithm the verifier cannot compute yields an unverifiable chain. | 22, 23 | +| D-11 | `data_class_unknown` | unverifiable | A hop whose child or parent `data_class` is absent from `data_class_lattice` (the same string at both ends included) yields an unverifiable chain. | 25, 26 | Provenance outranks authorization in the reported classification. A chain whose structure is broken has no established parent for a credential to be judged against, so reporting an @@ -188,11 +189,11 @@ freshness policy on top of this profile; no rule here reads it. ## 5. The conformance corpus -`examples/delegation-link/` holds 24 vectors. Each is one scenario: a complete record set, +`examples/delegation-link/` holds 26 vectors. Each is one scenario: a complete record set, the verifier context to judge it under, and the classification and codes it expects, so a third party can score an implementation without running anything from this repository. -Twenty of the twenty-four score the ten rules above, two per rule with a +Twenty-two of the twenty-six score the eleven rules above, two per rule with a declared implementation defect that one vector catches and the other misses. Three more (`01` to `03`) are clean chains, expected to verify with no rule firing. The remaining vector, `24-parent-key-supplementary-plane.json`, is @@ -222,10 +223,11 @@ Coverage is held to the discipline argued in #124 and executed in `tests/test_vector_completeness.py`: two load-bearing vectors per rule, and: the part that is not satisfied by writing the same vector twice: at least one declared implementation defect that one vector catches and the other misses. `tests/test_delegation_completeness.py` -declares those defects, all ten of them modelling a real shortcut: verifying the leaf only, -anchoring on any trusted key found, an off-by-one bound, case-insensitive identifier lookup, -issuer and holder compared to the wrong ends, half a validity window, narrowing checked at -one hop. +declares those defects, all eleven of them modelling a real shortcut: verifying the leaf only, +anchoring on any trusted key found, an off-by-one bound, reading only the first link's +digest algorithm, resolving over the signed body, case-insensitive identifier lookup, +issuer and holder compared to the wrong ends, half a validity window, checking only the +child's data class against the lattice, and narrowing checked at one hop. Vector identifiers are `TRACE-DELEG-NNN` and are never reused. They are deliberately not `ACTION-*`: that namespace belongs to ca2a's conformance set, whose own rule is that its diff --git a/examples/delegation-link/25-data-class-unknown-at-leaf.json b/examples/delegation-link/25-data-class-unknown-at-leaf.json new file mode 100644 index 00000000..86b7b04d --- /dev/null +++ b/examples/delegation-link/25-data-class-unknown-at-leaf.json @@ -0,0 +1,134 @@ +{ + "id": "TRACE-DELEG-025", + "name": "data-class-unknown-at-leaf", + "description": "The leaf declares `top-secret`, which is schema-valid and absent from the verifier's `data_class_lattice`. Because a class outside the supplied ordering cannot be compared against its parent, the hop's narrowing is `unverifiable` rather than silently accepted as `verified`.", + "spec": "docs/rfcs/a2a-delegation-profile.md", + "profile": "trace.a2a.delegation-link.v0", + "context": { + "leaf": "sha256:043e46131b38758706e4298e7e3d2d70bea5f44a23796ad56eb7ff62888a3e78", + "now": 1785000000, + "max_depth": 4, + "supported_digest_algorithms": [ + "sha256" + ], + "data_class_lattice": [ + "public", + "internal", + "confidential", + "restricted" + ], + "trusted_root_keys": [ + { + "kty": "OKP", + "crv": "Ed25519", + "x": "pNTZUXlAITEWbtVbHk6zGRwVD73s0BEakqkKKaFQyZ4" + } + ], + "credentials": { + "cred:orchestrator-to-planner": { + "issuer": "spiffe://acme.example/agent/orchestrator", + "holder": "spiffe://acme.example/agent/planner", + "not_before": 1784913600, + "not_after": 1785086400 + }, + "cred:planner-to-executor": { + "issuer": "spiffe://acme.example/agent/planner", + "holder": "spiffe://acme.example/agent/executor", + "not_before": 1784913600, + "not_after": 1785086400 + }, + "cred:executor-to-courier": { + "issuer": "spiffe://acme.example/agent/executor", + "holder": "spiffe://acme.example/agent/courier", + "not_before": 1784913600, + "not_after": 1785086400 + }, + "cred:courier-to-auditor": { + "issuer": "spiffe://acme.example/agent/courier", + "holder": "spiffe://acme.example/agent/auditor", + "not_before": 1784913600, + "not_after": 1785086400 + } + } + }, + "records": [ + { + "eat_profile": "tag:agentrust-io.com,2026:trace-v0.2", + "iat": 1785000000, + "subject": "spiffe://acme.example/agent/planner", + "model": { + "provider": "anthropic", + "model_id": "claude-sonnet-4-6" + }, + "runtime": { + "platform": "software-only", + "measurement": "sha256:0000000000000000000000000000000000000000000000000000000000000000" + }, + "policy": { + "bundle_hash": "sha256:aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa", + "enforcement_mode": "enforce" + }, + "data_class": "top-secret", + "build_provenance": { + "slsa_level": 0, + "digest": "sha256:bbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbb" + }, + "appraisal": { + "status": "affirming", + "verifier": "https://verifier.example/v1" + }, + "delegation": { + "parent_record_hash": "sha256:b1d65c65d7191e1c6f57f8902548475671edad51035f1542f76a4e45337052b6", + "credential_id": "cred:orchestrator-to-planner" + }, + "cnf": { + "jwk": { + "kty": "OKP", + "crv": "Ed25519", + "x": "W2BsRtHZW8BVdiCsCxrNxvv4LOfgBhuS_2v7TU7TyHU" + } + }, + "signature": "lFd-i3GNlYdytslfkVq8Zm2RDPznm2BkEO-7opbCz8LEHugruH3i5NmhpnDbwFxcCdjpYm_cJFkkyPsqv6v7DA" + }, + { + "eat_profile": "tag:agentrust-io.com,2026:trace-v0.2", + "iat": 1785000000, + "subject": "spiffe://acme.example/agent/orchestrator", + "model": { + "provider": "anthropic", + "model_id": "claude-sonnet-4-6" + }, + "runtime": { + "platform": "software-only", + "measurement": "sha256:0000000000000000000000000000000000000000000000000000000000000000" + }, + "policy": { + "bundle_hash": "sha256:aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa", + "enforcement_mode": "enforce" + }, + "data_class": "restricted", + "build_provenance": { + "slsa_level": 0, + "digest": "sha256:bbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbb" + }, + "appraisal": { + "status": "affirming", + "verifier": "https://verifier.example/v1" + }, + "cnf": { + "jwk": { + "kty": "OKP", + "crv": "Ed25519", + "x": "pNTZUXlAITEWbtVbHk6zGRwVD73s0BEakqkKKaFQyZ4" + } + }, + "signature": "m3TAE_Z_chc0gs8HMh1KAN4NECFLP09kiqyQPBDz9w9G0_iYoQ5-I94gpptC1a4xiRloSFLyTdprEmr9CFD_DA" + } + ], + "expected": { + "classification": "unverifiable", + "codes": [ + "data_class_unknown" + ] + } +} diff --git a/examples/delegation-link/26-data-class-unknown-at-parent.json b/examples/delegation-link/26-data-class-unknown-at-parent.json new file mode 100644 index 00000000..2ae6f4c0 --- /dev/null +++ b/examples/delegation-link/26-data-class-unknown-at-parent.json @@ -0,0 +1,134 @@ +{ + "id": "TRACE-DELEG-026", + "name": "data-class-unknown-at-parent", + "description": "The parent declares `top-secret`, absent from `data_class_lattice`, while the leaf declares `confidential`, which is in the lattice. Narrowing compares both ends of the hop: a verifier that checks only whether the child's class is in the lattice passes this chain as `verified` while catching vector 25.", + "spec": "docs/rfcs/a2a-delegation-profile.md", + "profile": "trace.a2a.delegation-link.v0", + "context": { + "leaf": "sha256:7a1b046fe79faafc8e9515c90afc26da3a2e05d0eb0a0090b06599fba7127141", + "now": 1785000000, + "max_depth": 4, + "supported_digest_algorithms": [ + "sha256" + ], + "data_class_lattice": [ + "public", + "internal", + "confidential", + "restricted" + ], + "trusted_root_keys": [ + { + "kty": "OKP", + "crv": "Ed25519", + "x": "pNTZUXlAITEWbtVbHk6zGRwVD73s0BEakqkKKaFQyZ4" + } + ], + "credentials": { + "cred:orchestrator-to-planner": { + "issuer": "spiffe://acme.example/agent/orchestrator", + "holder": "spiffe://acme.example/agent/planner", + "not_before": 1784913600, + "not_after": 1785086400 + }, + "cred:planner-to-executor": { + "issuer": "spiffe://acme.example/agent/planner", + "holder": "spiffe://acme.example/agent/executor", + "not_before": 1784913600, + "not_after": 1785086400 + }, + "cred:executor-to-courier": { + "issuer": "spiffe://acme.example/agent/executor", + "holder": "spiffe://acme.example/agent/courier", + "not_before": 1784913600, + "not_after": 1785086400 + }, + "cred:courier-to-auditor": { + "issuer": "spiffe://acme.example/agent/courier", + "holder": "spiffe://acme.example/agent/auditor", + "not_before": 1784913600, + "not_after": 1785086400 + } + } + }, + "records": [ + { + "eat_profile": "tag:agentrust-io.com,2026:trace-v0.2", + "iat": 1785000000, + "subject": "spiffe://acme.example/agent/planner", + "model": { + "provider": "anthropic", + "model_id": "claude-sonnet-4-6" + }, + "runtime": { + "platform": "software-only", + "measurement": "sha256:0000000000000000000000000000000000000000000000000000000000000000" + }, + "policy": { + "bundle_hash": "sha256:aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa", + "enforcement_mode": "enforce" + }, + "data_class": "confidential", + "build_provenance": { + "slsa_level": 0, + "digest": "sha256:bbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbb" + }, + "appraisal": { + "status": "affirming", + "verifier": "https://verifier.example/v1" + }, + "delegation": { + "parent_record_hash": "sha256:e075b86c9588fb5addd9380646ec063f6f85a453ecf0b49aeaaaca4cc55caabb", + "credential_id": "cred:orchestrator-to-planner" + }, + "cnf": { + "jwk": { + "kty": "OKP", + "crv": "Ed25519", + "x": "W2BsRtHZW8BVdiCsCxrNxvv4LOfgBhuS_2v7TU7TyHU" + } + }, + "signature": "PkPQBkfZzYGVDa9O6fj4bwmAUK6VfbZ5bg8qkdEt_WVONX6W5Tu2QO88m_LyxF6HKuxiZflJOZBiPdKHU5__DA" + }, + { + "eat_profile": "tag:agentrust-io.com,2026:trace-v0.2", + "iat": 1785000000, + "subject": "spiffe://acme.example/agent/orchestrator", + "model": { + "provider": "anthropic", + "model_id": "claude-sonnet-4-6" + }, + "runtime": { + "platform": "software-only", + "measurement": "sha256:0000000000000000000000000000000000000000000000000000000000000000" + }, + "policy": { + "bundle_hash": "sha256:aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa", + "enforcement_mode": "enforce" + }, + "data_class": "top-secret", + "build_provenance": { + "slsa_level": 0, + "digest": "sha256:bbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbb" + }, + "appraisal": { + "status": "affirming", + "verifier": "https://verifier.example/v1" + }, + "cnf": { + "jwk": { + "kty": "OKP", + "crv": "Ed25519", + "x": "pNTZUXlAITEWbtVbHk6zGRwVD73s0BEakqkKKaFQyZ4" + } + }, + "signature": "SovhFNqtpvIkmDdgLtW4Tsrb_KLzjiQ1gbiI3bxgUh5GwFNYr18ZQ7YZOwqM6rhSIocskYJNj6zVwuJjzgXZCg" + } + ], + "expected": { + "classification": "unverifiable", + "codes": [ + "data_class_unknown" + ] + } +} diff --git a/examples/delegation-link/README.md b/examples/delegation-link/README.md index 680e6103..a17c2c6c 100644 --- a/examples/delegation-link/README.md +++ b/examples/delegation-link/README.md @@ -61,7 +61,7 @@ and compares bytes on every run, with no entry in its `NOT_GENERATED` ledger. ## Coverage -Ten rules, two load-bearing vectors each, and for every rule at least one declared +Eleven rules, two load-bearing vectors each, and for every rule at least one declared implementation defect that one of its vectors catches and the other misses. `tests/test_delegation_completeness.py` enforces all three, and records the margins so they cannot silently thin. @@ -69,8 +69,9 @@ cannot silently thin. The pairs are not two views of the same mistake. They are built around a specific shortcut a real implementation takes — verifying the leaf only, anchoring on any trusted key it finds, an off-by-one bound, case-insensitive lookup of an opaque identifier, issuer and holder -compared to the wrong ends of the hop, half a validity window, narrowing checked at one hop, -the link algorithm read once and assumed uniform. +compared to the wrong ends of the hop, half a validity window, checking only the child's +data class against the lattice, narrowing checked at one hop, the link algorithm read once +and assumed uniform. Vector 05 is the one to read first. It is a complete, correctly signed chain whose only defect is which bytes its link was computed over, and it is the whole of the difference diff --git a/examples/delegation-link/gen_delegation_vectors.py b/examples/delegation-link/gen_delegation_vectors.py index 6e2e9010..6efd5c9f 100644 --- a/examples/delegation-link/gen_delegation_vectors.py +++ b/examples/delegation-link/gen_delegation_vectors.py @@ -689,6 +689,39 @@ def add(filename: str, doc: dict[str, Any]) -> None: classification="verified", codes=[], credentials=WELL_FORMED, )) + # -- data_class_unknown ----------------------------------------------------- + + add("25-data-class-unknown-at-leaf.json", vector( + "TRACE-DELEG-025", "data-class-unknown-at-leaf", + "The leaf declares `top-secret`, which is schema-valid and absent from " + "the verifier's `data_class_lattice`. Because a class outside the " + "supplied ordering cannot be compared against its parent, the hop's " + "narrowing is `unverifiable` rather than silently accepted as `verified`.", + build_chain([ + {"role": "orchestrator", "data_class": "restricted"}, + {"role": "planner", "data_class": "top-secret", + "credential_id": CRED_PLANNER[0]}, + ]), + classification="unverifiable", codes=["data_class_unknown"], + credentials=WELL_FORMED, + )) + + add("26-data-class-unknown-at-parent.json", vector( + "TRACE-DELEG-026", "data-class-unknown-at-parent", + "The parent declares `top-secret`, absent from `data_class_lattice`, " + "while the leaf declares `confidential`, which is in the lattice. " + "Narrowing compares both ends of the hop: a verifier that checks only " + "whether the child's class is in the lattice passes this chain as " + "`verified` while catching vector 25.", + build_chain([ + {"role": "orchestrator", "data_class": "top-secret"}, + {"role": "planner", "data_class": "confidential", + "credential_id": CRED_PLANNER[0]}, + ]), + classification="unverifiable", codes=["data_class_unknown"], + credentials=WELL_FORMED, + )) + for name, doc in sorted(out): (OUT / name).write_text( json.dumps(doc, indent=2, ensure_ascii=False) + "\n", encoding="utf-8", newline="\n" diff --git a/src/agentrust_trace/__init__.py b/src/agentrust_trace/__init__.py index befacae2..b8da47ee 100644 --- a/src/agentrust_trace/__init__.py +++ b/src/agentrust_trace/__init__.py @@ -128,3 +128,7 @@ from agentrust_trace import intent_bridge as intent_bridge # noqa: E402 __all__ = [*__all__, "intent_bridge"] +from agentrust_trace import delegation as delegation # noqa: E402 + +__all__ = [*__all__, "delegation"] + diff --git a/src/agentrust_trace/delegation.py b/src/agentrust_trace/delegation.py new file mode 100644 index 00000000..96d1af0e --- /dev/null +++ b/src/agentrust_trace/delegation.py @@ -0,0 +1,644 @@ +"""Delegation-link digest, single-hop verification, and A2A chain appraisal. + +Section 3.1.3 of ``spec/trace-v0.2.md`` defines the normative single-hop delegation +commitment: a child Trust Record's ``delegation.parent_record_hash`` is the +``sha256`` or ``sha384`` digest over the complete RFC 8785 (JCS) canonical form of +the parent Trust Record, with its ``signature`` member present: + + delegation.parent_record_hash = ":" + hex(H(JCS(parent_record))) + +``parent_record_hash`` and ``verify_delegation_link`` implement that normative +single-hop link digest and mirror ``trust-record-verify/src/delegation.ts``. + +``verify_delegation_chain`` executes the eleven-rule A2A delegation-link verification +walk from the draft profile in ``docs/rfcs/a2a-delegation-profile.md`` (which is +informative and binds no implementation yet), scored by the 26 test vectors under +``examples/delegation-link/``. +""" + +from __future__ import annotations + +import hashlib +import hmac +from collections.abc import Callable, Iterable, Sequence +from dataclasses import dataclass, field, replace +from typing import Any + +import rfc8785 +from cryptography.exceptions import InvalidSignature +from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PublicKey +from jsonschema import ValidationError + +from agentrust_trace.sign import _b64url_decode, _canonical_bytes, _integer_value +from agentrust_trace.validate import validate_json + +DELEGATION_DIGEST_ALGORITHMS: tuple[str, ...] = ("sha256", "sha384") +"""Digest algorithms permitted for ``delegation.parent_record_hash`` (section 3.1.3).""" + +CLASSIFICATIONS: frozenset[str] = frozenset( + {"verified", "provenance-invalid", "authorization-invalid", "unverifiable"} +) +"""Three-way A2A chain appraisal classifications plus ``verified``.""" + + +class DelegationError(ValueError): + """Raised when a delegation link or chain verification input is malformed or invalid.""" + + +@dataclass(frozen=True) +class DelegationLink: + """Verified single-hop commitment from a child Trust Record to its parent.""" + + parent_record_hash: str + algorithm: str + credential_id: str + + +@dataclass(frozen=True) +class ChainResult: + """Appraisal outcome for an A2A delegation chain.""" + + classification: str + failures: list[str] + warnings: list[str] + depth: int + + def __post_init__(self) -> None: + if self.classification not in CLASSIFICATIONS: + raise DelegationError(f"undeclared outcome {self.classification!r}") + + @property + def codes(self) -> list[str]: + return sorted(set(self.failures) | set(self.warnings)) + + +def parent_record_hash(record: dict[str, Any], algorithm: str = "sha256") -> str: + """Return the section 3.1.3 chain digest of *record* as ``":"``. + + The pre-image is the complete *record* as published, ``signature`` included, + canonicalised with RFC 8785 (JCS). Raises :class:`DelegationError` if *record* + is not a JSON object, cannot be canonicalised under RFC 8785, or if *algorithm* + is not in :data:`DELEGATION_DIGEST_ALGORITHMS`. + """ + if not isinstance(algorithm, str) or algorithm not in DELEGATION_DIGEST_ALGORITHMS: + raise DelegationError(f"unsupported chain digest algorithm {algorithm!r}") + if not isinstance(record, dict): + raise DelegationError( + f"record must be a JSON object, got {type(record).__name__}" + ) + try: + canonical = _canonical_bytes(record) + except (rfc8785.CanonicalizationError, ValueError, TypeError) as exc: + raise DelegationError( + f"record has no RFC 8785 canonical form: {exc}" + ) from exc + return f"{algorithm}:{hashlib.new(algorithm, canonical).hexdigest()}" + + +def _normalize_supported_algorithms(value: Any) -> tuple[str, ...]: + if isinstance(value, (str, bytes, bytearray, dict)) or not isinstance( + value, Iterable + ): + raise DelegationError( + "supported_digest_algorithms must be an iterable of algorithm names" + ) + items = tuple(value) + if not items or not all(isinstance(alg, str) and alg for alg in items): + raise DelegationError( + "supported_digest_algorithms must contain non-empty algorithm strings" + ) + return items + + +def verify_delegation_link( + child: dict[str, Any], + parent: dict[str, Any], + *, + supported_digest_algorithms: Iterable[str] = DELEGATION_DIGEST_ALGORITHMS, +) -> DelegationLink: + """Verify that *parent* is the record *child* commits to in ``child["delegation"]``. + + Raises :class:`DelegationError` when *child* or *parent* is not a JSON object, + when *child* carries no ``delegation`` block, when the block is malformed, + when the digest algorithm prefix is not in *supported_digest_algorithms* or + :data:`DELEGATION_DIGEST_ALGORITHMS`, or when the recomputed digest of *parent* + does not match ``child["delegation"]["parent_record_hash"]``. + """ + supported = _normalize_supported_algorithms(supported_digest_algorithms) + if not isinstance(child, dict): + raise DelegationError( + f"child record must be a JSON object, got {type(child).__name__}" + ) + if not isinstance(parent, dict): + raise DelegationError( + f"parent record must be a JSON object, got {type(parent).__name__}" + ) + delegation = child.get("delegation") + if delegation is None: + raise DelegationError( + "the record has no delegation block, so it claims no parent hop" + ) + if not isinstance(delegation, dict): + raise DelegationError("the record's delegation member is not an object") + if set(delegation) != {"parent_record_hash", "credential_id"}: + raise DelegationError( + "the delegation block must contain only parent_record_hash and credential_id" + ) + stated = delegation.get("parent_record_hash") + credential_id = delegation.get("credential_id") + if ( + not isinstance(stated, str) + or not stated + or not isinstance(credential_id, str) + or not credential_id + ): + raise DelegationError( + "the delegation block needs a parent_record_hash and a credential_id" + ) + separator = stated.find(":") + algorithm = "" if separator == -1 else stated[:separator] + if algorithm not in supported: + raise DelegationError( + f"the link names digest algorithm {algorithm!r}, which this verifier does not " + "implement; section 3.1.3 forbids falling back to another one" + ) + if algorithm not in DELEGATION_DIGEST_ALGORITHMS: + raise DelegationError( + f"digest algorithm {algorithm!r} is not one this build can compute" + ) + computed = parent_record_hash(parent, algorithm) + if not (stated.isascii() and hmac.compare_digest(computed, stated)): + raise DelegationError( + "the candidate parent record does not digest to the value the child commits to" + ) + return DelegationLink( + parent_record_hash=stated, + algorithm=algorithm, + credential_id=credential_id, + ) + + +def _signature_invalid(record: dict[str, Any]) -> bool: + """Does *record* fail verification under the key it advertises in ``cnf.jwk``? + + The claimed key, not a trusted one: whether the key is trusted is a separate + question with its own rule, and merging them produces a verifier that cannot + tell "forged" from "signed by someone you have not heard of". + """ + cnf = record.get("cnf") + jwk = cnf.get("jwk") if isinstance(cnf, dict) else None + if not isinstance(jwk, dict): + return True + if ( + jwk.get("kty") != "OKP" + or jwk.get("crv") != "Ed25519" + or not isinstance(jwk.get("x"), str) + ): + return True + signature = record.get("signature") + if not isinstance(signature, str): + return True + body = {k: v for k, v in record.items() if k != "signature"} + try: + canonical_body = _canonical_bytes(body) + raw_x = _b64url_decode(jwk["x"], field="cnf.jwk.x") + raw_sig = _b64url_decode(signature, field="signature") + Ed25519PublicKey.from_public_bytes(raw_x).verify( + raw_sig, canonical_body + ) + except ( + InvalidSignature, + ValueError, + TypeError, + KeyError, + rfc8785.CanonicalizationError, + ): + return True + return False + + +def _jwk_key(jwk: dict[str, Any]) -> tuple[Any, ...]: + """Hashable identity for a JWK over key material only.""" + return (jwk.get("kty"), jwk.get("crv"), jwk.get("x"), jwk.get("y")) + + +@dataclass(frozen=True) +class Hop: + """One position in the delegation walk, carrying the state a rule inspects.""" + + record: dict[str, Any] + context: dict[str, Any] + index: dict[str, dict[str, Any]] + depth: int + parent: dict[str, Any] | None = None + + @property + def delegation(self) -> dict[str, Any]: + return self.record["delegation"] # type: ignore[no-any-return] + + @property + def link_algorithm(self) -> str: + return str(self.delegation["parent_record_hash"]).split(":", 1)[0] + + @property + def credential(self) -> dict[str, Any] | None: + creds: dict[str, dict[str, Any]] = self.context["credentials"] + return creds.get(self.delegation["credential_id"]) + + +@dataclass(frozen=True) +class Rule: + """One named obligation in the A2A delegation profile.""" + + code: str + severity: str + klass: str + path: str + check: Callable[[Hop], bool] = field(compare=False) + + +def _record_signature_invalid(hop: Hop) -> bool: + return _signature_invalid(hop.record) + + +def _root_key_untrusted(hop: Hop) -> bool: + trusted = {_jwk_key(j) for j in hop.context["trusted_root_keys"]} + return _jwk_key(hop.record.get("cnf", {}).get("jwk", {})) not in trusted + + +def _depth_exceeded(hop: Hop) -> bool: + return bool(hop.depth > hop.context["max_depth"]) + + +def _digest_algorithm_unsupported(hop: Hop) -> bool: + return hop.link_algorithm not in hop.context["supported_digest_algorithms"] + + +def _parent_not_found(hop: Hop) -> bool: + if _digest_algorithm_unsupported(hop): + return False + return hop.delegation["parent_record_hash"] not in hop.index + + +def _credential_unknown(hop: Hop) -> bool: + return hop.credential is None + + +def _credential_issuer_mismatch(hop: Hop) -> bool: + cred = hop.credential + if cred is None or hop.parent is None: + return False + return bool(cred["issuer"] != hop.parent["subject"]) + + +def _credential_holder_mismatch(hop: Hop) -> bool: + cred = hop.credential + if cred is None: + return False + return bool(cred["holder"] != hop.record["subject"]) + + +def _credential_window(hop: Hop) -> bool: + cred = hop.credential + if cred is None: + return False + # Judged at the hop's own `iat`. A chain does not expire because it is being + # read late, and a verifier using its own clock reports a different answer + # every day for the same evidence. + iat = _integer_value(hop.record["iat"], "iat") + return not bool(cred["not_before"] <= iat <= cred["not_after"]) + + +def _data_class_unknown(hop: Hop) -> bool: + if hop.parent is None: + return False + lattice: list[str] = hop.context["data_class_lattice"] + return ( + hop.record["data_class"] not in lattice + or hop.parent["data_class"] not in lattice + ) + + +def _data_class_widened(hop: Hop) -> bool: + if hop.parent is None or _data_class_unknown(hop): + return False + lattice: list[str] = hop.context["data_class_lattice"] + return lattice.index(hop.record["data_class"]) > lattice.index( + hop.parent["data_class"] + ) + + +RULES: tuple[Rule, ...] = ( + Rule( + "record_signature_invalid", + "failure", + "provenance", + "record", + _record_signature_invalid, + ), + Rule("root_key_untrusted", "failure", "provenance", "root", _root_key_untrusted), + Rule("depth_exceeded", "failure", "authorization", "resolve", _depth_exceeded), + Rule( + "digest_algorithm_unsupported", + "warning", + "unverifiable", + "resolve", + _digest_algorithm_unsupported, + ), + Rule("parent_not_found", "failure", "provenance", "resolve", _parent_not_found), + Rule( + "credential_unknown", "failure", "authorization", "link", _credential_unknown + ), + Rule( + "credential_issuer_mismatch", + "failure", + "authorization", + "link", + _credential_issuer_mismatch, + ), + Rule( + "credential_holder_mismatch", + "failure", + "authorization", + "link", + _credential_holder_mismatch, + ), + Rule("credential_window", "failure", "authorization", "link", _credential_window), + Rule( + "data_class_unknown", "warning", "unverifiable", "link", _data_class_unknown + ), + Rule( + "data_class_widened", "failure", "authorization", "link", _data_class_widened + ), +) + + +def _evaluate( + hop: Hop, rules: Sequence[Rule], path: str +) -> tuple[list[str], list[str]]: + failures: list[str] = [] + warnings: list[str] = [] + for rule in rules: + if rule.path == path and rule.check(hop): + (failures if rule.severity == "failure" else warnings).append(rule.code) + return failures, warnings + + +def _validate_record_shape(record: Any) -> dict[str, Any]: + if not isinstance(record, dict): + raise DelegationError( + f"each record must be a JSON object, got {type(record).__name__}" + ) + try: + validate_json(record) + except ValidationError as exc: + raise DelegationError( + f"record does not conform to the TRACE v0.2 schema: {exc.message}" + ) from exc + try: + _integer_value(record.get("iat"), "iat") + except ValueError as exc: + raise DelegationError(str(exc)) from exc + return record + + +def _validate_context(context: Any) -> dict[str, Any]: + if not isinstance(context, dict): + raise DelegationError( + f"context must be a JSON object, got {type(context).__name__}" + ) + required = ( + "trusted_root_keys", + "credentials", + "data_class_lattice", + "max_depth", + "supported_digest_algorithms", + ) + missing = [key for key in required if key not in context] + if missing: + raise DelegationError(f"context is missing required field(s): {missing}") + + supported = _normalize_supported_algorithms( + context["supported_digest_algorithms"] + ) + for alg in supported: + if alg not in DELEGATION_DIGEST_ALGORITHMS: + raise DelegationError( + f"context lists unsupported digest algorithm {alg!r}" + ) + + max_depth = context["max_depth"] + if isinstance(max_depth, bool) or not isinstance(max_depth, int) or max_depth < 0: + raise DelegationError("context.max_depth must be a non-negative integer") + + lattice = context["data_class_lattice"] + if ( + isinstance(lattice, (str, bytes, bytearray, dict)) + or not isinstance(lattice, Sequence) + or not all(isinstance(item, str) and item for item in lattice) + ): + raise DelegationError( + "context.data_class_lattice must be a sequence of non-empty strings" + ) + + trusted_keys = context["trusted_root_keys"] + if ( + isinstance(trusted_keys, (str, bytes, bytearray, dict)) + or not isinstance(trusted_keys, Iterable) + ): + raise DelegationError( + "context.trusted_root_keys must be an iterable of JWK objects" + ) + trusted_list = list(trusted_keys) + if not all(isinstance(k, dict) for k in trusted_list): + raise DelegationError( + "context.trusted_root_keys must contain only JWK objects" + ) + for jwk in trusted_list: + for member in ("kty", "crv", "x"): + val = jwk.get(member) + if not isinstance(val, str) or not val: + raise DelegationError( + f"context.trusted_root_keys entry must carry non-empty string {member!r}" + ) + if "y" in jwk and (not isinstance(jwk["y"], str) or not jwk["y"]): + raise DelegationError( + "context.trusted_root_keys entry 'y' must be a non-empty string when present" + ) + + credentials = context["credentials"] + if not isinstance(credentials, dict): + raise DelegationError("context.credentials must be a JSON object") + for cid, cred in credentials.items(): + if not isinstance(cid, str) or not cid or not isinstance(cred, dict): + raise DelegationError( + "context.credentials must map credential IDs to credential objects" + ) + if not isinstance(cred.get("issuer"), str) or not isinstance( + cred.get("holder"), str + ): + raise DelegationError( + f"credential {cid!r} must carry string issuer and holder" + ) + for bound in ("not_before", "not_after"): + val = cred.get(bound) + if isinstance(val, bool) or not isinstance(val, int): + raise DelegationError( + f"credential {cid!r} must carry integer {bound}" + ) + + return { + **context, + "supported_digest_algorithms": list(supported), + "data_class_lattice": list(lattice), + "trusted_root_keys": trusted_list, + } + + +def verify_delegation_chain( + records: Iterable[dict[str, Any]], + leaf: dict[str, Any] | str, + context: dict[str, Any], + *, + rules: Sequence[Rule] = RULES, +) -> ChainResult: + """Walk a delegation chain from *leaf* towards its root under *context*. + + The record set is a set: it arrives in no useful order, the leaf is named by + digest (or passed as the leaf record dict itself) rather than by position, + and a parent is found by looking its digest up rather than by taking the next + element. An implementation that reads ``records[0]`` as the root agrees with + this one on nothing. + + No cycle check, deliberately. A cycle needs each record's delegation block to + carry a digest covering the block that names it back, which is a hash + collision; the reachable analogue is an unbounded chain, and that is what + ``depth_exceeded`` is for. The RFC states this rather than leaving a reader to + wonder which of the two was forgotten. + + Raises :class:`DelegationError` when the caller's inputs (*records*, *leaf*, + *context*, or *rules*) are malformed, when any record fails schema validation, + or when *leaf* is not present in *records* under a supported digest algorithm. + """ + if isinstance(records, (str, bytes, bytearray, dict)) or not isinstance( + records, Iterable + ): + raise DelegationError("records must be an iterable of Trust Record objects") + record_list = [_validate_record_shape(r) for r in records] + if not record_list: + raise DelegationError("records must not be empty") + + validated_context = _validate_context(context) + + if isinstance(rules, (str, bytes, bytearray, dict)) or not isinstance( + rules, Sequence + ): + raise DelegationError("rules must be a sequence of Rule objects") + rule_tuple = tuple(rules) + if not all(isinstance(r, Rule) for r in rule_tuple): + raise DelegationError("rules must contain only Rule instances") + + supported = validated_context["supported_digest_algorithms"] + index: dict[str, dict[str, Any]] = {} + for algorithm in supported: + for record in record_list: + index[parent_record_hash(record, algorithm)] = record + + if isinstance(leaf, str): + if ":" not in leaf: + raise DelegationError("leaf digest must have an ':' prefix") + current = index.get(leaf) + elif isinstance(leaf, dict): + _validate_record_shape(leaf) + current = index.get(parent_record_hash(leaf, supported[0])) + else: + raise DelegationError( + f"leaf must be a Trust Record object or digest string, got {type(leaf).__name__}" + ) + + if current is None: + raise DelegationError( + "the leaf record is not in records under a supported digest algorithm" + ) + + failures: list[str] = [] + warnings: list[str] = [] + depth = 0 + visited: set[int] = set() + + while True: + hop = Hop( + record=current, + context=validated_context, + index=index, + depth=depth, + ) + + found, warned = _evaluate(hop, rule_tuple, "record") + failures += found + warnings += warned + + if "delegation" not in current: + found, warned = _evaluate(hop, rule_tuple, "root") + failures += found + warnings += warned + break + + depth += 1 + hop = replace(hop, depth=depth) + found, warned = _evaluate(hop, rule_tuple, "resolve") + failures += found + warnings += warned + + # Whether the walk can continue is a structural fact about the records, + # asked here directly. The registry says *why* a link could not be + # followed; it does not decide whether one was. Deriving the control flow + # from "did any rule fire" instead couples the walk to the registry's + # contents, and the completeness suite's whole method is to run this + # function with rules removed: under which that walk stepped off the end + # of the index and raised, rather than reporting a changed outcome. + # One reason only: the parent could not be resolved. The depth bound is + # deliberately *not* repeated here. Repeating it stops the walk at the same + # record whether or not `depth_exceeded` is registered, so a weakened bound + # never gets to walk further than a correct one, and the two depth vectors + # move together under every mutation: margin without independence, which + # is exactly what #124 says a second vector must not be. + parent = None + if hop.link_algorithm in supported: + parent = index.get(current["delegation"]["parent_record_hash"]) + if parent is None: + break + + hop = replace(hop, parent=parent) + found, warned = _evaluate(hop, rule_tuple, "link") + failures += found + warnings += warned + + # Not a conformance rule, and it cannot fire on a chain anyone can build: + # a cycle would need a record's delegation block to hold a digest of the + # block that names it back. It is here so that this function terminates on + # any input at all, including one hand-edited into a shape the hashes + # forbid, rather than terminating because the corpus happens to be honest. + if id(parent) in visited: + break + visited.add(id(parent)) + + current = parent + + if any( + rule.klass == "provenance" for rule in rule_tuple if rule.code in failures + ): + classification = "provenance-invalid" + elif failures: + classification = "authorization-invalid" + elif warnings: + classification = "unverifiable" + else: + classification = "verified" + + return ChainResult( + classification=classification, + failures=failures, + warnings=warnings, + depth=depth, + ) diff --git a/tests/delegation_margins.json b/tests/delegation_margins.json index f1ed3fbe..081269c0 100644 --- a/tests/delegation_margins.json +++ b/tests/delegation_margins.json @@ -3,6 +3,7 @@ "credential_issuer_mismatch": 2, "credential_unknown": 2, "credential_window": 2, + "data_class_unknown": 2, "data_class_widened": 2, "depth_exceeded": 2, "digest_algorithm_unsupported": 2, diff --git a/tests/test_delegation_completeness.py b/tests/test_delegation_completeness.py index c43686f5..1056a731 100644 --- a/tests/test_delegation_completeness.py +++ b/tests/test_delegation_completeness.py @@ -43,19 +43,21 @@ import pytest import rfc8785 -from tests.test_delegation_vectors import ( +from agentrust_trace.delegation import ( CLASSIFICATIONS, - VECTOR_PATHS, + RULES, Hop, Rule, - RULES, _jwk_key, _signature_invalid, +) +from tests.test_delegation_vectors import ( + VECTOR_PATHS, verify_chain, ) TESTS_DIR = Path(__file__).parent -WALK_MODULE = TESTS_DIR / "test_delegation_vectors.py" +WALK_MODULE = TESTS_DIR.parent / "src" / "agentrust_trace" / "delegation.py" MARGINS_FILE = TESTS_DIR / "delegation_margins.json" RULE_CODES = tuple(rule.code for rule in RULES) @@ -152,6 +154,12 @@ def _expiry_only(hop: Hop) -> bool: return hop.record["iat"] > credential["not_after"] +def _child_data_class_only(hop: Hop) -> bool: + """Check only the child's data_class against the lattice and assume the parent's is.""" + assert hop.parent is not None + return hop.record["data_class"] not in hop.context["data_class_lattice"] + + def _narrowing_checked_at_the_first_hop_only(hop: Hop) -> bool: """Compare the leaf against its parent and call the chain narrowed.""" if hop.depth != 1: @@ -177,6 +185,7 @@ def _narrowing_checked_at_the_first_hop_only(hop: Hop) -> bool: "holder_compared_to_the_parent": _holder_compared_to_the_parent }, "credential_window": {"expiry_only": _expiry_only}, + "data_class_unknown": {"child_class_only": _child_data_class_only}, "data_class_widened": {"first_hop_only": _narrowing_checked_at_the_first_hop_only}, } diff --git a/tests/test_delegation_module.py b/tests/test_delegation_module.py new file mode 100644 index 00000000..324336d7 --- /dev/null +++ b/tests/test_delegation_module.py @@ -0,0 +1,297 @@ +"""Verification tests for the Python delegation-link and A2A chain module.""" + +from __future__ import annotations + +import base64 +import copy +import json +from pathlib import Path +from typing import Any + +import pytest + +from agentrust_trace import delegation, generate_key, key_to_jwk, sign_record, verify_record + +VECTOR_DIR = Path(__file__).resolve().parents[1] / "examples" / "delegation-link" +VECTOR_PATHS = sorted(VECTOR_DIR.glob("*.json")) + + +def _load(name: str) -> dict[str, Any]: + return dict(json.loads((VECTOR_DIR / name).read_text(encoding="utf-8"))) + + +def test_parent_record_hash_matches_published_vectors() -> None: + vec01 = _load("01-valid-single-hop.json") + leaf, parent = vec01["records"][0], vec01["records"][1] + assert delegation.parent_record_hash(leaf, "sha256") == vec01["context"]["leaf"] + assert ( + delegation.parent_record_hash(parent, "sha256") + == leaf["delegation"]["parent_record_hash"] + ) + + vec22 = _load("22-leaf-link-uses-sha384.json") + leaf22, parent22 = vec22["records"][0], vec22["records"][1] + assert ( + delegation.parent_record_hash(parent22, "sha384") + == leaf22["delegation"]["parent_record_hash"] + ) + + vec24 = _load("24-parent-key-supplementary-plane.json") + leaf24, parent24 = vec24["records"][0], vec24["records"][1] + assert ( + delegation.parent_record_hash(parent24, "sha256") + == leaf24["delegation"]["parent_record_hash"] + ) + + +def test_parent_record_hash_refuses_invalid_inputs() -> None: + bad_algs: tuple[Any, ...] = ("sha512", "md5", "", 123, None) + for bad_alg in bad_algs: + with pytest.raises(delegation.DelegationError, match="unsupported chain digest"): + delegation.parent_record_hash({"a": 1}, bad_alg) + bad_records: tuple[Any, ...] = (None, "not-a-dict", [1, 2], 42) + for bad_record in bad_records: + with pytest.raises(delegation.DelegationError, match="JSON object"): + delegation.parent_record_hash(bad_record) + for uncanonical in ( + {"x": float("nan")}, + {"x": 10**20}, + {"\ud800": 1}, + {"x": "\ud800"}, + ): + with pytest.raises(delegation.DelegationError, match="RFC 8785"): + delegation.parent_record_hash(uncanonical) + + +def test_verify_delegation_link_succeeds_for_sha256_and_sha384() -> None: + vec01 = _load("01-valid-single-hop.json") + link01 = delegation.verify_delegation_link(vec01["records"][0], vec01["records"][1]) + assert link01 == delegation.DelegationLink( + parent_record_hash=vec01["records"][0]["delegation"]["parent_record_hash"], + algorithm="sha256", + credential_id="cred:orchestrator-to-planner", + ) + + vec22 = _load("22-leaf-link-uses-sha384.json") + link22 = delegation.verify_delegation_link(vec22["records"][0], vec22["records"][1]) + assert link22.algorithm == "sha384" + assert ( + link22.parent_record_hash + == vec22["records"][0]["delegation"]["parent_record_hash"] + ) + + +def test_verify_delegation_link_refuses_absent_malformed_unsupported_or_mismatched() -> None: + vec01 = _load("01-valid-single-hop.json") + leaf, parent = vec01["records"][0], vec01["records"][1] + + with pytest.raises(delegation.DelegationError, match="no delegation block"): + delegation.verify_delegation_link(parent, leaf) + + for bad_deleg in ("not-a-dict", {}, {"parent_record_hash": "sha256:00", "credential_id": ""}): + child = {**leaf, "delegation": bad_deleg} + with pytest.raises(delegation.DelegationError): + delegation.verify_delegation_link(child, parent) + + vec22 = _load("22-leaf-link-uses-sha384.json") + with pytest.raises(delegation.DelegationError, match="does not implement"): + delegation.verify_delegation_link( + vec22["records"][0], + vec22["records"][1], + supported_digest_algorithms=("sha256",), + ) + + sha512_child = { + **leaf, + "delegation": { + "parent_record_hash": "sha512:" + "a" * 128, + "credential_id": "cred:orchestrator-to-planner", + }, + } + with pytest.raises(delegation.DelegationError, match="not one this build can compute"): + delegation.verify_delegation_link( + sha512_child, + parent, + supported_digest_algorithms=("sha512",), + ) + + bad_supported_cases: tuple[Any, ...] = ("sha256", b"sha256", {"sha256": 1}, (), [123]) + for bad_supported in bad_supported_cases: + with pytest.raises(delegation.DelegationError, match="supported_digest_algorithms"): + delegation.verify_delegation_link( + leaf, parent, supported_digest_algorithms=bad_supported + ) + + vec05 = _load("05-link-over-signed-body.json") + with pytest.raises(delegation.DelegationError, match="does not digest"): + delegation.verify_delegation_link(vec05["records"][0], vec05["records"][1]) + + for algorithm, width in (("sha256", 64), ("sha384", 96)): + for bad_char in ("\u00e9", "\ud800"): + non_ascii_child = { + **leaf, + "delegation": { + **leaf["delegation"], + "parent_record_hash": f"{algorithm}:{bad_char}{'0' * (width - 1)}", + }, + } + with pytest.raises(delegation.DelegationError, match="does not digest"): + delegation.verify_delegation_link(non_ascii_child, parent) + + +@pytest.mark.parametrize("path", VECTOR_PATHS, ids=lambda p: p.stem) +def test_verify_delegation_chain_matches_every_vector(path: Path) -> None: + vector = json.loads(path.read_text(encoding="utf-8")) + expected = vector["expected"] + + by_digest = delegation.verify_delegation_chain( + vector["records"], vector["context"]["leaf"], vector["context"] + ) + assert by_digest.classification == expected["classification"] + assert by_digest.codes == sorted(expected["codes"]) + + by_record = delegation.verify_delegation_chain( + vector["records"], vector["records"][0], vector["context"] + ) + assert by_record == by_digest + + reversed_records = list(reversed(vector["records"])) + by_reversed = delegation.verify_delegation_chain( + reversed_records, vector["context"]["leaf"], vector["context"] + ) + assert by_reversed.classification == by_digest.classification + assert by_reversed.codes == by_digest.codes + + +def test_verify_delegation_chain_refuses_malformed_caller_inputs() -> None: + vec01 = _load("01-valid-single-hop.json") + records = vec01["records"] + leaf = vec01["context"]["leaf"] + context = vec01["context"] + + bad_records_cases: tuple[Any, ...] = ( + "not-iterable-of-dicts", + {}, + [], + [{"subject": "x"}], + ) + for bad_records in bad_records_cases: + with pytest.raises(delegation.DelegationError): + delegation.verify_delegation_chain(bad_records, leaf, context) + + bad_leaf_cases: tuple[Any, ...] = ( + None, + 123, + "no-colon", + "sha256:" + "0" * 64, + {}, + ) + for bad_leaf in bad_leaf_cases: + with pytest.raises(delegation.DelegationError): + delegation.verify_delegation_chain(records, bad_leaf, context) + + for field_name, bad_val in ( + ("max_depth", -1), + ("max_depth", True), + ("supported_digest_algorithms", ("sha512",)), + ("supported_digest_algorithms", "sha256"), + ("data_class_lattice", "public"), + ("trusted_root_keys", "not-a-list"), + ("trusted_root_keys", [{"kty": "OKP", "crv": "Ed25519", "x": ["not-a-str"]}]), + ("trusted_root_keys", [{}]), + ("credentials", "not-a-dict"), + ): + bad_ctx = copy.deepcopy(context) + bad_ctx[field_name] = bad_val + with pytest.raises(delegation.DelegationError): + delegation.verify_delegation_chain(records, leaf, bad_ctx) + + bad_rules_cases: tuple[Any, ...] = ("not-rules", [123]) + for bad_rules in bad_rules_cases: + with pytest.raises(delegation.DelegationError): + delegation.verify_delegation_chain(records, leaf, context, rules=bad_rules) + + +@pytest.mark.parametrize( + ("vector_name", "lattice", "expected"), + [ + ( + "01-valid-single-hop.json", + ["public", "internal", "confidential", "restricted"], + "verified", + ), + ("01-valid-single-hop.json", ["public"], "unverifiable"), + ("01-valid-single-hop.json", [], "unverifiable"), + ("18-data-class-widened-at-leaf.json", ["public"], "unverifiable"), + ("18-data-class-widened-at-leaf.json", ["public", "internal"], "unverifiable"), + ("18-data-class-widened-at-leaf.json", ["public", "restricted"], "unverifiable"), + ], +) +def test_unlisted_data_class_reports_unverifiable_and_never_verified( + vector_name: str, lattice: list[str], expected: str +) -> None: + vector = _load(vector_name) + vector["context"]["data_class_lattice"] = lattice + result = delegation.verify_delegation_chain( + vector["records"], vector["context"]["leaf"], vector["context"] + ) + assert result.classification == expected + assert ("data_class_unknown" in result.codes) == (expected == "unverifiable") + + +def test_verify_delegation_chain_decides_iat_by_value() -> None: + """Section 3.2.2 (#404): a whole-valued float `iat` such as 1785000000.0 + has the same JCS form and signature as 1785000000, and must be accepted by + `verify_delegation_chain` just as `verify_record` accepts it.""" + vec01 = _load("01-valid-single-hop.json") + leaf = copy.deepcopy(vec01["records"][0]) + parent = vec01["records"][1] + leaf["iat"] = float(leaf["iat"]) + records = [leaf, parent] + + verify_record(leaf, leaf["cnf"]["jwk"], now=vec01["context"]["now"]) + + by_digest = delegation.verify_delegation_chain( + records, vec01["context"]["leaf"], vec01["context"] + ) + assert by_digest.classification == "verified" + assert by_digest.codes == [] + + by_dict = delegation.verify_delegation_chain(records, leaf, vec01["context"]) + assert by_dict == by_digest + + +def test_verify_delegation_chain_refuses_non_canonical_signature_and_unknown_members() -> None: + """Section 3.2.2 and schema validation: a signature whose final base64url + character has non-zero unused bits (decoding to the same 64 bytes) or a + record carrying an unknown top-level member must be refused rather than + reported as `verified`.""" + vec01 = _load("01-valid-single-hop.json") + leaf, parent = vec01["records"][0], vec01["records"][1] + context = vec01["context"] + + sig = leaf["signature"] + assert sig.endswith("g") + non_canonical_sig = sig[:-1] + "h" + assert base64.urlsafe_b64decode(sig + "==") == base64.urlsafe_b64decode( + non_canonical_sig + "==" + ) + mutated_leaf = {**leaf, "signature": non_canonical_sig} + + assert delegation._signature_invalid(mutated_leaf) is True + with pytest.raises(delegation.DelegationError, match="schema"): + delegation.verify_delegation_chain([mutated_leaf, parent], mutated_leaf, context) + + key = generate_key() + unsigned = { + k: v for k, v in leaf.items() if k not in ("signature", "cnf") + } + unsigned["extra_top_level_member"] = "forbidden-by-schema" + signed_with_extra = sign_record(unsigned, key) + signed_with_extra["cnf"] = {"jwk": key_to_jwk(key)} + with pytest.raises(delegation.DelegationError, match="schema"): + delegation.verify_delegation_chain( + [signed_with_extra, parent], signed_with_extra, context + ) + + diff --git a/tests/test_delegation_vectors.py b/tests/test_delegation_vectors.py index 05a51213..7e5f218c 100644 --- a/tests/test_delegation_vectors.py +++ b/tests/test_delegation_vectors.py @@ -1,16 +1,15 @@ -"""Run the delegation-link corpus through a reference walk of the proposed A2A profile. +"""Run the delegation-link corpus through `agentrust_trace.delegation`. The `delegation` block is normative in v0.2; what a verifier does with a chain of -them is not, and `docs/rfcs/a2a-delegation-profile.md` is the proposal that would -make it so. This module is that proposal executed: every rule below corresponds to -one numbered requirement in the RFC, and every vector under -`examples/delegation-link/` declares the outcome it expects before this code runs. +them is proposed in `docs/rfcs/a2a-delegation-profile.md` and implemented in +`agentrust_trace.delegation`. Every rule in `RULES` corresponds to one numbered +requirement in the RFC, and every vector under `examples/delegation-link/` +declares the outcome it expects before this code runs. -The registry is the same shape as `test_action_receipt_fixtures.RULES`, for the -same reason: a check that is not registered never runs, so it cannot exist quietly -outside the inventory that `test_delegation_completeness.py` mutates. Nothing here -is exported from `agentrust_trace`: the package gains no public API for rules -that are not yet normative. +The rule registry in `agentrust_trace.delegation` uses the same shape as +`test_action_receipt_fixtures.RULES`, for the same reason: a check that is not +registered never runs, so it cannot exist quietly outside the inventory that +`test_delegation_completeness.py` mutates. Two orderings in the walk carry weight and are not incidental: @@ -28,352 +27,36 @@ from __future__ import annotations -import base64 import hashlib import json -from collections.abc import Callable, Sequence -from dataclasses import dataclass, field, replace +from collections.abc import Sequence from pathlib import Path from typing import Any import pytest -import rfc8785 -from cryptography.exceptions import InvalidSignature -from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PublicKey +from agentrust_trace.delegation import ( + CLASSIFICATIONS, + RULES, + ChainResult, + Rule, + _jwk_key, + parent_record_hash as _digest, + verify_delegation_chain, +) +from agentrust_trace.sign import _canonical_bytes as _canonical from agentrust_trace.validate import iter_errors VECTOR_DIR = Path(__file__).resolve().parents[1] / "examples" / "delegation-link" -CLASSIFICATIONS = frozenset( - {"verified", "provenance-invalid", "authorization-invalid", "unverifiable"} -) -"""Every outcome the walk can return. ca2a's Group 7 three-way classification with -`unverifiable` kept distinct from both invalid kinds; collapsing any two of these -is the nonconformance the classification exists to name.""" - - -@dataclass(frozen=True) -class ChainResult: - classification: str - failures: list[str] - warnings: list[str] - depth: int - - def __post_init__(self) -> None: - assert self.classification in CLASSIFICATIONS, ( - f"undeclared outcome {self.classification!r}" - ) - - @property - def codes(self) -> list[str]: - return sorted(set(self.failures) | set(self.warnings)) - - -# --------------------------------------------------------------------------- -# Shared helpers -# --------------------------------------------------------------------------- - - -def _canonical(value: dict[str, Any]) -> bytes: - return rfc8785.dumps(value) - - -def _digest(record: dict[str, Any], alg: str) -> str: - """The profile's preimage: the complete record, signature included. - - The alternative reading, the signed body alone, is what vector 05 is built - on, and the RFC states why it is rejected: a body digest does not bind the - parent's signer, so anyone may re-sign identical bytes and satisfy the child's - commitment. - """ - return f"{alg}:" + hashlib.new(alg, _canonical(record)).hexdigest() - - -def _decode_b64u(value: str) -> bytes: - return base64.urlsafe_b64decode(value + "=" * (-len(value) % 4)) - - -def _signature_invalid(record: dict[str, Any]) -> bool: - """Does `record` fail verification under the key it advertises in `cnf.jwk`? - - The claimed key, not a trusted one: whether the key is trusted is a separate - question with its own rule, and merging them produces a verifier that cannot - tell "forged" from "signed by someone you have not heard of".""" - jwk = record.get("cnf", {}).get("jwk", {}) - if jwk.get("kty") != "OKP" or jwk.get("crv") != "Ed25519" or "x" not in jwk: - return True - body = {k: v for k, v in record.items() if k != "signature"} - try: - Ed25519PublicKey.from_public_bytes(_decode_b64u(jwk["x"])).verify( - _decode_b64u(record["signature"]), _canonical(body) - ) - except (InvalidSignature, ValueError, KeyError): - return True - return False - - -def _jwk_key(jwk: dict[str, Any]) -> tuple[Any, ...]: - """A hashable identity for a JWK, over key material only. - - `kid` and other members are advisory; two records naming the same curve point - hold the same key whatever else they say about it.""" - return (jwk.get("kty"), jwk.get("crv"), jwk.get("x"), jwk.get("y")) - - -# --------------------------------------------------------------------------- -# The rule registry -# --------------------------------------------------------------------------- - - -@dataclass(frozen=True) -class Hop: - """One position in the walk, with everything a rule may read. - - `parent` is populated only on the `link` path, after the parent record has - actually been resolved. A rule that needs the parent and runs before it exists - would be reasoning about a hop that was never demonstrated.""" - - record: dict[str, Any] - context: dict[str, Any] - index: dict[str, dict[str, Any]] - depth: int - parent: dict[str, Any] | None = None - - @property - def delegation(self) -> dict[str, Any]: - return self.record["delegation"] - - @property - def link_algorithm(self) -> str: - return self.delegation["parent_record_hash"].split(":", 1)[0] - - @property - def credential(self) -> dict[str, Any] | None: - return self.context["credentials"].get(self.delegation["credential_id"]) - - -@dataclass(frozen=True) -class Rule: - """One named obligation. ``check`` returns True when the defect it guards - against is observed at this hop: True means the code is emitted.""" - - code: str - severity: str # "failure" | "warning" - klass: str # "provenance" | "authorization" | "unverifiable" - path: str # "record" | "root" | "resolve" | "link" - check: Callable[[Hop], bool] = field(compare=False) - - -def _record_signature_invalid(hop: Hop) -> bool: - return _signature_invalid(hop.record) - - -def _root_key_untrusted(hop: Hop) -> bool: - trusted = {_jwk_key(j) for j in hop.context["trusted_root_keys"]} - return _jwk_key(hop.record.get("cnf", {}).get("jwk", {})) not in trusted - - -def _depth_exceeded(hop: Hop) -> bool: - return hop.depth > hop.context["max_depth"] - - -def _digest_algorithm_unsupported(hop: Hop) -> bool: - return hop.link_algorithm not in hop.context["supported_digest_algorithms"] - - -def _parent_not_found(hop: Hop) -> bool: - # Guarded on support: a link this verifier cannot compute is unread, not - # broken, and saying "not found" about it would be a finding nobody made. - if _digest_algorithm_unsupported(hop): - return False - return hop.delegation["parent_record_hash"] not in hop.index - - -def _credential_unknown(hop: Hop) -> bool: - return hop.credential is None - - -def _credential_issuer_mismatch(hop: Hop) -> bool: - cred = hop.credential - if cred is None: - return False - assert hop.parent is not None - return cred["issuer"] != hop.parent["subject"] - - -def _credential_holder_mismatch(hop: Hop) -> bool: - cred = hop.credential - if cred is None: - return False - return cred["holder"] != hop.record["subject"] - - -def _credential_window(hop: Hop) -> bool: - cred = hop.credential - if cred is None: - return False - # Judged at the hop's own `iat`. A chain does not expire because it is being - # read late, and a verifier using its own clock reports a different answer - # every day for the same evidence. - return not (cred["not_before"] <= hop.record["iat"] <= cred["not_after"]) - - -def _data_class_widened(hop: Hop) -> bool: - assert hop.parent is not None - lattice: list[str] = hop.context["data_class_lattice"] - if hop.record["data_class"] not in lattice or hop.parent["data_class"] not in lattice: - # A class outside the supplied ordering is not comparable. It is not this - # rule's business to invent a position for it. - return False - return lattice.index(hop.record["data_class"]) > lattice.index(hop.parent["data_class"]) - - -RULES: tuple[Rule, ...] = ( - # -- every record on the chain ---------------------------------------------- - Rule("record_signature_invalid", "failure", "provenance", "record", - _record_signature_invalid), - # -- the record with no delegation block ------------------------------------- - Rule("root_key_untrusted", "failure", "provenance", "root", _root_key_untrusted), - # -- following a link, before the parent is known ---------------------------- - Rule("depth_exceeded", "failure", "authorization", "resolve", _depth_exceeded), - Rule("digest_algorithm_unsupported", "warning", "unverifiable", "resolve", - _digest_algorithm_unsupported), - Rule("parent_not_found", "failure", "provenance", "resolve", _parent_not_found), - # -- the hop, once its parent has been resolved ------------------------------ - Rule("credential_unknown", "failure", "authorization", "link", _credential_unknown), - Rule("credential_issuer_mismatch", "failure", "authorization", "link", - _credential_issuer_mismatch), - Rule("credential_holder_mismatch", "failure", "authorization", "link", - _credential_holder_mismatch), - Rule("credential_window", "failure", "authorization", "link", _credential_window), - Rule("data_class_widened", "failure", "authorization", "link", _data_class_widened), -) - - -def _evaluate( - hop: Hop, rules: Sequence[Rule], path: str -) -> tuple[list[str], list[str]]: - """The single point where rule codes are emitted. - - Everything the walk reports flows through this loop, which is what makes the - registry authoritative. `test_delegation_completeness` asserts by AST that no - other function in this module appends to a failure or warning list. - """ - failures: list[str] = [] - warnings: list[str] = [] - for rule in rules: - if rule.path == path and rule.check(hop): - (failures if rule.severity == "failure" else warnings).append(rule.code) - return failures, warnings - - -# --------------------------------------------------------------------------- -# The walk -# --------------------------------------------------------------------------- - def verify_chain(vector: dict[str, Any], rules: Sequence[Rule] = RULES) -> ChainResult: - """Walk `vector` from its designated leaf towards the root. - - The record set is a set: it arrives in no useful order, the leaf is named by - digest rather than by position, and a parent is found by looking its digest up - rather than by taking the next element. An implementation that reads - `records[0]` as the root agrees with this one on nothing. - - No cycle check, deliberately. A cycle needs each record's delegation block to - carry a digest covering the block that names it back, which is a hash - collision; the reachable analogue is an unbounded chain, and that is what - `depth_exceeded` is for. The RFC states this rather than leaving a reader to - wonder which of the two was forgotten. - """ - context = vector["context"] - records = vector["records"] - - index: dict[str, dict[str, Any]] = {} - for algorithm in context["supported_digest_algorithms"]: - for record in records: - index[_digest(record, algorithm)] = record - - current = index.get(context["leaf"]) - assert current is not None, ( - f"{vector['id']}: the record under appraisal is not in the vector's own " - "record set, or is not addressable under a supported digest algorithm" - ) - - failures: list[str] = [] - warnings: list[str] = [] - depth = 0 - visited: set[int] = set() - - while True: - hop = Hop(record=current, context=context, index=index, depth=depth) - - found, warned = _evaluate(hop, rules, "record") - failures += found - warnings += warned - - if "delegation" not in current: - found, warned = _evaluate(hop, rules, "root") - failures += found - warnings += warned - break - - depth += 1 - hop = replace(hop, depth=depth) - found, warned = _evaluate(hop, rules, "resolve") - failures += found - warnings += warned - - # Whether the walk can continue is a structural fact about the records, - # asked here directly. The registry says *why* a link could not be - # followed; it does not decide whether one was. Deriving the control flow - # from "did any rule fire" instead couples the walk to the registry's - # contents, and the completeness suite's whole method is to run this - # function with rules removed: under which that walk stepped off the end - # of the index and raised, rather than reporting a changed outcome. - # One reason only: the parent could not be resolved. The depth bound is - # deliberately *not* repeated here. Repeating it stops the walk at the same - # record whether or not `depth_exceeded` is registered, so a weakened bound - # never gets to walk further than a correct one, and the two depth vectors - # move together under every mutation: margin without independence, which - # is exactly what #124 says a second vector must not be. - parent = None - if hop.link_algorithm in context["supported_digest_algorithms"]: - parent = index.get(current["delegation"]["parent_record_hash"]) - if parent is None: - break - - hop = replace(hop, parent=parent) - found, warned = _evaluate(hop, rules, "link") - failures += found - warnings += warned - - # Not a conformance rule, and it cannot fire on a chain anyone can build: - # a cycle would need a record's delegation block to hold a digest of the - # block that names it back. It is here so that this function terminates on - # any input at all, including one hand-edited into a shape the hashes - # forbid, rather than terminating because the corpus happens to be honest. - if id(parent) in visited: - break - visited.add(id(parent)) - - current = parent - - if any(rule.klass == "provenance" for rule in rules if rule.code in failures): - classification = "provenance-invalid" - elif failures: - classification = "authorization-invalid" - elif warnings: - classification = "unverifiable" - else: - classification = "verified" - - return ChainResult( - classification=classification, - failures=failures, - warnings=warnings, - depth=depth, + """Walk `vector` from its designated leaf towards the root via `agentrust_trace.delegation`.""" + return verify_delegation_chain( + vector["records"], + vector["context"]["leaf"], + vector["context"], + rules=rules, ) @@ -415,6 +98,8 @@ def test_vector_set_is_complete() -> None: "22-leaf-link-uses-sha384.json", "23-deep-link-uses-sha384.json", "24-parent-key-supplementary-plane.json", + "25-data-class-unknown-at-leaf.json", + "26-data-class-unknown-at-parent.json", ] diff --git a/tests/test_public_functions_raise_what_they_document.py b/tests/test_public_functions_raise_what_they_document.py index 16f7cf0e..13f839d9 100644 --- a/tests/test_public_functions_raise_what_they_document.py +++ b/tests/test_public_functions_raise_what_they_document.py @@ -52,8 +52,8 @@ import pytest import agentrust_trace as at -from agentrust_trace import (citation, content_marking, generate_key, intent_bridge, key_to_jwk, - platform_measurement, provenance, revocation, sign, +from agentrust_trace import (citation, content_marking, delegation, generate_key, intent_bridge, + key_to_jwk, platform_measurement, provenance, revocation, sign, successor_observation, validate) #: Values a caller can supply where an object, a string, a key or bytes is expected. @@ -69,6 +69,7 @@ #: contract does not catch it. DOCUMENTED: dict[str, tuple[str, ...]] = { "citation": ("ValueError",), + "delegation": ("DelegationError",), "platform_measurement": ("ValueError",), "content_marking": ("ContentMarkingError", "RecordMismatch"), "intent_bridge": ("IntentBridgeError", "AuthorizationDenied", "AuthorizationMismatch"), @@ -81,6 +82,15 @@ "adapters": ("ValueError", "ValidationError"), } +_DELEG_VECTOR = json.loads( + (pathlib.Path(__file__).resolve().parents[1] / "examples" / "delegation-link" + / "01-valid-single-hop.json").read_text(encoding="utf-8") +) +_DELEG_CTX = _DELEG_VECTOR["context"] +_DELEG_RECORDS = _DELEG_VECTOR["records"] +_DELEG_LEAF = _DELEG_RECORDS[0] +_DELEG_PARENT = _DELEG_RECORDS[1] + _KEY = generate_key() _JWK = key_to_jwk(_KEY) _RECORD_BYTES = json.dumps( @@ -91,6 +101,11 @@ #: name -> a call varying only the first positional argument. CALLS: dict[str, Callable[[Any], Any]] = { "citation.check_citations": lambda v: citation.check_citations(v, None), + "delegation.parent_record_hash": delegation.parent_record_hash, + "delegation.verify_delegation_link": + lambda v: delegation.verify_delegation_link(v, _DELEG_PARENT), + "delegation.verify_delegation_chain": + lambda v: delegation.verify_delegation_chain(v, _DELEG_LEAF, _DELEG_CTX), "platform_measurement.check_platform_measurement": lambda v: platform_measurement.check_platform_measurement(v, None), "content_marking.build_assertion": @@ -206,6 +221,25 @@ "citation.check_citations": ( lambda: {"record": _TRUST_RECORD, "resolver": None}, ("resolver",), ), + "delegation.parent_record_hash": ( + lambda: {"record": _DELEG_PARENT, "algorithm": "sha256"}, ("algorithm",), + ), + "delegation.verify_delegation_link": ( + lambda: { + "child": _DELEG_LEAF, + "parent": _DELEG_PARENT, + "supported_digest_algorithms": ("sha256",), + }, + ("parent", "supported_digest_algorithms"), + ), + "delegation.verify_delegation_chain": ( + lambda: { + "records": _DELEG_RECORDS, + "leaf": _DELEG_LEAF, + "context": _DELEG_CTX, + }, + ("leaf", "context", "rules"), + ), "platform_measurement.check_platform_measurement": ( lambda: {"record": _TRUST_RECORD, "appraiser": None}, ("appraiser",), ), @@ -363,6 +397,7 @@ def test_every_public_function_is_either_swept_or_declared_unsweepable() -> None def _module_of(name: str) -> Any: return {"citation": citation, "content_marking": content_marking, + "delegation": delegation, "intent_bridge": intent_bridge, "platform_measurement": platform_measurement, "provenance": provenance, "revocation": revocation, "sign": sign, "successor_observation": successor_observation, @@ -436,6 +471,9 @@ def test_no_keyword_argument_leaks_an_undocumented_exception(name: str, param: s #: come back as the documented refusal, so a clean sweep above means the call arrived. KEYWORD_REACHES: dict[str, tuple[str, Any, str]] = { "citation.check_citations": ("resolver", "a-string", "ValueError"), + "delegation.parent_record_hash": ("algorithm", 123, "DelegationError"), + "delegation.verify_delegation_link": ("parent", None, "DelegationError"), + "delegation.verify_delegation_chain": ("context", None, "DelegationError"), "platform_measurement.check_platform_measurement": ("appraiser", "a-string", "ValueError"), "content_marking.build_assertion": ("alg", 123, "ContentMarkingError"), "content_marking.verify_assertion": ("record_bytes", None, "ContentMarkingError"), @@ -654,6 +692,9 @@ def test_no_public_function_raises_an_undocumented_exception(name: str) -> None: #: of the function rather than evidence the call is wired up. An explicit witness is. REACHES: dict[str, tuple[Any, str]] = { "citation.check_citations": (None, "ValueError"), + "delegation.parent_record_hash": (None, "DelegationError"), + "delegation.verify_delegation_link": (None, "DelegationError"), + "delegation.verify_delegation_chain": (None, "DelegationError"), "platform_measurement.check_platform_measurement": (None, "ValueError"), "content_marking.build_assertion": (None, "ContentMarkingError"), "content_marking.verify_assertion": (None, "ContentMarkingError"),