From 7b412ac3b11016eeb5150fb7d974c26a231e422a Mon Sep 17 00:00:00 2001 From: "devin-ai-integration[bot]" <158243242+devin-ai-integration[bot]@users.noreply.github.com> Date: Mon, 20 Jul 2026 23:45:11 +0000 Subject: [PATCH] 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 --- pyproject.toml | 5 + strix/interface/main.py | 7 +- strix/interface/tui/app.py | 10 +- strix/viewer/auth.py | 30 ++++- strix/viewer/cli.py | 27 +++-- strix/viewer/server.py | 123 +++++++++++++++++---- tests/test_viewer.py | 221 +++++++++++++++++++++++++++++++++++-- tests/test_viewer_auth.py | 28 ++++- 8 files changed, 401 insertions(+), 50 deletions(-) diff --git a/pyproject.toml b/pyproject.toml index 2495b815..39806bde 100644 --- a/pyproject.toml +++ b/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"] diff --git a/strix/interface/main.py b/strix/interface/main.py index a23ca427..e307d61c 100644 --- a/strix/interface/main.py +++ b/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) diff --git a/strix/interface/tui/app.py b/strix/interface/tui/app.py index efcbf6d6..5f350898 100644 --- a/strix/interface/tui/app.py +++ b/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 diff --git a/strix/viewer/auth.py b/strix/viewer/auth.py index 353b9e30..5a51c37a 100644 --- a/strix/viewer/auth.py +++ b/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: diff --git a/strix/viewer/cli.py b/strix/viewer/cli.py index eb37951a..852fbfcc 100644 --- a/strix/viewer/cli.py +++ b/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}.[/]") diff --git a/strix/viewer/server.py b/strix/viewer/server.py index cdd1da2a..4d177fab 100644 --- a/strix/viewer/server.py +++ b/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"] diff --git a/tests/test_viewer.py b/tests/test_viewer.py index 834987e2..fb629e1d 100644 --- a/tests/test_viewer.py +++ b/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("
", 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("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("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") diff --git a/tests/test_viewer_auth.py b/tests/test_viewer_auth.py index 7449c923..b4f99393 100644 --- a/tests/test_viewer_auth.py +++ b/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)