From aaa3a0bed62e73e44760567d644fd6f75c463f6b Mon Sep 17 00:00:00 2001 From: Imran Siddique Date: Tue, 30 Jun 2026 12:15:32 -0700 Subject: [PATCH] fix(verify): require a trusted key, enforce freshness, validate JWK type Harden verify_record so it fails closed: - Require an explicit trusted key. verify_record no longer falls back to the key embedded in record.cnf.jwk by default, so a record can no longer vouch for itself. The embedded-key path is now gated behind an explicit allow_embedded_key=True opt-in that emits a loud UserWarning; with no trusted key and the opt-in off, it raises ValueError. - Enforce freshness. Add a max_age_seconds check (default 86400 = 24h, configurable, None to disable) against iat, and an optional constant-time expected_nonce comparison against runtime.nonce. Stale or mismatched records fail verification. - Validate JWK type before key reconstruction. Assert kty=="OKP" and crv=="Ed25519" before building an Ed25519PublicKey, so an EC key is never silently treated as Ed25519. - Robust base64url decode. Factor padding into one helper that raises ValueError (not binascii.Error) on malformed input, applied to both the signature and the JWK x field. Tests updated to pass a trusted key (or opt in), plus new tests for the no-key, wrong-key, expired, non-OKP, nonce, and malformed-signature paths. Doc code examples updated to verify against a trusted key. RFC 8785/JCS canonicalization and revocation enforcement are out of scope and tracked separately. Co-Authored-By: Claude Opus 4.8 (1M context) --- docs/quickstart.md | 6 +- .../hardware-attestation-platforms.md | 4 +- .../signing-your-first-trust-record.md | 6 +- docs/tutorials/verifying-a-trust-record.md | 22 ++-- src/agentrust_trace/sign.py | 124 +++++++++++++++--- tests/test_agt_adapter.py | 6 +- tests/test_sign.py | 93 ++++++++++++- 7 files changed, 216 insertions(+), 45 deletions(-) diff --git a/docs/quickstart.md b/docs/quickstart.md index 5264e3e..4d732e4 100644 --- a/docs/quickstart.md +++ b/docs/quickstart.md @@ -148,9 +148,11 @@ with open("session.trace.json") as f: # Schema check validate_json(signed_record) # raises jsonschema.ValidationError if malformed -# Signature check — uses the cnf.jwk embedded in the record +# Signature check — verify against a pinned trusted key in production. +# allow_embedded_key=True trusts the cnf.jwk in the record itself, which +# only proves internal consistency, not that the record came from a trusted issuer. try: - verify_record(signed_record) + verify_record(signed_record, allow_embedded_key=True) print("Signature valid (Ed25519)") print(f" subject: {signed_record['subject']}") print(f" policy: {signed_record['policy']['bundle_hash'][:24]}... " diff --git a/docs/tutorials/hardware-attestation-platforms.md b/docs/tutorials/hardware-attestation-platforms.md index 62d6d8f..853dcc4 100644 --- a/docs/tutorials/hardware-attestation-platforms.md +++ b/docs/tutorials/hardware-attestation-platforms.md @@ -162,9 +162,9 @@ Step 3 proves the key that signed the TRACE record was generated by the expected ```python from agentrust_trace import verify_record, validate_json -# Step 1: verify the TRACE record structure and signature +# Step 1: verify the TRACE record structure and signature against a trusted key validate_json(record) -verify_record(record) +verify_record(record, trusted_jwk) # trusted_jwk obtained out-of-band # Step 2 + 3: platform-specific — outside the scope of agentrust-trace # For cMCP-issued records, use cmcp-verify which handles the full chain. diff --git a/docs/tutorials/signing-your-first-trust-record.md b/docs/tutorials/signing-your-first-trust-record.md index 1cc74de..1278696 100644 --- a/docs/tutorials/signing-your-first-trust-record.md +++ b/docs/tutorials/signing-your-first-trust-record.md @@ -124,14 +124,14 @@ The spec (section 3.2.2) requires JCS canonical form. Do not reimplement this by ## Verify the Signed Record -`verify_record()` extracts `cnf.jwk`, recomputes the canonical bytes, and checks the Ed25519 signature. It raises `cryptography.exceptions.InvalidSignature` if the record was tampered with, and returns `None` on success. +`verify_record()` requires a trusted key, recomputes the canonical bytes, and checks the Ed25519 signature. It raises `cryptography.exceptions.InvalidSignature` if the record was tampered with, and returns `None` on success. Pass the public JWK of the key you trust — here, the `jwk` from earlier in the tutorial. ```python from agentrust_trace import verify_record from cryptography.exceptions import InvalidSignature try: - verify_record(signed) + verify_record(signed, jwk) # jwk is the public key you trust print("signature valid") except InvalidSignature: print("tampered — do not trust this record") @@ -146,7 +146,7 @@ tampered = copy.deepcopy(signed) tampered["data_class"] = "public" # change a field after signing try: - verify_record(tampered) + verify_record(tampered, jwk) except InvalidSignature: print("correctly rejected") # this branch runs ``` diff --git a/docs/tutorials/verifying-a-trust-record.md b/docs/tutorials/verifying-a-trust-record.md index acf77d2..efc5d37 100644 --- a/docs/tutorials/verifying-a-trust-record.md +++ b/docs/tutorials/verifying-a-trust-record.md @@ -21,15 +21,16 @@ pip install agentrust-trace ## Understand What verify_record() Does -When you call `verify_record(record)` without a second argument, the library: +When you call `verify_record(record, trusted_key)` the library: 1. Reads `record["signature"]` and base64url-decodes it to raw bytes -2. Reads `record["cnf"]["jwk"]` and reconstructs an `Ed25519PublicKey` from the `x` field -3. Rebuilds the canonical payload: all fields except `signature`, serialized with sorted keys and no whitespace -4. Calls `Ed25519PublicKey.verify(sig_bytes, payload_bytes)` from the `cryptography` library -5. Returns `None` on success, raises `cryptography.exceptions.InvalidSignature` on failure +2. Resolves the trusted key you supplied, checking `kty == "OKP"` and `crv == "Ed25519"` before reconstructing an `Ed25519PublicKey` from the `x` field +3. Enforces freshness: rejects records whose `iat` is older than `max_age_seconds` (default 24h), and, if you pass `expected_nonce`, compares it in constant time to `runtime.nonce` +4. Rebuilds the canonical payload: all fields except `signature`, serialized with sorted keys and no whitespace +5. Calls `Ed25519PublicKey.verify(sig_bytes, payload_bytes)` from the `cryptography` library +6. Returns `None` on success, raises `cryptography.exceptions.InvalidSignature` on a bad signature and `ValueError` on every other rejection -The public key is embedded in the record itself via `cnf.jwk`. This means any party can verify the record offline, without contacting the issuer. +A trusted key is required. A record cannot authenticate itself with the key it embeds, so `verify_record` will not fall back to `cnf.jwk` unless you explicitly opt in with `allow_embedded_key=True` (which emits a `UserWarning`, since it proves only internal consistency, not authenticity). ```python import json @@ -39,16 +40,17 @@ from cryptography.exceptions import InvalidSignature with open("session.trace.json") as f: record = json.load(f) +# trusted_jwk obtained out-of-band (see "Verify Against a Pinned Public Key" below) try: - verify_record(record) + verify_record(record, trusted_jwk) print("signature valid") except InvalidSignature: print("signature invalid — record may have been tampered with") except ValueError as e: - print(f"record malformed: {e}") + print(f"verification failed: {e}") ``` -`ValueError` is raised when the record is missing a `signature` field or the `cnf.jwk` cannot be decoded. Treat both as verification failure. +`ValueError` is raised when no trusted key is supplied, the record is missing a `signature` field, a key or signature cannot be decoded, the JWK type is not Ed25519, or the record is stale. Treat all of these as verification failure. --- @@ -201,7 +203,7 @@ from agentrust_trace import verify_record, validate_json, iter_errors from cryptography.exceptions import InvalidSignature import jsonschema -def verify_trust_record(path: str, trusted_jwk: dict | None = None) -> dict: +def verify_trust_record(path: str, trusted_jwk: dict) -> dict: with open(path) as f: record = json.load(f) diff --git a/src/agentrust_trace/sign.py b/src/agentrust_trace/sign.py index 1bd37a3..bed10e4 100644 --- a/src/agentrust_trace/sign.py +++ b/src/agentrust_trace/sign.py @@ -9,6 +9,7 @@ from __future__ import annotations import base64 +import binascii import json import os import warnings @@ -61,6 +62,21 @@ def _canonical_bytes(d: dict[str, Any]) -> bytes: return json.dumps(d, sort_keys=True, separators=(",", ":"), ensure_ascii=True).encode() +def _b64url_decode(value: str, *, field: str) -> bytes: + """Decode an unpadded base64url string, raising ValueError on malformed input. + + Restores the padding the encoder stripped and surfaces ``binascii`` decode + failures as ``ValueError`` so callers see one consistent failure type. + """ + if not isinstance(value, str): + raise ValueError(f"{field} must be a base64url string") + padded = value + "=" * (-len(value) % 4) + try: + return base64.urlsafe_b64decode(padded) + except (binascii.Error, ValueError) as exc: + raise ValueError(f"{field} is not valid base64url: {exc}") from exc + + def sign_record(record: dict[str, Any], key: Ed25519PrivateKey) -> dict[str, Any]: """Return a copy of *record* with ``cnf.jwk`` populated and a ``signature`` field added. @@ -80,45 +96,115 @@ def sign_record(record: dict[str, Any], key: Ed25519PrivateKey) -> dict[str, Any return {**payload, "signature": sig_b64} -def verify_record(record: dict[str, Any], public_key_or_jwk: Any = None) -> None: - """Verify an Ed25519 signature on a signed TRACE Trust Record. +def _pubkey_from_jwk(jwk: dict[str, Any]) -> Any: + """Reconstruct an Ed25519 public key from a JWK, rejecting other key types. + + Asserts ``kty == "OKP"`` and ``crv == "Ed25519"`` before building the key so + that, for example, an EC key is never silently treated as Ed25519. + """ + from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PublicKey - Raises InvalidSignature if the signature does not verify. - Raises ValueError if the record has no signature field or the key cannot - be decoded. + kty = jwk.get("kty") + crv = jwk.get("crv") + if kty != "OKP": + raise ValueError(f"unsupported JWK kty {kty!r}; expected 'OKP' for Ed25519") + if crv != "Ed25519": + raise ValueError(f"unsupported JWK crv {crv!r}; expected 'Ed25519'") + + x_b64 = jwk.get("x") + if not x_b64: + raise ValueError("JWK missing 'x' field") + x_bytes = _b64url_decode(x_b64, field="JWK 'x'") + return Ed25519PublicKey.from_public_bytes(x_bytes) + + +def verify_record( + record: dict[str, Any], + public_key_or_jwk: Any = None, + *, + allow_embedded_key: bool = False, + max_age_seconds: int | None = 86400, + expected_nonce: str | None = None, +) -> None: + """Verify an Ed25519 signature on a signed TRACE Trust Record. - If public_key_or_jwk is None, the public key is taken from record["cnf"]["jwk"]. - Pass an Ed25519PublicKey or a JWK dict to verify against an explicit key. + A trusted key is REQUIRED. Pass an ``Ed25519PublicKey`` or a JWK dict via + *public_key_or_jwk* to verify against a key the caller already trusts. + + Raises ``InvalidSignature`` if the signature does not verify, and ``ValueError`` + for every other rejection (no signature, no trusted key, malformed input, + unsupported JWK type, stale record, or nonce mismatch). Returns ``None`` on + success. All checks fail closed. + + Trust anchoring (fail closed): + Without a trusted key, the record cannot vouch for itself, so verification + is refused. Set ``allow_embedded_key=True`` to opt in to verifying against + ``record["cnf"]["jwk"]`` — this only proves internal consistency, not + authenticity, and emits a loud ``UserWarning``. + + Freshness (fail closed): + ``max_age_seconds`` (default 86400 = 24h) bounds how old ``record["iat"]`` + may be relative to now; pass ``None`` to disable the age check. If + ``expected_nonce`` is given, it is compared in constant time against + ``record["runtime"]["nonce"]``. A stale record or nonce mismatch raises + ``ValueError``. """ + import time + from hmac import compare_digest + from cryptography.exceptions import InvalidSignature as _InvalidSignature # noqa: F401 - from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PublicKey sig_b64 = record.get("signature") if not sig_b64: raise ValueError("record has no 'signature' field") - # Decode signature - pad = 4 - len(sig_b64) % 4 - sig_bytes = base64.urlsafe_b64decode(sig_b64 + "=" * (pad % 4)) + sig_bytes = _b64url_decode(sig_b64, field="signature") - # Resolve public key + # Resolve the trusted public key. A trusted key is required: a record cannot + # authenticate itself with the key it embeds. if public_key_or_jwk is None: + if not allow_embedded_key: + raise ValueError( + "verify_record requires a trusted key. Pass an Ed25519PublicKey or " + "JWK dict, or set allow_embedded_key=True to (insecurely) trust the " + "key embedded in record.cnf.jwk." + ) jwk = record.get("cnf", {}).get("jwk", {}) if not jwk: raise ValueError("record has no cnf.jwk and no public key was supplied") + warnings.warn( + "verify_record is trusting the key embedded in record.cnf.jwk " + "(allow_embedded_key=True). This proves the record is internally " + "consistent, NOT that it came from a trusted issuer. Verify against a " + "pinned trusted key in production.", + UserWarning, + stacklevel=2, + ) public_key_or_jwk = jwk if isinstance(public_key_or_jwk, dict): - jwk = public_key_or_jwk - x_b64 = jwk.get("x") - if not x_b64: - raise ValueError("JWK missing 'x' field") - pad = 4 - len(x_b64) % 4 - x_bytes = base64.urlsafe_b64decode(x_b64 + "=" * (pad % 4)) - pub = Ed25519PublicKey.from_public_bytes(x_bytes) + pub = _pubkey_from_jwk(public_key_or_jwk) else: pub = public_key_or_jwk + # Freshness: bound the age of the record against its issued-at timestamp. + if max_age_seconds is not None: + iat = record.get("iat") + if not isinstance(iat, (int, float)) or isinstance(iat, bool): + raise ValueError("record has no valid integer 'iat' for freshness check") + age = time.time() - iat + if age > max_age_seconds: + raise ValueError( + f"record is stale: iat is {int(age)}s old, exceeds max_age_seconds=" + f"{max_age_seconds}" + ) + + # Freshness: bind to a caller-supplied nonce when provided. + if expected_nonce is not None: + actual_nonce = record.get("runtime", {}).get("nonce") + if not isinstance(actual_nonce, str) or not compare_digest(actual_nonce, expected_nonce): + raise ValueError("record runtime.nonce does not match expected_nonce") + # Canonical bytes: record without "signature" key record_no_sig = {k: v for k, v in record.items() if k != "signature"} msg = _canonical_bytes(record_no_sig) diff --git a/tests/test_agt_adapter.py b/tests/test_agt_adapter.py index 5b154d4..f338eab 100644 --- a/tests/test_agt_adapter.py +++ b/tests/test_agt_adapter.py @@ -8,7 +8,7 @@ import pytest from pydantic import ValidationError -from agentrust_trace import TrustRecord, sign_record, generate_key, verify_record +from agentrust_trace import TrustRecord, sign_record, generate_key, key_to_jwk, verify_record from agentrust_trace.adapters import AGTSessionResult, TraceAGTAdapter # --------------------------------------------------------------------------- @@ -144,8 +144,8 @@ def test_sign_and_verify_round_trip() -> None: record = adapter.build_trust_record(session) key = generate_key() signed = sign_record(record, key) - # Must not raise - verify_record(signed) + # Must not raise — verify against the trusted signing key. + verify_record(signed, key_to_jwk(key)) # Structural validation of signed record TrustRecord.model_validate(signed) diff --git a/tests/test_sign.py b/tests/test_sign.py index 6a57748..341692a 100644 --- a/tests/test_sign.py +++ b/tests/test_sign.py @@ -1,6 +1,7 @@ """Tests for agentrust_trace.sign.""" import base64 +import time import pytest from cryptography.exceptions import InvalidSignature @@ -120,21 +121,101 @@ def test_sign_record_spiffe_subject(): assert validated.subject.startswith("spiffe://") +def _trusted_jwk(record: dict) -> dict: + """Return the public JWK of the key that signed *record* (the trust anchor).""" + return record["cnf"]["jwk"] + + +def _fresh_record() -> dict: + """A minimal record with a recent iat so the freshness check passes.""" + record = _minimal_record() + record["iat"] = int(time.time()) + return record + + def test_verify_record_passes_for_valid_signature(): key = generate_key() - record = sign_record(_minimal_record(), key) - verify_record(record) # must not raise + record = sign_record(_fresh_record(), key) + verify_record(record, key_to_jwk(key)) # must not raise def test_verify_record_raises_for_tampered_record(): key = generate_key() - record = sign_record(_minimal_record(), key) - record["iat"] = record["iat"] + 1 # tamper + record = sign_record(_fresh_record(), key) + trusted = key_to_jwk(key) + record["iat"] = record["iat"] + 1 # tamper (and still fresh) with pytest.raises(InvalidSignature): - verify_record(record) + verify_record(record, trusted) def test_verify_record_raises_for_missing_signature(): - record = dict(_minimal_record()) + record = dict(_fresh_record()) + key = generate_key() with pytest.raises(ValueError, match="no 'signature' field"): + verify_record(record, key_to_jwk(key)) + + +def test_verify_record_requires_trusted_key_by_default(): + key = generate_key() + record = sign_record(_fresh_record(), key) + with pytest.raises(ValueError, match="requires a trusted key"): verify_record(record) + + +def test_verify_record_embedded_key_opt_in_warns(): + key = generate_key() + record = sign_record(_fresh_record(), key) + with pytest.warns(UserWarning, match="cnf.jwk"): + verify_record(record, allow_embedded_key=True) + + +def test_verify_record_rejects_wrong_trusted_key(): + key_a = generate_key() + key_b = generate_key() + record = sign_record(_fresh_record(), key_a) + # Signed by A, verified against B's public key — must not verify. + with pytest.raises(InvalidSignature): + verify_record(record, key_to_jwk(key_b)) + + +def test_verify_record_rejects_expired_record(): + key = generate_key() + record = _minimal_record() + record["iat"] = int(time.time()) - 90000 # ~25h old, beyond 24h default + record = sign_record(record, key) + with pytest.raises(ValueError, match="stale"): + verify_record(record, key_to_jwk(key)) + + +def test_verify_record_expired_allowed_when_max_age_none(): + key = generate_key() + record = _minimal_record() + record["iat"] = int(time.time()) - 90000 + record = sign_record(record, key) + verify_record(record, key_to_jwk(key), max_age_seconds=None) # must not raise + + +def test_verify_record_rejects_non_okp_jwk(): + key = generate_key() + record = sign_record(_fresh_record(), key) + ec_jwk = {"kty": "EC", "crv": "P-256", "x": "abc", "y": "def"} + with pytest.raises(ValueError, match="kty"): + verify_record(record, ec_jwk) + + +def test_verify_record_nonce_match(): + key = generate_key() + record = _fresh_record() + record["runtime"]["nonce"] = "abc123" + record = sign_record(record, key) + verify_record(record, key_to_jwk(key), expected_nonce="abc123") # must not raise + with pytest.raises(ValueError, match="nonce"): + verify_record(record, key_to_jwk(key), expected_nonce="wrong") + + +def test_verify_record_rejects_malformed_signature(): + key = generate_key() + record = sign_record(_fresh_record(), key) + record["signature"] = "!!!not base64!!!" + with pytest.raises(ValueError, match="base64url"): + verify_record(record, key_to_jwk(key))