diff --git a/docs/integration/pic-trace-bridge-v1.md b/docs/integration/pic-trace-bridge-v1.md index 25e03906..b20e90cf 100644 --- a/docs/integration/pic-trace-bridge-v1.md +++ b/docs/integration/pic-trace-bridge-v1.md @@ -34,9 +34,68 @@ objects. The verifier checks that the executed tool is in `scope.tools`, the declaration impact is in `scope.impacts`, and both bridge-specific digests match. If `transcript_required` is true, a complete `before` and `after` transcript is -required, and `before.tool_call` must equal the executed call. This binds the -authorization to the call and its execution evidence without claiming that -TRACE proves the real-world outcome of the call. +required. `before.tool_call` must equal the executed call, and `after` must be +the successor-observation envelope whose RFC 8785 / SHA-256 digest equals the +signed `authorization.successor_observation_digest`. Because that digest is +inside the signed authorization, a caller cannot substitute both a new +observation and a matching expected digest. This binds the authorization to the +exact call and exact successor envelope without claiming that TRACE proves the +real-world outcome of the call. + +### Successor-observation binding + +When `transcript_required` is true, `transcript.after` is the successor envelope and +has exactly three fields: + +~~~json +{ + "observation": {"application": "defined"}, + "observer": "observer-identity", + "observed_at": 1750000000 +} +~~~ + +The bridge identity relation is the SHA-256 digest of the RFC 8785 canonical bytes of +that complete envelope. The expected digest is carried in the signed +`authorization.successor_observation_digest`; it is not supplied independently by the +caller. The binding therefore covers the observation content, observer identity, and +observation timestamp together. Relabelling a genuine observation to a +different observer, retiming it, or altering its content changes the binding. + +A matching binding establishes **integrity**, not **sufficiency**. It does not by itself +establish that the requested transition occurred. A verifier evaluating a successor +claim separately applies its configured trust, freshness, and observation-source +policy and then an application- or profile-defined transition predicate. + +The successor-evaluation surface has three evidence outcomes: + +- `established`: trusted, bound successor evidence satisfies the transition predicate; +- `contradicted`: trusted, bound successor evidence contradicts the transition predicate; +- `not-established`: the available evidence is absent or insufficient to justify + either conclusion. + +Malformed successor artifacts and binding failures are refusals, not a fourth evidence +outcome. At the bridge layer, an absent `after` is a refusal when +`transcript_required` is true because the signed authorization explicitly requires the +successor binding. At the separate successor-evaluation surface, where an observation may +be absent before bridge verification is attempted, absence remains +`not-established` and never becomes a positive conclusion. + +Observation independence is policy, not a universal rule. Where verifier policy +requires an observer independent of the executing principal, executor-supplied +successor evidence alone is insufficient and yields `not-established`. Where the +applicable policy permits deterministic local evidence, the same-principal observation +is not rejected merely because observer and executor are equal. + +A successful successor binding permits the verifier to conclude only that it is +evaluating the exact bound observation envelope and, after policy evaluation, that the +envelope's source/freshness/trust properties are acceptable. A transition-level +positive conclusion additionally requires the defined predicate over the relevant +predecessor, action/execution, and successor evidence. A bound `after` object alone +does not prove a real-world outcome. + +The surface-local three-state result is an instance of the evidence discipline tracked +in #279; it does not introduce a repository-wide status enum. The reference implementation is `agentrust_trace.intent_bridge`; the versioned schema is `schema/pic-trace-bridge-v1.json`. diff --git a/schema/pic-trace-bridge-v1.json b/schema/pic-trace-bridge-v1.json index 03193b67..b5f9004c 100644 --- a/schema/pic-trace-bridge-v1.json +++ b/schema/pic-trace-bridge-v1.json @@ -2,32 +2,159 @@ "$schema": "https://json-schema.org/draft/2020-12/schema", "$id": "https://trace.agentrust-io.com/schema/pic-trace-bridge-v1.json", "title": "PIC/TRACE Bridge Authorization v1", - "type": "object", "additionalProperties": false, - "required": ["profile", "authorization", "signature"], + "type": "object", + "additionalProperties": false, + "required": [ + "profile", + "authorization", + "signature" + ], "properties": { - "profile": {"const": "tag:agentrust-io.com,2026:pic-trace-bridge-v1"}, - "signature": {"type": "string", "pattern": "^[A-Za-z0-9_-]{86}$"}, + "profile": { + "const": "tag:agentrust-io.com,2026:pic-trace-bridge-v1" + }, + "signature": { + "type": "string", + "pattern": "^[A-Za-z0-9_-]{86}$" + }, "authorization": { - "type": "object", "additionalProperties": false, - "required": ["authorization_id", "decision", "authorizer", "authorizer_key_id", "authorized_at", "expires_at", "scope", "pic", "declaration_digest", "tool_call_digest", "transcript_required"], + "type": "object", + "additionalProperties": false, + "required": [ + "authorization_id", + "decision", + "authorizer", + "authorizer_key_id", + "authorized_at", + "expires_at", + "scope", + "pic", + "declaration_digest", + "tool_call_digest", + "transcript_required" + ], "properties": { - "authorization_id": {"type": "string", "minLength": 1}, - "decision": {"enum": ["allow", "deny"]}, - "authorizer": {"type": "string", "minLength": 1}, - "authorizer_key_id": {"type": "string", "minLength": 1}, - "authorized_at": {"type": "integer", "minimum": 0, "maximum": 9007199254740991}, - "expires_at": {"type": "integer", "minimum": 0, "maximum": 9007199254740991}, - "scope": {"type": "object", "additionalProperties": false, "required": ["tools", "impacts"], "properties": { - "tools": {"type": "array", "minItems": 1, "uniqueItems": true, "items": {"type": "string", "minLength": 1}}, - "impacts": {"type": "array", "minItems": 1, "uniqueItems": true, "items": {"type": "string", "minLength": 1}} - }}, - "pic": {"type": "object", "additionalProperties": false, "required": ["profile", "intent_digest", "args_digest"], "properties": { - "profile": {"const": "PIC-CJSON/1.0"}, "intent_digest": {"$ref": "#/$defs/digest"}, "args_digest": {"$ref": "#/$defs/digest"} - }}, - "declaration_digest": {"$ref": "#/$defs/digest"}, "tool_call_digest": {"$ref": "#/$defs/digest"}, - "transcript_required": {"type": "boolean"} - } + "authorization_id": { + "type": "string", + "minLength": 1 + }, + "decision": { + "enum": [ + "allow", + "deny" + ] + }, + "authorizer": { + "type": "string", + "minLength": 1 + }, + "authorizer_key_id": { + "type": "string", + "minLength": 1 + }, + "authorized_at": { + "type": "integer", + "minimum": 0, + "maximum": 9007199254740991 + }, + "expires_at": { + "type": "integer", + "minimum": 0, + "maximum": 9007199254740991 + }, + "scope": { + "type": "object", + "additionalProperties": false, + "required": [ + "tools", + "impacts" + ], + "properties": { + "tools": { + "type": "array", + "minItems": 1, + "uniqueItems": true, + "items": { + "type": "string", + "minLength": 1 + } + }, + "impacts": { + "type": "array", + "minItems": 1, + "uniqueItems": true, + "items": { + "type": "string", + "minLength": 1 + } + } + } + }, + "pic": { + "type": "object", + "additionalProperties": false, + "required": [ + "profile", + "intent_digest", + "args_digest" + ], + "properties": { + "profile": { + "const": "PIC-CJSON/1.0" + }, + "intent_digest": { + "$ref": "#/$defs/digest" + }, + "args_digest": { + "$ref": "#/$defs/digest" + } + } + }, + "declaration_digest": { + "$ref": "#/$defs/digest" + }, + "tool_call_digest": { + "$ref": "#/$defs/digest" + }, + "successor_observation_digest": { + "$ref": "#/$defs/digest" + }, + "transcript_required": { + "type": "boolean" + } + }, + "allOf": [ + { + "if": { + "properties": { + "transcript_required": { + "const": true + } + }, + "required": [ + "transcript_required" + ] + }, + "then": { + "required": [ + "successor_observation_digest" + ] + }, + "else": { + "not": { + "required": [ + "successor_observation_digest" + ] + } + } + } + ] } }, - "$defs": {"digest": {"type": "string", "pattern": "^sha256:[0-9a-f]{64}$"}} + "$defs": { + "digest": { + "type": "string", + "pattern": "^sha256:[0-9a-f]{64}$" + } + } } diff --git a/src/agentrust_trace/intent_bridge.py b/src/agentrust_trace/intent_bridge.py index 2d91c91e..fe428bb5 100644 --- a/src/agentrust_trace/intent_bridge.py +++ b/src/agentrust_trace/intent_bridge.py @@ -11,7 +11,12 @@ import rfc8785 from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PrivateKey -from agentrust_trace.sign import _b64url_decode, _canonical_bytes, _pubkey_from_jwk +from agentrust_trace.sign import ( + JCS_SAFE_INTEGER, + _b64url_decode, + _canonical_bytes, + _pubkey_from_jwk, +) BRIDGE_PROFILE = "tag:agentrust-io.com,2026:pic-trace-bridge-v1" PIC_PROFILE = "PIC-CJSON/1.0" @@ -105,6 +110,55 @@ def _nonempty_string(value: Any, field: str) -> str: return value +def _bind_successor_observation( + after: Any, expected_successor_digest: str +) -> dict[str, Any]: + """Bind the exact successor envelope using the bridge identity relation. + + The digest covers observation content, observer identity, and observation time. + This establishes integrity only. Trust, freshness, independence, and predicate + sufficiency are deliberately evaluated separately. + """ + if not isinstance(after, dict): + raise AuthorizationMismatch("transcript.after must be a successor observation object") + required = {"observation", "observer", "observed_at"} + missing = required - set(after) + unknown = set(after) - required + if missing: + raise AuthorizationMismatch( + f"transcript.after is missing successor fields: {sorted(missing)}" + ) + if unknown: + raise AuthorizationMismatch( + f"transcript.after contains unknown successor fields: {sorted(unknown)}" + ) + if not isinstance(after["observation"], dict): + raise AuthorizationMismatch("transcript.after.observation must be an object") + _nonempty_string(after["observer"], "transcript.after.observer") + observed_at = after["observed_at"] + if ( + not isinstance(observed_at, int) + or isinstance(observed_at, bool) + or observed_at < 0 + or observed_at > JCS_SAFE_INTEGER + ): + raise IntentBridgeError( + "transcript.after.observed_at must be a non-negative integer within " + "the JCS safe-integer range" + ) + expected = _digest(expected_successor_digest, "expected_successor_digest") + try: + actual = digest_jcs(after) + except IntentBridgeError: + raise AuthorizationMismatch( + "transcript.after has no RFC 8785 canonical form" + ) from None + if not compare_digest(expected, actual): + raise AuthorizationMismatch( + "transcript.after does not match the expected digest binding" + ) + return after + def _decision(value: Any) -> str: """Return a valid authorization decision or refuse a malformed value.""" @@ -158,10 +212,11 @@ def verify_bridge( fields = { "authorization_id", "decision", "authorizer", "authorizer_key_id", "authorized_at", "expires_at", "scope", "pic", "declaration_digest", - "tool_call_digest", "transcript_required", + "tool_call_digest", "successor_observation_digest", "transcript_required", } + required_fields = fields - {"successor_observation_digest"} authorization = _object(root.get("authorization"), "authorization", fields) - missing = fields - set(authorization) + missing = required_fields - set(authorization) if missing: raise IntentBridgeError(f"authorization is missing fields: {sorted(missing)}") for field in ("authorization_id", "authorizer", "authorizer_key_id"): @@ -236,6 +291,17 @@ def verify_bridge( if not isinstance(authorization["transcript_required"], bool): raise IntentBridgeError("transcript_required must be boolean") + successor_digest_present = "successor_observation_digest" in authorization + if authorization["transcript_required"] and not successor_digest_present: + raise IntentBridgeError( + "authorization.successor_observation_digest is required when " + "transcript_required is true" + ) + if not authorization["transcript_required"] and successor_digest_present: + raise IntentBridgeError( + "authorization.successor_observation_digest must be absent when " + "transcript_required is false" + ) if authorization["transcript_required"]: if not isinstance(transcript, dict) or set(transcript) != {"before", "after"}: raise AuthorizationMismatch("a full before/after transcript is required") @@ -255,6 +321,10 @@ def verify_bridge( ) from None if not compare_digest(before_digest, tool_call_digest): raise AuthorizationMismatch("transcript.before.tool_call does not match execution") - if not isinstance(transcript.get("after"), dict): - raise AuthorizationMismatch("transcript.after must contain the execution result") + after = transcript.get("after") + expected_successor_digest = _digest( + authorization["successor_observation_digest"], + "authorization.successor_observation_digest", + ) + _bind_successor_observation(after, expected_successor_digest) return authorization diff --git a/src/agentrust_trace/successor_observation.py b/src/agentrust_trace/successor_observation.py new file mode 100644 index 00000000..6be0a36f --- /dev/null +++ b/src/agentrust_trace/successor_observation.py @@ -0,0 +1,141 @@ +"""Evaluate bound PIC/TRACE successor observations without conflating integrity and closure. + +The bridge-local binding establishes the exact successor envelope. This module applies +trust, freshness, independence policy, and an application-defined transition predicate +to produce the surface-local three-state evidence result defined by #338. +""" + +from __future__ import annotations + +from collections.abc import Callable, Collection +from dataclasses import dataclass +from typing import Any, Literal + +from agentrust_trace.intent_bridge import ( + AuthorizationMismatch, + IntentBridgeError, + _bind_successor_observation, +) +from agentrust_trace.sign import JCS_SAFE_INTEGER + +SuccessorStatus = Literal["established", "contradicted", "not-established"] + + +class SuccessorObservationError(ValueError): + """The successor artifact or its binding is malformed or inconsistent.""" + + +@dataclass(frozen=True) +class SuccessorOutcome: + """What the verifier may conclude from a bound successor observation.""" + + status: SuccessorStatus + reason: str + + +def _digest(value: Any, field: str) -> str: + if not isinstance(value, str) or not value.startswith("sha256:"): + raise SuccessorObservationError(f"{field} must be a sha256 digest") + tail = value[7:] + if len(tail) != 64 or any(c not in "0123456789abcdef" for c in tail): + raise SuccessorObservationError( + f"{field} must contain 64 lowercase hexadecimal characters" + ) + return value + + +def _not_established(reason: str) -> SuccessorOutcome: + return SuccessorOutcome("not-established", reason) + + +def evaluate_successor_observation( + after: dict[str, Any] | None, + *, + expected_successor_digest: str, + trusted_observers: Collection[str], + executor_id: str | None, + independence_required: bool, + now: int, + max_age_seconds: int, + predicate: Callable[[dict[str, Any]], bool | None], +) -> SuccessorOutcome: + """Evaluate a successor observation without conflating binding with closure. + + `expected_successor_digest` is the authenticated binding value established by + the bridge profile. In the PIC/TRACE bridge it comes from the signed + `authorization.successor_observation_digest`, not from the observed envelope. + + The predicate is application-defined and returns True when the requested + transition is established by the observation, False when trusted evidence + contradicts it, and None when the observation itself does not decide it. + """ + + expected = _digest(expected_successor_digest, "expected_successor_digest") + if not isinstance(independence_required, bool): + raise SuccessorObservationError("independence_required must be boolean") + if ( + not isinstance(trusted_observers, Collection) + or isinstance(trusted_observers, (str, bytes, bytearray, dict)) + or not all(isinstance(item, str) and item for item in trusted_observers) + ): + raise SuccessorObservationError( + "trusted_observers must be a collection of non-empty observer identity strings" + ) + if executor_id is not None and (not isinstance(executor_id, str) or not executor_id): + raise SuccessorObservationError("executor_id must be a non-empty string or None") + if not callable(predicate): + raise SuccessorObservationError("predicate must be callable") + for field, value in (("now", now), ("max_age_seconds", max_age_seconds)): + if ( + not isinstance(value, int) + or isinstance(value, bool) + or value < 0 + or value > JCS_SAFE_INTEGER + ): + raise SuccessorObservationError( + f"{field} must be a non-negative integer within the JCS safe-integer range" + ) + + if after is None: + return _not_established("successor observation is absent") + + try: + bound = _bind_successor_observation(after, expected) + except (IntentBridgeError, AuthorizationMismatch) as exc: + raise SuccessorObservationError(str(exc)) from exc + + observation = bound["observation"] + observer = bound["observer"] + observed_at = bound["observed_at"] + + if observer not in trusted_observers: + return _not_established("successor observer is not trusted by verifier policy") + if observed_at > now: + return _not_established("successor observation is dated in the future") + if now - observed_at > max_age_seconds: + return _not_established("successor observation is stale") + + if independence_required: + if executor_id is None: + return _not_established( + "independent observation is required but executor identity is unavailable" + ) + if observer == executor_id: + return _not_established( + "independent observation is required but observer is the executor" + ) + + result = predicate(observation) + if result is True: + return SuccessorOutcome( + "established", + "trusted bound successor evidence satisfies the transition predicate", + ) + if result is False: + return SuccessorOutcome( + "contradicted", + "trusted bound successor evidence contradicts the transition predicate", + ) + if result is None: + return _not_established("transition predicate cannot decide from this observation") + raise SuccessorObservationError("transition predicate must return True, False, or None") diff --git a/tests/test_intent_bridge.py b/tests/test_intent_bridge.py index 3f6b62f8..ac0c5c2b 100644 --- a/tests/test_intent_bridge.py +++ b/tests/test_intent_bridge.py @@ -24,6 +24,11 @@ def _fixture( key = Ed25519PrivateKey.generate() declaration = declaration or {"impact": "external-side-effect", "purpose": "send invoice"} tool_call = tool_call or {"name": "send_invoice", "arguments": {"invoice_id": "INV-7"}} + after = { + "observation": {"status": "accepted"}, + "observer": "observer-1", + "observed_at": 150, + } authorization = { "authorization_id": "auth-7", "decision": "allow", @@ -39,10 +44,11 @@ def _fixture( }, "declaration_digest": digest_jcs(declaration), "tool_call_digest": digest_jcs(tool_call), + "successor_observation_digest": digest_jcs(after), "transcript_required": True, } bridge = sign_bridge(authorization, key) - transcript = {"before": {"tool_call": tool_call}, "after": {"status": "accepted"}} + transcript = {"before": {"tool_call": tool_call}, "after": after} return ( bridge, key, declaration, authorization["pic"]["intent_digest"], authorization["pic"]["args_digest"], tool_call, transcript, @@ -59,6 +65,49 @@ def test_verify_bridge_accepts_authorized_bound_execution() -> None: assert result["authorization_id"] == "auth-7" +def test_successor_observation_is_bound_to_signed_authorization() -> None: + bridge, key, declaration, intent, args, tool_call, transcript = _fixture() + substituted = copy.deepcopy(transcript) + substituted["after"]["observation"]["status"] = "different" + with pytest.raises(AuthorizationMismatch, match="expected digest binding"): + verify_bridge( + bridge, {**key_to_jwk(key), "kid": "key-7"}, declaration=declaration, + pic_intent_digest=intent, pic_args_digest=args, tool_call=tool_call, + transcript=substituted, now=150, + ) + + +def test_successor_envelope_rejects_unknown_fields_even_when_signed() -> None: + bridge, key, declaration, intent, args, tool_call, transcript = _fixture() + extended = copy.deepcopy(transcript) + extended["after"]["extra"] = "signed-but-not-part-of-profile" + + authorization = copy.deepcopy(bridge["authorization"]) + authorization["successor_observation_digest"] = digest_jcs(extended["after"]) + resigned = sign_bridge(authorization, key) + + with pytest.raises(AuthorizationMismatch, match="unknown successor fields"): + verify_bridge( + resigned, {**key_to_jwk(key), "kid": "key-7"}, declaration=declaration, + pic_intent_digest=intent, pic_args_digest=args, tool_call=tool_call, + transcript=extended, now=150, + ) + + +def test_caller_cannot_substitute_successor_and_matching_digest_without_resigning() -> None: + bridge, key, declaration, intent, args, tool_call, transcript = _fixture() + tampered = copy.deepcopy(bridge) + substituted = copy.deepcopy(transcript) + substituted["after"]["observation"]["status"] = "different" + tampered["authorization"]["successor_observation_digest"] = digest_jcs(substituted["after"]) + with pytest.raises(IntentBridgeError, match="authorization signature is invalid"): + verify_bridge( + tampered, {**key_to_jwk(key), "kid": "key-7"}, declaration=declaration, + pic_intent_digest=intent, pic_args_digest=args, tool_call=tool_call, + transcript=substituted, now=150, + ) + + @pytest.mark.parametrize("field", ["authorization", "signature"]) def test_tampering_is_rejected(field: str) -> None: bridge, key, declaration, intent, args, tool_call, transcript = _fixture() @@ -202,6 +251,41 @@ def test_a_declaration_with_no_impact_at_all_is_refused() -> None: _verify(*_fixture(declaration={"purpose": "send invoice"})) +def test_successor_digest_is_required_iff_transcript_is_required() -> None: + bridge, key, declaration, intent, args, tool_call, transcript = _fixture() + + missing = copy.deepcopy(bridge["authorization"]) + del missing["successor_observation_digest"] + missing_bridge = sign_bridge(missing, key) + with pytest.raises(IntentBridgeError, match="successor_observation_digest is required"): + verify_bridge( + missing_bridge, {**key_to_jwk(key), "kid": "key-7"}, + declaration=declaration, pic_intent_digest=intent, pic_args_digest=args, + tool_call=tool_call, transcript=transcript, now=150, + ) + + no_transcript = copy.deepcopy(bridge["authorization"]) + no_transcript["transcript_required"] = False + del no_transcript["successor_observation_digest"] + no_transcript_bridge = sign_bridge(no_transcript, key) + result = verify_bridge( + no_transcript_bridge, {**key_to_jwk(key), "kid": "key-7"}, + declaration=declaration, pic_intent_digest=intent, pic_args_digest=args, + tool_call=tool_call, transcript=None, now=150, + ) + assert result["transcript_required"] is False + + contradictory = copy.deepcopy(no_transcript) + contradictory["successor_observation_digest"] = digest_jcs(transcript["after"]) + contradictory_bridge = sign_bridge(contradictory, key) + with pytest.raises(IntentBridgeError, match="must be absent"): + verify_bridge( + contradictory_bridge, {**key_to_jwk(key), "kid": "key-7"}, + declaration=declaration, pic_intent_digest=intent, pic_args_digest=args, + tool_call=tool_call, transcript=None, now=150, + ) + + def test_expiry_and_required_transcript_are_enforced() -> None: bridge, key, declaration, intent, args, tool_call, transcript = _fixture() with pytest.raises(IntentBridgeError, match="expired"): @@ -317,7 +401,7 @@ def test_transcript_call_is_compared_over_canonical_bytes_not_python_equality( ) -> None: """#317: `True == 1` in Python, so `!=` accepted a JSON-distinct transcript call.""" call = {"name": "send_invoice", "arguments": executed} - bridge, key, declaration, intent, args, tool_call, _ = _fixture(tool_call=call) + bridge, key, declaration, intent, args, tool_call, transcript = _fixture(tool_call=call) substituted = {"name": "send_invoice", "arguments": transcribed} assert substituted == tool_call, "the substitution must be Python-equal to be a regression" assert digest_jcs(substituted) != digest_jcs(tool_call) @@ -325,7 +409,7 @@ def test_transcript_call_is_compared_over_canonical_bytes_not_python_equality( verify_bridge( bridge, {**key_to_jwk(key), "kid": "key-7"}, declaration=declaration, pic_intent_digest=intent, pic_args_digest=args, tool_call=tool_call, - transcript={"before": {"tool_call": substituted}, "after": {"status": "accepted"}}, + transcript={"before": {"tool_call": substituted}, "after": transcript["after"]}, now=150, ) @@ -347,11 +431,12 @@ def test_transcript_call_that_is_not_an_object_stays_an_authorization_mismatch( bad: object, ) -> None: """Comparing digests must not turn a malformed transcript into a different class.""" - bridge, key, declaration, intent, args, tool_call, _ = _fixture() + bridge, key, declaration, intent, args, tool_call, transcript = _fixture() + transcript_after = transcript["after"] with pytest.raises(AuthorizationMismatch, match="transcript.before.tool_call"): verify_bridge( bridge, {**key_to_jwk(key), "kid": "key-7"}, declaration=declaration, pic_intent_digest=intent, pic_args_digest=args, tool_call=tool_call, - transcript={"before": {"tool_call": bad}, "after": {"status": "accepted"}}, + transcript={"before": {"tool_call": bad}, "after": transcript_after}, now=150, ) diff --git a/tests/test_public_functions_raise_what_they_document.py b/tests/test_public_functions_raise_what_they_document.py index 3fdbaea7..fd94f278 100644 --- a/tests/test_public_functions_raise_what_they_document.py +++ b/tests/test_public_functions_raise_what_they_document.py @@ -53,7 +53,7 @@ import agentrust_trace as at from agentrust_trace import (content_marking, generate_key, intent_bridge, key_to_jwk, - provenance, revocation, sign, validate) + provenance, revocation, sign, successor_observation, validate) #: Values a caller can supply where an object, a string, a key or bytes is expected. #: The last five are the ones that separate a strict canonicalizer from a permissive @@ -72,6 +72,7 @@ "provenance": ("ProvenanceError", "ToolCatalogMismatch"), "revocation": ("ValueError", "UnanchorableValue"), "sign": ("ValueError", "UnanchorableValue", "InvalidSignature"), + "successor_observation": ("SuccessorObservationError",), "validate": ("ValueError", "ValidationError"), "models": ("ValidationError",), "adapters": ("ValueError", "ValidationError"), @@ -109,6 +110,18 @@ "provenance.sign_record": lambda v: provenance.sign_record(v, _KEY), "provenance.tool_catalog_hash": provenance.tool_catalog_hash, "provenance.verify_record": lambda v: provenance.verify_record(v, _JWK), + "successor_observation.evaluate_successor_observation": lambda v: ( + successor_observation.evaluate_successor_observation( + v, + expected_successor_digest="sha256:" + "0" * 64, + trusted_observers={"observer-1"}, + executor_id="executor-1", + independence_required=True, + now=160, + max_age_seconds=30, + predicate=lambda obs: obs.get("reachable") is True, + ) + ), "revocation.bundle_digest": revocation.bundle_digest, "revocation.check_bundle": lambda v: revocation.check_bundle( v, trusted_key_identifiers=[], trusted_bundle_keys=[_JWK], now=1785000000, @@ -139,6 +152,12 @@ _ASSERTION = content_marking.build_assertion(_RECORD_JSON, url="https://r.example/r.json") _DECLARATION = {"impact": "external-side-effect", "purpose": "send invoice"} _TOOL_CALL = {"name": "send_invoice", "arguments": {"approved": 1}} +_SUCCESSOR_AFTER = { + "observation": {"commit": "abc123", "reachable": True}, + "observer": "observer-1", + "observed_at": 150, +} +_SUCCESSOR_DIGEST = intent_bridge.digest_jcs(_SUCCESSOR_AFTER) _BRIDGE_AUTH = { "authorization_id": "auth-1", "decision": "allow", "authorizer": "finance-policy", "authorizer_key_id": "key-1", "authorized_at": 100, "expires_at": 200, @@ -147,10 +166,11 @@ "args_digest": "sha256:" + "2" * 64}, "declaration_digest": intent_bridge.digest_jcs(_DECLARATION), "tool_call_digest": intent_bridge.digest_jcs(_TOOL_CALL), + "successor_observation_digest": _SUCCESSOR_DIGEST, "transcript_required": True, } _BRIDGE = intent_bridge.sign_bridge(_BRIDGE_AUTH, _KEY) -_TRANSCRIPT = {"before": {"tool_call": dict(_TOOL_CALL)}, "after": {"status": "accepted"}} +_TRANSCRIPT = {"before": {"tool_call": dict(_TOOL_CALL)}, "after": _SUCCESSOR_AFTER} _TOOLS = [{"name": "search", "description": "search", "input_schema": {"type": "object"}}] _ARTIFACT = {"package": "pkg:npm/%40acme/mcp-search@2.1.0", "digest": "sha256:" + "0" * 64} _PROVENANCE = provenance.build_record( @@ -189,6 +209,20 @@ ("bridge", "trusted_authorizer_jwk", "declaration", "pic_intent_digest", "pic_args_digest", "tool_call", "transcript", "now"), ), + "successor_observation.evaluate_successor_observation": ( + lambda: { + "after": _SUCCESSOR_AFTER, + "expected_successor_digest": _SUCCESSOR_DIGEST, + "trusted_observers": {"observer-1"}, + "executor_id": "executor-1", + "independence_required": True, + "now": 160, + "max_age_seconds": 30, + "predicate": lambda obs: obs.get("reachable") is True, + }, + ("expected_successor_digest", "trusted_observers", "executor_id", + "independence_required", "now", "max_age_seconds", "predicate"), + ), "provenance.build_record": ( lambda: {"kind": "publisher-asserted", "publisher": "did:web:acme.example", "tools": _TOOLS, "artifact": _ARTIFACT, "endpoint": None, "attestation": None, @@ -286,6 +320,7 @@ def test_every_public_function_is_either_swept_or_declared_unsweepable() -> None def _module_of(name: str) -> Any: return {"content_marking": content_marking, "intent_bridge": intent_bridge, "provenance": provenance, "revocation": revocation, "sign": sign, + "successor_observation": successor_observation, "validate": validate}[name.split(".")[0]] @@ -359,6 +394,8 @@ def test_no_keyword_argument_leaks_an_undocumented_exception(name: str, param: s "content_marking.verify_assertion": ("record_bytes", None, "ContentMarkingError"), "intent_bridge.sign_bridge": ("key", None, "IntentBridgeError"), "intent_bridge.verify_bridge": ("now", "a-string", "IntentBridgeError"), + "successor_observation.evaluate_successor_observation": + ("expected_successor_digest", None, "SuccessorObservationError"), "provenance.build_record": ("publisher", 123, "ProvenanceError"), "provenance.check_tool_catalog": ("tools", None, "ProvenanceError"), "provenance.sign_record": ("key", None, "ProvenanceError"), @@ -567,6 +604,8 @@ def test_no_public_function_raises_an_undocumented_exception(name: str) -> None: "provenance.sign_record": (None, "ProvenanceError"), "provenance.tool_catalog_hash": (None, "ProvenanceError"), "provenance.verify_record": (None, "ProvenanceError"), + "successor_observation.evaluate_successor_observation": + ("a-string", "SuccessorObservationError"), "revocation.bundle_digest": (None, "ValueError"), "sign.anchor_bytes": (b"bytes", "UnanchorableValue"), "sign.jwk_thumbprint": (None, "ValueError"), diff --git a/tests/test_safe_integer_range.py b/tests/test_safe_integer_range.py index aeb69907..074bb32f 100644 --- a/tests/test_safe_integer_range.py +++ b/tests/test_safe_integer_range.py @@ -436,6 +436,7 @@ def _positions(node: Any, prefix: tuple[Any, ...] = ()) -> list[tuple[tuple[Any, }, "declaration_digest": "sha256:" + "c" * 64, "tool_call_digest": "sha256:" + "d" * 64, + "successor_observation_digest": "sha256:" + "e" * 64, "transcript_required": True, }, "signature": "x" * 86, diff --git a/tests/test_successor_observation.py b/tests/test_successor_observation.py new file mode 100644 index 00000000..0a1cfb3f --- /dev/null +++ b/tests/test_successor_observation.py @@ -0,0 +1,204 @@ +from __future__ import annotations + +import pytest + +from agentrust_trace.intent_bridge import digest_jcs +from agentrust_trace.successor_observation import ( + SuccessorObservationError, + evaluate_successor_observation, +) + + +def _after( + observation: dict | None = None, + *, + observer: str = "observer-1", + observed_at: int = 150, +) -> dict: + return { + "observation": ( + observation + if observation is not None + else {"commit": "abc123", "reachable": True} + ), + "observer": observer, + "observed_at": observed_at, + } + + +def _evaluate( + after: dict | None, + *, + expected: str | None = None, + trusted: set[str] | None = None, + executor_id: str | None = "executor-1", + independence_required: bool = True, + now: int = 160, + max_age_seconds: int = 30, + predicate=lambda obs: obs.get("reachable") is True, +): + if expected is None: + expected = digest_jcs(after) if after is not None else "sha256:" + "0" * 64 + return evaluate_successor_observation( + after, + expected_successor_digest=expected, + trusted_observers=trusted or {"observer-1"}, + executor_id=executor_id, + independence_required=independence_required, + now=now, + max_age_seconds=max_age_seconds, + predicate=predicate, + ) + + +def test_trusted_bound_successor_can_establish_transition() -> None: + outcome = _evaluate(_after()) + assert outcome.status == "established" + + +def test_trusted_bound_successor_can_contradict_transition() -> None: + after = _after({"commit": "abc123", "reachable": False}) + outcome = _evaluate(after, predicate=lambda obs: obs.get("reachable") is True) + assert outcome.status == "contradicted" + + +def test_missing_successor_is_not_established() -> None: + outcome = _evaluate(None) + assert outcome.status == "not-established" + assert "absent" in outcome.reason + + +def test_substituted_successor_fails_the_binding() -> None: + original = _after() + expected = digest_jcs(original) + substituted = _after({"commit": "def456", "reachable": True}) + with pytest.raises(SuccessorObservationError, match="expected digest binding"): + _evaluate(substituted, expected=expected) + + +def test_observer_metadata_is_inside_the_binding() -> None: + original = _after(observer="observer-1") + expected = digest_jcs(original) + relabelled = _after(observer="trusted-observer") + with pytest.raises(SuccessorObservationError, match="expected digest binding"): + _evaluate(relabelled, expected=expected, trusted={"trusted-observer"}) + + +def test_observation_time_is_inside_the_binding() -> None: + original = _after(observed_at=150) + expected = digest_jcs(original) + retimed = _after(observed_at=159) + with pytest.raises(SuccessorObservationError, match="expected digest binding"): + _evaluate(retimed, expected=expected) + + +@pytest.mark.parametrize( + "observation", + [ + {"value": 2**60}, + {"value": float("nan")}, + ], +) +def test_uncanonicalizable_successor_is_malformed(observation: dict) -> None: + after = _after(observation) + with pytest.raises(SuccessorObservationError, match="canonical form"): + _evaluate(after, expected="sha256:" + "0" * 64) + + +def test_stale_successor_is_not_established() -> None: + outcome = _evaluate(_after(observed_at=100), now=160, max_age_seconds=30) + assert outcome.status == "not-established" + assert "stale" in outcome.reason + + + +def test_future_successor_is_not_established() -> None: + outcome = _evaluate(_after(observed_at=161), now=160, max_age_seconds=30) + assert outcome.status == "not-established" + assert "future" in outcome.reason + + +def test_zero_age_accepts_only_same_instant_observation() -> None: + same = _evaluate(_after(observed_at=160), now=160, max_age_seconds=0) + assert same.status == "established" + + older = _evaluate(_after(observed_at=159), now=160, max_age_seconds=0) + assert older.status == "not-established" + assert "stale" in older.reason + + +def test_empty_trusted_observer_set_is_valid_but_establishes_nothing() -> None: + after = _after() + outcome = evaluate_successor_observation( + after, + expected_successor_digest=digest_jcs(after), + trusted_observers=set(), + executor_id="executor-1", + independence_required=True, + now=160, + max_age_seconds=30, + predicate=lambda obs: True, + ) + assert outcome.status == "not-established" + assert "not trusted" in outcome.reason + + +def test_untrusted_successor_is_not_established() -> None: + outcome = _evaluate(_after(observer="observer-2"), trusted={"observer-1"}) + assert outcome.status == "not-established" + assert "not trusted" in outcome.reason + + +def test_self_observation_does_not_close_when_independence_is_required() -> None: + after = _after(observer="executor-1") + outcome = _evaluate(after, trusted={"executor-1"}, independence_required=True) + assert outcome.status == "not-established" + assert "observer is the executor" in outcome.reason + + +def test_self_observation_can_close_when_policy_does_not_require_independence() -> None: + after = _after(observer="executor-1") + outcome = _evaluate(after, trusted={"executor-1"}, independence_required=False) + assert outcome.status == "established" + + +def test_missing_executor_identity_blocks_required_independence() -> None: + outcome = _evaluate(_after(), executor_id=None, independence_required=True) + assert outcome.status == "not-established" + assert "executor identity is unavailable" in outcome.reason + + +def test_indeterminate_predicate_is_not_established() -> None: + outcome = _evaluate(_after(), predicate=lambda obs: None) + assert outcome.status == "not-established" + assert "cannot decide" in outcome.reason + + +def test_predicate_must_return_three_state_value() -> None: + with pytest.raises(SuccessorObservationError, match="True, False, or None"): + _evaluate(_after(), predicate=lambda obs: "yes") + + +@pytest.mark.parametrize( + "trusted", + [None, "observer-1", b"observer-1", {"observer-1": True}, {"", "observer-1"}], +) +def test_trusted_observers_must_be_an_actual_identity_collection(trusted) -> None: + after = _after() + with pytest.raises(SuccessorObservationError, match="trusted_observers"): + evaluate_successor_observation( + after, + expected_successor_digest=digest_jcs(after), + trusted_observers=trusted, + executor_id="executor-1", + independence_required=True, + now=160, + max_age_seconds=30, + predicate=lambda obs: True, + ) + + +def test_executor_identity_shape_is_validated() -> None: + after = _after() + with pytest.raises(SuccessorObservationError, match="executor_id"): + _evaluate(after, executor_id="")