From 47216c20b60bc5cdcd89f41c9307493266e4b707 Mon Sep 17 00:00:00 2001 From: Raghuvar Karthik Durga <177662415+drk-ram@users.noreply.github.com> Date: Fri, 6 Nov 2026 10:18:42 -0600 Subject: [PATCH] fix(trace): reject boolean timestamps Signed-off-by: Raghuvar Karthik Durga <177662415+drk-ram@users.noreply.github.com> --- src/agentrust_telemetry/trace_adapter.py | 51 ++++++++++++++--- tests/test_trace_adapter.py | 70 ++++++++++++++++++++++++ 2 files changed, 112 insertions(+), 9 deletions(-) diff --git a/src/agentrust_telemetry/trace_adapter.py b/src/agentrust_telemetry/trace_adapter.py index 56932a9..f0fc1b3 100644 --- a/src/agentrust_telemetry/trace_adapter.py +++ b/src/agentrust_telemetry/trace_adapter.py @@ -4,7 +4,7 @@ import hashlib from dataclasses import dataclass -from typing import Any, Literal +from typing import Any, Literal, cast import rfc8785 @@ -50,7 +50,9 @@ def finalize_trace( data_class = _highest_data_class(events, config) tool_transcript = _tool_transcript(snapshot) appraisal_status = _appraisal(events) - issued_at = max(int(event["time_unix_nano"]) for event in events) // 1_000_000_000 + issued_at = max( + _unix_nano_timestamp(event["time_unix_nano"]) for event in events + ) // 1_000_000_000 except TraceFinalizationError: raise except (KeyError, TypeError, ValueError, AttributeError) as exc: @@ -260,6 +262,21 @@ def _appraisal(events: list[dict[str, Any]]) -> str: ActionBinding = Literal["bound", "unresolved", "contradicted"] +def _unix_nano_timestamp(value: Any) -> int: + if not isinstance(value, str): + raise TypeError("Unix nanosecond timestamps must be decimal strings") + if ( + not value.isascii() + or not value.isdecimal() + or len(value) > 20 + or (len(value) > 1 and value.startswith("0")) + ): + raise ValueError( + "Unix nanosecond timestamps must be canonical non-negative decimals" + ) + return int(value) + + def _action_approval_binding( events: list[dict[str, Any]], decisions: dict[str, Any], @@ -289,7 +306,13 @@ def _action_approval_binding( continue approval_id = action.get("approval_id") policy_event_id = action.get("policy_event_id") - if approval_id is None and decisions.get(policy_event_id) != "challenge": + if ( + approval_id is None + and ( + policy_event_id is None + or decisions.get(policy_event_id) != "challenge" + ) + ): continue candidates = [ event @@ -300,10 +323,18 @@ def _action_approval_binding( else event.get("policy_event_id") == policy_event_id ) ] - completed_at = int(action["time_unix_nano"]) + try: + completed_at = _unix_nano_timestamp(action["time_unix_nano"]) + except (TypeError, ValueError): + result = "unresolved" + continue contradicted = satisfied = False for approval in candidates: approval_policy = approval.get("policy_event_id") + try: + approved_at = _unix_nano_timestamp(approval["time_unix_nano"]) + except (TypeError, ValueError): + continue if ( not _same_digest(approval["action_digest"], action["action_digest"]) or ( @@ -311,7 +342,7 @@ def _action_approval_binding( and approval_policy is not None and approval_policy != policy_event_id ) - or int(approval["time_unix_nano"]) > completed_at + or approved_at > completed_at ): contradicted = True elif id(approval) in bound_ids: @@ -362,9 +393,11 @@ def _bound_approvals(approvals: list[dict[str, Any]]) -> list[dict[str, Any]]: ): continue try: - requested_at = int(request["requested_at_unix_nano"]) - approved_at = int(approval["time_unix_nano"]) - expires_at = int(request["expires_at_unix_nano"]) + requested_at = _unix_nano_timestamp( + request["requested_at_unix_nano"] + ) + approved_at = _unix_nano_timestamp(approval["time_unix_nano"]) + expires_at = _unix_nano_timestamp(request["expires_at_unix_nano"]) except (TypeError, ValueError): continue if requested_at <= approved_at <= expires_at: @@ -381,7 +414,7 @@ def _tool_transcript(snapshot: EvidenceSnapshot) -> dict[str, Any] | None: ] if not actions: return None - canonical = rfc8785.dumps(actions) + canonical = rfc8785.dumps(cast(Any, actions)) return { "hash": "sha256:" + hashlib.sha256(canonical).hexdigest(), "call_count": len(actions), diff --git a/tests/test_trace_adapter.py b/tests/test_trace_adapter.py index 6aa91ad..03edae7 100644 --- a/tests/test_trace_adapter.py +++ b/tests/test_trace_adapter.py @@ -8,6 +8,7 @@ ROOT = Path(__file__).resolve().parents[1] sys.path.insert(0, str(ROOT / "src")) +from agentrust_telemetry import trace_adapter # noqa: E402 from agentrust_telemetry import ( # noqa: E402 EvidenceAccumulator, SchemaValidator, @@ -195,6 +196,75 @@ def finalize_events(self, events): accumulator.seal(completeness="complete"), self.config, signing_key=self.key ) + def test_boolean_event_timestamp_cannot_become_issued_at(self): + from agentrust_telemetry.evidence import EvidenceEntry, _entry_digest + + for timestamp in (True, False): + with self.subTest(timestamp=timestamp): + snapshot = self.snapshot() + entries = [] + previous = None + for position, original in enumerate(snapshot.entries): + event = dict(original.event) + if position == len(snapshot.entries) - 1: + event["time_unix_nano"] = timestamp + digest = _entry_digest(position, previous, event) + entries.append( + EvidenceEntry( + position, event["event_id"], previous, digest, event + ) + ) + previous = digest + malformed = replace( + snapshot, entries=tuple(entries), chain_digest=previous + ) + with self.assertRaisesRegex( + TraceFinalizationError, "decimal strings" + ): + finalize_trace(malformed, self.config, signing_key=self.key) + + def test_boolean_approval_and_action_timestamps_do_not_bind(self): + for timestamp in (True, False): + with self.subTest(timestamp=timestamp): + request = fixture("approval.json") + request.update( + { + "event_type": "approval.requested", + "event_id": "018f0f7d-7a13-7cc2-8000-000000000010", + "policy_event_id": "018f0f7d-7a13-7cc2-8000-000000000001", + "chain_id": "operators", + "chain_version": "3", + "requested_at_unix_nano": timestamp, + "expires_at_unix_nano": timestamp, + "time_unix_nano": timestamp, + } + ) + approval = fixture("approval.json") + approval.update( + { + "event_type": "approval.approved", + "policy_event_id": request["policy_event_id"], + "chain_id": request["chain_id"], + "chain_version": request["chain_version"], + "requested_at_unix_nano": timestamp, + "expires_at_unix_nano": timestamp, + "time_unix_nano": timestamp, + } + ) + action = fixture("action.json") + action["time_unix_nano"] = timestamp + approvals = [request, approval] + decisions = {request["policy_event_id"]: "challenge"} + bound = trace_adapter._bound_approvals(approvals) + + self.assertEqual(bound, []) + self.assertEqual( + trace_adapter._action_approval_binding( + [action], decisions, approvals, bound + ), + "unresolved", + ) + def test_executed_action_must_match_the_digest_its_approval_approved(self): case = json.loads( (APPRAISAL_DIR / "approved-action-matches.json").read_text(encoding="utf-8")