From 658eca654f4f05e8904fca19f193d6ae871b0556 Mon Sep 17 00:00:00 2001 From: Srinivasa Consultants <78673153+dinakarjs@users.noreply.github.com> Date: Tue, 3 Nov 2026 19:57:32 +0530 Subject: [PATCH] fix(trace): reject noncanonical approval timestamps (#126) Signed-off-by: Srinivasa Dinakar Jainti --- CHANGELOG.md | 8 ++ src/agentrust_telemetry/trace_adapter.py | 46 +++++-- tests/test_trace_timestamps.py | 155 +++++++++++++++++++++++ 3 files changed, 201 insertions(+), 8 deletions(-) create mode 100644 tests/test_trace_timestamps.py diff --git a/CHANGELOG.md b/CHANGELOG.md index da2c0ae..d1c81c3 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -2,6 +2,14 @@ ## Unreleased +### Fixed + +- Python TRACE finalization rejects noncanonical timestamps before appraisal or + signing, even in a self-consistent imported evidence chain. Approval binding + no longer coerces booleans, numbers, or noncanonical strings into timestamps + (#126). This enforces the existing decimal-string schema; valid event streams + and the wire contract are unchanged. + ### Changed - CI checks the existing four job-level OIDC grants across all workflow files, diff --git a/src/agentrust_telemetry/trace_adapter.py b/src/agentrust_telemetry/trace_adapter.py index 56932a9..863cf7e 100644 --- a/src/agentrust_telemetry/trace_adapter.py +++ b/src/agentrust_telemetry/trace_adapter.py @@ -3,6 +3,7 @@ from __future__ import annotations import hashlib +import re from dataclasses import dataclass from typing import Any, Literal @@ -11,7 +12,6 @@ from .errors import TraceFinalizationError from .evidence import CANONICALIZATION_PROFILE, EvidenceSnapshot, _entry_digest - OriginKind = Literal["self", "third-party-control-plane", "log-import"] @@ -46,11 +46,20 @@ def finalize_trace( events = [entry.event for entry in snapshot.entries] try: + # A consistent evidence hash does not imply schema-valid timestamps. + # Validate before appraisal or signing, including approval windows. + for event in events: + _timestamp(event["time_unix_nano"]) + for field in ("requested_at_unix_nano", "expires_at_unix_nano"): + if field in event: + _timestamp(event[field]) bundle_hash, enforcement_mode = _policy_binding(events) 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( + _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 +269,13 @@ def _appraisal(events: list[dict[str, Any]]) -> str: ActionBinding = Literal["bound", "unresolved", "contradicted"] +def _timestamp(value: Any) -> int: + """Parse the canonical decimal-string form in common.schema.json.""" + if not isinstance(value, str) or re.fullmatch(r"0|[1-9][0-9]{0,19}", value) is None: + raise ValueError("timestamp must be a canonical decimal string of at most 20 digits") + return int(value) + + def _action_approval_binding( events: list[dict[str, Any]], decisions: dict[str, Any], @@ -300,9 +316,16 @@ def _action_approval_binding( else event.get("policy_event_id") == policy_event_id ) ] - completed_at = int(action["time_unix_nano"]) + try: + completed_at = _timestamp(action["time_unix_nano"]) + except (KeyError, ValueError): + return "contradicted" contradicted = satisfied = False for approval in candidates: + try: + approved_at = _timestamp(approval["time_unix_nano"]) + except (KeyError, ValueError): + return "contradicted" approval_policy = approval.get("policy_event_id") if ( not _same_digest(approval["action_digest"], action["action_digest"]) @@ -311,7 +334,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,10 +385,17 @@ 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"]) - except (TypeError, ValueError): + # Check both sides, including the request event time. Equality + # alone is not validation (e.g. True == 1 in imported data). + for event in (request, approval): + for field in ( + "time_unix_nano", "requested_at_unix_nano", "expires_at_unix_nano" + ): + _timestamp(event[field]) + requested_at = _timestamp(request["requested_at_unix_nano"]) + approved_at = _timestamp(approval["time_unix_nano"]) + expires_at = _timestamp(request["expires_at_unix_nano"]) + except (KeyError, ValueError): continue if requested_at <= approved_at <= expires_at: resolved.append(approval) diff --git a/tests/test_trace_timestamps.py b/tests/test_trace_timestamps.py new file mode 100644 index 0000000..17ec39e --- /dev/null +++ b/tests/test_trace_timestamps.py @@ -0,0 +1,155 @@ +"""Timestamp trust boundaries, including self-consistent imported snapshots.""" + +import copy +import json +import sys +import unittest +from pathlib import Path +from unittest import mock + +ROOT = Path(__file__).resolve().parents[1] +sys.path.insert(0, str(ROOT / "src")) + +from agentrust_telemetry import ( + TraceConfiguration, + TraceFinalizationError, + finalize_trace, +) +from agentrust_telemetry.evidence import ( + CANONICALIZATION_PROFILE, + EvidenceEntry, + EvidenceSnapshot, + _entry_digest, +) +from agentrust_telemetry.trace_adapter import _action_approval_binding, _bound_approvals + +INVALID = (True, False, 1, 1.0, None, "", "01", "+1", "-1", " 1", "1 ", + "1\n", "1.0", "1e0", "١", "１", "100000000000000000000") +FIELDS = ("time_unix_nano", "requested_at_unix_nano", "expires_at_unix_nano") + + +def events(): + return json.loads( + (ROOT / "conformance/appraisal/approved-action-matches.json").read_text() + )["events"] + + +def snapshot(rows): + # Deliberately bypass schema ingestion and recompute every digest. These + # tests exercise the public finalizer, not a stale-chain refusal. + entries, previous = [], None + for sequence, event in enumerate(rows): + digest = _entry_digest(sequence, previous, event) + entries.append(EvidenceEntry(sequence, event["event_id"], previous, digest, event)) + previous = digest + return EvidenceSnapshot( + run_id=rows[0]["run_id"], entries=tuple(entries), chain_digest=previous, + canonicalization_profile=CANONICALIZATION_PROFILE, + completeness="complete", sealed=True, + ) + + +class TimestampBindingTests(unittest.TestCase): + def binding(self, rows): + approvals = rows[1:3] + return _action_approval_binding( + rows, {rows[0]["event_id"]: "challenge"}, approvals, + _bound_approvals(approvals), + ) + + def test_canonical_window_boundaries_and_expiration(self): + for value in ("0", "1", "99999999999999999999"): + rows = events() + for row in rows: + for field in FIELDS: + if field in row: + row[field] = value + with self.subTest(value=value): + self.assertEqual(len(_bound_approvals(rows[1:3])), 1) + self.assertEqual(self.binding(rows), "bound") + rows = events() + rows[2]["time_unix_nano"] = rows[1]["expires_at_unix_nano"] + rows[-1]["time_unix_nano"] = rows[2]["time_unix_nano"] + self.assertEqual(self.binding(rows), "bound") + rows[2]["time_unix_nano"] = str(int(rows[1]["expires_at_unix_nano"]) + 1) + self.assertEqual(_bound_approvals(rows[1:3]), []) + self.assertEqual(self.binding(rows), "contradicted") + rows = events() + rows[2]["time_unix_nano"] = str(int(rows[1]["requested_at_unix_nano"]) - 1) + self.assertEqual(_bound_approvals(rows[1:3]), []) + self.assertEqual(self.binding(rows), "unresolved") + + def test_noncanonical_triples_never_bind(self): + for value in INVALID: + rows = events() + for row in rows: + for field in FIELDS: + if field in row: + row[field] = value + with self.subTest(value=value): + self.assertEqual(_bound_approvals(rows[1:3]), []) + self.assertEqual(self.binding(rows), "contradicted") + + def test_each_request_and_resolution_timestamp_is_checked(self): + for index in (1, 2): + for field in FIELDS: + rows = events() + # All canonical times are 1, so bool/int equality cannot hide + # behind an unrelated mismatched binding field. + for row in rows: + for name in FIELDS: + if name in row: + row[name] = "1" + rows[index][field] = True + with self.subTest(index=index, field=field): + self.assertEqual(_bound_approvals(rows[1:3]), []) + + def test_each_action_or_approval_time_is_checked(self): + for index in (2, 4): + for value in INVALID: + rows = events() + rows[index]["time_unix_nano"] = value + with self.subTest(index=index, value=value): + self.assertEqual(self.binding(rows), "contradicted") + + +@unittest.skipIf(sys.version_info < (3, 11), "TRACE requires Python 3.11+") +class TimestampFinalizationTests(unittest.TestCase): + @classmethod + def setUpClass(cls): + from agentrust_trace import generate_key + cls.key = generate_key() + cls.config = TraceConfiguration( + subject="spiffe://example.test/agent/workflow", model_provider="example", + model_id="example-model", build_digest="sha256:" + "b" * 64, + origin_kind="self", origin_producer="example-runtime", + appraisal_verifier="https://example.test/verifier", + classification_taxonomy="example.enterprise.v1", + classification_order=("public", "internal", "confidential", "restricted"), + ) + + def test_valid_snapshot_signs_with_expected_iat_and_appraisal(self): + rows = events() + record = finalize_trace(snapshot(rows), self.config, signing_key=self.key) + self.assertEqual(record["iat"], 1787079650) + self.assertEqual(record["appraisal"]["status"], "affirming") + + def test_rehashed_invalid_timestamps_refuse_before_signing(self): + baseline = events() + for index, row in enumerate(baseline): + for field in FIELDS: + if field not in row: + continue + for value in INVALID: + rows = copy.deepcopy(baseline) + rows[index][field] = value + with self.subTest(index=index, field=field, value=value), mock.patch( + "agentrust_trace.sign_record" + ) as sign: + with self.assertRaisesRegex(TraceFinalizationError, "timestamp"): + finalize_trace(snapshot(rows), self.config, signing_key=self.key) + sign.assert_not_called() + + +if __name__ == "__main__": + unittest.main()