Accept epoch-seconds relay expiry so valid verification is not misread

This commit is contained in:
yoni
2026-07-21 00:00:43 +00:00
parent a1e0f20d6e
commit ab840c31ae
2 changed files with 40 additions and 3 deletions
+21 -3
View File
@@ -61,13 +61,23 @@ def read_auth() -> dict[str, Any] | None:
def _expiry(record: dict[str, Any]) -> datetime | None:
"""Parse ``verified_at`` (the relay's ``expires_at``) into an aware UTC datetime.
Returns None when it is absent or unparseable. The local gate fails closed on
such records (see ``is_verified``), matching the relay, which rejects a token
with no valid expiry on report send.
Accepts both ISO 8601 strings and epoch seconds (as a number or numeric
string) so a valid relay expiry is not misread as missing. Returns None only
when it is genuinely absent or unparseable; the local gate then fails closed
(see ``is_verified``), matching the relay, which rejects a token with no valid
expiry on report send.
"""
raw = record.get("verified_at")
if isinstance(raw, bool):
return None
if isinstance(raw, int | float):
return _from_epoch(raw)
if not isinstance(raw, str) or not raw:
return None
try:
return _from_epoch(float(raw))
except ValueError:
pass
try:
parsed = datetime.fromisoformat(raw.replace("Z", "+00:00"))
except ValueError:
@@ -75,6 +85,14 @@ def _expiry(record: dict[str, Any]) -> datetime | None:
return parsed if parsed.tzinfo else parsed.replace(tzinfo=UTC)
def _from_epoch(seconds: float) -> datetime | None:
"""Epoch seconds → aware UTC datetime, or None if out of range."""
try:
return datetime.fromtimestamp(seconds, tz=UTC)
except (OverflowError, OSError, ValueError):
return None
def is_verified() -> bool:
"""True when a usable email + token record with a valid future expiry exists.
+19
View File
@@ -67,6 +67,25 @@ def test_is_verified_fails_closed_when_expiry_absent_or_unparseable() -> None:
assert auth.is_verified() is False
def test_is_verified_accepts_epoch_expiry() -> None:
# A relay expiry expressed as epoch seconds must not be misread as missing.
future = (datetime.now(UTC) + timedelta(hours=1)).timestamp()
past = (datetime.now(UTC) - timedelta(hours=1)).timestamp()
# As a numeric string (how write_auth persists it).
auth.write_auth(email="a@b.com", token="t", verified_at=str(future))
assert auth.is_verified() is True
auth.write_auth(email="a@b.com", token="t", verified_at=str(past))
assert auth.is_verified() is False
# As a raw JSON number, if a record is written that way.
auth.AUTH_PATH.parent.mkdir(parents=True, exist_ok=True)
auth.AUTH_PATH.write_text(
f'{{"email": "a@b.com", "token": "t", "verified_at": {future}}}', encoding="utf-8"
)
assert auth.is_verified() is True
def test_write_auth_is_0600() -> None:
auth.write_auth(email="a@b.com", token="t", verified_at="")
mode = stat.S_IMODE(auth.AUTH_PATH.stat().st_mode)