Harden local viewer authorization (session capability, history gating, expiry) (#818)

* Harden viewer authz: session capability, history gating, expiry

* Set session cookie whenever index.html is served

* Gate session capability behind bootstrap token; sync auth status with expiry

---------

Co-authored-by: yoni <yoni@usestrix.com>

pyproject.toml

@@ -214,6 +214,11 @@ 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 telemetry import to avoid importing PostHog before the viewer starts.
+"strix/viewer/cli.py" = ["PLC0415"]
 # Lazy imports inside functions to avoid circular dependency with
 # strix.telemetry / strix.report.dedupe / cvss.
 "strix/tools/notes/tools.py" = ["PLC0415", "TC002"]

strix/interface/main.py

@@ -981,11 +981,14 @@ def main() -> None:
     viewer_httpd = None
     web_url = None
     if not args.non_interactive and sys.stdout.isatty():
-        from strix.viewer.server import bundle_is_built, serve
+        from strix.viewer.server import authorized_url, bundle_is_built, serve
 
         if bundle_is_built():
             try:
-                viewer_httpd, web_url = serve(results_path, open_browser=False)
+                viewer_httpd, base_url, token = serve(results_path, open_browser=False)
+                # The completion panel's "View in web" link must authorize the
+                # browser, so hand it the tokened URL rather than the bare host.
+                web_url = authorized_url(base_url, token)
                 posthog.viewer_opened(source="post_scan", live=False)
             except Exception:
                 logger.debug("could not start local viewer", exc_info=True)

strix/interface/tui/app.py

@@ -1838,7 +1838,7 @@ class StrixTUIApp(App):  # type: ignore[misc]
                 webbrowser.open(self._viewer_url)
             return
         try:
-            from strix.viewer.server import bundle_is_built, serve
+            from strix.viewer.server import authorized_url, bundle_is_built, serve
 
             if not bundle_is_built():
                 self._set_viewer_cta("[#eab308]Viewer UI not built[/]")
@@ -1856,14 +1856,16 @@ class StrixTUIApp(App):  # type: ignore[misc]
                     message=message,
                 )
 
-            httpd, url = serve(run_dir, open_browser=True, steer_handler=_viewer_steer)
+            httpd, url, token = serve(run_dir, open_browser=True, steer_handler=_viewer_steer)
         except Exception:
             logger.debug("failed to start local viewer", exc_info=True)
             self._set_viewer_cta("[red]Viewer failed to start[/]")
             return
         self._viewer_httpd = httpd
-        self._viewer_url = url
-        self._set_viewer_cta(self._viewer_cta_markup(url))
+        # Store the tokened URL so reopening the CTA re-authorizes the browser
+        # (this viewer carries a steer handler, so the session is required).
+        self._viewer_url = authorized_url(url, token)
+        self._set_viewer_cta(self._viewer_cta_markup(self._viewer_url))
 
         with contextlib.suppress(Exception):
             from strix.telemetry import posthog

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/cli.py

@@ -16,7 +16,7 @@ from strix.core.paths import (
     run_record_path,
     runs_base_dir,
 )
-from strix.viewer.server import bundle_is_built, serve
+from strix.viewer.server import authorized_url, bundle_is_built, serve
 from strix.viewer.transcript import read_run_summary
 
 
@@ -64,25 +64,30 @@ def run_view(argv: list[str]) -> None:
 
     run_dir = _resolve_run_dir(args.run, console)
 
-    httpd, url = serve(
+    httpd, url, token = serve(
         run_dir,
         host=args.host,
         port=args.port,
         open_browser=not args.no_open,
     )
+    # The tokened URL is what authorizes the browser (steering, report sending,
+    # history). Print it rather than the bare URL so the operator -- and only
+    # the operator -- can open or share an authorized link.
+    open_url = authorized_url(url, token)
 
     run_name = run_dir.name
     summary = read_run_summary(run_dir)
     live = not summary.get("finished", False)
 
-    from strix.telemetry import posthog  # noqa: PLC0415
+    from strix.telemetry import posthog
 
     posthog.viewer_opened(source="cli", live=live)
 
     state_label = "[#eab308]live[/]" if live else "[#22c55e]finished[/]"
     console.print()
-    console.print(f"Serving [bold white]{run_name}[/] ({state_label}) at [#60a5fa]{url}[/]")
-    console.print("[dim]Press Ctrl-C to stop the viewer.[/]")
+    console.print(f"Serving [bold white]{run_name}[/] ({state_label}) at [#60a5fa]{open_url}[/]")
+    console.print("[dim]This link authorizes the browser; anyone you share it with can steer[/]")
+    console.print("[dim]a live scan and browse history. Press Ctrl-C to stop the viewer.[/]")
     console.print()
 
     try:
@@ -110,10 +115,14 @@ def _resolve_run_dir(run: str | None, console: Console) -> Path:
 
 def _fail_no_run(console: Console, *, requested: str | None) -> NoReturn:
     base = runs_base_dir()
-    available = sorted(
-        (child.name for child in base.iterdir() if run_record_path(child).is_file()),
-        reverse=True,
-    ) if base.is_dir() else []
+    available = (
+        sorted(
+            (child.name for child in base.iterdir() if run_record_path(child).is_file()),
+            reverse=True,
+        )
+        if base.is_dir()
+        else []
+    )
 
     if requested:
         console.print(f"[bold red]No run named '{requested}' under ./{RUNS_DIR_NAME}.[/]")

strix/viewer/server.py

@@ -17,13 +17,14 @@ from __future__ import annotations
 import json
 import logging
 import mimetypes
+import secrets
 import threading
 import webbrowser
 from http import HTTPStatus
 from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
 from pathlib import Path
 from typing import TYPE_CHECKING, Any
-from urllib.parse import parse_qs, unquote, urlsplit
+from urllib.parse import parse_qs, unquote, urlencode, urlsplit
 
 from strix.core.paths import run_record_path
 from strix.viewer import auth
@@ -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,14 @@ 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. It is minted here, printed/opened
+        # for the operator who started the server (see ``authorized_url``), and
+        # exchanged for a session cookie only when presented on the initial page
+        # load. It is the request-level authorization the review asked for:
+        # reachability of the port (e.g. when bound with ``--host``) is not
+        # enough to steer a live scan, trigger a report, or browse history --
+        # the token is never handed to a caller who merely reaches ``/``.
+        self.session_token = secrets.token_urlsafe(32)
 
 
 def _make_handler(state: _ViewerState) -> type[BaseHTTPRequestHandler]:
@@ -138,7 +151,7 @@ def _make_handler(state: _ViewerState) -> type[BaseHTTPRequestHandler]:
                 if path.startswith("/api/"):
                     self._handle_api(path, parse_qs(parts.query))
                 else:
-                    self._handle_static(path)
+                    self._handle_static(path, parse_qs(parts.query))
             except BrokenPipeError:
                 # The browser closed the connection mid-response (e.g. it
                 # navigated away between polls). Not an error.
@@ -195,20 +208,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,16 +231,17 @@ 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":
+                # Report verification through is_verified() so an expired record
+                # is advertised as unverified -- otherwise the SPA would suppress
+                # re-verification while history stays locked, stranding the user.
                 record = auth.read_auth()
                 self._send_json(
                     HTTPStatus.OK,
                     {
-                        "verified": record is not None,
+                        "verified": auth.is_verified(),
                         "email": record.get("email") if record else None,
                     },
                 )
@@ -244,6 +254,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 +310,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 +323,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 +346,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 +364,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,12 +386,41 @@ 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 _handle_static(self, path: str) -> None:
+        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 _token_presented(self, query: dict[str, list[str]]) -> bool:
+            """True when the request carries the correct bootstrap token.
+
+            The token reaches the operator's browser through the URL printed /
+            opened by the process that started the server, a channel an
+            arbitrary network caller on an exposed port cannot observe.
+            """
+            supplied = (query.get("token") or [""])[0]
+            return bool(supplied) and secrets.compare_digest(supplied, state.session_token)
+
+        def _handle_static(self, path: str, query: dict[str, list[str]]) -> None:
             target = self._resolve_asset(path)
             if target is None:
                 # SPA fallback: unknown non-asset routes render index.html so
                 # client-side deep links work.
                 target = state.assets_dir / "index.html"
+            is_index = target.name == "index.html"
             if not target.is_file():
                 self._send_json(HTTPStatus.NOT_FOUND, {"error": "not found"})
                 return
@@ -378,6 +429,16 @@ 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 and self._token_presented(query):
+                # Exchange the bootstrap token for the per-process session
+                # capability. Issued only when the correct token is presented,
+                # so a caller who merely reaches ``/`` never obtains it.
+                # 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)
 
@@ -404,6 +465,17 @@ def _make_handler(state: _ViewerState) -> type[BaseHTTPRequestHandler]:
     return ViewerHandler
 
 
+def authorized_url(base_url: str, token: str) -> str:
+    """URL that bootstraps the viewer session for the operator.
+
+    Presenting ``token`` on the initial page load is what mints the session
+    cookie, so this URL is printed / opened only for the operator who started
+    the server. Sharing it (rather than the bare ``base_url``) is what lets a
+    trusted remote user authorize when the viewer is exposed with ``--host``.
+    """
+    return f"{base_url}/?{urlencode({'token': token})}"
+
+
 def serve(
     run_dir: Path,
     *,
@@ -411,8 +483,11 @@ def serve(
     port: int = 0,
     open_browser: bool = True,
     steer_handler: Callable[[str, str], bool] | None = None,
-) -> tuple[ThreadingHTTPServer, str]:
-    """Start the viewer server on a background thread and return (server, url).
+) -> tuple[ThreadingHTTPServer, str, str]:
+    """Start the viewer server on a background thread; return (server, url, token).
+
+    ``url`` is the bare base; pass it through ``authorized_url(url, token)`` to
+    build the operator link that authorizes the browser.
 
     Binds an ephemeral port by default. If a fixed ``port`` is requested but in
     use, falls back to an ephemeral port. Reused by both the ``strix view``
@@ -442,9 +517,9 @@ def serve(
     thread.start()
 
     if open_browser:
-        _open_browser(url)
+        _open_browser(authorized_url(url, state.session_token))
 
-    return httpd, url
+    return httpd, url, state.session_token
 
 
 def _open_browser(url: str) -> None:
@@ -454,4 +529,4 @@ def _open_browser(url: str) -> None:
         logger.debug("could not open browser for %s", url, exc_info=True)
 
 
-__all__ = ["bundle_dir", "bundle_is_built", "serve"]
+__all__ = ["authorized_url", "bundle_dir", "bundle_is_built", "serve"]

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"
@@ -100,7 +100,7 @@ def test_server_serves_api_and_static(
     (assets / "assets" / "app.js").write_text("console.log(1)", encoding="utf-8")
     monkeypatch.setattr("strix.viewer.server.bundle_dir", lambda: assets)
 
-    httpd, url = serve(run_dir, open_browser=False)
+    httpd, url, _ = serve(run_dir, open_browser=False)
     try:
         status, ctype, body = _get(f"{url}/api/run")
         assert status == 200
@@ -138,7 +138,7 @@ def test_server_event_endpoint_forwards_cta(
         lambda cta, surface=None: seen.append((cta, surface)),
     )
 
-    httpd, url = serve(run_dir, open_browser=False)
+    httpd, url, _ = serve(run_dir, open_browser=False)
     try:
         body = json.dumps(
             {"event": "cta_clicked", "cta": "PR reviews", "surface": "sidebar_nav"}
@@ -169,7 +169,7 @@ def test_server_event_endpoint_forwards_email_funnel(
         lambda step, purpose=None: seen.append((step, purpose)),
     )
 
-    httpd, url = serve(run_dir, open_browser=False)
+    httpd, url, _ = serve(run_dir, open_browser=False)
     try:
         # A whitelisted funnel event is forwarded; an unknown event is ignored.
         for payload, expected in (
@@ -189,9 +189,214 @@ 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, token: str) -> str:
+    """Bootstrap a session via the tokened URL and return its ``name=value`` cookie."""
+    bootstrap = f"{url}/?token={token}"
+    with urllib.request.urlopen(bootstrap) 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_capability_issued_only_for_tokened_bootstrap(
     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, token = serve(run_dir, open_browser=False)
+    try:
+        # A bare index load -- all a reachable client can do -- hands out nothing.
+        with urllib.request.urlopen(url + "/") as resp:  # noqa: S310
+            assert resp.headers.get("Set-Cookie") is None
+
+        # A wrong token is likewise refused the capability.
+        with urllib.request.urlopen(f"{url}/?token=wrong") as resp:  # noqa: S310
+            assert resp.headers.get("Set-Cookie") is None
+
+        # Only the correct bootstrap token mints the session cookie.
+        with urllib.request.urlopen(f"{url}/?token={token}") as resp:  # noqa: S310
+            cookie = str(resp.headers.get("Set-Cookie", ""))
+        assert "strix_viewer_session=" in cookie
+        assert "HttpOnly" in cookie and "SameSite=Strict" in cookie
+
+        # Static assets never carry it.
+        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_unauthorized_client_cannot_acquire_capability(
+    tmp_path: Path, monkeypatch: pytest.MonkeyPatch
+) -> None:
+    run_dir = _make_run(tmp_path, "exposed", 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:
+        # A direct network client can reach the page but is handed no capability,
+        # so replaying an empty/guessed cookie cannot steer a live scan.
+        with urllib.request.urlopen(url + "/") as resp:  # noqa: S310
+            assert resp.headers.get("Set-Cookie") is None
+        status, _ = _post(
+            url,
+            "/api/agents/steer",
+            {"agent_id": "root", "message": "pwn"},
+            cookie="strix_viewer_session=",
+        )
+        assert status == 403
+        assert delivered == []
+    finally:
+        httpd.shutdown()
+        httpd.server_close()
+
+
+def test_auth_status_reflects_expiry(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
+    run_dir = _make_run(tmp_path, "status", status="running", end_time=None)
+    _bundle(tmp_path, monkeypatch)
+    monkeypatch.setattr("strix.viewer.auth.read_auth", lambda: {"email": "a@b.com", "token": "t"})
+    verified = {"value": True}
+    monkeypatch.setattr("strix.viewer.auth.is_verified", lambda: verified["value"])
+
+    httpd, url, _ = serve(run_dir, open_browser=False)
+    try:
+        _, _, body = _get(f"{url}/api/auth/status")
+        assert json.loads(body) == {"verified": True, "email": "a@b.com"}
+
+        # Once expired, status must advertise unverified so the SPA re-prompts.
+        verified["value"] = False
+        _, _, body = _get(f"{url}/api/auth/status")
+        assert json.loads(body)["verified"] is False
+    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, token = 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, token))
+        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, token = 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, token)
+        )
+        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")
@@ -201,7 +406,7 @@ def test_server_rejects_path_traversal(
     (assets / "index.html").write_text("<!doctype html>index", encoding="utf-8")
     monkeypatch.setattr("strix.viewer.server.bundle_dir", lambda: assets)
 
-    httpd, url = serve(run_dir, open_browser=False)
+    httpd, url, _ = serve(run_dir, open_browser=False)
     try:
         # A traversal target must never leak the file; it falls back to index.html.
         _, _, body = _get(f"{url}/..%2f..%2fsecret.txt")

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)