mirror of
https://github.com/usestrix/strix.git
synced 2026-08-17 09:29:49 +02:00
Add viewer email auth state, relay client, and encrypted PDF reports
This commit is contained in:
Jonathan Singer
@@ -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",
|
||||
]
|
||||
@@ -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", "<br/>")
|
||||
|
||||
|
||||
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"<b>Target:</b> {_esc(primary_target(record) or 'unknown target')}",
|
||||
f"<b>Scan mode:</b> {_esc(record.get('scan_mode') or 'n/a')}",
|
||||
f"<b>Status:</b> {_esc(record.get('status') or 'n/a')}",
|
||||
f"<b>Started:</b> {_esc(record.get('start_time') or 'n/a')}",
|
||||
f"<b>Ended:</b> {_esc(record.get('end_time') or 'n/a')}",
|
||||
f"<b>Duration:</b> {_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'<font color="{_SEVERITY_COLORS[name].hexval()}"><b>{name.title()}:</b> '
|
||||
f"{counts[name]}</font>"
|
||||
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"<b>Severity:</b> {_esc(severity)}"]
|
||||
if vuln.get("cvss") is not None:
|
||||
meta_bits.append(f"<b>CVSS:</b> {_esc(vuln.get('cvss'))}")
|
||||
meta_bits.extend(
|
||||
f"<b>{key.title()}:</b> {_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",
|
||||
]
|
||||
@@ -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",
|
||||
]
|
||||
|
||||
Reference in New Issue
Block a user