fix(delegation): reject private JWKs in trusted_root_keys, duplicate lattice entries, and invalid credential bounds (#647)

src/agentrust_trace/delegation.py

@@ -29,7 +29,13 @@
 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.sign import (
+    JCS_SAFE_INTEGER,
+    _JWK_PRIVATE_PARAMS,
+    _b64url_decode,
+    _canonical_bytes,
+    _integer_value,
+)
 from agentrust_trace.validate import validate_json
 
 DELEGATION_DIGEST_ALGORITHMS: tuple[str, ...] = ("sha256", "sha384")
@@ -428,8 +434,14 @@ def _validate_context(context: Any) -> dict[str, Any]:
             )
 
     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")
+    if (
+        isinstance(max_depth, bool)
+        or not isinstance(max_depth, int)
+        or not 0 <= max_depth <= JCS_SAFE_INTEGER
+    ):
+        raise DelegationError(
+            "context.max_depth must be a non-negative integer within the JCS safe-integer range"
+        )
 
     lattice = context["data_class_lattice"]
     if (
@@ -440,6 +452,11 @@ def _validate_context(context: Any) -> dict[str, Any]:
         raise DelegationError(
             "context.data_class_lattice must be a sequence of non-empty strings"
         )
+    lattice_list = list(lattice)
+    if len(set(lattice_list)) != len(lattice_list):
+        raise DelegationError(
+            "context.data_class_lattice must not contain duplicate entries"
+        )
 
     trusted_keys = context["trusted_root_keys"]
     if (
@@ -455,6 +472,13 @@ def _validate_context(context: Any) -> dict[str, Any]:
             "context.trusted_root_keys must contain only JWK objects"
         )
     for jwk in trusted_list:
+        private_members = sorted(_JWK_PRIVATE_PARAMS.intersection(jwk))
+        if private_members:
+            raise DelegationError(
+                f"context.trusted_root_keys entry contains private key material "
+                f"({', '.join(repr(m) for m in private_members)}); "
+                "expected a public verification key"
+            )
         for member in ("kty", "crv", "x"):
             val = jwk.get(member)
             if not isinstance(val, str) or not val:
@@ -474,23 +498,36 @@ def _validate_context(context: Any) -> dict[str, Any]:
             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
+        issuer = cred.get("issuer")
+        holder = cred.get("holder")
+        if (
+            not isinstance(issuer, str)
+            or not issuer
+            or not isinstance(holder, str)
+            or not holder
         ):
             raise DelegationError(
-                f"credential {cid!r} must carry string issuer and holder"
+                f"credential {cid!r} must carry non-empty string issuer and holder"
             )
         for bound in ("not_before", "not_after"):
             val = cred.get(bound)
-            if isinstance(val, bool) or not isinstance(val, int):
+            if (
+                isinstance(val, bool)
+                or not isinstance(val, int)
+                or not 0 <= val <= JCS_SAFE_INTEGER
+            ):
                 raise DelegationError(
-                    f"credential {cid!r} must carry integer {bound}"
+                    f"credential {cid!r} must carry integer {bound} within [0, {JCS_SAFE_INTEGER}]"
                 )
+        if cred["not_before"] > cred["not_after"]:
+            raise DelegationError(
+                f"credential {cid!r} not_before must not exceed not_after"
+            )
 
     return {
         **context,
         "supported_digest_algorithms": list(supported),
-        "data_class_lattice": list(lattice),
+        "data_class_lattice": lattice_list,
         "trusted_root_keys": trusted_list,
     }
 
@@ -564,7 +601,7 @@ def verify_delegation_chain(
     failures: list[str] = []
     warnings: list[str] = []
     depth = 0
-    visited: set[int] = set()
+    visited: set[int] = {id(current)}
 
     while True:
         hop = Hop(

tests/test_delegation_module.py

@@ -295,3 +295,96 @@ def test_verify_delegation_chain_refuses_non_canonical_signature_and_unknown_mem
         )
 
 
+@pytest.mark.parametrize("member", ["d", "p", "q", "dp", "dq", "qi", "oth", "k"])
+def test_verify_delegation_chain_rejects_private_jwk_members_in_trusted_root_keys(
+    member: str,
+) -> None:
+    vec01 = _load("01-valid-single-hop.json")
+    assert (
+        delegation.verify_delegation_chain(
+            vec01["records"], vec01["context"]["leaf"], vec01["context"]
+        ).classification
+        == "verified"
+    )
+    sentinel = "secret-material-do-not-echo-999"
+    bad_ctx = copy.deepcopy(vec01["context"])
+    bad_ctx["trusted_root_keys"][0][member] = sentinel
+    with pytest.raises(
+        delegation.DelegationError, match="private key material"
+    ) as exc_info:
+        delegation.verify_delegation_chain(
+            vec01["records"], vec01["context"]["leaf"], bad_ctx
+        )
+    assert member in str(exc_info.value)
+    assert sentinel not in str(exc_info.value)
+
+
+def test_verify_delegation_chain_rejects_max_depth_above_safe_int() -> None:
+    vec01 = _load("01-valid-single-hop.json")
+    ok_ctx = copy.deepcopy(vec01["context"])
+    ok_ctx["max_depth"] = 2**53 - 1
+    assert (
+        delegation.verify_delegation_chain(
+            vec01["records"], ok_ctx["leaf"], ok_ctx
+        ).classification
+        == "verified"
+    )
+    bad_ctx = copy.deepcopy(vec01["context"])
+    bad_ctx["max_depth"] = 2**53
+    with pytest.raises(delegation.DelegationError, match="max_depth"):
+        delegation.verify_delegation_chain(
+            vec01["records"], bad_ctx["leaf"], bad_ctx
+        )
+
+
+def test_verify_delegation_chain_rejects_duplicate_data_class_lattice_entries() -> None:
+    vec01 = _load("01-valid-single-hop.json")
+    bad_ctx = copy.deepcopy(vec01["context"])
+    bad_ctx["data_class_lattice"] = ["public", "confidential", "public"]
+    with pytest.raises(
+        delegation.DelegationError, match="must not contain duplicate entries"
+    ):
+        delegation.verify_delegation_chain(
+            vec01["records"], bad_ctx["leaf"], bad_ctx
+        )
+
+
+@pytest.mark.parametrize(
+    ("mutated_fields", "expected_match"),
+    [
+        ({"issuer": ""}, "non-empty string issuer and holder"),
+        ({"holder": ""}, "non-empty string issuer and holder"),
+        ({"not_before": -1}, "not_before"),
+        ({"not_before": 2**53}, "not_before"),
+        ({"not_after": -1, "not_before": 0}, "not_after"),
+        ({"not_after": 2**53}, "not_after"),
+        ({"not_before": 2000, "not_after": 1000}, "not_before must not exceed not_after"),
+    ],
+)
+def test_verify_delegation_chain_rejects_invalid_credential_identities_and_windows(
+    mutated_fields: dict[str, Any], expected_match: str
+) -> None:
+    vec01 = _load("01-valid-single-hop.json")
+    bad_ctx = copy.deepcopy(vec01["context"])
+    cid = next(iter(bad_ctx["credentials"]))
+    bad_ctx["credentials"][cid].update(mutated_fields)
+    with pytest.raises(delegation.DelegationError, match=expected_match):
+        delegation.verify_delegation_chain(
+            vec01["records"], bad_ctx["leaf"], bad_ctx
+        )
+
+
+def test_verify_delegation_chain_stops_at_first_return_to_leaf_in_cycle(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    vec01 = _load("01-valid-single-hop.json")
+    leaf = copy.deepcopy(vec01["records"][0])
+    target_hash = leaf["delegation"]["parent_record_hash"]
+    monkeypatch.setattr(delegation, "parent_record_hash", lambda rec, alg="sha256": target_hash)
+    result = delegation.verify_delegation_chain(
+        [leaf], target_hash, vec01["context"]
+    )
+    assert result.depth == 1
+
+
+