From 6f1b1a111101779ce7016a9a5e0371abf7052063 Mon Sep 17 00:00:00 2001 From: Josh Stevenson Date: Sat, 3 Oct 2026 21:48:55 -0700 Subject: [PATCH] fix(authority): compare temporal scope bounds as instants --- ares_runtime/authority.py | 22 ++++- .../test_authority_fractional_time.py | 84 +++++++++++++++++++ 2 files changed, 103 insertions(+), 3 deletions(-) create mode 100644 tests/ares_runtime/test_authority_fractional_time.py diff --git a/ares_runtime/authority.py b/ares_runtime/authority.py index 89ebf1cc36bd4..7c2e5c98a89d0 100644 --- a/ares_runtime/authority.py +++ b/ares_runtime/authority.py @@ -39,6 +39,11 @@ def _normalized_time(value: Any, field: str) -> str: return instant.astimezone(timezone.utc).isoformat().replace("+00:00", "Z") +def _time_instant(value: str) -> datetime: + """Compare normalized bounds as instants, preserving their wire spelling.""" + return datetime.fromisoformat(value.replace("Z", "+00:00")) + + def normalize_scope(raw: Any) -> dict[str, Any]: """Canonicalize a scope record; reject unknown fields and bad types.""" if not isinstance(raw, Mapping): @@ -64,7 +69,12 @@ def normalize_scope(raw: Any) -> dict[str, Any]: if not value: raise ContractError("INVALID_TIME_SCOPE") normalized_time = {bound: _normalized_time(value[bound], "time." + bound) for bound in ("not_before", "not_after") if bound in value} - if "not_before" in normalized_time and "not_after" in normalized_time and normalized_time["not_before"] > normalized_time["not_after"]: + if ( + "not_before" in normalized_time + and "not_after" in normalized_time + and _time_instant(normalized_time["not_before"]) + > _time_instant(normalized_time["not_after"]) + ): raise ContractError("INVALID_TIME_SCOPE") normalized[field] = normalized_time else: @@ -108,9 +118,15 @@ def is_subset_scope(subset: Mapping[str, Any], superset: Mapping[str, Any]) -> b return False elif field == "time": sa, sb = a[field], b[field] - if "not_before" in sb and ("not_before" not in sa or sa["not_before"] < sb["not_before"]): + if "not_before" in sb and ( + "not_before" not in sa + or _time_instant(sa["not_before"]) < _time_instant(sb["not_before"]) + ): return False - if "not_after" in sb and ("not_after" not in sa or sa["not_after"] > sb["not_after"]): + if "not_after" in sb and ( + "not_after" not in sa + or _time_instant(sa["not_after"]) > _time_instant(sb["not_after"]) + ): return False else: if a[field] != b[field]: diff --git a/tests/ares_runtime/test_authority_fractional_time.py b/tests/ares_runtime/test_authority_fractional_time.py new file mode 100644 index 0000000000000..f602281a38981 --- /dev/null +++ b/tests/ares_runtime/test_authority_fractional_time.py @@ -0,0 +1,84 @@ +"""Temporal attenuation compares UTC instants without changing wire digests.""" + +import pytest + +from ares_runtime.authority import ( + AuthorityScopeV1, + ContractError, + is_subset_scope, + normalize_scope, + scope_fingerprint, +) + + +WHOLE = "2026-10-04T00:00:00Z" +FRACTION = "2026-10-04T00:00:00.100000Z" + + +@pytest.mark.parametrize( + "start,end", + [ + (WHOLE, FRACTION), + ("2026-10-04T01:00:00+01:00", FRACTION), + (FRACTION, "2026-10-04T00:00:00.200000Z"), + (WHOLE, "2026-10-04T00:00:00.000000Z"), + ], +) +def test_interval_accepts_ordered_same_second_instants(start, end): + scope = normalize_scope({"time": {"not_before": start, "not_after": end}}) + assert set(scope["time"]) == {"not_before", "not_after"} + + +@pytest.mark.parametrize( + "start,end", + [ + (FRACTION, WHOLE), + (FRACTION, "2026-10-04T01:00:00+01:00"), + ("2026-10-04T00:00:00.200000Z", FRACTION), + ], +) +def test_interval_rejects_reversed_same_second_instants(start, end): + with pytest.raises(ContractError, match="INVALID_TIME_SCOPE"): + normalize_scope({"time": {"not_before": start, "not_after": end}}) + + +@pytest.mark.parametrize( + "bound,parent,child,contained", + [ + ("not_after", WHOLE, FRACTION, False), + ("not_after", FRACTION, WHOLE, True), + ("not_before", FRACTION, WHOLE, False), + ("not_before", WHOLE, FRACTION, True), + ("not_after", WHOLE, "2026-10-04T01:00:00.000000+01:00", True), + ("not_before", FRACTION, "2026-10-04T01:00:00.100000+01:00", True), + ], +) +def test_same_second_subset_respects_each_bound(bound, parent, child, contained): + assert is_subset_scope( + {"time": {bound: child}}, {"time": {bound: parent}} + ) is contained + + +@pytest.mark.parametrize( + "bound,parent_time,child_time", + [("not_after", WHOLE, FRACTION), ("not_before", FRACTION, WHOLE)], +) +def test_rejected_temporal_widening_does_not_spend_delegation( + bound, parent_time, child_time +): + parent = AuthorityScopeV1( + scope={"tool": "read_file", "time": {bound: parent_time}, "use_count": 1}, + generation=1, + ) + with pytest.raises(ContractError, match="ATTENUATION_ESCALATION"): + parent.attenuate({"time": {bound: child_time}}, child_generation=2) + child = parent.attenuate({"time": {bound: parent_time}}, child_generation=2) + assert child.subset_witness(parent)["contained"] is True + + +def test_existing_canonical_timestamp_and_fingerprint_are_preserved(): + scope = {"time": {"not_after": WHOLE}} + equivalent = {"time": {"not_after": "2026-10-04T01:00:00.000000+01:00"}} + assert normalize_scope(scope) == scope + assert normalize_scope(equivalent) == scope + assert scope_fingerprint(equivalent) == scope_fingerprint(scope)