Harden viewer authz: session capability, history gating, expiry

pyproject.toml

@@ -214,6 +214,9 @@ ignore = [
 # args they intentionally ignore.
 "tests/test_viewer_auth.py" = ["S105", "S106", "ARG001"]
 "tests/test_report_pdf.py" = ["S105", "S106"]
+# Stdlib HTTP handler overrides (do_GET/do_POST) and lazy imports that avoid a
+# circular dependency with strix.telemetry / strix.viewer.report_pdf.
+"strix/viewer/server.py" = ["N802", "PLC0415"]
 # Lazy imports inside functions to avoid circular dependency with
 # strix.telemetry / strix.report.dedupe / cvss.
 "strix/tools/notes/tools.py" = ["PLC0415", "TC002"]

strix/viewer/auth.py

@@ -17,6 +17,7 @@ import json
 import logging
 import urllib.error
 import urllib.request
+from datetime import UTC, datetime
 from pathlib import Path
 from typing import Any
 
@@ -57,9 +58,34 @@ def read_auth() -> dict[str, Any] | None:
     return data
 
 
+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).
+    """
+    raw = record.get("verified_at")
+    if not isinstance(raw, str) or not raw:
+        return None
+    try:
+        parsed = datetime.fromisoformat(raw.replace("Z", "+00:00"))
+    except ValueError:
+        return None
+    return parsed if parsed.tzinfo else parsed.replace(tzinfo=UTC)
+
+
 def is_verified() -> bool:
-    """True when a usable email + token record exists locally."""
-    return read_auth() is not None
+    """True when a usable, unexpired email + token record exists locally.
+
+    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).
+    """
+    record = read_auth()
+    if record is None:
+        return False
+    expiry = _expiry(record)
+    return expiry is None or expiry > datetime.now(UTC)
 
 
 def write_auth(email: str, token: str, verified_at: str) -> None:

strix/viewer/server.py

@@ -17,6 +17,7 @@ from __future__ import annotations
 import json
 import logging
 import mimetypes
+import secrets
 import threading
 import webbrowser
 from http import HTTPStatus
@@ -106,6 +107,10 @@ def resolve_run_dir(base_dir: Path, run_param: str | None, default_run_dir: Path
     return candidate
 
 
+# Name of the cookie carrying the per-process session capability.
+SESSION_COOKIE = "strix_viewer_session"
+
+
 class _ViewerState:
     def __init__(
         self,
@@ -122,6 +127,13 @@ class _ViewerState:
         # launcher), which can deliver a message to a running agent. Absent for
         # standalone ``strix view`` / finished runs, so steering is unavailable.
         self.steer_handler = steer_handler
+        # Unguessable per-process capability handed to the local browser (via a
+        # cookie on index.html) and required on the sensitive routes. It is the
+        # request-level authorization Greptile asked for: reachability of the
+        # port (e.g. when bound with ``--host``) is no longer sufficient to
+        # steer a live scan or trigger a report; only the browser that loaded
+        # the page this process served holds the token.
+        self.session_token = secrets.token_urlsafe(32)
 
 
 def _make_handler(state: _ViewerState) -> type[BaseHTTPRequestHandler]:
@@ -195,20 +207,16 @@ def _make_handler(state: _ViewerState) -> type[BaseHTTPRequestHandler]:
             # only the whitelisted event names and their known props are passed.
             event = body.get("event")
             if event == "cta_clicked":
-                from strix.telemetry import posthog  # noqa: PLC0415
+                from strix.telemetry import posthog
 
                 cta = str(body.get("cta") or "unknown")
                 surface = body.get("surface")
-                posthog.viewer_cta_clicked(
-                    cta, surface=str(surface) if surface else None
-                )
+                posthog.viewer_cta_clicked(cta, surface=str(surface) if surface else None)
             elif event in self._EMAIL_EVENTS:
-                from strix.telemetry import posthog  # noqa: PLC0415
+                from strix.telemetry import posthog
 
                 purpose = body.get("purpose")
-                posthog.viewer_email_event(
-                    str(event), purpose=str(purpose) if purpose else None
-                )
+                posthog.viewer_email_event(str(event), purpose=str(purpose) if purpose else None)
             self.send_response(HTTPStatus.NO_CONTENT)
             self.end_headers()
 
@@ -222,9 +230,7 @@ def _make_handler(state: _ViewerState) -> type[BaseHTTPRequestHandler]:
             if path == "/api/capabilities":
                 # Steering is only possible when the viewer shares a live scan's
                 # coordinator + event loop (the TUI launcher wires a handler).
-                self._send_json(
-                    HTTPStatus.OK, {"can_steer": state.steer_handler is not None}
-                )
+                self._send_json(HTTPStatus.OK, {"can_steer": state.steer_handler is not None})
                 return
             if path == "/api/auth/status":
                 record = auth.read_auth()
@@ -244,6 +250,14 @@ def _make_handler(state: _ViewerState) -> type[BaseHTTPRequestHandler]:
                 self._send_json(HTTPStatus.NOT_FOUND, {"error": "unknown run"})
                 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
+            # 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 path == "/api/run":
                 self._send_json(HTTPStatus.OK, read_run_summary(run_dir))
             elif path == "/api/vulnerabilities":
@@ -292,6 +306,9 @@ def _make_handler(state: _ViewerState) -> type[BaseHTTPRequestHandler]:
             self._send_json(HTTPStatus.OK, {"ok": True})
 
         def _handle_report_send(self) -> None:
+            if not self._has_session():
+                self._send_json(HTTPStatus.FORBIDDEN, {"error": "forbidden"})
+                return
             record = auth.read_auth()
             if record is None:
                 self._send_json(HTTPStatus.UNAUTHORIZED, {"error": "unverified"})
@@ -302,7 +319,7 @@ def _make_handler(state: _ViewerState) -> type[BaseHTTPRequestHandler]:
                 self._send_json(HTTPStatus.NOT_FOUND, {"error": "unknown run"})
                 return
 
-            from strix.viewer.report_pdf import build_encrypted_report  # noqa: PLC0415
+            from strix.viewer.report_pdf import build_encrypted_report
 
             pdf_bytes, password, filename = build_encrypted_report(run_dir)
             summary = read_run_summary(run_dir)
@@ -325,6 +342,9 @@ def _make_handler(state: _ViewerState) -> type[BaseHTTPRequestHandler]:
         _STEER_MESSAGE_MAX = 4000
 
         def _handle_steer(self) -> None:
+            if not self._has_session():
+                self._send_json(HTTPStatus.FORBIDDEN, {"error": "forbidden"})
+                return
             body = self._read_body()
             agent_id = body.get("agent_id")
             message = body.get("message")
@@ -340,9 +360,7 @@ def _make_handler(state: _ViewerState) -> type[BaseHTTPRequestHandler]:
                 return
             if state.steer_handler is None:
                 # Standalone / finished-run viewing has no live scan to steer.
-                self._send_json(
-                    HTTPStatus.FORBIDDEN, {"error": "steering_unavailable"}
-                )
+                self._send_json(HTTPStatus.FORBIDDEN, {"error": "steering_unavailable"})
                 return
             delivered = state.steer_handler(agent_id, message)
             if delivered:
@@ -364,8 +382,27 @@ def _make_handler(state: _ViewerState) -> type[BaseHTTPRequestHandler]:
             status = status_by_code.get(exc.code, HTTPStatus.BAD_GATEWAY)
             self._send_json(status, {"error": exc.code})
 
+        def _cookies(self) -> dict[str, str]:
+            jar: dict[str, str] = {}
+            for chunk in (self.headers.get("Cookie") or "").split(";"):
+                name, sep, value = chunk.strip().partition("=")
+                if sep:
+                    jar[name] = value
+            return jar
+
+        def _has_session(self) -> bool:
+            """True when the request carries this process's session capability.
+
+            The cookie is set only when the SPA is served (index.html), so only
+            the browser this process handed the page to can pass. A direct
+            caller on an exposed port has no cookie and is rejected.
+            """
+            supplied = self._cookies().get(SESSION_COOKIE, "")
+            return bool(supplied) and secrets.compare_digest(supplied, state.session_token)
+
         def _handle_static(self, path: str) -> None:
             target = self._resolve_asset(path)
+            is_index = target is None
             if target is None:
                 # SPA fallback: unknown non-asset routes render index.html so
                 # client-side deep links work.
@@ -378,6 +415,14 @@ def _make_handler(state: _ViewerState) -> type[BaseHTTPRequestHandler]:
             self.send_response(HTTPStatus.OK)
             self.send_header("Content-Type", content_type or "application/octet-stream")
             self.send_header("Content-Length", str(len(content)))
+            if is_index:
+                # Hand the loading browser the per-process session capability.
+                # HttpOnly (JS never needs it; fetch sends it automatically) and
+                # SameSite=Strict (never sent from a cross-site context).
+                self.send_header(
+                    "Set-Cookie",
+                    f"{SESSION_COOKIE}={state.session_token}; Path=/; HttpOnly; SameSite=Strict",
+                )
             self.end_headers()
             self.wfile.write(content)
 

tests/test_viewer.py

@@ -4,6 +4,7 @@ from __future__ import annotations
 
 import json
 import os
+import urllib.error
 import urllib.request
 from typing import TYPE_CHECKING
 
@@ -18,6 +19,7 @@ from strix.viewer.transcript import (
 
 
 if TYPE_CHECKING:
+    from collections.abc import Mapping
     from pathlib import Path
 
     import pytest
@@ -89,9 +91,7 @@ def _get(url: str) -> tuple[int, str, bytes]:
         return resp.status, resp.headers.get("Content-Type", ""), resp.read()
 
 
-def test_server_serves_api_and_static(
-    tmp_path: Path, monkeypatch: pytest.MonkeyPatch
-) -> None:
+def test_server_serves_api_and_static(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
     run_dir = _make_run(tmp_path, "served", status="completed", end_time="2026-01-01T00:00:00Z")
 
     assets = tmp_path / "bundle"
@@ -189,9 +189,151 @@ def test_server_event_endpoint_forwards_email_funnel(
         httpd.server_close()
 
 
-def test_server_rejects_path_traversal(
+def _post(
+    url: str, path: str, payload: Mapping[str, object], *, cookie: str | None = None
+) -> tuple[int, bytes]:
+    headers = {"Content-Type": "application/json"}
+    if cookie:
+        headers["Cookie"] = cookie
+    req = urllib.request.Request(  # noqa: S310 - localhost test server
+        url + path, data=json.dumps(payload).encode(), headers=headers, method="POST"
+    )
+    try:
+        with urllib.request.urlopen(req) as resp:  # noqa: S310
+            return resp.status, resp.read()
+    except urllib.error.HTTPError as exc:
+        return exc.code, exc.read()
+
+
+def _session_cookie(url: str) -> str:
+    """Fetch index.html and return its ``name=value`` session cookie."""
+    with urllib.request.urlopen(url + "/") as resp:  # noqa: S310 - localhost test server
+        raw = str(resp.headers.get("Set-Cookie", ""))
+    return raw.split(";", 1)[0]
+
+
+def _get_status(url: str) -> int:
+    try:
+        with urllib.request.urlopen(url) as resp:  # noqa: S310 - localhost test server
+            return int(resp.status)
+    except urllib.error.HTTPError as exc:
+        return int(exc.code)
+
+
+def _bundle(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
+    assets = tmp_path / "bundle"
+    assets.mkdir()
+    (assets / "index.html").write_text("<!doctype html><div id=root></div>", encoding="utf-8")
+    monkeypatch.setattr("strix.viewer.server.bundle_dir", lambda: assets)
+
+
+def test_index_sets_session_cookie_but_assets_do_not(
     tmp_path: Path, monkeypatch: pytest.MonkeyPatch
 ) -> None:
+    run_dir = _make_run(tmp_path, "cookie", status="running", end_time=None)
+    assets = tmp_path / "bundle"
+    (assets / "assets").mkdir(parents=True)
+    (assets / "index.html").write_text("<!doctype html>index", encoding="utf-8")
+    (assets / "assets" / "app.js").write_text("1", encoding="utf-8")
+    monkeypatch.setattr("strix.viewer.server.bundle_dir", lambda: assets)
+
+    httpd, url = serve(run_dir, open_browser=False)
+    try:
+        with urllib.request.urlopen(url + "/") as resp:  # noqa: S310
+            cookie = resp.headers.get("Set-Cookie", "")
+        assert "strix_viewer_session=" in cookie
+        assert "HttpOnly" in cookie and "SameSite=Strict" in cookie
+
+        with urllib.request.urlopen(url + "/assets/app.js") as resp:  # noqa: S310
+            assert resp.headers.get("Set-Cookie") is None
+    finally:
+        httpd.shutdown()
+        httpd.server_close()
+
+
+def test_steer_requires_session_cookie(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
+    run_dir = _make_run(tmp_path, "steer", status="running", end_time=None)
+    _bundle(tmp_path, monkeypatch)
+
+    delivered: list[tuple[str, str]] = []
+
+    def handler(agent_id: str, message: str) -> bool:
+        delivered.append((agent_id, message))
+        return True
+
+    httpd, url = serve(run_dir, open_browser=False, steer_handler=handler)
+    try:
+        body = {"agent_id": "root", "message": "focus on auth"}
+        # No cookie: rejected before reaching the live coordinator.
+        status, _ = _post(url, "/api/agents/steer", body)
+        assert status == 403
+        assert delivered == []
+
+        # With the session cookie the message is delivered.
+        status, raw = _post(url, "/api/agents/steer", body, cookie=_session_cookie(url))
+        assert status == 200
+        assert json.loads(raw)["ok"] is True
+        assert delivered == [("root", "focus on auth")]
+    finally:
+        httpd.shutdown()
+        httpd.server_close()
+
+
+def test_report_send_requires_session_cookie(
+    tmp_path: Path, monkeypatch: pytest.MonkeyPatch
+) -> None:
+    run_dir = _make_run(tmp_path, "report", status="completed", end_time="2026-01-01T00:00:00Z")
+    _bundle(tmp_path, monkeypatch)
+
+    # A verified machine token exists, but that alone must not authorize a caller.
+    monkeypatch.setattr("strix.viewer.auth.read_auth", lambda: {"email": "a@b.com", "token": "t"})
+
+    httpd, url = serve(run_dir, open_browser=False)
+    try:
+        # No cookie: forbidden before the machine token is ever consulted.
+        status, _ = _post(url, "/api/report/send", {})
+        assert status == 403
+
+        # With the cookie the request clears the session gate; it then reaches
+        # the run resolver, so an unknown run is a 404 rather than a 403.
+        status, _ = _post(
+            url, "/api/report/send", {"run": "does-not-exist"}, cookie=_session_cookie(url)
+        )
+        assert status == 404
+    finally:
+        httpd.shutdown()
+        httpd.server_close()
+
+
+def test_historical_run_data_requires_verification(
+    tmp_path: Path, monkeypatch: pytest.MonkeyPatch
+) -> None:
+    launched = _make_run(tmp_path, "launched", status="completed", end_time="2026-01-01T00:00:00Z")
+    _make_run(tmp_path, "other", status="completed", end_time="2026-01-01T00:00:00Z")
+    _bundle(tmp_path, monkeypatch)
+
+    verified = {"value": False}
+    monkeypatch.setattr("strix.viewer.auth.is_verified", lambda: verified["value"])
+
+    httpd, url = serve(launched, open_browser=False)
+    try:
+        # The launched run is always viewable, no verification required.
+        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
+
+        # Once verified, the historical run resolves.
+        verified["value"] = True
+        status, _, _ = _get(f"{url}/api/run?run=other")
+        assert status == 200
+    finally:
+        httpd.shutdown()
+        httpd.server_close()
+
+
+def test_server_rejects_path_traversal(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
     run_dir = _make_run(tmp_path, "guard", status="completed", end_time="2026-01-01T00:00:00Z")
     secret = tmp_path / "secret.txt"
     secret.write_text("top secret", encoding="utf-8")

tests/test_viewer_auth.py

@@ -3,6 +3,7 @@
 from __future__ import annotations
 
 import stat
+from datetime import UTC, datetime, timedelta
 from typing import TYPE_CHECKING, Any
 
 import pytest
@@ -10,6 +11,10 @@ import pytest
 from strix.viewer import auth
 
 
+def _iso(delta: timedelta) -> str:
+    return (datetime.now(UTC) + delta).isoformat()
+
+
 if TYPE_CHECKING:
     from pathlib import Path
 
@@ -25,7 +30,7 @@ def test_write_read_forget_roundtrip() -> None:
     assert auth.read_auth() is None
     assert auth.is_verified() is False
 
-    auth.write_auth(email="user@example.com", token="tok-123", verified_at="2026-07-20T00:00:00Z")
+    auth.write_auth(email="user@example.com", token="tok-123", verified_at=_iso(timedelta(days=30)))
 
     record = auth.read_auth()
     assert record is not None
@@ -40,6 +45,27 @@ def test_write_read_forget_roundtrip() -> None:
     auth.forget()
 
 
+def test_is_verified_enforces_expiry() -> None:
+    # An expired record still reads back, but no longer unlocks history.
+    auth.write_auth(email="a@b.com", token="t", verified_at=_iso(timedelta(hours=-1)))
+    assert auth.read_auth() is not None
+    assert auth.is_verified() is False
+
+    # A future expiry unlocks it.
+    auth.write_auth(email="a@b.com", token="t", verified_at=_iso(timedelta(hours=1)))
+    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.
+    auth.write_auth(email="a@b.com", token="t", verified_at="")
+    assert auth.is_verified() is True
+
+    # Garbage expiry is ignored rather than locking the user out.
+    auth.write_auth(email="a@b.com", token="t", verified_at="not-a-date")
+    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)