From 76cacd81a18c40597e7088aaae3c05059071d754 Mon Sep 17 00:00:00 2001 From: Raghuvar Karthik Durga <177662415+drk-ram@users.noreply.github.com> Date: Thu, 29 Oct 2026 12:13:47 -0500 Subject: [PATCH] fix(intent_bridge,successor_observation): validate caller options and JWKs at entry and deep-copy observation (#617) Signed-off-by: Raghuvar Karthik Durga <177662415+drk-ram@users.noreply.github.com> --- src/agentrust_trace/intent_bridge.py | 69 ++--- src/agentrust_trace/successor_observation.py | 31 ++- tests/test_intent_bridge.py | 260 ++++++++++++++++++- tests/test_successor_observation.py | 45 ++++ 4 files changed, 371 insertions(+), 34 deletions(-) diff --git a/src/agentrust_trace/intent_bridge.py b/src/agentrust_trace/intent_bridge.py index d0e573d0..dba33587 100644 --- a/src/agentrust_trace/intent_bridge.py +++ b/src/agentrust_trace/intent_bridge.py @@ -140,6 +140,17 @@ def verify_successor_artifact( decides trust and independence on that name, so a key the verifier accepts for one observer must not be able to sign under another observer's name. """ + if not isinstance(trusted_observer_jwk, dict): + raise IntentBridgeError("trusted_observer_jwk must be an object") + try: + pub = _pubkey_from_jwk(trusted_observer_jwk) + except ValueError as exc: + raise IntentBridgeError(f"trusted_observer_jwk is unusable: {exc}") from exc + trusted_kid = trusted_observer_jwk.get("kid") + if not isinstance(trusted_kid, str) or not trusted_kid: + raise IntentBridgeError("trusted_observer_jwk kid must be a non-empty string") + expected_observer = _nonempty_string(trusted_observer, "trusted_observer") + expected_authorization_id = _nonempty_string(authorization_id, "authorization_id") root = _object( artifact, "successor artifact", @@ -155,7 +166,6 @@ def verify_successor_artifact( ) if root.get("profile") != SUCCESSOR_PROFILE: raise IntentBridgeError("unknown successor profile; best-effort parsing is refused") - expected_authorization_id = _nonempty_string(authorization_id, "authorization_id") actual_authorization_id = _nonempty_string( root.get("authorization_id"), "successor.authorization_id" ) @@ -167,9 +177,6 @@ def verify_successor_artifact( observer_key_id = _nonempty_string( root.get("observer_key_id"), "successor.observer_key_id" ) - if not isinstance(trusted_observer_jwk, dict): - raise IntentBridgeError("trusted_observer_jwk must be an object") - expected_observer = _nonempty_string(trusted_observer, "trusted_observer") # By value, not by spelling (spec section 3.2.2), as in the envelope it is # compared with below: the signed body is its RFC 8785 form, which writes # `150.0` as `150`, so a type test here refused a spelling the signature covers. @@ -185,8 +192,7 @@ def verify_successor_artifact( "successor.observed_at must be a non-negative integer within " f"the JCS safe-integer range: it is {observed_at}" ) - trusted_kid = trusted_observer_jwk.get("kid") - if not isinstance(trusted_kid, str) or trusted_kid != observer_key_id: + if trusted_kid != observer_key_id: raise IntentBridgeError("observer_key_id does not identify the trusted observer key") if observer != expected_observer: raise IntentBridgeError( @@ -203,10 +209,6 @@ def verify_successor_artifact( signature = _b64url_decode(signature_value, field="successor.signature") except ValueError as exc: raise IntentBridgeError(str(exc)) from exc - try: - pub = _pubkey_from_jwk(trusted_observer_jwk) - except ValueError as exc: - raise IntentBridgeError(f"trusted_observer_jwk is unusable: {exc}") from exc signed = { "profile": SUCCESSOR_PROFILE, "authorization_id": actual_authorization_id, @@ -261,6 +263,7 @@ def _bind_successor_observation( This establishes integrity only. Trust, freshness, independence, and predicate sufficiency are deliberately evaluated separately. """ + expected = _digest(expected_successor_digest, "expected_successor_digest") if not isinstance(after, dict): raise AuthorizationMismatch("successor envelope must be a successor observation object") required = {"observation", "observer", "observed_at"} @@ -291,7 +294,6 @@ def _bind_successor_observation( "successor envelope observed_at must be a non-negative integer within " f"the JCS safe-integer range: it is {observed_at}" ) - expected = _digest(expected_successor_digest, "expected_successor_digest") try: actual = digest_jcs(after) except IntentBridgeError: @@ -337,6 +339,29 @@ def verify_bridge( transcript is required, its pre-execution/dispatch before half must be supplied. Post-execution successor evidence is verified as a separate observer-signed artifact. """ + if not isinstance(trusted_authorizer_jwk, dict): + raise IntentBridgeError("trusted_authorizer_jwk must be an object") + try: + pub = _pubkey_from_jwk(trusted_authorizer_jwk) + except ValueError as exc: + raise IntentBridgeError(f"trusted_authorizer_jwk is unusable: {exc}") from exc + trusted_kid = trusted_authorizer_jwk.get("kid") + if not isinstance(trusted_kid, str) or not trusted_kid: + raise IntentBridgeError("trusted_authorizer_jwk kid must be a non-empty string") + invalid_now_type = not isinstance(now, (int, type(None))) or isinstance(now, bool) + if invalid_now_type or (now is not None and (now < 0 or now > JCS_SAFE_INTEGER)): + raise IntentBridgeError( + "now must be a non-negative integer Unix timestamp within the JCS safe-integer range" + ) + if not isinstance(declaration, dict): + raise IntentBridgeError("declaration must be an object") + if not isinstance(tool_call, dict): + raise IntentBridgeError("tool_call must be an object") + _digest(pic_intent_digest, "intent_digest") + _digest(pic_args_digest, "args_digest") + declaration_digest = digest_jcs(declaration) + tool_call_digest = digest_jcs(tool_call) + root = _object(bridge, "bridge", {"profile", "authorization", "signature"}) if root.get("profile") != BRIDGE_PROFILE: raise IntentBridgeError("unknown bridge profile; best-effort parsing is refused") @@ -368,15 +393,8 @@ def verify_bridge( for field in ("authorization_id", "authorizer", "authorizer_key_id"): _nonempty_string(authorization[field], f"authorization.{field}") _decision(authorization["decision"]) - if not isinstance(trusted_authorizer_jwk, dict): - raise IntentBridgeError("trusted_authorizer_jwk must be an object") - trusted_kid = trusted_authorizer_jwk.get("kid") - if not isinstance(trusted_kid, str) or trusted_kid != authorization["authorizer_key_id"]: + if trusted_kid != authorization["authorizer_key_id"]: raise IntentBridgeError("authorizer_key_id does not identify the trusted key") - try: - pub = _pubkey_from_jwk(trusted_authorizer_jwk) - except ValueError as exc: - raise IntentBridgeError(f"trusted_authorizer_jwk is unusable: {exc}") from exc # Hoisted out of the try below. Inside it, an authorization JCS cannot serialize # was reported as "the signature is invalid", which is a different fact and sends @@ -406,11 +424,6 @@ def verify_bridge( f"{field} must be a non-negative integer Unix timestamp: it is {timestamp}" ) normalized_auth[field] = timestamp - invalid_now_type = not isinstance(now, (int, type(None))) or isinstance(now, bool) - if invalid_now_type or (now is not None and (now < 0 or now > JCS_SAFE_INTEGER)): - raise IntentBridgeError( - "now must be a non-negative integer Unix timestamp within the JCS safe-integer range" - ) instant = int(time.time()) if now is None else now if instant < normalized_auth["authorized_at"]: raise IntentBridgeError("authorization is not yet valid") @@ -422,12 +435,8 @@ def verify_bridge( scope = _object(authorization["scope"], "authorization.scope", {"tools", "impacts"}) tools = _unique_nonempty_strings(scope.get("tools"), "authorization.scope.tools") impacts = _unique_nonempty_strings(scope.get("impacts"), "authorization.scope.impacts") - if not isinstance(tool_call, dict): - raise IntentBridgeError("tool_call must be an object") if tool_call.get("name") not in tools: raise AuthorizationMismatch("executed tool is outside the authorized tool scope") - if not isinstance(declaration, dict): - raise IntentBridgeError("declaration must be an object") if declaration.get("impact") not in impacts: raise AuthorizationMismatch("declaration impact is outside the authorized impact scope") @@ -438,13 +447,11 @@ def verify_bridge( raise IntentBridgeError("authorization.pic.profile is not PIC-CJSON/1.0") for name, actual in (("intent_digest", pic_intent_digest), ("args_digest", pic_args_digest)): expected = _digest(pic.get(name), f"authorization.pic.{name}") - _digest(actual, name) if not compare_digest(expected, actual): raise AuthorizationMismatch(f"PIC {name} does not match the signed authorization") - tool_call_digest = digest_jcs(tool_call) digest_pairs = ( - ("declaration_digest", digest_jcs(declaration)), + ("declaration_digest", declaration_digest), ("tool_call_digest", tool_call_digest), ) for name, actual in digest_pairs: diff --git a/src/agentrust_trace/successor_observation.py b/src/agentrust_trace/successor_observation.py index 4a99ea72..d078f473 100644 --- a/src/agentrust_trace/successor_observation.py +++ b/src/agentrust_trace/successor_observation.py @@ -48,6 +48,35 @@ def _not_established(reason: str) -> SuccessorOutcome: return SuccessorOutcome("not-established", reason) +def _clone_observation(root: dict[str, Any]) -> dict[str, Any]: + """Iteratively deep-copy a JSON-compatible observation dict without recursion limits.""" + cloned_root: dict[str, Any] = {} + stack: list[tuple[Any, Any]] = [(root, cloned_root)] + tuple_fixups: list[tuple[Any, Any, list[Any]]] = [] + while stack: + src, dst = stack.pop() + items = src.items() if isinstance(src, dict) else enumerate(src) + for k, v in items: + if isinstance(v, dict): + child_dict: dict[str, Any] = {} + dst[k] = child_dict + stack.append((v, child_dict)) + elif isinstance(v, list): + child_list: list[Any] = [None] * len(v) + dst[k] = child_list + stack.append((v, child_list)) + elif isinstance(v, tuple): + staged: list[Any] = [None] * len(v) + dst[k] = staged + tuple_fixups.append((dst, k, staged)) + stack.append((v, staged)) + else: + dst[k] = v + for parent, key, staged in reversed(tuple_fixups): + parent[key] = tuple(staged) + return cloned_root + + def evaluate_successor_observation( after: dict[str, Any] | None, *, @@ -125,7 +154,7 @@ def evaluate_successor_observation( "independent observation is required but observer is the executor" ) - result = predicate(observation) + result = predicate(_clone_observation(observation)) if result is True: return SuccessorOutcome( "established", diff --git a/tests/test_intent_bridge.py b/tests/test_intent_bridge.py index 3644a6d1..6332325d 100644 --- a/tests/test_intent_bridge.py +++ b/tests/test_intent_bridge.py @@ -12,6 +12,7 @@ AuthorizationDenied, AuthorizationMismatch, IntentBridgeError, + _bind_successor_observation, digest_jcs, sign_bridge, sign_successor_artifact, @@ -528,7 +529,7 @@ def test_runtime_verifier_enforces_schema_scalar_constraints( bridge, key, declaration, intent, args, tool_call, transcript = _fixture() bridge["authorization"][field] = value bridge = sign_bridge(bridge["authorization"], key) - trusted = {**key_to_jwk(key), "kid": "" if field == "authorizer_key_id" else "key-7"} + trusted = {**key_to_jwk(key), "kid": "key-7"} with pytest.raises(IntentBridgeError, match=message): verify_bridge( bridge, trusted, declaration=declaration, pic_intent_digest=intent, @@ -669,7 +670,6 @@ def test_bridge_naming_another_authorizer_key_id_is_rejected() -> None: for mismatched_jwk in ( {**key_to_jwk(other_key), "kid": "key-other"}, {**key_to_jwk(key), "kid": "key-other"}, - key_to_jwk(key), ): with pytest.raises(IntentBridgeError, match="does not identify the trusted key"): verify_bridge( @@ -758,3 +758,259 @@ def test_sign_bridge_requires_authorization_to_be_a_json_object(bad_auth: object with pytest.raises(IntentBridgeError, match="authorization must be a JSON object"): sign_bridge(bad_auth, Ed25519PrivateKey.generate()) # type: ignore[arg-type] + +@pytest.mark.parametrize( + "bad_jwk", + [ + {"kty": "RSA", "kid": "wrong-kid"}, + {"kty": "OKP", "crv": "X25519", "kid": "wrong-kid"}, + {"kty": "OKP", "crv": "Ed25519", "kid": "wrong-kid"}, + {"kty": "OKP", "crv": "Ed25519", "x": "!!!", "kid": "wrong-kid"}, + {"kty": "OKP", "crv": "Ed25519", "x": "AA", "kid": "wrong-kid"}, + {"kty": "OKP", "crv": "Ed25519", "x": "AA", "d": "secret-private-key", "kid": "wrong-kid"}, + ], +) +def test_verify_bridge_rejects_unusable_trusted_authorizer_jwk_before_kid_mismatch( + bad_jwk: dict, +) -> None: + bridge, key, declaration, intent, args, tool_call, transcript = _fixture() + if "d" in bad_jwk: + bad_jwk = {**key_to_jwk(key), "d": "secret-private-key", "kid": "wrong-kid"} + with pytest.raises(IntentBridgeError, match="trusted_authorizer_jwk is unusable") as excinfo: + verify_bridge( + bridge, + bad_jwk, + declaration=declaration, + pic_intent_digest=intent, + pic_args_digest=args, + tool_call=tool_call, + transcript=transcript, + now=150, + ) + assert "secret-private-key" not in str(excinfo.value) + + +@pytest.mark.parametrize("bad_jwk", [None, "not-a-jwk", 123, ["kid"]]) +def test_verify_successor_artifact_rejects_non_object_trusted_observer_jwk( + bad_jwk: object, +) -> None: + observer_key = Ed25519PrivateKey.generate() + after = _after() + successor = sign_successor_artifact("auth-other", after, "observer-key-1", observer_key) + with pytest.raises( + IntentBridgeError, match="trusted_observer_jwk must be an object" + ) as excinfo: + verify_successor_artifact( + successor, + bad_jwk, # type: ignore[arg-type] + trusted_observer="observer-1", + authorization_id="auth-7", + after=after, + ) + assert not isinstance(excinfo.value, AuthorizationMismatch) + + +@pytest.mark.parametrize( + "bad_jwk", + [ + {"kty": "RSA", "kid": "wrong-kid"}, + {"kty": "OKP", "crv": "X25519", "kid": "wrong-kid"}, + {"kty": "OKP", "crv": "Ed25519", "kid": "wrong-kid"}, + {"kty": "OKP", "crv": "Ed25519", "x": "!!!", "kid": "wrong-kid"}, + {"kty": "OKP", "crv": "Ed25519", "x": "AA", "kid": "wrong-kid"}, + {"kty": "OKP", "crv": "Ed25519", "x": "AA", "d": "secret-private-key", "kid": "wrong-kid"}, + ], +) +def test_verify_successor_artifact_rejects_unusable_jwk_before_kid_or_link_mismatch( + bad_jwk: dict, +) -> None: + observer_key = Ed25519PrivateKey.generate() + after = _after() + successor = sign_successor_artifact("auth-other", after, "observer-key-1", observer_key) + if "d" in bad_jwk: + bad_jwk = {**key_to_jwk(observer_key), "d": "secret-private-key", "kid": "wrong-kid"} + with pytest.raises(IntentBridgeError, match="trusted_observer_jwk is unusable") as excinfo: + verify_successor_artifact( + successor, + bad_jwk, + trusted_observer="other-observer", + authorization_id="auth-7", + after=after, + ) + assert not isinstance(excinfo.value, AuthorizationMismatch) + assert "secret-private-key" not in str(excinfo.value) + + +@pytest.mark.parametrize( + ("param", "bad_value", "expected_match"), + [ + ("trusted_observer", "", "trusted_observer must be a non-empty string"), + ("trusted_observer", None, "trusted_observer must be a non-empty string"), + ("authorization_id", "", "authorization_id must be a non-empty string"), + ("authorization_id", None, "authorization_id must be a non-empty string"), + ], +) +def test_verify_successor_artifact_validates_caller_options_at_entry( + param: str, bad_value: object, expected_match: str +) -> None: + observer_key = Ed25519PrivateKey.generate() + after = _after() + successor = sign_successor_artifact("auth-other", after, "observer-key-1", observer_key) + wrong_profile = {**successor, "profile": "tag:example.com,2026:other"} + for candidate in (successor, wrong_profile): + kwargs: dict[str, object] = { + "trusted_observer": "observer-1", + "authorization_id": "auth-7", + "after": after, + } + kwargs[param] = bad_value + with pytest.raises(IntentBridgeError, match=expected_match) as excinfo: + verify_successor_artifact( + candidate, + {**key_to_jwk(observer_key), "kid": "observer-key-1"}, + **kwargs, # type: ignore[arg-type] + ) + assert not isinstance(excinfo.value, AuthorizationMismatch) + + +@pytest.mark.parametrize( + ("param", "bad_value", "expected_match"), + [ + ("now", -1, "now must be a non-negative integer"), + ("now", True, "now must be a non-negative integer"), + ("now", 150.0, "now must be a non-negative integer"), + ("declaration", "not-a-dict", "declaration must be an object"), + ( + "declaration", + {"impact": "external-side-effect", "bad": float("nan")}, + "no RFC 8785 canonical form", + ), + ("tool_call", 123, "tool_call must be an object"), + ( + "tool_call", + {"name": "send_invoice", "bad": float("nan")}, + "no RFC 8785 canonical form", + ), + ("pic_intent_digest", "not-a-digest", "intent_digest must be a sha256 digest"), + ( + "pic_args_digest", + "sha256:xyz", + "args_digest must contain 64 lowercase hexadecimal characters", + ), + ], +) +def test_verify_bridge_validates_caller_options_before_signature_and_denial( + param: str, bad_value: object, expected_match: str +) -> None: + bridge, key, declaration, intent, args, tool_call, transcript = _fixture() + trusted = {**key_to_jwk(key), "kid": "key-7"} + + denied_auth = copy.deepcopy(bridge["authorization"]) + denied_auth["decision"] = "deny" + denied_bridge = sign_bridge(denied_auth, key) + + bad_sig_bridge = copy.deepcopy(bridge) + head = bad_sig_bridge["signature"][0] + bad_sig_bridge["signature"] = ("B" if head == "A" else "A") + bad_sig_bridge["signature"][1:] + + for candidate_bridge in (denied_bridge, bad_sig_bridge): + kwargs: dict[str, object] = { + "declaration": declaration, + "pic_intent_digest": intent, + "pic_args_digest": args, + "tool_call": tool_call, + "transcript": transcript, + "now": 150, + } + kwargs[param] = bad_value + with pytest.raises(IntentBridgeError, match=expected_match) as excinfo: + verify_bridge( + candidate_bridge, + trusted, + **kwargs, # type: ignore[arg-type] + ) + assert not isinstance(excinfo.value, (AuthorizationDenied, AuthorizationMismatch)) + + +def test_bind_successor_observation_validates_expected_digest_before_envelope() -> None: + with pytest.raises(IntentBridgeError, match="expected_successor_digest") as excinfo: + _bind_successor_observation("not-an-envelope", "not-a-digest") + assert not isinstance(excinfo.value, AuthorizationMismatch) + + +@pytest.mark.parametrize("bad_kid", [None, "", 7, ["kid"]]) +def test_verify_bridge_rejects_bad_trusted_authorizer_kid_at_entry( + bad_kid: object, +) -> None: + bridge, key, declaration, intent, args, tool_call, transcript = _fixture() + trusted = key_to_jwk(key) if bad_kid is None else {**key_to_jwk(key), "kid": bad_kid} + for malformed_bridge in ({"profile": "wrong"}, "not-an-object"): + with pytest.raises( + IntentBridgeError, match="trusted_authorizer_jwk kid must be a non-empty string" + ): + verify_bridge( + malformed_bridge, # type: ignore[arg-type] + trusted, + declaration=declaration, + pic_intent_digest=intent, + pic_args_digest=args, + tool_call=tool_call, + transcript=transcript, + now=150, + ) + + +@pytest.mark.parametrize("bad_kid", [None, "", 7, ["kid"]]) +def test_verify_successor_artifact_rejects_bad_trusted_observer_kid_at_entry( + bad_kid: object, +) -> None: + observer_key = Ed25519PrivateKey.generate() + after = _after() + mismatched_successor = sign_successor_artifact( + "auth-other", after, "observer-key-1", observer_key + ) + trusted = ( + key_to_jwk(observer_key) + if bad_kid is None + else {**key_to_jwk(observer_key), "kid": bad_kid} + ) + for candidate in (mismatched_successor, {"profile": "wrong"}, "not-an-object"): + with pytest.raises( + IntentBridgeError, match="trusted_observer_jwk kid must be a non-empty string" + ) as excinfo: + verify_successor_artifact( + candidate, # type: ignore[arg-type] + trusted, + trusted_observer="observer-1", + authorization_id="auth-7", + after=after, + ) + assert not isinstance(excinfo.value, AuthorizationMismatch) + + +def test_evaluate_successor_observation_clones_600_deep_observation() -> None: + import json + from agentrust_trace.successor_observation import evaluate_successor_observation + + deep_obs = json.loads('{"k":' * 600 + '{"leaf": 1}' + "}" * 600) + after = _after(deep_obs, observer="observer-1", observed_at=160) + + def mutating_predicate(obs: dict) -> bool: + obs["mutated"] = True + return "k" in obs + + outcome = evaluate_successor_observation( + after, + expected_successor_digest=digest_jcs(after), + trusted_observers={"observer-1"}, + executor_id="executor-1", + independence_required=True, + now=170, + max_age_seconds=60, + predicate=mutating_predicate, + ) + assert outcome.status == "established" + assert "mutated" not in after["observation"] + + + diff --git a/tests/test_successor_observation.py b/tests/test_successor_observation.py index 0a1cfb3f..dd9963c3 100644 --- a/tests/test_successor_observation.py +++ b/tests/test_successor_observation.py @@ -1,5 +1,7 @@ from __future__ import annotations +import copy + import pytest from agentrust_trace.intent_bridge import digest_jcs @@ -202,3 +204,46 @@ def test_executor_identity_shape_is_validated() -> None: after = _after() with pytest.raises(SuccessorObservationError, match="executor_id"): _evaluate(after, executor_id="") + + +def test_predicate_receives_deep_copy_and_cannot_mutate_after_observation() -> None: + after = _after( + { + "commit": "abc123", + "reachable": True, + "nested": {"checks": ["ci", "lint"]}, + } + ) + snapshot = copy.deepcopy(after) + + def mutating_predicate(obs: dict) -> bool: + obs["reachable"] = False + obs["nested"]["checks"].append("tampered") + obs["injected"] = True + return True + + outcome = _evaluate(after, predicate=mutating_predicate) + assert outcome.status == "established" + assert after == snapshot + + +def test_predicate_cannot_mutate_nested_objects_inside_tuple_array() -> None: + after = _after( + { + "tuple_array": ({"secret": "original"}, (["inner"],)), + } + ) + + def mutating_predicate(obs: dict) -> bool: + assert isinstance(obs["tuple_array"], tuple) + assert isinstance(obs["tuple_array"][1], tuple) + obs["tuple_array"][0]["secret"] = "MUTATED" + obs["tuple_array"][1][0].append("MUTATED") + return True + + outcome = _evaluate(after, predicate=mutating_predicate) + assert outcome.status == "established" + assert after["observation"]["tuple_array"][0]["secret"] == "original" + assert after["observation"]["tuple_array"][1][0] == ["inner"] + +