diff --git a/.gitignore b/.gitignore index d05f9e5..d861581 100644 --- a/.gitignore +++ b/.gitignore @@ -31,3 +31,6 @@ agt-evidence.json agt-attestation.json sbom.spdx.json scorecard-results.sarif + +# Generated example output (ephemeral keys + timestamps) +examples/trace-dag/dag.json diff --git a/CHANGELOG.md b/CHANGELOG.md index 70eb40e..a18d17a 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -42,6 +42,7 @@ built/stubbed boundary. - AGT governance gate is now **blocking** (was advisory): CI installs `agent-governance-toolkit[full]`, so the OWASP ASI 2026 coverage modules load and `agt verify` reports 10/10 coverage with 6/6 runtime checks (COMPLETE). The enforcement descriptor now declares cA2A's `governed_capabilities`, reported as the registered-tools inventory. A governance or coverage regression now fails CI. - Real Cedar policy engine binding: `ca2a_runtime.cedar.CedarPolicy` (backed by `cedarpy`, the engine cMCP runs) evaluates each capability as a Cedar authorization request. A new `ca2a_runtime.policy.Policy` protocol makes `LocalPolicy` (allow set) and `CedarPolicy` interchangeable in the peer path. Adds the `cedarpy` dependency. - Transport-agnostic inbound peer request handler: `ca2a_runtime.peer.handle_peer_request` with `PeerRequest` / `PeerResult`. Composes the full pipeline (verify chain, intersect scope and enforce, open a sealed payload with the enclave key, emit a linked provenance record) fail-closed. A transport parses its wire format into a `PeerRequest`; cA2A does not define the transport (profile, not protocol). +- TRACE binding for the delegation DAG (Tier 2): `ca2a_runtime.trace_binding` lifts each delegation hop into a signed TRACE Trust Record carrying the A2A profile `delegation` block (`build_trace_record`, `sign_trace_record`, `emit_dag`, `trace_record_hash`, `HopContext`). Records are produced and signed with `agentrust-trace` (Ed25519 over RFC 8785), so TRACE canonicalization and signing are reused, not reimplemented. `ca2a_verify.verify_trace_dag` verifies a signed root-to-leaf DAG offline (structural validity, trusted-key signature, unbroken parent links over the full signed parent record) and `cross_check_trace_dag` ties it to the delegation chain; new error `TRACE_RECORD_INVALID`. Software-mode records are Level 0 (platform `software-only`); a hardware TEE run is what lifts them to Level 1. Adds the `agentrust-trace` dependency (and `agentrust-trace-tests` for dev). See `examples/trace-dag/`. - RFC 8785 (JSON Canonicalization Scheme) canonicalization: `ca2a_runtime.canonical.canonicalize`. Credential and provenance bodies are now signed over the JCS encoding (UTF-16 key ordering, JCS string escaping, literal non-ASCII, shortest-decimal integers), so cA2A signatures are cross-verifiable with agent-manifest. ASCII credentials are byte-identical to the previous encoding, so existing signatures still verify. - Repository scaffold: governance, CI/CD, docs framework, and packaging at parity with the agentrust-io house standard diff --git a/ROADMAP.md b/ROADMAP.md index af79966..f68477f 100644 --- a/ROADMAP.md +++ b/ROADMAP.md @@ -24,7 +24,7 @@ Already implemented and tested elsewhere; cA2A depends on it rather than reimple - Runtime peer-delegation enforcement: **decision core landed** (`ca2a_runtime.peer.enforce_peer_call`: verify chain, intersect delegated scope with local policy, enforce, emit provenance record; claim C3 validated), now with a **real Cedar policy engine** option (`ca2a_runtime.cedar.CedarPolicy`) alongside the allow-set `LocalPolicy`. Remaining: wire the decision core to a live A2A transport (Tier 2) - Sealed peer channel: **landed** (`ca2a_runtime.channel`: HPKE-style X25519 -> HKDF-SHA256 -> ChaCha20-Poly1305 sealing to the peer's attested key; claim C4 validated). Remaining: bind the seal to a verified attestation report on a live call, and rely on the enclave to hold the private key (hardware property) -- Linked runtime evidence: each hop's TRACE record references the parent record hash and delegation credential id, producing a verifiable delegation DAG (Tier 2) +- Linked runtime evidence: **landed** (`ca2a_runtime.trace_binding` emits a signed TRACE record per hop with the A2A `delegation` block; `ca2a_verify.verify_trace_dag` verifies the DAG offline, each link committing to the parent's full signed record). Built on `agentrust-trace` (Ed25519 + RFC 8785), reused not reimplemented. Software-mode records are Level 0; a hardware TEE run lifts them to Level 1. See `examples/trace-dag/`. ## Critical path, sequenced first (Tier 3) diff --git a/docs/spec/trace-a2a-profile.md b/docs/spec/trace-a2a-profile.md index 4d3a580..1d367ca 100644 --- a/docs/spec/trace-a2a-profile.md +++ b/docs/spec/trace-a2a-profile.md @@ -25,4 +25,4 @@ This is done offline, from signed records alone, without trusting the operators ## Status -The delegation-link field is on the trace-spec roadmap as the "A2A profile, pending A2A protocol stability." A2A is now stable at v1.x, which clears that blocker. The DAG verifier lands with the Tier 2 provenance work in this repo. See [ROADMAP.md](../../ROADMAP.md). +The delegation-link field is on the trace-spec roadmap as the "A2A profile, pending A2A protocol stability." A2A is now stable at v1.x, which clears that blocker. The `delegation` block ships in the TRACE v0.1 schema. cA2A emits and verifies the DAG today: `ca2a_runtime.trace_binding` produces a signed TRACE record per hop with the block, and `ca2a_verify.verify_trace_dag` reconstructs and verifies the DAG offline, each `parent_record_hash` committing to the parent's full signed record. Software-mode records are TRACE Level 0 (platform `software-only`); an end-to-end run on confidential-computing hardware is what lifts them to Level 1. See [ROADMAP.md](../../ROADMAP.md) and `examples/trace-dag/`. diff --git a/examples/trace-dag/README.md b/examples/trace-dag/README.md new file mode 100644 index 0000000..b78efd1 --- /dev/null +++ b/examples/trace-dag/README.md @@ -0,0 +1,36 @@ +# TRACE delegation DAG + +Emit a signed [TRACE](https://github.com/agentrust-io/trace-spec) Trust Record per +delegation hop, linked into a verifiable DAG via the A2A profile's `delegation` +block, then verify the DAG offline. + +```bash +python examples/trace-dag/demo.py +``` + +The demo builds a three-hop chain (orchestrator to researcher to retriever), +lifts each hop into a TRACE record, verifies the DAG from the signed records +alone, cross-checks it against the delegation chain, confirms each record passes +the TRACE conformance suite at **Level 0**, and writes `dag.json`. + +This is the **software-attestation** path (`runtime.platform` = `software-only`), +so the records are honestly Level 0. The appraisal and transparency URIs use the +reserved `.invalid` TLD so a software-mode record can never be mistaken for a +hardware-attested one. A run on confidential-computing hardware (a real TEE +measurement) is what lifts these records to Level 1. + +## What each piece does + +- `ca2a_runtime.trace_binding.emit_dag` — build and sign the linked records. +- `ca2a_verify.verify_trace_dag` — verify the DAG offline against trusted keys: + each record valid and signed, and each `delegation.parent_record_hash` equal to + the hash of the parent's **full signed record**. +- `ca2a_verify.cross_check_trace_dag` — tie the DAG to the delegation chain + (each non-root hop acted under the credential the chain names). + +Validate a single record with the TRACE tooling directly: + +```bash +python -c "import json; json.dump(json.load(open('examples/trace-dag/dag.json'))[1], open('rec.json','w'))" +trace-tests verify --record rec.json --level 0 +``` diff --git a/examples/trace-dag/demo.py b/examples/trace-dag/demo.py new file mode 100644 index 0000000..3f6a1f5 --- /dev/null +++ b/examples/trace-dag/demo.py @@ -0,0 +1,101 @@ +"""Emit a TRACE delegation DAG from a cA2A chain and verify it offline. + +Run: python examples/trace-dag/demo.py + +Builds a three-hop delegation chain (A -> B -> C), lifts each hop into a signed +TRACE Trust Record carrying the A2A ``delegation`` block, verifies the resulting +DAG from the signed records alone, cross-checks it against the chain, confirms +each record passes the TRACE conformance suite at Level 0, and writes the DAG to +``dag.json``. This is the software-attestation path: records are honestly Level 0 +(platform ``software-only``); a hardware TEE run is what lifts them to Level 1. +""" + +from __future__ import annotations + +import json +import time +from pathlib import Path + +from agentrust_trace import generate_key +from trace_tests.runner import run as run_conformance + +from ca2a_runtime.delegation import DelegationCredential, new_keypair +from ca2a_runtime.trace_binding import HopContext, HopSpec, digest, emit_dag +from ca2a_verify import cross_check_trace_dag, verify_trace_dag + +SUBJECTS = [ + "spiffe://ca2a.example/agent/orchestrator", + "spiffe://ca2a.example/agent/researcher", + "spiffe://ca2a.example/agent/retriever", +] +SCOPES = [ + frozenset({"task:read", "task:write", "tool:search"}), + frozenset({"task:read", "tool:search"}), + frozenset({"tool:search"}), +] + + +def build_chain(scopes: list[frozenset[str]]) -> list[DelegationCredential]: + """A correctly signed, narrowing root-to-leaf chain (one hop per scope).""" + chain: list[DelegationCredential] = [] + priv, pub = new_keypair() + parent_id: str | None = None + for depth, scope in enumerate(scopes): + next_priv, next_pub = new_keypair() + cred = DelegationCredential( + credential_id=f"cred-{depth}", + issuer=pub, + subject=next_pub, + scope=scope, + depth=depth, + parent_id=parent_id, + ).sign(priv) + chain.append(cred) + parent_id = cred.credential_id + priv, pub = next_priv, next_pub + return chain + + +def main() -> None: + now = int(time.time()) + chain = build_chain(SCOPES) + keys = [generate_key() for _ in chain] + + hops = [ + HopSpec( + subject=SUBJECTS[i], + signing_key=keys[i], + context=HopContext.software( + model_provider="anthropic", + model_id="claude-opus-4-8", + image_label=f"ca2a-peer:{SUBJECTS[i].rsplit('/', 1)[-1]}", + policy_bundle_hash=digest(b"demo-policy-bundle"), + ), + iat=now, + credential_id=cred.credential_id, + ) + for i, cred in enumerate(chain) + ] + + records = emit_dag(hops) + print(f"emitted {len(records)} linked TRACE records (root -> leaf)") + + trusted = [k.public_key() for k in keys] + result = verify_trace_dag(records, trusted_keys=trusted) + cross_check_trace_dag(records, chain) + print(f"DAG verified offline: {result.hops} hops, {result.root_subject} -> {result.leaf_subject}") + + for i, record in enumerate(records): + findings = run_conformance(record, "trace", level=0) + failed = [f for group in findings.values() for f in group if f.failed()] + status = "PASS" if not failed else f"FAIL {failed}" + block = "root (no delegation block)" if i == 0 else record["delegation"]["credential_id"] + print(f" record {i}: TRACE Level 0 {status} [{block}]") + + out = Path(__file__).parent / "dag.json" + out.write_text(json.dumps(records, indent=2), encoding="utf-8") + print(f"wrote {out}") + + +if __name__ == "__main__": + main() diff --git a/pyproject.toml b/pyproject.toml index d9d7135..6f8a4e9 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -27,6 +27,11 @@ dependencies = [ "cryptography>=42.0", "pyyaml>=6.0", "cedarpy>=4.8", + # TRACE record models + RFC 8785 signing, reused rather than reimplemented + # (see ROADMAP Tier 0). Pin a compatible minor; cross-repo skew is a known + # risk documented in LIMITATIONS.md. + "agentrust-trace>=0.3,<0.4", + "rfc8785>=0.1", ] [project.optional-dependencies] @@ -39,6 +44,8 @@ dev = [ "types-pyyaml", "bandit[toml]>=1.7", "pip-audit>=2.6", + # Conformance suite: asserts emitted records pass the TRACE Level checks. + "agentrust-trace-tests>=0.2,<0.3", ] [project.scripts] @@ -73,6 +80,7 @@ ignore = ["E501"] "src/ca2a_runtime/cli.py" = ["T201"] "scripts/gen_example_chain.py" = ["T201"] "scripts/gen_agt_evidence.py" = ["T201"] +"examples/**/*.py" = ["T201"] [tool.bandit] skips = ["B101"] diff --git a/src/ca2a_runtime/errors.py b/src/ca2a_runtime/errors.py index 3a98e76..80a351f 100644 --- a/src/ca2a_runtime/errors.py +++ b/src/ca2a_runtime/errors.py @@ -93,3 +93,15 @@ class TransportError(CA2AError): code = "TRANSPORT_ERROR" http_status = 400 + + +class TraceRecordInvalid(CA2AError): + """A TRACE record is structurally invalid or its signature does not verify. + + Raised by the TRACE DAG verifier when a hop's record fails schema validation, + is signed by a key that is not trusted, or its embedded key does not match the + one supplied. Distinct from ProvenanceLinkBroken, which covers the parent-link + chaining between otherwise valid records.""" + + code = "TRACE_RECORD_INVALID" + http_status = 422 diff --git a/src/ca2a_runtime/trace_binding.py b/src/ca2a_runtime/trace_binding.py new file mode 100644 index 0000000..709e1d9 --- /dev/null +++ b/src/ca2a_runtime/trace_binding.py @@ -0,0 +1,248 @@ +"""Bind a delegation hop to a signed TRACE Trust Record. + +cA2A's native provenance record (:class:`ca2a_runtime.provenance.DelegationRecord`) +records *that* a hop happened and links it to its parent. The TRACE binding lifts +each hop into a TRACE Trust Record (the agentrust-io interchange format) carrying +the A2A profile's ``delegation`` block, so a chain of records is verifiable with +the standard TRACE tooling (``agentrust-trace``, ``trace-tests``, the ``/trace`` +report) rather than only cA2A's own verifier. + +The record is produced, signed, and hashed with the ``agentrust-trace`` reference +implementation (Ed25519 over RFC 8785 JCS canonical bytes), so cA2A does not +reimplement TRACE canonicalization or signing. See the TRACE A2A profile in +``docs/spec/trace-a2a-profile.md``. + +The delegation link (``delegation.parent_record_hash``) commits to the parent's +**full signed record**, including its signature: any change to the parent, its +signature included, breaks the child's link. A root hop carries no delegation +block, matching the profile. +""" + +from __future__ import annotations + +import hashlib +from dataclasses import dataclass +from typing import Any + +import rfc8785 +from agentrust_trace import sign_record +from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PrivateKey + +# The frozen TRACE v0.1 profile URI. Every record's eat_profile equals this. +EAT_PROFILE = "tag:agentrust.io,2026:trace-v0.1" + +# The platform value for a software-attestation (no hardware TEE) run. Records +# emitted with this platform are honestly Level 0 only: trace-tests fails a +# software-only record at Level 1 by design. +SOFTWARE_PLATFORM = "software-only" + + +def digest(data: bytes) -> str: + """Return the ``sha256:``-prefixed lowercase-hex digest TRACE expects. + + cA2A's own provenance hashes are bare hex; TRACE digests carry an algorithm + prefix (``sha256:...``). This is the one-line conversion between the two. + """ + return "sha256:" + hashlib.sha256(data).hexdigest() + + +@dataclass(frozen=True) +class HopContext: + """The per-hop runtime, model, and policy facts a TRACE record needs. + + A :class:`~ca2a_runtime.provenance.DelegationRecord` carries the delegation + identity (credential id, subject, scope) but none of the runtime/attestation + context a full TRACE record requires. The callee supplies that context here; + :func:`build_trace_record` composes the two into a record. + + Use :meth:`software` for a software-attestation run (no hardware TEE). All + digest fields must be ``sha256:``/``sha384:`` strings (see :func:`digest`); + the URI fields must be URIs. + """ + + model_provider: str + model_id: str + runtime_measurement: str + policy_bundle_hash: str + build_digest: str + appraisal_verifier: str + transparency: str + runtime_platform: str = SOFTWARE_PLATFORM + enforcement_mode: str = "enforce" + data_class: str = "confidential" + slsa_level: int = 0 + appraisal_status: str = "none" + runtime_nonce: str | None = None + model_version: str | None = None + policy_version: str | None = None + + @classmethod + def software( + cls, + *, + model_provider: str, + model_id: str, + image_label: str, + policy_bundle_hash: str, + data_class: str = "confidential", + enforcement_mode: str = "enforce", + runtime_nonce: str | None = None, + ) -> HopContext: + """A software-mode context with clearly non-production placeholders. + + The runtime measurement is the digest of ``image_label`` (a real digest + of the software identity, not a hardware measurement). The appraisal and + transparency URIs use the reserved ``.invalid`` TLD so a reader can never + mistake a software-mode record for a hardware-attested one. Records built + from this context are Level 0. + """ + return cls( + model_provider=model_provider, + model_id=model_id, + runtime_platform=SOFTWARE_PLATFORM, + runtime_measurement=digest(image_label.encode("utf-8")), + policy_bundle_hash=policy_bundle_hash, + build_digest=digest(f"software-build:{image_label}".encode()), + appraisal_status="none", + appraisal_verifier="https://appraisal.invalid/software-mode", + transparency="https://transparency.invalid/software-mode", + data_class=data_class, + enforcement_mode=enforcement_mode, + slsa_level=0, + runtime_nonce=runtime_nonce, + ) + + +def build_trace_record( + *, + subject: str, + iat: int, + context: HopContext, + credential_id: str | None = None, + parent_record_hash: str | None = None, +) -> dict[str, Any]: + """Compose an unsigned TRACE record for one delegation hop. + + ``subject`` is the hop's TRACE identity (``spiffe://`` or ``did:``); note this + is a distinct identifier from the raw-hex public key a + :class:`DelegationCredential` uses as its subject. + + A non-root hop supplies both ``credential_id`` and ``parent_record_hash`` and + gets a ``delegation`` block; a root hop supplies neither. Supplying exactly + one is a programming error and raises ``ValueError``. + + The returned dict has no ``cnf`` or ``signature`` yet; pass it to + :func:`sign_trace_record` to get a schema-valid, signed record. + """ + if (parent_record_hash is None) != (credential_id is None): + raise ValueError( + "the delegation block needs both parent_record_hash and credential_id; " + "a root hop supplies neither" + ) + + runtime: dict[str, Any] = { + "platform": context.runtime_platform, + "measurement": context.runtime_measurement, + } + if context.runtime_nonce is not None: + runtime["nonce"] = context.runtime_nonce + + model: dict[str, Any] = { + "provider": context.model_provider, + "model_id": context.model_id, + } + if context.model_version is not None: + model["version"] = context.model_version + + policy: dict[str, Any] = { + "bundle_hash": context.policy_bundle_hash, + "enforcement_mode": context.enforcement_mode, + } + if context.policy_version is not None: + policy["version"] = context.policy_version + + record: dict[str, Any] = { + "eat_profile": EAT_PROFILE, + "iat": int(iat), + "subject": subject, + "model": model, + "runtime": runtime, + "policy": policy, + "data_class": context.data_class, + "build_provenance": { + "slsa_level": context.slsa_level, + "digest": context.build_digest, + }, + "appraisal": { + "status": context.appraisal_status, + "verifier": context.appraisal_verifier, + }, + "transparency": context.transparency, + } + if parent_record_hash is not None: + record["delegation"] = { + "parent_record_hash": parent_record_hash, + "credential_id": credential_id, + } + return record + + +def sign_trace_record(record: dict[str, Any], key: Ed25519PrivateKey) -> dict[str, Any]: + """Sign a TRACE record with ``key`` (Ed25519 over RFC 8785 JCS). + + Thin pass-through to ``agentrust_trace.sign_record``: it sets ``cnf.jwk`` to + the public key and adds the embedded ``signature``. Returned dict is + schema-valid and ready to hash, store, or verify. + """ + return sign_record(record, key) + + +def trace_record_hash(signed_record: dict[str, Any]) -> str: + """The ``sha256:`` digest a child hop puts in ``delegation.parent_record_hash``. + + Hashes the **full signed record**, signature included, over its RFC 8785 JCS + canonical bytes, so the link commits to the exact signed parent. Compute this + only on a signed record. + """ + return "sha256:" + hashlib.sha256(rfc8785.dumps(signed_record)).hexdigest() + + +@dataclass(frozen=True) +class HopSpec: + """One hop's inputs for assembling a linked TRACE DAG. + + ``credential_id`` is the delegation credential this hop acted under; it is + recorded in the ``delegation`` block of every non-root hop and ignored for + the root (whose credential is the chain root and needs no back-link). + """ + + subject: str + signing_key: Ed25519PrivateKey + context: HopContext + iat: int + credential_id: str = "" + + +def emit_dag(hops: list[HopSpec]) -> list[dict[str, Any]]: + """Build and sign an ordered root-to-leaf TRACE DAG, linking each hop. + + Each hop's record links to the previous hop's signed record by hash. The + first hop is the root (no delegation block); every later hop carries a + ``delegation`` block referencing its parent's ``trace_record_hash`` and its + own ``credential_id``. Returns the signed records in root-to-leaf order. + """ + records: list[dict[str, Any]] = [] + parent_hash: str | None = None + for i, hop in enumerate(hops): + credential_id = None if i == 0 else hop.credential_id + record = build_trace_record( + subject=hop.subject, + iat=hop.iat, + context=hop.context, + credential_id=credential_id, + parent_record_hash=parent_hash, + ) + signed = sign_trace_record(record, hop.signing_key) + records.append(signed) + parent_hash = trace_record_hash(signed) + return records diff --git a/src/ca2a_verify/__init__.py b/src/ca2a_verify/__init__.py index 8bd8eb7..05ad6ad 100644 --- a/src/ca2a_verify/__init__.py +++ b/src/ca2a_verify/__init__.py @@ -4,6 +4,11 @@ the issuers' public keys and the attenuation invariants alone. """ +from ca2a_verify.dag import ( + TraceDagResult, + cross_check_trace_dag, + verify_trace_dag, +) from ca2a_verify.verify import ( ChainResult, VerificationError, @@ -14,7 +19,10 @@ __version__ = "0.1.0" __all__ = [ "ChainResult", + "TraceDagResult", "VerificationError", + "cross_check_trace_dag", "verify_chain_file", "verify_delegation_chain", + "verify_trace_dag", ] diff --git a/src/ca2a_verify/dag.py b/src/ca2a_verify/dag.py new file mode 100644 index 0000000..5d9f170 --- /dev/null +++ b/src/ca2a_verify/dag.py @@ -0,0 +1,223 @@ +"""Offline verification of a cA2A TRACE delegation DAG. + +Given the signed TRACE records a workflow emitted (one per delegation hop), this +verifier confirms, from the signed records alone and a set of trusted keys: + +1. each record is structurally valid TRACE and signed by a trusted key; +2. the records form an unbroken root-to-leaf chain, each hop's + ``delegation.parent_record_hash`` equal to the hash of the parent's full + signed record (see :func:`ca2a_runtime.trace_binding.trace_record_hash`); +3. the root carries no delegation block and no record hash repeats. + +The base TRACE record is validated with ``agentrust-trace``. The A2A profile's +``delegation`` block is validated here, in cA2A, because it is the profile's own +extension: a published ``agentrust-trace`` whose bundled schema predates the +block would otherwise reject any delegated record. cA2A owns the profile, so it +owns the block's validation and stays robust to that cross-repo version skew. + +Scope non-escalation across hops is a property of the delegation *credentials*, +verified by :func:`ca2a_runtime.delegation.verify_chain`; this verifier covers +the record linkage and authenticity. :func:`cross_check_trace_dag` ties the two +together by matching each hop's recorded credential id to the chain. + +Everything fails closed: the first violation raises, and no partial result is +returned. +""" + +from __future__ import annotations + +import base64 +import binascii +import re +from collections.abc import Iterable +from dataclasses import dataclass +from typing import Any + +from agentrust_trace import validate_json, verify_record +from cryptography.exceptions import InvalidSignature +from cryptography.hazmat.primitives import serialization +from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PublicKey + +from ca2a_runtime.delegation import DelegationCredential +from ca2a_runtime.errors import ProvenanceLinkBroken, TraceRecordInvalid +from ca2a_runtime.trace_binding import trace_record_hash + + +@dataclass(frozen=True) +class TraceDagResult: + """The outcome of a successful TRACE DAG verification.""" + + hops: int + root_subject: str + leaf_subject: str + subjects: list[str] + + +# The TRACE digest form (sha256:/sha384: + lowercase hex), same as the schema. +_DIGEST_RE = re.compile(r"^sha(256:[0-9a-f]{64}|384:[0-9a-f]{96})$") + + +def _validate_delegation_block(block: Any, index: int) -> None: + """Validate the A2A profile delegation block on a record (None means root). + + Mirrors the ``delegation`` object in the TRACE v0.1 schema: exactly + ``parent_record_hash`` (a digest) and ``credential_id`` (a non-empty string), + no other keys. Validated in cA2A so a stale published ``agentrust-trace`` + schema cannot make a well-formed delegated record fail structural checks. + """ + if block is None: + return + if not isinstance(block, dict): + raise TraceRecordInvalid(f"record {index} delegation block is not an object") + if set(block) != {"parent_record_hash", "credential_id"}: + raise TraceRecordInvalid( + f"record {index} delegation block has unexpected fields", + detail=f"got {sorted(block)}, expected parent_record_hash + credential_id", + ) + parent_hash = block["parent_record_hash"] + if not isinstance(parent_hash, str) or not _DIGEST_RE.match(parent_hash): + raise TraceRecordInvalid(f"record {index} delegation.parent_record_hash is not a digest") + credential_id = block["credential_id"] + if not isinstance(credential_id, str) or not credential_id: + raise TraceRecordInvalid(f"record {index} delegation.credential_id is missing") + + +def _b64url_decode(value: str) -> bytes: + padded = value + "=" * (-len(value) % 4) + return base64.urlsafe_b64decode(padded) + + +def _jwk_raw(jwk: dict[str, Any]) -> bytes: + """Raw Ed25519 public-key bytes from an OKP JWK, or raise TraceRecordInvalid.""" + if jwk.get("kty") != "OKP" or jwk.get("crv") != "Ed25519": + raise TraceRecordInvalid( + "unsupported cnf.jwk key type", detail="expected an OKP/Ed25519 key" + ) + x = jwk.get("x") + if not isinstance(x, str): + raise TraceRecordInvalid("cnf.jwk is missing its 'x' parameter") + try: + return _b64url_decode(x) + except (binascii.Error, ValueError) as exc: + raise TraceRecordInvalid("cnf.jwk 'x' is not valid base64url", detail=str(exc)) from exc + + +def _trusted_raw_keys(trusted_keys: Iterable[Any]) -> set[bytes]: + """Normalize trusted keys (Ed25519PublicKey or JWK dict) to raw-byte identities.""" + raw: set[bytes] = set() + for key in trusted_keys: + if isinstance(key, Ed25519PublicKey): + raw.add( + key.public_bytes( + encoding=serialization.Encoding.Raw, + format=serialization.PublicFormat.Raw, + ) + ) + elif isinstance(key, dict): + raw.add(_jwk_raw(key)) + else: + raise TypeError(f"unsupported trusted key type: {type(key).__name__}") + return raw + + +def verify_trace_dag( + records: list[dict[str, Any]], + *, + trusted_keys: Iterable[Any], + max_age_seconds: int | None = None, +) -> TraceDagResult: + """Verify a signed root-to-leaf TRACE delegation DAG offline. + + ``trusted_keys`` is the set of keys the caller trusts to have produced hops, + each an ``Ed25519PublicKey`` or an OKP JWK dict. A record is accepted only if + the key it embeds is trusted *and* its signature verifies against that key, so + a record signed by an untrusted key (or a trusted key spoofed into ``cnf``) is + rejected. ``max_age_seconds`` bounds each record's freshness; pass ``None`` + (the default, suited to offline audit of historical records) to skip the age + check. + + Raises ``TraceRecordInvalid`` for a structurally invalid, untrusted, or + badly-signed record, and ``ProvenanceLinkBroken`` for a broken parent link, + a mislabeled root, or a repeated record. Returns a summary on success. + """ + if not records: + raise ProvenanceLinkBroken("empty TRACE DAG") + + trusted = _trusted_raw_keys(trusted_keys) + prev_hash: str | None = None + seen_hashes: set[str] = set() + subjects: list[str] = [] + + for i, record in enumerate(records): + # Validate the base record with agentrust-trace, and the A2A profile's + # delegation block here (see the module docstring): the base schema is + # version-stable, the profile extension is cA2A's own. + base_record = {k: v for k, v in record.items() if k != "delegation"} + try: + validate_json(base_record) + except Exception as exc: # jsonschema.ValidationError + raise TraceRecordInvalid( + f"record {i} is not a valid TRACE record", detail=str(exc) + ) from exc + _validate_delegation_block(record.get("delegation"), i) + + jwk = record.get("cnf", {}).get("jwk", {}) + if _jwk_raw(jwk) not in trusted: + raise TraceRecordInvalid(f"record {i} is signed by an untrusted key") + try: + verify_record(record, jwk, max_age_seconds=max_age_seconds) + except InvalidSignature as exc: + raise TraceRecordInvalid(f"record {i} signature does not verify") from exc + except ValueError as exc: + raise TraceRecordInvalid( + f"record {i} could not be verified", detail=str(exc) + ) from exc + + record_hash = trace_record_hash(record) + if record_hash in seen_hashes: + raise ProvenanceLinkBroken(f"record {i} repeats an earlier record") + seen_hashes.add(record_hash) + + delegation = record.get("delegation") + if i == 0: + if delegation is not None: + raise ProvenanceLinkBroken("root record must not carry a delegation block") + else: + if delegation is None: + raise ProvenanceLinkBroken(f"record {i} is missing its delegation block") + if delegation.get("parent_record_hash") != prev_hash: + raise ProvenanceLinkBroken( + f"record {i} parent link does not match the previous record's hash", + detail="a tampered or reparented record was detected", + ) + + subjects.append(record["subject"]) + prev_hash = record_hash + + return TraceDagResult( + hops=len(records), + root_subject=subjects[0], + leaf_subject=subjects[-1], + subjects=subjects, + ) + + +def cross_check_trace_dag( + records: list[dict[str, Any]], chain: list[DelegationCredential] +) -> None: + """Tie a verified TRACE DAG to the delegation chain it should reflect. + + Confirms the DAG has one record per credential and that every non-root hop + acted under the credential the chain names at that position (the root hop + records no credential id, per the profile). Raises ProvenanceLinkBroken on + any mismatch. Run ``verify_trace_dag`` and ``verify_chain`` first; this only + checks that they line up. + """ + if len(records) != len(chain): + raise ProvenanceLinkBroken( + f"DAG length {len(records)} does not match chain length {len(chain)}" + ) + for i in range(1, len(records)): + delegation = records[i].get("delegation") or {} + if delegation.get("credential_id") != chain[i].credential_id: + raise ProvenanceLinkBroken(f"record {i} credential_id does not match the chain") diff --git a/src/ca2a_verify/verify.py b/src/ca2a_verify/verify.py index 1d03f71..f9d54cc 100644 --- a/src/ca2a_verify/verify.py +++ b/src/ca2a_verify/verify.py @@ -2,7 +2,7 @@ Thin wrapper over ``ca2a_runtime.delegation.verify_chain`` that loads a chain from JSON and returns a structured result. The delegation DAG verifier (linking -each hop's TRACE record to its parent) lands with the Tier 2 provenance work. +each hop's TRACE record to its parent) lives in ``ca2a_verify.dag``. """ from __future__ import annotations diff --git a/tests/unit/test_trace_binding.py b/tests/unit/test_trace_binding.py new file mode 100644 index 0000000..9ac861c --- /dev/null +++ b/tests/unit/test_trace_binding.py @@ -0,0 +1,213 @@ +"""Tests for the delegation-hop -> TRACE record binding and the TRACE DAG verifier. + +These exercise the software-attestation path end to end: build a linked TRACE DAG +from a delegation chain, verify it offline, confirm each record passes the TRACE +conformance suite at Level 0 (software-only records are honestly Level 0), and +confirm tampering and untrusted keys fail closed. +""" + +from __future__ import annotations + +import time + +import pytest +from agentrust_trace import generate_key, sign_record, validate_json +from trace_tests.runner import run as run_conformance + +from ca2a_runtime.delegation import DelegationCredential +from ca2a_runtime.errors import ProvenanceLinkBroken, TraceRecordInvalid +from ca2a_runtime.trace_binding import ( + EAT_PROFILE, + HopContext, + HopSpec, + build_trace_record, + digest, + emit_dag, + sign_trace_record, + trace_record_hash, +) +from ca2a_verify import cross_check_trace_dag, verify_trace_dag +from tests.unit.conftest import build_chain + +_NOW = int(time.time()) + + +def _software_context(label: str = "ca2a-peer") -> HopContext: + return HopContext.software( + model_provider="anthropic", + model_id="claude-opus-4-8", + image_label=label, + policy_bundle_hash=digest(b"policy-bundle"), + ) + + +def _hops(chain: list[DelegationCredential], keys: list) -> list[HopSpec]: + """One HopSpec per credential, each with its own signing key and subject.""" + return [ + HopSpec( + subject=f"spiffe://ca2a.example/peer/{i}", + signing_key=keys[i], + context=_software_context(f"peer-{i}"), + iat=_NOW, + credential_id=cred.credential_id, + ) + for i, cred in enumerate(chain) + ] + + +def _trusted(keys: list) -> list: + return [k.public_key() for k in keys] + + +# --- record construction --------------------------------------------------- + + +def test_root_record_is_schema_valid_when_signed() -> None: + record = build_trace_record(subject="spiffe://ca2a.example/root", iat=_NOW, context=_software_context()) + assert "delegation" not in record + assert record["eat_profile"] == EAT_PROFILE + signed = sign_trace_record(record, generate_key()) + validate_json(signed) # does not raise + assert "signature" in signed and signed["cnf"]["jwk"]["kty"] == "OKP" + + +def test_non_root_record_carries_delegation_block() -> None: + parent = sign_trace_record( + build_trace_record(subject="spiffe://ca2a.example/root", iat=_NOW, context=_software_context()), + generate_key(), + ) + child = build_trace_record( + subject="spiffe://ca2a.example/child", + iat=_NOW, + context=_software_context(), + credential_id="cred-1", + parent_record_hash=trace_record_hash(parent), + ) + assert child["delegation"] == { + "parent_record_hash": trace_record_hash(parent), + "credential_id": "cred-1", + } + + +def test_build_requires_both_credential_id_and_parent_hash() -> None: + with pytest.raises(ValueError, match="root hop supplies neither"): + build_trace_record( + subject="spiffe://ca2a.example/x", + iat=_NOW, + context=_software_context(), + credential_id="cred-1", # parent_record_hash missing + ) + + +def test_digest_has_sha256_prefix() -> None: + d = digest(b"x") + assert d.startswith("sha256:") and len(d) == len("sha256:") + 64 + + +# --- conformance ----------------------------------------------------------- + + +def test_each_record_passes_trace_level0() -> None: + keys = [generate_key() for _ in range(3)] + chain = build_chain( + [frozenset({"cap:a", "cap:b"}), frozenset({"cap:a", "cap:b"}), frozenset({"cap:a"})] + ) + records = emit_dag(_hops(chain, keys)) + for i, record in enumerate(records): + results = run_conformance(record, "trace", level=0) + failed = [f for findings in results.values() for f in findings if f.failed()] + assert not failed, f"record {i} failed Level 0: {failed}" + + +# --- DAG verification ------------------------------------------------------ + + +def test_emit_dag_verifies_offline() -> None: + keys = [generate_key() for _ in range(3)] + chain = build_chain( + [frozenset({"cap:a", "cap:b"}), frozenset({"cap:a", "cap:b"}), frozenset({"cap:a"})] + ) + records = emit_dag(_hops(chain, keys)) + result = verify_trace_dag(records, trusted_keys=_trusted(keys)) + assert result.hops == 3 + assert result.root_subject == "spiffe://ca2a.example/peer/0" + assert result.leaf_subject == "spiffe://ca2a.example/peer/2" + + +def test_cross_check_ties_dag_to_chain() -> None: + keys = [generate_key() for _ in range(3)] + chain = build_chain( + [frozenset({"cap:a", "cap:b"}), frozenset({"cap:a", "cap:b"}), frozenset({"cap:a"})] + ) + records = emit_dag(_hops(chain, keys)) + cross_check_trace_dag(records, chain) # does not raise + + +def test_cross_check_rejects_credential_mismatch() -> None: + keys = [generate_key() for _ in range(2)] + chain = build_chain([frozenset({"cap:a"}), frozenset({"cap:a"})]) + hops = _hops(chain, keys) + hops[1] = HopSpec( + subject=hops[1].subject, + signing_key=hops[1].signing_key, + context=hops[1].context, + iat=hops[1].iat, + credential_id="cred-forged", + ) + records = emit_dag(hops) + with pytest.raises(ProvenanceLinkBroken, match="credential_id does not match"): + cross_check_trace_dag(records, chain) + + +# --- fail-closed behavior -------------------------------------------------- + + +def test_tampered_record_fails_signature() -> None: + keys = [generate_key() for _ in range(2)] + chain = build_chain([frozenset({"cap:a"}), frozenset({"cap:a"})]) + records = emit_dag(_hops(chain, keys)) + records[1]["data_class"] = "public" # mutate after signing + with pytest.raises(TraceRecordInvalid, match="signature does not verify"): + verify_trace_dag(records, trusted_keys=_trusted(keys)) + + +def test_resigned_parent_breaks_child_link() -> None: + keys = [generate_key() for _ in range(2)] + chain = build_chain([frozenset({"cap:a"}), frozenset({"cap:a"})]) + records = emit_dag(_hops(chain, keys)) + # Re-sign the root with a *valid* signature but altered content: signature + # verifies, but the record hash changes, so the child's link no longer holds. + mutated_root = {k: v for k, v in records[0].items() if k != "signature"} + mutated_root["data_class"] = "public" + records[0] = sign_record(mutated_root, keys[0]) + with pytest.raises(ProvenanceLinkBroken, match="parent link does not match"): + verify_trace_dag(records, trusted_keys=_trusted(keys)) + + +def test_untrusted_key_rejected() -> None: + keys = [generate_key() for _ in range(2)] + chain = build_chain([frozenset({"cap:a"}), frozenset({"cap:a"})]) + records = emit_dag(_hops(chain, keys)) + with pytest.raises(TraceRecordInvalid, match="untrusted key"): + verify_trace_dag(records, trusted_keys=[generate_key().public_key()]) + + +def test_root_with_delegation_block_rejected() -> None: + key = generate_key() + rogue_root = sign_trace_record( + build_trace_record( + subject="spiffe://ca2a.example/root", + iat=_NOW, + context=_software_context(), + credential_id="cred-1", + parent_record_hash=digest(b"phantom-parent"), + ), + key, + ) + with pytest.raises(ProvenanceLinkBroken, match="root record must not carry"): + verify_trace_dag([rogue_root], trusted_keys=[key.public_key()]) + + +def test_empty_dag_rejected() -> None: + with pytest.raises(ProvenanceLinkBroken, match="empty TRACE DAG"): + verify_trace_dag([], trusted_keys=[])