Add tests for PDF reports, viewer auth, and run history gating

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_auth.py

@@ -0,0 +1,110 @@
+"""Tests for viewer auth state and the relay client mapping."""
+
+from __future__ import annotations
+
+import stat
+from typing import TYPE_CHECKING, Any
+
+import pytest
+
+from strix.viewer import auth
+
+
+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="2026-07-20T00:00:00Z")
+
+    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_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:
+    _stub_post(monkeypatch, 200, {"token": "t", "email": "a@b.com", "expires_at": "later"})
+    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_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