From b2918c14cf1ea2a83eb0f288a247dc33c446aa3c Mon Sep 17 00:00:00 2001 From: Raghuvar Karthik Durga <177662415+drk-ram@users.noreply.github.com> Date: Fri, 30 Oct 2026 09:59:28 -0500 Subject: [PATCH] fix(delegation): reject private JWKs in trusted_root_keys, duplicate lattice entries, and invalid credential bounds (#647) - Reject RFC 7517/7518 private or symmetric JWK parameters in context.trusted_root_keys without echoing secret values. - Bound context.max_depth to JCS_SAFE_INTEGER and reject duplicate entries in context.data_class_lattice. - Require non-empty credential issuer and holder strings, integer not_before and not_after in [0, JCS_SAFE_INTEGER], and not_before <= not_after. - Seed visited with id(current) in verify_delegation_chain so a cycle returning to leaf terminates on the first return. Closes #647 Signed-off-by: Raghuvar Karthik Durga <177662415+drk-ram@users.noreply.github.com> --- src/agentrust_trace/delegation.py | 57 +++++++++++++++---- tests/test_delegation_module.py | 93 +++++++++++++++++++++++++++++++ 2 files changed, 140 insertions(+), 10 deletions(-) diff --git a/src/agentrust_trace/delegation.py b/src/agentrust_trace/delegation.py index 96d1af0e..f8cf2c78 100644 --- a/src/agentrust_trace/delegation.py +++ b/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( diff --git a/tests/test_delegation_module.py b/tests/test_delegation_module.py index 324336d7..8f522211 100644 --- a/tests/test_delegation_module.py +++ b/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 + + +