From 98c2e0be4ee4c7b653ce147fb3825806b3136145 Mon Sep 17 00:00:00 2001 From: Jonathan Singer Date: Mon, 20 Jul 2026 11:52:46 -0400 Subject: [PATCH] Add viewer email auth state, relay client, and encrypted PDF reports --- strix/viewer/auth.py | 191 +++++++++++++++++++++++ strix/viewer/report_pdf.py | 308 +++++++++++++++++++++++++++++++++++++ strix/viewer/transcript.py | 34 ++++ 3 files changed, 533 insertions(+) create mode 100644 strix/viewer/auth.py create mode 100644 strix/viewer/report_pdf.py 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", ]