diff --git a/strix/viewer/auth.py b/strix/viewer/auth.py
new file mode 100644
index 00000000..e830a478
--- /dev/null
+++ b/strix/viewer/auth.py
@@ -0,0 +1,191 @@
+"""Viewer email verification state and the relay client.
+
+The local viewer proxies email verification and encrypted-report delivery to
+the Strix relay (``STRIX_APP_URL``). The browser never talks to the relay
+directly, and the report password generated locally is never sent to it.
+
+State lives in ``~/.strix/viewer-auth.json`` (0600). ``is_verified`` is a local
+flag that unlocks browsing the run history list; the relay still enforces token
+expiry when a report is actually sent.
+"""
+
+from __future__ import annotations
+
+import base64
+import contextlib
+import json
+import logging
+import urllib.error
+import urllib.request
+from pathlib import Path
+from typing import Any
+
+from strix.config.loader import load_settings
+
+
+logger = logging.getLogger(__name__)
+
+AUTH_PATH = Path.home() / ".strix" / "viewer-auth.json"
+
+_OTP_TIMEOUT = 15
+_SEND_TIMEOUT = 30
+
+
+class RelayError(Exception):
+ """A relay call failed. ``code`` is a stable, machine-readable reason."""
+
+ def __init__(self, code: str, message: str | None = None) -> None:
+ self.code = code
+ super().__init__(message or code)
+
+
+# --- local state ------------------------------------------------------------
+
+
+def read_auth() -> dict[str, Any] | None:
+ """Return the stored ``{email, token, verified_at}`` record, or None."""
+ try:
+ data = json.loads(AUTH_PATH.read_text(encoding="utf-8"))
+ except (OSError, json.JSONDecodeError):
+ return None
+ if not isinstance(data, dict):
+ return None
+ email = data.get("email")
+ token = data.get("token")
+ if not isinstance(email, str) or not email or not isinstance(token, str) or not token:
+ return None
+ return data
+
+
+def is_verified() -> bool:
+ """True when a usable email + token record exists locally."""
+ return read_auth() is not None
+
+
+def write_auth(email: str, token: str, verified_at: str) -> None:
+ """Atomically persist the auth record with 0600 permissions."""
+ AUTH_PATH.parent.mkdir(parents=True, exist_ok=True)
+ payload = json.dumps({"email": email, "token": token, "verified_at": verified_at})
+ tmp = AUTH_PATH.with_suffix(".json.tmp")
+ tmp.write_text(payload, encoding="utf-8")
+ with contextlib.suppress(OSError):
+ tmp.chmod(0o600)
+ tmp.replace(AUTH_PATH)
+ with contextlib.suppress(OSError):
+ AUTH_PATH.chmod(0o600)
+
+
+def forget() -> None:
+ """Delete the stored auth record. No-op if it is absent."""
+ with contextlib.suppress(OSError):
+ AUTH_PATH.unlink()
+
+
+# --- relay client -----------------------------------------------------------
+
+
+def _app_url() -> str:
+ return load_settings().viewer.app_url.rstrip("/")
+
+
+def _post_json(path: str, payload: dict[str, Any], *, timeout: int) -> tuple[int, dict[str, Any]]:
+ """POST JSON to the relay. Returns (status, parsed body).
+
+ Raises RelayError("unavailable") for network/transport failures. HTTP
+ error responses (4xx/5xx) are returned as (status, body) for the caller to
+ map, not raised.
+ """
+ url = f"{_app_url()}{path}"
+ body = json.dumps(payload).encode("utf-8")
+ request = urllib.request.Request( # noqa: S310 - fixed https relay URL
+ url,
+ data=body,
+ headers={"Content-Type": "application/json", "Accept": "application/json"},
+ method="POST",
+ )
+ try:
+ with urllib.request.urlopen(request, timeout=timeout) as response: # noqa: S310
+ return response.status, _parse_body(response.read())
+ except urllib.error.HTTPError as exc:
+ return exc.code, _parse_body(exc.read())
+ except (urllib.error.URLError, TimeoutError, OSError) as exc:
+ logger.warning("relay request to %s failed: %s", path, exc)
+ raise RelayError("unavailable") from exc
+
+
+def _parse_body(raw: bytes) -> dict[str, Any]:
+ try:
+ data = json.loads(raw or b"{}")
+ except json.JSONDecodeError:
+ return {}
+ return data if isinstance(data, dict) else {}
+
+
+def otp_start(email: str) -> None:
+ """Ask the relay to email a verification code. Raises RelayError on failure."""
+ status, _ = _post_json("/api/oss/otp/start", {"email": email}, timeout=_OTP_TIMEOUT)
+ if status == 200:
+ return
+ if status == 429:
+ raise RelayError("rate_limited")
+ if status == 400:
+ raise RelayError("invalid_email")
+ raise RelayError("unavailable")
+
+
+def otp_verify(email: str, code: str) -> dict[str, Any]:
+ """Verify a code. Returns ``{token, email, expires_at}`` or raises RelayError."""
+ status, data = _post_json(
+ "/api/oss/otp/verify",
+ {"email": email, "code": code},
+ timeout=_OTP_TIMEOUT,
+ )
+ if status == 200 and isinstance(data.get("token"), str):
+ return data
+ if status == 403:
+ raise RelayError("invalid_code")
+ raise RelayError("unavailable")
+
+
+def report_send(
+ token: str,
+ pdf_bytes: bytes,
+ filename: str,
+ run_name: str,
+ target: str,
+) -> None:
+ """Forward the encrypted PDF to the relay for delivery.
+
+ The report password is NEVER part of this payload; only the encrypted PDF
+ bytes travel to the relay.
+ """
+ payload = {
+ "token": token,
+ "pdf_base64": base64.b64encode(pdf_bytes).decode("ascii"),
+ "filename": filename,
+ "run_name": run_name,
+ "target": target,
+ }
+ status, _ = _post_json("/api/oss/report/send", payload, timeout=_SEND_TIMEOUT)
+ if status == 200:
+ return
+ if status == 401:
+ raise RelayError("reverify")
+ if status == 413:
+ raise RelayError("too_large")
+ if status == 403:
+ raise RelayError("forbidden")
+ raise RelayError("unavailable")
+
+
+__all__ = [
+ "AUTH_PATH",
+ "RelayError",
+ "forget",
+ "is_verified",
+ "otp_start",
+ "otp_verify",
+ "read_auth",
+ "report_send",
+ "write_auth",
+]
diff --git a/strix/viewer/report_pdf.py b/strix/viewer/report_pdf.py
new file mode 100644
index 00000000..5dc6a623
--- /dev/null
+++ b/strix/viewer/report_pdf.py
@@ -0,0 +1,308 @@
+"""Build and encrypt a branded PDF report for a run.
+
+The PDF carries FULL finding detail, including proof-of-concept scripts, so it
+is encrypted end to end with AES-256. The password is generated locally with a
+CSPRNG, shown only to the local browser, and never leaves the machine except in
+the user's own hands. Strix cannot read the delivered report.
+"""
+
+from __future__ import annotations
+
+import html
+import secrets
+from datetime import datetime
+from io import BytesIO
+from typing import TYPE_CHECKING, Any
+
+from pypdf import PdfReader, PdfWriter
+from reportlab.lib import colors
+from reportlab.lib.enums import TA_LEFT
+from reportlab.lib.pagesizes import letter
+from reportlab.lib.styles import ParagraphStyle, getSampleStyleSheet
+from reportlab.lib.units import inch
+from reportlab.platypus import (
+ HRFlowable,
+ PageBreak,
+ Paragraph,
+ SimpleDocTemplate,
+ Spacer,
+)
+
+from strix.viewer.transcript import (
+ primary_target,
+ read_run_summary,
+ read_vulnerabilities,
+ severity_counts,
+)
+
+
+if TYPE_CHECKING:
+ from pathlib import Path
+
+ from reportlab.platypus import Flowable
+
+
+_BRAND = colors.HexColor("#6d28d9")
+_INK = colors.HexColor("#111827")
+_MUTED = colors.HexColor("#6b7280")
+
+_SEVERITY_COLORS = {
+ "critical": colors.HexColor("#b91c1c"),
+ "high": colors.HexColor("#ea580c"),
+ "medium": colors.HexColor("#ca8a04"),
+ "low": colors.HexColor("#2563eb"),
+}
+
+
+def _esc(value: Any) -> str:
+ """Escape a value for reportlab's Paragraph markup."""
+ return html.escape(str(value)).replace("\n", "
")
+
+
+def _styles() -> dict[str, ParagraphStyle]:
+ base = getSampleStyleSheet()
+ styles: dict[str, ParagraphStyle] = {}
+ styles["title"] = ParagraphStyle(
+ "StrixTitle",
+ parent=base["Title"],
+ textColor=_BRAND,
+ fontSize=26,
+ leading=30,
+ alignment=TA_LEFT,
+ )
+ styles["subtitle"] = ParagraphStyle(
+ "StrixSubtitle",
+ parent=base["Normal"],
+ textColor=_MUTED,
+ fontSize=11,
+ leading=15,
+ )
+ styles["h2"] = ParagraphStyle(
+ "StrixH2",
+ parent=base["Heading2"],
+ textColor=_INK,
+ fontSize=16,
+ leading=20,
+ spaceBefore=16,
+ spaceAfter=6,
+ )
+ styles["finding"] = ParagraphStyle(
+ "StrixFinding",
+ parent=base["Heading3"],
+ textColor=_INK,
+ fontSize=13,
+ leading=17,
+ spaceBefore=14,
+ spaceAfter=2,
+ )
+ styles["label"] = ParagraphStyle(
+ "StrixLabel",
+ parent=base["Normal"],
+ textColor=_BRAND,
+ fontSize=9,
+ leading=12,
+ spaceBefore=8,
+ )
+ styles["body"] = ParagraphStyle(
+ "StrixBody",
+ parent=base["Normal"],
+ textColor=_INK,
+ fontSize=10,
+ leading=14,
+ )
+ styles["code"] = ParagraphStyle(
+ "StrixCode",
+ parent=base["Code"],
+ textColor=_INK,
+ backColor=colors.HexColor("#f3f4f6"),
+ fontSize=8,
+ leading=11,
+ borderPadding=6,
+ leftIndent=6,
+ )
+ return styles
+
+
+def _parse_time(raw: Any) -> datetime | None:
+ if not isinstance(raw, str) or not raw:
+ return None
+ text = raw.strip().replace(" UTC", "Z").replace(" ", "T")
+ if text.endswith("Z"):
+ text = text[:-1] + "+00:00"
+ try:
+ return datetime.fromisoformat(text)
+ except ValueError:
+ return None
+
+
+def _duration(start: Any, end: Any) -> str:
+ start_dt = _parse_time(start)
+ end_dt = _parse_time(end)
+ if not start_dt or not end_dt:
+ return "n/a"
+ seconds = int((end_dt - start_dt).total_seconds())
+ if seconds < 0:
+ return "n/a"
+ hours, remainder = divmod(seconds, 3600)
+ minutes, secs = divmod(remainder, 60)
+ if hours:
+ return f"{hours}h {minutes}m {secs}s"
+ if minutes:
+ return f"{minutes}m {secs}s"
+ return f"{secs}s"
+
+
+def _field_block(
+ styles: dict[str, ParagraphStyle], label: str, value: Any, *, code: bool = False
+) -> list[Flowable]:
+ if value is None or (isinstance(value, str) and not value.strip()):
+ return []
+ flowables: list[Flowable] = [Paragraph(label.upper(), styles["label"])]
+ flowables.append(Paragraph(_esc(value), styles["code"] if code else styles["body"]))
+ return flowables
+
+
+def generate_report_pdf(run_dir: Path) -> bytes:
+ """Render a branded, full-detail PDF report for the run at ``run_dir``."""
+ record = read_run_summary(run_dir)
+ vulns = read_vulnerabilities(run_dir)
+ counts = severity_counts(vulns)
+
+ styles = _styles()
+ buffer = BytesIO()
+ doc = SimpleDocTemplate(
+ buffer,
+ pagesize=letter,
+ title="Strix Security Report",
+ author="Strix",
+ leftMargin=0.9 * inch,
+ rightMargin=0.9 * inch,
+ topMargin=0.9 * inch,
+ bottomMargin=0.9 * inch,
+ )
+
+ story: list[Flowable] = []
+ story.append(Paragraph("Strix Security Report", styles["title"]))
+ story.append(Spacer(1, 4))
+ run_name = record.get("run_name") or run_dir.name
+ story.append(Paragraph(f"Run {_esc(run_name)}", styles["subtitle"]))
+ story.append(Spacer(1, 6))
+ story.append(HRFlowable(width="100%", thickness=1, color=_BRAND))
+ story.append(Spacer(1, 10))
+
+ meta_lines = [
+ f"Target: {_esc(primary_target(record) or 'unknown target')}",
+ f"Scan mode: {_esc(record.get('scan_mode') or 'n/a')}",
+ f"Status: {_esc(record.get('status') or 'n/a')}",
+ f"Started: {_esc(record.get('start_time') or 'n/a')}",
+ f"Ended: {_esc(record.get('end_time') or 'n/a')}",
+ f"Duration: {_esc(_duration(record.get('start_time'), record.get('end_time')))}",
+ ]
+ story.extend(Paragraph(line, styles["body"]) for line in meta_lines)
+
+ story.append(Paragraph("Findings by severity", styles["h2"]))
+ severity_line = " ".join(
+ f'{name.title()}: '
+ f"{counts[name]}"
+ for name in ("critical", "high", "medium", "low")
+ )
+ story.append(Paragraph(severity_line, styles["body"]))
+ story.append(Paragraph(f"Total findings: {len(vulns)}", styles["body"]))
+
+ scan_results = record.get("scan_results")
+ if isinstance(scan_results, dict):
+ summary = scan_results.get("executive_summary")
+ if isinstance(summary, str) and summary.strip():
+ story.append(Paragraph("Executive summary", styles["h2"]))
+ story.append(Paragraph(_esc(summary), styles["body"]))
+ for label, key in (
+ ("Methodology", "methodology"),
+ ("Technical analysis", "technical_analysis"),
+ ("Recommendations", "recommendations"),
+ ):
+ value = scan_results.get(key)
+ if isinstance(value, str) and value.strip():
+ story.append(Paragraph(label, styles["h2"]))
+ story.append(Paragraph(_esc(value), styles["body"]))
+
+ if vulns:
+ story.append(PageBreak())
+ story.append(Paragraph("Detailed findings", styles["h2"]))
+ for index, vuln in enumerate(vulns, start=1):
+ if not isinstance(vuln, dict):
+ continue
+ story.extend(_finding_flowables(styles, index, vuln))
+ else:
+ story.append(Paragraph("No findings were recorded for this run.", styles["body"]))
+
+ doc.build(story)
+ return buffer.getvalue()
+
+
+def _finding_flowables(
+ styles: dict[str, ParagraphStyle], index: int, vuln: dict[str, Any]
+) -> list[Flowable]:
+ title = vuln.get("title") or "Untitled finding"
+ severity = str(vuln.get("severity") or "").lower().strip() or "unknown"
+ story: list[Flowable] = [Paragraph(f"{index}. {_esc(title)}", styles["finding"])]
+
+ meta_bits = [f"Severity: {_esc(severity)}"]
+ if vuln.get("cvss") is not None:
+ meta_bits.append(f"CVSS: {_esc(vuln.get('cvss'))}")
+ meta_bits.extend(
+ f"{key.title()}: {_esc(vuln.get(key))}"
+ for key in ("target", "endpoint", "method")
+ if vuln.get(key)
+ )
+ story.append(Paragraph(" ".join(meta_bits), styles["body"]))
+
+ story.extend(_field_block(styles, "Description", vuln.get("description")))
+ story.extend(_field_block(styles, "Impact", vuln.get("impact")))
+ story.extend(_field_block(styles, "Technical analysis", vuln.get("technical_analysis")))
+ story.extend(_field_block(styles, "Proof of concept", vuln.get("poc_description")))
+ story.extend(_field_block(styles, "PoC script", vuln.get("poc_script_code"), code=True))
+ story.extend(_field_block(styles, "Evidence", vuln.get("evidence"), code=True))
+
+ remediation = vuln.get("remediation_steps")
+ if isinstance(remediation, list):
+ remediation = "\n".join(str(step) for step in remediation)
+ story.extend(_field_block(styles, "Remediation", remediation))
+
+ story.append(Spacer(1, 4))
+ story.append(HRFlowable(width="100%", thickness=0.5, color=_MUTED))
+ return story
+
+
+def generate_password() -> str:
+ """Return a >=20 character URL-safe password from a CSPRNG."""
+ return secrets.token_urlsafe(16)
+
+
+def encrypt_pdf(pdf_bytes: bytes, password: str) -> bytes:
+ """Encrypt a PDF with AES-256 using ``password`` as the user password."""
+ reader = PdfReader(BytesIO(pdf_bytes))
+ writer = PdfWriter()
+ writer.append(reader)
+ writer.encrypt(user_password=password, algorithm="AES-256")
+ out = BytesIO()
+ writer.write(out)
+ return out.getvalue()
+
+
+def build_encrypted_report(run_dir: Path) -> tuple[bytes, str, str]:
+ """Build, encrypt, and name the report. Returns (pdf_bytes, password, filename)."""
+ record = read_run_summary(run_dir)
+ run_name = str(record.get("run_name") or run_dir.name)
+ pdf_bytes = generate_report_pdf(run_dir)
+ password = generate_password()
+ encrypted = encrypt_pdf(pdf_bytes, password)
+ filename = f"strix-report-{run_name}.pdf"
+ return encrypted, password, filename
+
+
+__all__ = [
+ "build_encrypted_report",
+ "encrypt_pdf",
+ "generate_password",
+ "generate_report_pdf",
+]
diff --git a/strix/viewer/transcript.py b/strix/viewer/transcript.py
index 781850ad..3f287168 100644
--- a/strix/viewer/transcript.py
+++ b/strix/viewer/transcript.py
@@ -17,6 +17,26 @@ logger = logging.getLogger(__name__)
_TERMINAL_STATUSES = {"completed", "stopped", "failed", "interrupted"}
+_KNOWN_SEVERITIES = ("critical", "high", "medium", "low")
+
+
+def severity_counts(vulns: list[Any]) -> dict[str, int]:
+ """Bucket vulnerabilities into critical/high/medium/low counts.
+
+ Mirrors the SPA's ``severityCounts``: severities are lowercased and
+ trimmed, and anything outside the four known buckets (``info``,
+ ``informational``, ``unknown``, missing, ...) folds into ``low`` so the
+ shared UI renders cleanly.
+ """
+ counts = dict.fromkeys(_KNOWN_SEVERITIES, 0)
+ for vuln in vulns:
+ raw = vuln.get("severity") if isinstance(vuln, dict) else None
+ severity = str(raw or "").lower().strip()
+ if severity not in counts:
+ severity = "low"
+ counts[severity] += 1
+ return counts
+
def build_run_state(run_dir: Path) -> dict[str, Any]:
"""Agent graph + full per-agent event/message stream.
@@ -42,6 +62,18 @@ def read_run_summary(run_dir: Path) -> dict[str, Any]:
return {**record, "finished": finished}
+def primary_target(record: dict[str, Any]) -> str | None:
+ """The first target's original string from a run record, or None."""
+ targets = record.get("targets_info")
+ if isinstance(targets, list):
+ for entry in targets:
+ if isinstance(entry, dict):
+ original = entry.get("original")
+ if isinstance(original, str) and original:
+ return original
+ return None
+
+
def read_vulnerabilities(run_dir: Path) -> list[Any]:
"""The ``vulnerabilities.json`` list (empty until a scan writes it)."""
data = _load_json(run_dir / "vulnerabilities.json", default=[])
@@ -66,7 +98,9 @@ def _load_json(path: Path, *, default: Any) -> Any:
__all__ = [
"build_run_state",
+ "primary_target",
"read_report_markdown",
"read_run_summary",
"read_vulnerabilities",
+ "severity_counts",
]