mirror of
https://github.com/usestrix/strix.git
synced 2026-08-21 18:52:47 +02:00
Session-gate historical run reads; fail closed on missing expiry
This commit is contained in:
@@ -61,8 +61,9 @@ 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, in which case expiry cannot be
|
||||
enforced locally (the relay still rejects an expired token on report send).
|
||||
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.
|
||||
"""
|
||||
raw = record.get("verified_at")
|
||||
if not isinstance(raw, str) or not raw:
|
||||
@@ -75,17 +76,19 @@ def _expiry(record: dict[str, Any]) -> datetime | None:
|
||||
|
||||
|
||||
def is_verified() -> bool:
|
||||
"""True when a usable, unexpired email + token record exists locally.
|
||||
"""True when a usable email + token record with a valid future expiry exists.
|
||||
|
||||
The expiry returned by OTP verification is enforced here so history stops
|
||||
unlocking once the token lapses, keeping the local gate in step with the
|
||||
relay (which rejects an expired token on report send).
|
||||
unlocking once the token lapses. It fails closed: a record whose expiry is
|
||||
absent, blank, or unparseable requires re-verification rather than unlocking
|
||||
forever, keeping the local gate in step with the relay (which rejects an
|
||||
expired token on report send).
|
||||
"""
|
||||
record = read_auth()
|
||||
if record is None:
|
||||
return False
|
||||
expiry = _expiry(record)
|
||||
return expiry is None or expiry > datetime.now(UTC)
|
||||
return expiry is not None and expiry > datetime.now(UTC)
|
||||
|
||||
|
||||
def write_auth(email: str, token: str, verified_at: str) -> None:
|
||||
|
||||
+10
-5
@@ -255,12 +255,17 @@ def _make_handler(state: _ViewerState) -> type[BaseHTTPRequestHandler]:
|
||||
return
|
||||
|
||||
# The launched run is always viewable. Any *other* run's data is part
|
||||
# of the gated history, so it requires the same email verification as
|
||||
# the /api/runs list — otherwise knowing a run name would leak its
|
||||
# of the gated history: it needs this process's session capability
|
||||
# (so merely reaching an exposed --host port is not enough) *and*
|
||||
# email verification -- otherwise knowing a run name would leak its
|
||||
# metadata, vulnerabilities, report, and transcript.
|
||||
if run_dir.resolve() != state.run_dir.resolve() and not auth.is_verified():
|
||||
self._send_json(HTTPStatus.UNAUTHORIZED, {"error": "unverified"})
|
||||
return
|
||||
if run_dir.resolve() != state.run_dir.resolve():
|
||||
if not self._has_session():
|
||||
self._send_json(HTTPStatus.FORBIDDEN, {"error": "forbidden"})
|
||||
return
|
||||
if not auth.is_verified():
|
||||
self._send_json(HTTPStatus.UNAUTHORIZED, {"error": "unverified"})
|
||||
return
|
||||
|
||||
if path == "/api/run":
|
||||
self._send_json(HTTPStatus.OK, read_run_summary(run_dir))
|
||||
|
||||
+18
-9
@@ -213,9 +213,11 @@ def _session_cookie(url: str, token: str) -> str:
|
||||
return raw.split(";", 1)[0]
|
||||
|
||||
|
||||
def _get_status(url: str) -> int:
|
||||
def _get_status(url: str, *, cookie: str | None = None) -> int:
|
||||
headers = {"Cookie": cookie} if cookie else {}
|
||||
req = urllib.request.Request(url, headers=headers) # noqa: S310 - localhost test server
|
||||
try:
|
||||
with urllib.request.urlopen(url) as resp: # noqa: S310 - localhost test server
|
||||
with urllib.request.urlopen(req) as resp: # noqa: S310
|
||||
return int(resp.status)
|
||||
except urllib.error.HTTPError as exc:
|
||||
return int(exc.code)
|
||||
@@ -378,19 +380,26 @@ def test_historical_run_data_requires_verification(
|
||||
verified = {"value": False}
|
||||
monkeypatch.setattr("strix.viewer.auth.is_verified", lambda: verified["value"])
|
||||
|
||||
httpd, url, _ = serve(launched, open_browser=False)
|
||||
httpd, url, token = serve(launched, open_browser=False)
|
||||
try:
|
||||
# The launched run is always viewable, no verification required.
|
||||
# The launched run is always viewable, no verification and no cookie.
|
||||
status, _, _ = _get(f"{url}/api/run")
|
||||
assert status == 200
|
||||
|
||||
# A different run's data is gated behind verification.
|
||||
assert _get_status(f"{url}/api/run?run=other") == 401
|
||||
cookie = _session_cookie(url, token)
|
||||
|
||||
# Once verified, the historical run resolves.
|
||||
# A different run needs the session capability first: a cookie-less
|
||||
# caller is forbidden even once the machine is verified.
|
||||
verified["value"] = True
|
||||
status, _, _ = _get(f"{url}/api/run?run=other")
|
||||
assert status == 200
|
||||
assert _get_status(f"{url}/api/run?run=other") == 403
|
||||
|
||||
# With the cookie but not verified, the history gate returns 401.
|
||||
verified["value"] = False
|
||||
assert _get_status(f"{url}/api/run?run=other", cookie=cookie) == 401
|
||||
|
||||
# With both the cookie and verification, the historical run resolves.
|
||||
verified["value"] = True
|
||||
assert _get_status(f"{url}/api/run?run=other", cookie=cookie) == 200
|
||||
finally:
|
||||
httpd.shutdown()
|
||||
httpd.server_close()
|
||||
|
||||
@@ -56,14 +56,15 @@ def test_is_verified_enforces_expiry() -> None:
|
||||
assert auth.is_verified() is True
|
||||
|
||||
|
||||
def test_is_verified_when_expiry_absent_or_unparseable() -> None:
|
||||
# No/blank expiry: cannot enforce locally, so treat as valid.
|
||||
def test_is_verified_fails_closed_when_expiry_absent_or_unparseable() -> None:
|
||||
# No/blank expiry: fail closed rather than unlocking history forever.
|
||||
auth.write_auth(email="a@b.com", token="t", verified_at="")
|
||||
assert auth.is_verified() is True
|
||||
assert auth.read_auth() is not None
|
||||
assert auth.is_verified() is False
|
||||
|
||||
# Garbage expiry is ignored rather than locking the user out.
|
||||
# Garbage expiry likewise requires re-verification.
|
||||
auth.write_auth(email="a@b.com", token="t", verified_at="not-a-date")
|
||||
assert auth.is_verified() is True
|
||||
assert auth.is_verified() is False
|
||||
|
||||
|
||||
def test_write_auth_is_0600() -> None:
|
||||
|
||||
Reference in New Issue
Block a user