Skip to content
Draft
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
22 changes: 19 additions & 3 deletions ares_runtime/authority.py
Original file line number Diff line number Diff line change
Expand Up @@ -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):
Expand All @@ -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:
Expand Down Expand Up @@ -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]:
Expand Down
84 changes: 84 additions & 0 deletions tests/ares_runtime/test_authority_fractional_time.py
Original file line number Diff line number Diff line change
@@ -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)
Loading