Local run viewer: email reports, run history, and the platform suite (#813)

tests/test_report_pdf.py

@@ -0,0 +1,105 @@
+"""Tests for building and encrypting the viewer PDF report."""
+
+from __future__ import annotations
+
+import json
+from io import BytesIO
+from typing import TYPE_CHECKING
+
+import pytest
+from pypdf import PdfReader
+from pypdf.errors import WrongPasswordError
+
+from strix.viewer.report_pdf import (
+    build_encrypted_report,
+    encrypt_pdf,
+    generate_password,
+    generate_report_pdf,
+)
+
+
+if TYPE_CHECKING:
+    from pathlib import Path
+
+
+def _make_run(base: Path, name: str = "sample") -> Path:
+    run_dir = base / "strix_runs" / name
+    run_dir.mkdir(parents=True)
+    record = {
+        "run_name": name,
+        "targets_info": [{"original": "https://example.com"}],
+        "scan_mode": "deep",
+        "status": "completed",
+        "start_time": "2026-01-01T00:00:00Z",
+        "end_time": "2026-01-01T01:02:03Z",
+        "scan_results": {
+            "executive_summary": "Summary with an ampersand & an <angle> bracket.",
+            "recommendations": "Patch things.",
+        },
+    }
+    (run_dir / "run.json").write_text(json.dumps(record), encoding="utf-8")
+    vulns = [
+        {
+            "title": "SQL Injection",
+            "severity": "CRITICAL",
+            "cvss": 9.8,
+            "description": "User input reaches the query.",
+            "impact": "Full database read.",
+            "technical_analysis": "Details here.",
+            "poc_description": "Send a crafted parameter.",
+            "poc_script_code": "print('exploit')",
+            "evidence": "HTTP 500 with SQL error.",
+            "remediation_steps": ["Use parameterized queries", "Validate input"],
+            "target": "https://example.com",
+            "endpoint": "/login",
+            "method": "POST",
+        },
+        {"title": "Informational note", "severity": "info"},
+    ]
+    (run_dir / "vulnerabilities.json").write_text(json.dumps(vulns), encoding="utf-8")
+    return run_dir
+
+
+def test_generate_report_pdf_has_pdf_header(tmp_path: Path) -> None:
+    run_dir = _make_run(tmp_path)
+    pdf = generate_report_pdf(run_dir)
+    assert pdf.startswith(b"%PDF-")
+    assert len(pdf) > 1000
+
+
+def test_generate_password_is_long_and_random() -> None:
+    first = generate_password()
+    second = generate_password()
+    assert len(first) >= 20
+    assert first != second
+
+
+def test_encrypt_pdf_roundtrip(tmp_path: Path) -> None:
+    run_dir = _make_run(tmp_path)
+    pdf = generate_report_pdf(run_dir)
+    password = generate_password()
+    encrypted = encrypt_pdf(pdf, password)
+
+    reader = PdfReader(BytesIO(encrypted))
+    assert reader.is_encrypted
+    assert reader.decrypt(password)
+    # A correct password unlocks the pages.
+    assert len(reader.pages) >= 1
+
+
+def test_wrong_password_is_rejected(tmp_path: Path) -> None:
+    run_dir = _make_run(tmp_path)
+    encrypted = encrypt_pdf(generate_report_pdf(run_dir), "correct-horse-battery")
+    with pytest.raises(WrongPasswordError):
+        PdfReader(BytesIO(encrypted), password="not-the-password")
+
+
+def test_build_encrypted_report(tmp_path: Path) -> None:
+    run_dir = _make_run(tmp_path, name="run-42")
+    pdf_bytes, password, filename = build_encrypted_report(run_dir)
+
+    assert filename == "strix-report-run-42.pdf"
+    assert len(password) >= 20
+    reader = PdfReader(BytesIO(pdf_bytes))
+    assert reader.is_encrypted
+    assert reader.decrypt(password)

tests/test_viewer.py

@@ -0,0 +1,503 @@
+"""Tests for the local run viewer (strix.viewer) and its path helpers."""
+
+from __future__ import annotations
+
+import json
+import os
+import urllib.error
+import urllib.request
+from typing import TYPE_CHECKING
+
+from strix.core.paths import latest_run_dir, runs_base_dir
+from strix.viewer.server import serve
+from strix.viewer.transcript import (
+    build_run_state,
+    read_report_markdown,
+    read_run_summary,
+    read_vulnerabilities,
+)
+
+
+if TYPE_CHECKING:
+    from collections.abc import Mapping
+    from pathlib import Path
+
+    import pytest
+
+
+def _make_run(base: Path, name: str, *, status: str, end_time: str | None) -> Path:
+    run_dir = base / "strix_runs" / name
+    state_dir = run_dir / ".state"
+    state_dir.mkdir(parents=True)
+    record = {"run_name": name, "status": status, "end_time": end_time}
+    (run_dir / "run.json").write_text(json.dumps(record), encoding="utf-8")
+    agents = {
+        "statuses": {"root": "completed", "child": "running"},
+        "names": {"root": "strix", "child": "recon"},
+        "parent_of": {"root": None, "child": "root"},
+    }
+    (state_dir / "agents.json").write_text(json.dumps(agents), encoding="utf-8")
+    return run_dir
+
+
+def test_latest_run_dir_none_when_no_runs(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
+    monkeypatch.chdir(tmp_path)
+    assert latest_run_dir() is None
+    assert runs_base_dir() == tmp_path / "strix_runs"
+
+
+def test_latest_run_dir_picks_newest_by_record_mtime(
+    tmp_path: Path, monkeypatch: pytest.MonkeyPatch
+) -> None:
+    monkeypatch.chdir(tmp_path)
+    older = _make_run(tmp_path, "old", status="completed", end_time="2026-01-01T00:00:00Z")
+    newer = _make_run(tmp_path, "new", status="running", end_time=None)
+    # Force a newer mtime on the second run's record.
+    os.utime(newer / "run.json", (2_000_000_000, 2_000_000_000))
+    os.utime(older / "run.json", (1_000_000_000, 1_000_000_000))
+    assert latest_run_dir() == newer
+
+
+def test_read_run_summary_finished_flag(tmp_path: Path) -> None:
+    finished = _make_run(tmp_path, "done", status="completed", end_time="2026-01-01T00:00:00Z")
+    live = _make_run(tmp_path, "live", status="running", end_time=None)
+    assert read_run_summary(finished)["finished"] is True
+    assert read_run_summary(live)["finished"] is False
+    # A terminal status without an end_time is not "finished".
+    partial = _make_run(tmp_path, "partial", status="failed", end_time=None)
+    assert read_run_summary(partial)["finished"] is False
+
+
+def test_read_missing_artifacts_return_defaults(tmp_path: Path) -> None:
+    run_dir = _make_run(tmp_path, "empty", status="running", end_time=None)
+    assert read_vulnerabilities(run_dir) == []
+    assert read_report_markdown(run_dir) == ""
+
+
+def test_build_run_state_from_agents_json(tmp_path: Path) -> None:
+    run_dir = _make_run(tmp_path, "graph", status="running", end_time=None)
+    state = build_run_state(run_dir)
+    ids = {a["id"] for a in state["agents"]}
+    assert ids == {"root", "child"}
+    child = next(a for a in state["agents"] if a["id"] == "child")
+    assert child["parent_id"] == "root"
+    assert child["name"] == "recon"
+    # No agents.db, so no message/tool events.
+    assert state["events"] == []
+
+
+def _get(url: str, *, cookie: str | None = None) -> tuple[int, str, bytes]:
+    headers = {"Cookie": cookie} if cookie else {}
+    req = urllib.request.Request(url, headers=headers)  # noqa: S310 - localhost test server
+    with urllib.request.urlopen(req) as resp:  # noqa: S310 - localhost test server
+        return resp.status, resp.headers.get("Content-Type", ""), resp.read()
+
+
+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"
+    (assets / "assets").mkdir(parents=True)
+    (assets / "index.html").write_text("<!doctype html><div id=root></div>", encoding="utf-8")
+    (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)
+    try:
+        status, ctype, body = _get(f"{url}/api/run")
+        assert status == 200
+        assert "application/json" in ctype
+        assert json.loads(body)["finished"] is True
+
+        status, _, body = _get(f"{url}/api/transcript")
+        assert {a["id"] for a in json.loads(body)["agents"]} == {"root", "child"}
+
+        # Real asset is served.
+        status, ctype, _ = _get(f"{url}/assets/app.js")
+        assert status == 200
+
+        # Unknown non-API route falls back to index.html (SPA routing).
+        status, ctype, body = _get(f"{url}/agents/root")
+        assert status == 200
+        assert b"<div id=root>" in body
+    finally:
+        httpd.shutdown()
+        httpd.server_close()
+
+
+def test_server_event_endpoint_forwards_cta(
+    tmp_path: Path, monkeypatch: pytest.MonkeyPatch
+) -> None:
+    run_dir = _make_run(tmp_path, "evt", status="running", end_time=None)
+    assets = tmp_path / "bundle"
+    assets.mkdir()
+    (assets / "index.html").write_text("x", encoding="utf-8")
+    monkeypatch.setattr("strix.viewer.server.bundle_dir", lambda: assets)
+
+    seen: list[tuple[str, str | None]] = []
+    monkeypatch.setattr(
+        "strix.telemetry.posthog.viewer_cta_clicked",
+        lambda cta, surface=None: seen.append((cta, surface)),
+    )
+
+    httpd, url, _ = serve(run_dir, open_browser=False)
+    try:
+        body = json.dumps(
+            {"event": "cta_clicked", "cta": "PR reviews", "surface": "sidebar_nav"}
+        ).encode()
+        req = urllib.request.Request(  # noqa: S310 - localhost test server
+            f"{url}/api/event", data=body, headers={"Content-Type": "application/json"}
+        )
+        with urllib.request.urlopen(req) as resp:  # noqa: S310
+            assert resp.status == 204
+        assert seen == [("PR reviews", "sidebar_nav")]
+    finally:
+        httpd.shutdown()
+        httpd.server_close()
+
+
+def test_server_event_endpoint_forwards_email_funnel(
+    tmp_path: Path, monkeypatch: pytest.MonkeyPatch
+) -> None:
+    run_dir = _make_run(tmp_path, "evt2", status="running", end_time=None)
+    assets = tmp_path / "bundle"
+    assets.mkdir()
+    (assets / "index.html").write_text("x", encoding="utf-8")
+    monkeypatch.setattr("strix.viewer.server.bundle_dir", lambda: assets)
+
+    seen: list[tuple[str, str | None]] = []
+    monkeypatch.setattr(
+        "strix.telemetry.posthog.viewer_email_event",
+        lambda step, purpose=None: seen.append((step, purpose)),
+    )
+
+    httpd, url, _ = serve(run_dir, open_browser=False)
+    try:
+        # A whitelisted funnel event is forwarded; an unknown event is ignored.
+        for payload, expected in (
+            ({"event": "email_verified", "purpose": "report"}, [("email_verified", "report")]),
+            ({"event": "not_a_real_event"}, [("email_verified", "report")]),
+        ):
+            req = urllib.request.Request(  # noqa: S310 - localhost test server
+                f"{url}/api/event",
+                data=json.dumps(payload).encode(),
+                headers={"Content-Type": "application/json"},
+            )
+            with urllib.request.urlopen(req) as resp:  # noqa: S310
+                assert resp.status == 204
+            assert seen == expected
+    finally:
+        httpd.shutdown()
+        httpd.server_close()
+
+
+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, *, 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(req) as resp:  # noqa: S310
+            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, token = serve(run_dir, open_browser=False)
+    try:
+        cookie = _session_cookie(url, token)
+        _, _, body = _get(f"{url}/api/auth/status", cookie=cookie)
+        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", cookie=cookie)
+        assert json.loads(body)["verified"] is False
+
+        # A cookie-less caller never sees the cached email or verified state.
+        verified["value"] = True
+        _, _, body = _get(f"{url}/api/auth/status")
+        assert json.loads(body) == {"verified": False, "email": None}
+    finally:
+        httpd.shutdown()
+        httpd.server_close()
+
+
+def test_auth_mutations_require_session(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
+    run_dir = _make_run(tmp_path, "authmut", status="running", end_time=None)
+    _bundle(tmp_path, monkeypatch)
+    forgotten = {"value": False}
+    monkeypatch.setattr("strix.viewer.auth.forget", lambda: forgotten.update(value=True))
+
+    httpd, url, _ = serve(run_dir, open_browser=False)
+    try:
+        for path in ("/api/auth/forget", "/api/auth/otp/start", "/api/auth/otp/verify"):
+            status, _ = _post(url, path, {"email": "a@b.com", "code": "123456"})
+            assert status == 403, path
+        assert forgotten["value"] 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_report_send_rejects_live_run(
+    tmp_path: Path, monkeypatch: pytest.MonkeyPatch
+) -> None:
+    # A running scan would only produce a partial report, so the endpoint must
+    # fail closed even for a verified, session-holding caller.
+    run_dir = _make_run(tmp_path, "live", status="running", end_time=None)
+    _bundle(tmp_path, monkeypatch)
+    monkeypatch.setattr("strix.viewer.auth.read_auth", lambda: {"email": "a@b.com", "token": "t"})
+
+    httpd, url, token = serve(run_dir, open_browser=False)
+    try:
+        status, _ = _post(
+            url, "/api/report/send", {}, cookie=_session_cookie(url, token)
+        )
+        assert status == 409
+    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, token = serve(launched, open_browser=False)
+    try:
+        # The launched run is always viewable, no verification and no cookie.
+        status, _, _ = _get(f"{url}/api/run")
+        assert status == 200
+
+        cookie = _session_cookie(url, token)
+
+        # A different run needs the session capability first: a cookie-less
+        # caller is forbidden even once the machine is verified.
+        verified["value"] = True
+        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()
+
+
+def test_runs_list_requires_session_and_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)
+
+    monkeypatch.setattr("strix.viewer.auth.is_verified", lambda: True)
+
+    def _runs(cookie: str | None) -> dict[str, object]:
+        headers = {"Cookie": cookie} if cookie else {}
+        req = urllib.request.Request(f"{url}/api/runs", headers=headers)  # noqa: S310
+        with urllib.request.urlopen(req) as resp:  # noqa: S310 - localhost test server
+            return dict(json.loads(resp.read()))
+
+    httpd, url, token = serve(launched, open_browser=False)
+    try:
+        # A cookie-less caller (even with the machine verified) only sees the
+        # teaser count, never the run entries.
+        payload = _runs(None)
+        assert payload["locked"] is True
+        assert payload["count"] == 2
+        assert payload["runs"] == []
+
+        # With the session cookie and verification, the entries unlock.
+        payload = _runs(_session_cookie(url, token))
+        assert payload["locked"] is False
+        assert {r["name"] for r in payload["runs"]} == {"launched", "other"}  # type: ignore[attr-defined]
+    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")
+
+    assets = tmp_path / "bundle"
+    assets.mkdir()
+    (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)
+    try:
+        # A traversal target must never leak the file; it falls back to index.html.
+        _, _, body = _get(f"{url}/..%2f..%2fsecret.txt")
+        assert b"top secret" not in body
+    finally:
+        httpd.shutdown()
+        httpd.server_close()

tests/test_viewer_auth.py

@@ -0,0 +1,167 @@
+"""Tests for viewer auth state and the relay client mapping."""
+
+from __future__ import annotations
+
+import stat
+from datetime import UTC, datetime, timedelta
+from typing import TYPE_CHECKING, Any
+
+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
+
+
+@pytest.fixture(autouse=True)
+def _tmp_auth(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> Path:
+    home = tmp_path / "home"
+    monkeypatch.setattr(auth, "AUTH_PATH", home / ".strix" / "viewer-auth.json")
+    return auth.AUTH_PATH
+
+
+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=_iso(timedelta(days=30)))
+
+    record = auth.read_auth()
+    assert record is not None
+    assert record["email"] == "user@example.com"
+    assert record["token"] == "tok-123"
+    assert auth.is_verified() is True
+
+    auth.forget()
+    assert auth.read_auth() is None
+    assert auth.is_verified() is False
+    # Forget is a no-op when the file is already gone.
+    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_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.read_auth() is not None
+    assert auth.is_verified() is False
+
+    # 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 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)
+    assert mode == 0o600
+
+
+def test_read_auth_rejects_incomplete_record() -> None:
+    auth.AUTH_PATH.parent.mkdir(parents=True, exist_ok=True)
+    auth.AUTH_PATH.write_text('{"email": "a@b.com"}', encoding="utf-8")
+    assert auth.read_auth() is None
+    assert auth.is_verified() is False
+
+
+def _stub_post(monkeypatch: pytest.MonkeyPatch, status: int, body: dict[str, Any]) -> None:
+    def fake(path: str, payload: dict[str, Any], *, timeout: int) -> tuple[int, dict[str, Any]]:
+        return status, body
+
+    monkeypatch.setattr(auth, "_post_json", fake)
+
+
+def test_otp_start_maps_errors(monkeypatch: pytest.MonkeyPatch) -> None:
+    _stub_post(monkeypatch, 200, {"ok": True})
+    auth.otp_start("a@b.com")  # no raise
+
+    _stub_post(monkeypatch, 429, {"error": "rate_limited"})
+    with pytest.raises(auth.RelayError) as exc:
+        auth.otp_start("a@b.com")
+    assert exc.value.code == "rate_limited"
+
+    _stub_post(monkeypatch, 400, {})
+    with pytest.raises(auth.RelayError) as exc:
+        auth.otp_start("bad")
+    assert exc.value.code == "invalid_email"
+
+
+def test_otp_verify_success_and_invalid(monkeypatch: pytest.MonkeyPatch) -> None:
+    expires = _iso(timedelta(hours=1))
+    _stub_post(monkeypatch, 200, {"token": "t", "email": "a@b.com", "expires_at": expires})
+    result = auth.otp_verify("a@b.com", "123456")
+    assert result["token"] == "t"
+
+    _stub_post(monkeypatch, 403, {"error": "invalid_code"})
+    with pytest.raises(auth.RelayError) as exc:
+        auth.otp_verify("a@b.com", "000000")
+    assert exc.value.code == "invalid_code"
+
+
+def test_otp_verify_rejects_token_without_usable_expiry(monkeypatch: pytest.MonkeyPatch) -> None:
+    # A 200 with a token but no valid expiry must not be reported as success,
+    # otherwise the caller would store a record that immediately reads unverified.
+    for expires in (None, "", "later"):
+        _stub_post(monkeypatch, 200, {"token": "t", "email": "a@b.com", "expires_at": expires})
+        with pytest.raises(auth.RelayError) as exc:
+            auth.otp_verify("a@b.com", "123456")
+        assert exc.value.code == "unavailable"
+
+
+def test_report_send_never_includes_password(monkeypatch: pytest.MonkeyPatch) -> None:
+    captured: dict[str, Any] = {}
+
+    def fake(path: str, payload: dict[str, Any], *, timeout: int) -> tuple[int, dict[str, Any]]:
+        captured["payload"] = payload
+        return 200, {"ok": True}
+
+    monkeypatch.setattr(auth, "_post_json", fake)
+    auth.report_send("tok", b"%PDF-fake", "strix-report-x.pdf", "x", "https://example.com")
+
+    payload = captured["payload"]
+    assert set(payload) == {"token", "pdf_base64", "filename", "run_name", "target"}
+    # The password is generated locally and must never appear in the relay body.
+    assert "password" not in payload
+    assert all("password" not in str(k).lower() for k in payload)
+
+
+def test_report_send_reverify_on_401(monkeypatch: pytest.MonkeyPatch) -> None:
+    _stub_post(monkeypatch, 401, {"error": "invalid_token"})
+    with pytest.raises(auth.RelayError) as exc:
+        auth.report_send("tok", b"x", "f.pdf", "r", "t")
+    assert exc.value.code == "reverify"

tests/test_viewer_runs_gating.py

@@ -0,0 +1,89 @@
+"""Tests for the /api/runs gating and the ?run= resolver (pure functions)."""
+
+from __future__ import annotations
+
+import json
+from typing import TYPE_CHECKING
+
+from strix.viewer.server import build_runs_payload, resolve_run_dir
+
+
+if TYPE_CHECKING:
+    from pathlib import Path
+
+
+def _make_run(base: Path, name: str, *, severity: str = "high") -> Path:
+    run_dir = base / "strix_runs" / name
+    run_dir.mkdir(parents=True)
+    record = {
+        "run_name": name,
+        "targets_info": [{"original": f"https://{name}.example.com"}],
+        "scan_mode": "deep",
+        "status": "completed",
+        "start_time": "2026-01-01T00:00:00Z",
+        "end_time": "2026-01-01T00:10:00Z",
+    }
+    (run_dir / "run.json").write_text(json.dumps(record), encoding="utf-8")
+    (run_dir / "vulnerabilities.json").write_text(
+        json.dumps([{"title": "v", "severity": severity}]), encoding="utf-8"
+    )
+    return run_dir
+
+
+def test_runs_payload_locked_when_unverified(tmp_path: Path) -> None:
+    base = tmp_path / "strix_runs"
+    _make_run(tmp_path, "alpha")
+    _make_run(tmp_path, "beta")
+
+    payload = build_runs_payload(base, verified=False)
+    assert payload["locked"] is True
+    assert payload["count"] == 2
+    assert payload["runs"] == []
+
+
+def test_runs_payload_lists_when_verified(tmp_path: Path) -> None:
+    base = tmp_path / "strix_runs"
+    _make_run(tmp_path, "alpha", severity="critical")
+    _make_run(tmp_path, "beta", severity="info")
+
+    payload = build_runs_payload(base, verified=True)
+    assert payload["locked"] is False
+    assert payload["count"] == 2
+    assert len(payload["runs"]) == 2
+    entry = next(r for r in payload["runs"] if r["name"] == "alpha")
+    assert entry["target"] == "https://alpha.example.com"
+    assert entry["severity_counts"]["critical"] == 1
+    # "info" folds into low, matching the SPA's bucketing.
+    beta = next(r for r in payload["runs"] if r["name"] == "beta")
+    assert beta["severity_counts"]["low"] == 1
+
+
+def test_runs_payload_empty_base(tmp_path: Path) -> None:
+    payload = build_runs_payload(tmp_path / "strix_runs", verified=True)
+    assert payload == {"locked": False, "count": 0, "runs": []}
+
+
+def test_resolve_run_dir_defaults_when_absent(tmp_path: Path) -> None:
+    base = tmp_path / "strix_runs"
+    default = _make_run(tmp_path, "alpha")
+    assert resolve_run_dir(base, None, default) == default
+    assert resolve_run_dir(base, "", default) == default
+
+
+def test_resolve_run_dir_valid_named_run(tmp_path: Path) -> None:
+    base = tmp_path / "strix_runs"
+    default = _make_run(tmp_path, "alpha")
+    other = _make_run(tmp_path, "beta")
+    assert resolve_run_dir(base, "beta", default) == other
+
+
+def test_resolve_run_dir_rejects_unknown_and_traversal(tmp_path: Path) -> None:
+    base = tmp_path / "strix_runs"
+    default = _make_run(tmp_path, "alpha")
+    secret = tmp_path / "secret"
+    secret.mkdir()
+    (secret / "run.json").write_text("{}", encoding="utf-8")
+
+    assert resolve_run_dir(base, "nope", default) is None
+    assert resolve_run_dir(base, "../secret", default) is None
+    assert resolve_run_dir(base, "../../etc", default) is None