Reject boolean timestamps in TRACE finalization

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,18 +323,26 @@ 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 (
                     policy_event_id is not None
                     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),

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")