From ee02e4817fe105428db83f5c3abd5f8074bdcc6c Mon Sep 17 00:00:00 2001 From: Raghuvar Karthik Durga <177662415+drk-ram@users.noreply.github.com> Date: Thu, 29 Oct 2026 12:14:35 -0500 Subject: [PATCH] fix(agt_data,data-flow): reject float classification tiers and pre-epoch evaluated_at timestamps in Python and TypeScript (#112) Signed-off-by: Raghuvar Karthik Durga <177662415+drk-ram@users.noreply.github.com> --- packages/typescript/src/data-flow.ts | 9 ++++--- packages/typescript/test/data-flow.test.ts | 2 ++ src/agentrust_telemetry/adapters/agt_data.py | 6 +++-- tests/test_agt_adapter.py | 28 ++++++++++++++++++++ 4 files changed, 39 insertions(+), 6 deletions(-) diff --git a/packages/typescript/src/data-flow.ts b/packages/typescript/src/data-flow.ts index 9e88956..7437ab4 100644 --- a/packages/typescript/src/data-flow.ts +++ b/packages/typescript/src/data-flow.ts @@ -34,14 +34,15 @@ export function agtDataClassification(source: Record): Classifi } export function agtDataAccessFlow(factory: EventFactory, decision: Record, options: {runId: string; direction: string; source: DataEndpoint; destination: DataEndpoint; purpose: string; contentDigest?: Digest; sizeBytes?: number; tokenCount?: number; mediaType?: string; transformation?: string; envelope?: Omit}): NormalizedEvent { - if (typeof decision.allowed !== "boolean") throw new Error("AGT data access allowed must be a boolean"); - const label = object(decision.data_label, "AGT data_label"); + const rec = object(decision, "AGT decision"); + if (typeof rec.allowed !== "boolean") throw new Error("AGT data access allowed must be a boolean"); + const label = object(rec.data_label, "AGT data_label"); const classification = agtDataClassification(label); - return classifiedDataFlow(factory, {classify: () => classification}, undefined, {...options, agentId: required(decision.agent_id, "AGT data access agent_id"), policyDecision: decision.allowed ? "allow" : "deny", envelope: {timeUnixNano: timestampNs(decision.evaluated_at, "evaluated_at"), ...options.envelope}}); + return classifiedDataFlow(factory, {classify: () => classification}, undefined, {...options, agentId: required(rec.agent_id, "AGT data access agent_id"), policyDecision: rec.allowed ? "allow" : "deny", envelope: {timeUnixNano: timestampNs(rec.evaluated_at, "evaluated_at"), ...options.envelope}}); } function identifier(value: unknown, field: string): string { if (typeof value !== "string" || !IDENTIFIER.test(value) || value.includes("://")) throw new Error(`${field} must be a 1-256 character metadata identifier without whitespace, URLs, query strings, fragments, or key/value delimiters`); return value; } function required(value: unknown, field: string): string { if (typeof value !== "string" || !value) throw new Error(`${field} must be a non-empty string`); return value; } function object(value: unknown, field: string): Record { if (typeof value !== "object" || value === null || Array.isArray(value)) throw new Error(`${field} must be an object`); return value as Record; } function present(key: string, value: unknown): Record { return value === undefined ? {} : {[key]: value}; } -function timestampNs(value: unknown, field: string): bigint { if (typeof value !== "string") throw new Error(`AGT data access ${field} must be an RFC 3339 UTC timestamp`); const match = /^(\d{4}-\d{2}-\d{2}T\d{2}:\d{2}:\d{2})(?:\.(\d{1,9}))?(?:Z|\+00:00)$/.exec(value); if (!match) throw new Error(`AGT data access ${field} must be an RFC 3339 UTC timestamp`); const milliseconds = Date.parse(`${match[1]}Z`); if (match[1]!.startsWith("0000") || !Number.isFinite(milliseconds) || new Date(milliseconds).toISOString().slice(0, 19) !== match[1]) throw new Error(`AGT data access ${field} is invalid`); return BigInt(milliseconds) * 1_000_000n + BigInt((match[2] ?? "").padEnd(9, "0") || "0"); } +function timestampNs(value: unknown, field: string): bigint { if (typeof value !== "string") throw new Error(`AGT data access ${field} must be an RFC 3339 UTC timestamp`); const match = /^(\d{4}-\d{2}-\d{2}T\d{2}:\d{2}:\d{2})(?:\.(\d{1,9}))?(?:Z|\+00:00)$/.exec(value); if (!match) throw new Error(`AGT data access ${field} must be an RFC 3339 UTC timestamp`); const milliseconds = Date.parse(`${match[1]}Z`); if (match[1]!.startsWith("0000") || !Number.isFinite(milliseconds) || milliseconds < 0 || new Date(milliseconds).toISOString().slice(0, 19) !== match[1]) throw new Error(`AGT data access ${field} is invalid`); return BigInt(milliseconds) * 1_000_000n + BigInt((match[2] ?? "").padEnd(9, "0") || "0"); } diff --git a/packages/typescript/test/data-flow.test.ts b/packages/typescript/test/data-flow.test.ts index c1a912b..6d10e30 100644 --- a/packages/typescript/test/data-flow.test.ts +++ b/packages/typescript/test/data-flow.test.ts @@ -19,4 +19,6 @@ test("AGT access maps only tier, decision, agent, and time", () => { const event = agtDataAccessFlow(factory, decision, {runId: "run-1", direction: "read", source: new DataEndpoint("database", "customer-records"), destination: new DataEndpoint("agent", "builder"), purpose: "requirements-analysis"}); assert.equal((event.classification as Record).value, "confidential"); assert.equal(event.policy_decision, "deny"); const serialized = JSON.stringify(event); for (const value of ["PII", "customer@example", "alice@example", "123-45-6789", "geography", "matched_policy"]) assert.equal(serialized.includes(value), false); + assert.throws(() => agtDataAccessFlow(factory, {...decision, evaluated_at: "1969-12-31T23:59:59Z"}, {runId: "run-1", direction: "read", source: new DataEndpoint("database", "customer-records"), destination: new DataEndpoint("agent", "builder"), purpose: "requirements-analysis"}), /evaluated_at is invalid/); + assert.throws(() => agtDataAccessFlow(factory, null as unknown as Record, {runId: "run-1", direction: "read", source: new DataEndpoint("database", "customer-records"), destination: new DataEndpoint("agent", "builder"), purpose: "requirements-analysis"}), /must be an object/); }); diff --git a/src/agentrust_telemetry/adapters/agt_data.py b/src/agentrust_telemetry/adapters/agt_data.py index a7d9a38..08f39c0 100644 --- a/src/agentrust_telemetry/adapters/agt_data.py +++ b/src/agentrust_telemetry/adapters/agt_data.py @@ -26,7 +26,7 @@ def agt_data_classification(source: Any) -> ClassificationResult: """Map only AGT's ordered sensitivity tier; omit auxiliary label content.""" classification = _field(source, "classification") raw = classification.value if isinstance(classification, Enum) else classification - if isinstance(raw, bool) or not isinstance(raw, (int, float)) or raw not in _LEVELS: + if type(raw) is not int or raw not in _LEVELS: raise ValueError(f"unsupported AGT data classification: {raw!r}") return ClassificationResult( taxonomy="agt.data_classification.v1", @@ -93,10 +93,12 @@ def _field(source: Any, name: str) -> Any: def _datetime_ns(value: Any) -> int: from datetime import datetime, timezone - if not isinstance(value, datetime) or value.tzinfo is None: + if not isinstance(value, datetime) or value.tzinfo is None or value.utcoffset() is None: raise ValueError("AGT data access evaluated_at must be a timezone-aware datetime") utc = value.astimezone(timezone.utc) epoch = datetime(1970, 1, 1, tzinfo=timezone.utc) + if utc < epoch: + raise ValueError("AGT data access evaluated_at must not predate the Unix epoch") delta = utc - epoch return ( delta.days * 86_400_000_000_000 diff --git a/tests/test_agt_adapter.py b/tests/test_agt_adapter.py index 0b40f45..ec12fc0 100644 --- a/tests/test_agt_adapter.py +++ b/tests/test_agt_adapter.py @@ -155,6 +155,34 @@ def test_sink_uses_source_result_sentinels_and_reports_projection_failure(self): ) self.assertEqual(sink.emit([FakeAgtEvent()]), Result.FAILURE) + def test_agt_data_rejects_float_tiers_and_pre_epoch_timestamps(self): + from datetime import datetime, timezone + from agentrust_telemetry.adapters.agt_data import ( + DataEndpoint, + agt_data_access_flow, + agt_data_classification, + ) + + for bad_tier in (0.0, 2.0, True, False, -1, 5, "public"): + with self.subTest(bad_tier=bad_tier): + with self.assertRaises(ValueError): + agt_data_classification({"classification": bad_tier}) + with self.assertRaises(ValueError): + agt_data_access_flow( + self.factory, + { + "allowed": True, + "agent_id": "agent-1", + "data_label": {"classification": 1}, + "evaluated_at": datetime(1969, 12, 31, 23, 59, 59, tzinfo=timezone.utc), + }, + run_id="run-1", + direction="read", + source=DataEndpoint("database", "db"), + destination=DataEndpoint("agent", "a1"), + purpose="analysis", + ) + if __name__ == "__main__": unittest.main()