fix(verify): require a trusted key, enforce freshness, validate JWK type

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

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.

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

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)
 

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)

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)