#!/usr/bin/env python3
"""
Standalone security worker: Supabase queue + Docker scans + Jinja/Playwright/SMTP reports.
Single-file bundle — deploy this directory (worker_security/) including Reports/.
Env: SUPABASE_URL, SUPABASE_SERVICE_ROLE_KEY; SMTP / SCAN_REPORT_* (see .env.example).
Requires: Docker (nuclei, ZAP, httpx, nmap, subfinder images), pip deps in requirements.txt,
playwright install chromium for PDF attachment.
"""
from __future__ import annotations
import argparse
import datetime
import html
import json
import os
import re
import smtplib
import subprocess
import sys
import tempfile
import time
import traceback
import uuid
import xml.etree.ElementTree as ET
from concurrent.futures import ThreadPoolExecutor, as_completed
from datetime import datetime, timezone
from email.message import EmailMessage
from pathlib import Path
from urllib.parse import urlparse
try:
from jinja2 import Environment, FileSystemLoader, select_autoescape
except ImportError as e:
raise SystemExit("Missing dependency: jinja2. pip install jinja2") from e
# Bundle root (this file lives here)
ROOT = Path(__file__).resolve().parent
if str(ROOT) not in sys.path:
sys.path.insert(0, str(ROOT))
from report_branding import report_logo_context
def parse_args():
parser = argparse.ArgumentParser(description="Web Vulnerability Scanner CLI")
parser.add_argument("--target", required=True, help="Target URL (e.g. https://example.com)")
parser.add_argument(
"--target-name",
default=None,
help="Client/site display name for reports (e.g. Overdrive, CSA). Stored in JSON as target_name.",
)
parser.add_argument("--output", default="update.json", help="Output JSON file")
parser.add_argument(
"--threads",
type=int,
default=4,
help="Number of parallel scans (nuclei, nmap, zap, httpx, subfinder)",
)
parser.add_argument(
"--timeout",
type=int,
default=1800,
help="Timeout per tool in seconds (Nuclei often needs 15–30+ min on first cold run)",
)
return parser.parse_args()
# -----------------------
# Utility
# -----------------------
def _scan_warn(warnings: list[str] | None, message: str) -> None:
print(f"[!] {message}")
if warnings is not None:
warnings.append(message)
def run_command(cmd, timeout, warnings: list[str] | None = None, tool: str = "command"):
try:
result = subprocess.run(
cmd,
shell=True,
capture_output=True,
text=True,
timeout=timeout
)
if not (result.stdout or "").strip() and result.returncode != 0:
tail = (result.stderr or "").strip()
msg = f"worker_security.py / {tool}: empty output (exit {result.returncode})"
if tail:
msg = f"{msg}; {tail[-400:]}"
_scan_warn(warnings, msg)
return result.stdout
except subprocess.TimeoutExpired:
_scan_warn(warnings, f"worker_security.py / {tool}: command timed out after {timeout}s")
return ""
def tool_timeout_sec(tool: str, ceiling: int) -> int:
"""Per-tool timeout so one blocked scanner does not hold the full scan."""
env_keys = {
"nmap": "SCAN_NMAP_TIMEOUT",
"zap": "SCAN_ZAP_TIMEOUT",
"httpx": "SCAN_HTTPX_TIMEOUT",
"subfinder": "SCAN_SUBFINDER_TIMEOUT",
"nuclei": "SCAN_NUCLEI_TIMEOUT",
}
raw = os.environ.get(env_keys.get(tool, ""), "").strip()
if raw:
return min(int(raw), ceiling)
defaults = {"nmap": 300, "zap": 600, "httpx": 120, "subfinder": 300, "nuclei": 600}
return min(defaults.get(tool, ceiling), ceiling)
def scan_sequential() -> bool:
"""Run tools one-by-one (default). Set SCAN_SEQUENTIAL=0 for parallel batch."""
return os.environ.get("SCAN_SEQUENTIAL", "1").strip().lower() not in ("0", "false", "no")
# -----------------------
# NUCLEI
# -----------------------
def nuclei_finding_from_data(data: dict) -> dict | None:
"""Map one Nuclei JSONL object to a normalized finding (Nuclei v3 uses -jsonl, not -json)."""
info = data.get("info") or {}
if not info and not data.get("template-id"):
return None
sev = str(info.get("severity") or "info").lower()
title = (info.get("name") or data.get("template-id") or "Nuclei finding").strip()
endpoint = data.get("matched-at") or data.get("url") or data.get("host") or ""
return {
"tool": "nuclei",
"type": data.get("type"),
"severity": sev,
"title": title,
"endpoint": endpoint,
"raw": data,
}
def nuclei_timeout_sec(default_timeout: int) -> int:
"""Default 10 minutes; override via SCAN_NUCLEI_TIMEOUT (seconds)."""
return int(os.environ.get("SCAN_NUCLEI_TIMEOUT", "600"))
def nuclei_docker_cmd(target: str) -> str:
"""Quote target for shell safety; optional templates volume via SCAN_NUCLEI_TEMPLATES_VOLUME."""
vol = os.environ.get("SCAN_NUCLEI_TEMPLATES_VOLUME", "").strip()
vol_arg = f"-v {vol}:/root/nuclei-templates " if vol else ""
severity = os.environ.get("SCAN_NUCLEI_SEVERITY", "critical,high,medium").strip()
rate_limit = os.environ.get("SCAN_NUCLEI_RATE_LIMIT", "25").strip()
concurrency = os.environ.get("SCAN_NUCLEI_CONCURRENCY", "10").strip()
sev_arg = f"-severity {severity} " if severity else ""
return (
f"docker run --rm {vol_arg}"
f"projectdiscovery/nuclei -u \"{target}\" "
f"{sev_arg}-rate-limit {rate_limit} -c {concurrency} -jsonl -silent"
)
def nuclei_delay_sec() -> int:
return int(os.environ.get("SCAN_NUCLEI_DELAY_SEC", "60"))
def nuclei_retry_on_empty() -> bool:
return os.environ.get("SCAN_NUCLEI_RETRY_ON_EMPTY", "1").strip().lower() not in (
"0",
"false",
"no",
)
def nuclei_min_runtime_for_retry() -> int:
"""Ignore instant empty exits (broken templates volume, docker error)."""
return int(os.environ.get("SCAN_NUCLEI_MIN_RUNTIME_FOR_RETRY", "120"))
def parse_nuclei_jsonl_text(text: str) -> list:
findings = []
for line in text.splitlines():
line = line.strip()
if not line or not line.startswith("{"):
continue
try:
row = nuclei_finding_from_data(json.loads(line))
if row:
findings.append(row)
except json.JSONDecodeError:
continue
return findings
def _run_nuclei_once(target, tool_timeout, warnings, script_prefix: str) -> tuple[list, int, subprocess.CompletedProcess | None]:
cmd = nuclei_docker_cmd(target)
proc = None
t0 = time.monotonic()
findings: list = []
try:
proc = subprocess.run(
cmd,
shell=True,
capture_output=True,
text=True,
timeout=tool_timeout,
)
findings = parse_nuclei_jsonl_text(proc.stdout or "")
if proc.returncode != 0 and not findings:
tail = (proc.stderr or proc.stdout or "").strip()
msg = f"{script_prefix} / nuclei: docker exited {proc.returncode}"
if tail:
msg = f"{msg}; {tail[-400:]}"
_scan_warn(warnings, msg)
except subprocess.TimeoutExpired as exc:
partial = exc.stdout if isinstance(exc.stdout, str) else (exc.stdout or b"").decode(
"utf-8", errors="replace"
)
findings = parse_nuclei_jsonl_text(partial or "")
if findings:
_scan_warn(
warnings,
f"{script_prefix} / nuclei: timed out after {tool_timeout}s; kept {len(findings)} partial finding(s)",
)
else:
_scan_warn(
warnings,
f"{script_prefix} / nuclei: timed out after {tool_timeout}s with 0 findings",
)
elapsed = int(time.monotonic() - t0)
return findings, elapsed, proc
def run_nuclei(target, timeout, warnings: list[str] | None = None):
tool_timeout = nuclei_timeout_sec(timeout)
script_prefix = "worker_security.py"
delay = nuclei_delay_sec()
min_retry_runtime = nuclei_min_runtime_for_retry()
if delay > 0:
print(f"[*] nuclei waiting {delay}s after other tools (SCAN_NUCLEI_DELAY_SEC)...", flush=True)
time.sleep(delay)
findings, elapsed, proc = _run_nuclei_once(target, tool_timeout, warnings, script_prefix)
if (
not findings
and nuclei_retry_on_empty()
and elapsed >= min_retry_runtime
and elapsed < 300
and tool_timeout - elapsed > 120
):
_scan_warn(
warnings,
f"{script_prefix} / nuclei: 0 findings after {elapsed}s; retrying once after 60s cooldown",
)
time.sleep(60)
retry_findings, retry_elapsed, retry_proc = _run_nuclei_once(
target, tool_timeout - elapsed - 60, warnings, script_prefix
)
if retry_findings:
findings = retry_findings
elapsed = elapsed + 60 + retry_elapsed
proc = retry_proc
if not findings:
tail = ""
if proc is not None:
tail = (proc.stderr or proc.stdout or "").strip()
if elapsed < min_retry_runtime:
msg = (
f"{script_prefix} / nuclei: 0 findings after {elapsed}s (fast exit)"
" — check SCAN_NUCLEI_TEMPLATES_VOLUME (leave unset unless pre-seeded);"
f" severity filter is {os.environ.get('SCAN_NUCLEI_SEVERITY', 'critical,high,medium')}"
)
else:
msg = (
f"{script_prefix} / nuclei: 0 findings after {elapsed}s (limit {tool_timeout}s)"
" — site may block automated scans, need longer SCAN_NUCLEI_TIMEOUT,"
" or no critical/high/medium templates matched"
)
if tail:
msg = f"{msg}; {tail[-400:]}"
_scan_warn(warnings, msg)
return findings
# -----------------------
# ZAP
# -----------------------
def _docker_host_mount_path(local_dir):
"""Windows drive letters break unquoted docker -v host:container; normalize + quote caller."""
p = os.path.abspath(local_dir)
if os.name == "nt":
p = p.replace("\\", "/")
return p
def resolve_zap_workdir(work_root: Path) -> Path:
"""Absolute host path for ZAP JSON/summary output (override via SCAN_ZAP_WORKDIR)."""
override = os.environ.get("SCAN_ZAP_WORKDIR", "").strip()
if override:
zap_dir = Path(override).expanduser().resolve()
else:
zap_dir = (work_root / "zap_wrk").resolve()
zap_dir.mkdir(parents=True, exist_ok=True)
try:
os.chmod(zap_dir, 0o777)
except OSError:
pass
return zap_dir
def _zap_docker_cmd(target: str, output_file: str, workdir: str, autooff: bool = False) -> str:
extra = " --autooff" if autooff else ""
return (
f'docker run --rm '
f'-e HOME=/zap/wrk -w /zap/wrk '
f'-v "{workdir}":/zap/wrk:rw '
f'-v "{workdir}":/home/zap:rw '
f'zaproxy/zap-stable:latest zap-baseline.py '
f'-t "{target}" -J {output_file} -I{extra}'
)
def _find_zap_report(zap_dir: Path, output_file: str) -> Path | None:
candidate = zap_dir / output_file
if candidate.is_file():
return candidate
reports = sorted(
(p for p in zap_dir.glob("zap_*.json") if p.name != "zap_out.json"),
key=lambda p: p.stat().st_mtime,
reverse=True,
)
return reports[0] if reports else None
def _zap_proc_output(proc: subprocess.CompletedProcess) -> str:
return f"{proc.stdout or ''}\n{proc.stderr or ''}".strip()
def _zap_target_blocked(proc: subprocess.CompletedProcess, target: str) -> bool:
"""True when ZAP could not reach the URL (WAF, firewall, down site)."""
text = _zap_proc_output(proc).lower()
if "zap failed to access" in text:
return True
host = urlparse(target).hostname or ""
return bool(host) and "failed to access" in text and host.lower() in text
def _zap_findings_from_file(out_full: Path, warnings: list[str] | None, script_prefix: str) -> list:
findings = []
try:
with open(out_full, encoding="utf-8") as f:
data = json.load(f)
for site in data.get("site", []):
for alert in site.get("alerts", []):
findings.append({
"tool": "zap",
"type": "web",
"severity": alert.get("riskdesc", "info").split(" ")[0].lower(),
"title": alert.get("alert"),
"description": alert.get("desc"),
"endpoint": site.get("name"),
"raw": alert,
})
out_full.unlink(missing_ok=True)
except (json.JSONDecodeError, OSError) as e:
_scan_warn(warnings, f"{script_prefix} / zap: parsing error: {e}")
return findings
def run_zap(target, timeout, work_root: Path, warnings: list[str] | None = None):
script_prefix = "worker_security.py"
output_file = f"zap_{uuid.uuid4()}.json"
zap_dir = resolve_zap_workdir(work_root)
workdir = _docker_host_mount_path(str(zap_dir))
print(f"[*] zap workdir (host): {zap_dir}", flush=True)
findings = []
proc = None
out_full: Path | None = None
for autooff in (False, True):
cmd = _zap_docker_cmd(target, output_file, workdir, autooff=autooff)
try:
proc = subprocess.run(
cmd,
shell=True,
capture_output=True,
text=True,
timeout=timeout,
)
except subprocess.TimeoutExpired:
_scan_warn(warnings, f"{script_prefix} / zap: timed out after {timeout}s")
return findings
out_full = _find_zap_report(zap_dir, output_file)
if out_full is not None:
if autooff:
_scan_warn(
warnings,
f"{script_prefix} / zap: used --autooff fallback; report at {out_full.name}",
)
break
if _zap_target_blocked(proc, target):
_scan_warn(
warnings,
f"{script_prefix} / zap: target unreachable or blocked ({target}); continuing",
)
return findings
if not autooff:
print(f"[*] zap: automation framework produced no report; retrying with --autooff", flush=True)
if out_full is not None and proc is not None:
findings = _zap_findings_from_file(out_full, warnings, script_prefix)
if proc.returncode not in (0, 1, 2) and not findings:
tail = _zap_proc_output(proc)[-400:]
msg = f"{script_prefix} / zap: docker exited {proc.returncode}"
if tail:
msg = f"{msg}; {tail}"
_scan_warn(warnings, msg)
return findings
tail = _zap_proc_output(proc)[-400:] if proc is not None else ""
msg = f"{script_prefix} / zap: no report written under {zap_dir}"
if proc is not None and proc.returncode != 0:
msg = f"{msg} (docker exit {proc.returncode})"
if tail:
msg = f"{msg}; {tail}"
_scan_warn(warnings, msg)
return findings
# -----------------------
# HTTPX
# -----------------------
def run_httpx(target, timeout, warnings: list[str] | None = None):
script_prefix = "worker_security.py"
cmd = f'docker run --rm projectdiscovery/httpx -u {target} -json'
output = run_command(cmd, timeout, warnings, tool="httpx")
findings = []
for line in output.splitlines():
try:
data = json.loads(line)
findings.append({
"tool": "httpx",
"type": "info",
"severity": "info",
"title": f"Status {data.get('status_code')}",
"endpoint": data.get("url"),
"raw": data
})
except:
continue
if not findings:
_scan_warn(
warnings,
f"{script_prefix} / httpx: 0 findings — target unreachable or blocked; continuing",
)
return findings
# -----------------------
# NMAP
# -----------------------
def _nmap_host_ports(target: str):
"""Host + extra TCP ports from URL (always include 80,443 when probing explicit port)."""
t = target.strip()
if "://" not in t:
t = "http://" + t
p = urlparse(t)
host = p.hostname
if not host:
return None, []
extra = [str(p.port)] if p.port else []
return host, extra
def _nmap_finding_from_port(addr, pel):
"""One finding dict for an open port element, or None."""
st = pel.find("state")
if st is None or st.get("state") != "open":
return None
pid = pel.get("portid", "")
proto = pel.get("protocol", "tcp")
svc_el = pel.find("service")
svc_name = (svc_el.get("name") if svc_el is not None else None) or "unknown"
product = (svc_el.get("product") if svc_el is not None else None) or ""
title = f"Open {proto}/{pid} ({svc_name})"
if product:
title = f"{title}: {product}"
return {
"tool": "nmap",
"type": "network",
"severity": "info",
"title": title,
"description": f"Service {svc_name} on {addr}:{pid}/{proto}",
"endpoint": f"{addr}:{pid}",
"raw": {
"host": addr,
"port": pid,
"protocol": proto,
"service": svc_name,
"product": product,
"extrainfo": svc_el.get("extrainfo") if svc_el is not None else None,
},
}
def _nmap_findings_from_xml(root, fallback_host):
out = []
for hel in root.findall("host"):
addrs = [a.get("addr", "") for a in hel.findall("address") if a.get("addr")]
addr = addrs[0] if addrs else fallback_host
ports_el = hel.find("ports")
if ports_el is None:
continue
for pel in ports_el.findall("port"):
row = _nmap_finding_from_port(addr, pel)
if row:
out.append(row)
return out
def run_nmap(target, timeout, warnings: list[str] | None = None):
host, extra_ports = _nmap_host_ports(target)
if not host:
_scan_warn(warnings, "worker_security.py / nmap: could not parse host from --target")
return []
# -F: fast top 100; if URL has a port, bias -p toward web + that port
port_set = {"80", "443", *extra_ports}
pflag = f'-p {",".join(sorted(port_set, key=int))}' if extra_ports else "-F"
cmd = (
f'docker run --rm instrumentisto/nmap -oX - -T4 -Pn {pflag} '
f'"{host}"'
)
try:
proc = subprocess.run(
cmd,
shell=True,
capture_output=True,
text=True,
timeout=timeout,
)
if not (proc.stdout or "").strip():
tail = (proc.stderr or "").strip()
msg = f"worker_security.py / nmap: empty output (exit {proc.returncode})"
if tail:
msg = f"{msg}; {tail[-400:]}"
_scan_warn(warnings, msg)
return []
root = ET.fromstring(proc.stdout)
except subprocess.TimeoutExpired:
_scan_warn(warnings, f"worker_security.py / nmap: timed out after {timeout}s")
return []
except ET.ParseError as e:
_scan_warn(warnings, f"worker_security.py / nmap: XML parse error: {e}")
return []
return _nmap_findings_from_xml(root, host)
# -----------------------
# SUBFINDER (passive subdomain enumeration)
# -----------------------
def run_subfinder(target, timeout, warnings: list[str] | None = None):
"""Docker subfinder: passive sources only (no -active). NDJSON via -json."""
host, _extra = _nmap_host_ports(target)
if not host:
_scan_warn(warnings, "worker_security.py / subfinder: could not parse host from --target")
return []
cmd = (
f'docker run --rm projectdiscovery/subfinder:latest '
f'-d "{host}" -silent -json'
)
output = run_command(cmd, timeout, warnings, tool="subfinder")
findings = []
for line in output.splitlines():
try:
data = json.loads(line)
sub_host = (data.get("host") or "").strip()
if not sub_host:
continue
src = (data.get("source") or "").strip()
title = f"Subdomain ({src})" if src else "Discovered subdomain"
findings.append({
"tool": "subfinder",
"type": "subdomain",
"severity": "info",
"title": title,
"endpoint": sub_host,
"raw": data,
})
except Exception:
continue
if not findings:
_scan_warn(
warnings,
"worker_security.py / subfinder: 0 findings — no subdomains or lookup blocked; continuing",
)
return findings
def _invoke_tool(name: str, target: str, timeout: int, work_root: Path, warnings: list[str]) -> list:
"""Run one scanner; never raise — caller always moves to the next tool."""
runners = {
"subfinder": lambda: run_subfinder(target, timeout, warnings),
"nmap": lambda: run_nmap(target, timeout, warnings),
"httpx": lambda: run_httpx(target, timeout, warnings),
"zap": lambda: run_zap(target, timeout, work_root, warnings),
}
return runners[name]()
# -----------------------
# SCAN RUNNER (sequential by default; parallel optional)
# -----------------------
def run_all_scans(target, threads, timeout, work_root: Path):
warnings: list[str] = []
script_prefix = "worker_security.py"
# Passive/network first, then light HTTP, then heavy HTTP. Nuclei always last.
tool_order = ("subfinder", "nmap", "httpx", "zap")
all_findings = []
if scan_sequential():
for name in tool_order:
tool_to = tool_timeout_sec(name, timeout)
print(f"[*] {name} starting (timeout {tool_to}s)...", flush=True)
try:
results = _invoke_tool(name, target, tool_to, work_root, warnings)
print(f"[+] {name} completed: {len(results)} findings", flush=True)
all_findings.extend(results)
except Exception as e:
_scan_warn(warnings, f"{script_prefix} / {name}: failed: {e}; continuing to next tool")
else:
functions = [
("subfinder", lambda t, to, w=warnings: run_subfinder(t, to, w)),
("nmap", lambda t, to, w=warnings: run_nmap(t, to, w)),
("httpx", lambda t, to, w=warnings: run_httpx(t, to, w)),
("zap", lambda t, to, w=warnings: run_zap(t, to, work_root, w)),
]
with ThreadPoolExecutor(max_workers=threads) as executor:
futures = {
executor.submit(func, target, tool_timeout_sec(name, timeout)): name
for name, func in functions
}
for future in as_completed(futures):
name = futures[future]
try:
results = future.result()
print(f"[+] {name} completed: {len(results)} findings", flush=True)
all_findings.extend(results)
except Exception as e:
_scan_warn(warnings, f"{script_prefix} / {name}: failed: {e}")
nuclei_to = tool_timeout_sec("nuclei", timeout)
print(f"[*] nuclei starting (timeout {nuclei_to}s, after other tools)...", flush=True)
try:
nuclei_results = run_nuclei(target, nuclei_to, warnings)
print(f"[+] nuclei completed: {len(nuclei_results)} findings", flush=True)
all_findings.extend(nuclei_results)
except Exception as e:
_scan_warn(warnings, f"{script_prefix} / nuclei: failed: {e}")
return all_findings, warnings
# -----------------------
# SUMMARY
# -----------------------
def build_summary(findings):
summary = {"critical": 0, "high": 0, "medium": 0, "low": 0, "info": 0}
for f in findings:
sev = f.get("severity", "info")
if sev in summary:
summary[sev] += 1
else:
summary["info"] += 1
return summary
# -----------------------
# run_scan_to_dict (same JSON shape as legacy scan.py --output)
# -----------------------
def run_scan_to_dict(
target: str,
target_name: str | None,
threads: int,
timeout: int,
work_root: Path,
) -> dict:
print(f"\n[*] Scanning: {target}\n")
findings, warnings = run_all_scans(
target=target, threads=threads, timeout=timeout, work_root=work_root
)
result = {
"id": str(uuid.uuid4()),
"target": target,
"scan_date": datetime.now(timezone.utc).isoformat(),
"total_findings": len(findings),
"summary": build_summary(findings),
"findings": findings,
}
if warnings:
result["warnings"] = warnings
if target_name:
result["target_name"] = str(target_name).strip()
print(f"\n[+] Done. Total findings: {len(findings)}")
print(f"[*] Summary: {result['summary']}\n")
return result
# --- Reports / email (from generate_reports_from_update.py) ---
REPORT_EMAIL_JACO = "jaco@overdrive.co.za"
REPORT_EMAIL_BCC = REPORT_EMAIL_JACO
OPS_ALERT_EMAIL = os.environ.get("SCAN_OPS_ALERT_EMAIL", REPORT_EMAIL_JACO)
REPORT_BRAND_NAME = os.environ.get("SCAN_REPORT_BRAND_NAME", "Silicon Overdrive")
def report_brand_name() -> str:
return (os.environ.get("SCAN_REPORT_BRAND_NAME") or REPORT_BRAND_NAME).strip() or "Silicon Overdrive"
SEVERITY_ORDER = ("critical", "high", "medium", "low", "info")
SEVERITY_RANK = {s: i for i, s in enumerate(SEVERITY_ORDER)}
PILL_CLASS = {
"critical": "crit",
"high": "high",
"medium": "med",
"low": "low",
"info": "info",
}
EMAIL_BADGE = {
"critical": ("#fbe9e7", "#8b1a1a"),
"high": ("#fdecdb", "#c2410c"),
"medium": ("#fbf2cf", "#a16207"),
"low": ("#e0e7ff", "#1d4ed8"),
"info": ("#e8eaef", "#475569"),
}
def _parse_iso(dt: str) -> datetime:
s = (dt or "").strip()
if s.endswith("Z"):
s = s[:-1] + "+00:00"
return datetime.fromisoformat(s)
def friendly_target_name(data: dict, target: str) -> str:
"""Display name: JSON target_name / client_name / project_name, else derived from host."""
for key in ("target_name", "client_name", "project_name"):
v = data.get(key)
if v and str(v).strip():
return str(v).strip()
h = host_display(target)
if not h:
return "Scan"
first = h.split(".")[0].replace("-", " ").replace("_", " ")
return first.title() if first else "Scan"
def slug_stem(name: str) -> str:
"""Filesystem-safe stem fragment from a display name (e.g. Shorts of Practise → Shorts_of_Practise)."""
s = re.sub(r"[^a-zA-Z0-9]+", "_", (name or "").strip()).strip("_")
return s or "Scan"
def file_stem_from_data(data: dict, scan_date: str) -> str:
"""{TargetNameSlug}_{YYYY-MM-DD} — starts with client/target name when set in JSON."""
target = str(data.get("target") or "")
display = friendly_target_name(data, target)
dt = _parse_iso(scan_date)
stamp = dt.strftime("%Y-%m-%d")
return f"{slug_stem(display)}_{stamp}"
def host_display(target: str) -> str:
p = urlparse((target or "").strip())
return re.sub(r"^www\.", "", p.hostname or p.netloc or target, flags=re.I)
def resolved_ip(findings: list, target: str) -> str:
for f in findings:
raw = f.get("raw") or {}
if f.get("tool") == "httpx" and raw.get("host_ip"):
return str(raw["host_ip"])
if f.get("tool") == "nmap" and raw.get("host"):
return str(raw["host"])
p = urlparse(target)
return p.hostname or "—"
def tools_line(findings: list) -> str:
tools = sorted({str(f.get("tool") or "?") for f in findings})
return " · ".join(tools)
def sort_findings(findings: list) -> list:
def key(f):
sev = (f.get("severity") or "info").lower()
return (SEVERITY_RANK.get(sev, 99), (f.get("title") or ""))
return sorted(findings, key=key)
def instance_count(f: dict) -> int:
raw = f.get("raw") or {}
if raw.get("count") is not None:
try:
return int(str(raw["count"]))
except ValueError:
pass
inst = raw.get("instances")
if isinstance(inst, list):
return len(inst)
return 1
def confidence_label(raw: dict) -> str:
c = str(raw.get("confidence", ""))
if c == "3":
return "High"
if c == "2":
return "Medium"
if c == "1":
return "Low"
return "—"
def difficulty_badge(raw: dict) -> tuple[str | None, str | None]:
sol = (raw.get("solution") or "").lower()
if any(x in sol for x in ("easy", "simple", "x-frame-options", "same-site")):
return ("easy", "Easy fix")
if any(x in sol for x in ("difficult", "complex", "hard to")):
return ("hard", "Hard fix")
if raw.get("solution"):
return ("mod", "Moderate fix")
return (None, None)
def wrap_plain_description(text: str) -> str:
t = (text or "").strip()
if not t:
return "(No description)
"
if "<" in t and ">" in t:
return t
return f"{html.escape(t)}
"
def evidence_blocks(f: dict) -> str:
raw = f.get("raw") or {}
parts: list[str] = []
inst = raw.get("instances")
if isinstance(inst, list) and inst:
for it in inst[:12]:
uri = html.escape(str(it.get("uri") or it.get("nodeName") or ""))
ev = it.get("evidence") or it.get("param") or ""
evs = html.escape(str(ev)[:2000])
parts.append(f'{uri}{evs}
')
if len(inst) > 12:
parts.append(f'… {len(inst) - 12} more instance(s)
')
elif f.get("tool") == "nuclei":
line = html.escape(json.dumps(raw, indent=2)[:4000])
parts.append(f'nuclei raw{line}
')
elif f.get("tool") == "nmap":
r = raw
line = f"{r.get('host')}:{r.get('port')}/{r.get('protocol')} {r.get('service')} {r.get('product') or ''}"
parts.append(f'{html.escape(line.strip())}
')
elif f.get("tool") == "subfinder":
r = raw
line = f"{r.get('host')} ← {r.get('source') or 'passive'}"
parts.append(f'{html.escape(line.strip())}
')
if not parts and f.get("endpoint"):
parts.append(f'{html.escape(str(f.get("endpoint")))}
')
if not parts:
return ""
return '' + "".join(parts) + "
"
def references_html(raw: dict) -> str:
ref = raw.get("reference") or ""
if not ref.strip():
return ""
# ZAP often wraps links in
if "
" in ref or " and split URLs
text = re.sub(r"]*>", "\n", ref, flags=re.I)
lines = [ln.strip() for ln in re.sub(r"<[^>]+>", "", text).splitlines() if ln.strip()]
if not lines:
return ""
items = "".join(f"{html.escape(ln)}" for ln in lines[:20])
return f''
def is_network_finding(f: dict) -> bool:
t = f.get("tool")
if t == "nmap":
return True
if t == "httpx":
return True
if t == "subfinder":
return True
return False
def is_detail_finding(f: dict) -> bool:
if f.get("tool") == "nmap":
return False
if f.get("tool") == "httpx":
return False
if f.get("tool") == "subfinder":
return False
if f.get("tool") == "zap" and (f.get("severity") or "").lower() == "info":
return False
return True
def is_info_zap(f: dict) -> bool:
return f.get("tool") == "zap" and (f.get("severity") or "").lower() == "info"
def build_detail_row(f: dict, display_id: str) -> dict:
raw = f.get("raw") or {}
sev = (f.get("severity") or "info").lower()
desc = f.get("description") or raw.get("desc") or f.get("title") or ""
sol = raw.get("solution") or ""
diff_c, diff_l = difficulty_badge(raw)
cweid = raw.get("cweid")
cwe = f"CVE-{cweid}" if cweid else ""
endpoint_line = f.get("endpoint") or ""
if not endpoint_line and isinstance(raw.get("instances"), list) and raw["instances"]:
endpoint_line = str(raw["instances"][0].get("uri") or "")
param_line = ""
if isinstance(raw.get("instances"), list) and raw["instances"]:
p0 = raw["instances"][0].get("param")
if p0 and len(str(p0)) < 200:
param_line = str(p0)
systemic = "Yes" if raw.get("systemic") else ("No" if raw.get("systemic") is False else "—")
return {
"display_id": display_id,
"title": f.get("title") or "Finding",
"severity_label": sev.title(),
"pill": PILL_CLASS.get(sev, "info"),
"diff_class": diff_c,
"diff_label": diff_l,
"tool": f.get("tool") or "?",
"cwe": cwe,
"confidence": confidence_label(raw),
"instances": instance_count(f),
"systemic": systemic,
"endpoint_line": endpoint_line,
"param_line": param_line,
"description_html": wrap_plain_description(str(desc)),
"evidence_html": evidence_blocks(f),
"solution_html": sol if sol else "",
"references_html": references_html(raw),
}
def next_display_id(sev: str, counters: dict) -> str:
counters[sev] = counters.get(sev, 0) + 1
prefix = {"critical": "C", "high": "H", "medium": "M", "low": "L", "info": "I"}.get(sev, "X")
return f"{prefix}-{counters[sev]:02d}"
def build_network_rows(findings: list) -> list[dict]:
rows = []
nf = [f for f in sort_findings(findings) if is_network_finding(f)]
for i, f in enumerate(nf, start=1):
raw = f.get("raw") or {}
if f.get("tool") == "nmap":
svc = str(raw.get("service") or "?")
detail = str(raw.get("product") or "") or str(f.get("description") or "")
ep = str(f.get("endpoint") or f"{raw.get('host')}:{raw.get('port')}")
rows.append(
{
"idx": f"N-{i:02d}",
"endpoint": ep,
"service": svc,
"detail": detail[:500],
}
)
elif f.get("tool") == "subfinder":
ep = str(f.get("endpoint") or raw.get("host") or "")
src = str(raw.get("source") or "passive")
rows.append(
{
"idx": f"N-{i:02d}",
"endpoint": ep,
"service": "subfinder",
"detail": f"source: {src}"[:500],
}
)
else:
title = str(f.get("title") or "httpx")
loc = raw.get("location") or ""
tech = ", ".join(raw.get("tech") or [])[:200]
ep = str(f.get("endpoint") or raw.get("url") or "")
detail = " ".join(x for x in (loc, tech) if x)
rows.append(
{
"idx": f"N-{i:02d}",
"endpoint": ep,
"service": title,
"detail": detail[:500],
}
)
return rows
def build_info_zap_rows(findings: list, target: str) -> list[dict]:
rows = []
for i, f in enumerate([x for x in sort_findings(findings) if is_info_zap(x)], start=1):
raw = f.get("raw") or {}
aff = str(f.get("endpoint") or raw.get("alert") or host_display(target) or "—")
rows.append(
{
"idx": f"I-{i:02d}",
"title": str(f.get("title") or "Finding"),
"affected": aff[:300],
"instances": instance_count(f),
}
)
return rows
def collect_top_mediums(findings: list, limit: int = 5) -> list[dict]:
"""Short list for the email_body highlight box."""
rows: list[dict] = []
n = 0
for f in sort_findings(findings):
if (f.get("severity") or "").lower() != "medium":
continue
n += 1
rows.append({"row_id": f"M-{n:02d}", "title": str(f.get("title") or "")})
if len(rows) >= limit:
break
return rows
def hook_line(summary: dict) -> str:
c = int(summary.get("critical") or 0)
h = int(summary.get("high") or 0)
m = int(summary.get("medium") or 0)
if c or h:
return "Critical or high items need urgent review; see the attached PDF for detail."
if m:
return (
f"There are {m} medium-severity item(s)—often header or configuration fixes. "
"No critical or high issues in this run."
)
return "No critical, high, or medium issues in this run; remaining findings are low or informational."
def summary_email_text(summary: dict, total: int) -> str:
c, h, m, l, i = (
summary.get("critical", 0),
summary.get("high", 0),
summary.get("medium", 0),
summary.get("low", 0),
summary.get("info", 0),
)
parts = [
f"The latest scan identified {total} findings across the public-facing infrastructure."
]
if c or h:
parts.append(f"Critical: {c}, High: {h} — review urgently.")
elif m:
parts.append(
f"No critical or high-severity issues were observed in this run. "
f"{m} medium-severity item(s) should be scheduled for the next change window."
)
else:
parts.append("No critical, high, or medium-severity issues were observed in this run.")
parts.append(f"Low: {l}, Informational: {i}.")
return " ".join(parts)
def build_info_nuclei_rows(findings: list, limit: int = 20) -> list[dict]:
"""Tabloid PDF rows for informational Nuclei matches (template detections, headers, etc.)."""
rows = []
nuclei_info = [
f
for f in sort_findings(findings)
if f.get("tool") == "nuclei" and (f.get("severity") or "").lower() == "info"
]
for i, f in enumerate(nuclei_info[:limit], start=1):
ep = str(f.get("endpoint") or "")[:120]
rows.append(
{
"row_id": f"N-{i:02d}",
"title": str(f.get("title") or "Finding"),
"subline": ep or "—",
}
)
return rows
def info_rollups(findings: list, resolved: str) -> tuple[str, str]:
nmap = [f for f in findings if f.get("tool") == "nmap"]
httpx = [f for f in findings if f.get("tool") == "httpx"]
nuclei = [f for f in findings if f.get("tool") == "nuclei"]
subfinder_f = [f for f in findings if f.get("tool") == "subfinder"]
zap_info = [f for f in findings if is_info_zap(f)]
ports = []
for f in nmap[:20]:
raw = f.get("raw") or {}
ports.append(f"{raw.get('service')}/{raw.get('port')}")
primary = (
f"{len(nmap)} network observation(s) on {resolved}"
+ (f": {', '.join(ports)}" if ports else "")
+ ("." if not httpx else f" plus httpx ({len(httpx)}).")
)
if nuclei:
primary += f" Nuclei: {len(nuclei)} detection(s)."
if subfinder_f:
primary = primary.rstrip(".") + f" plus subfinder ({len(subfinder_f)})."
secondary = ""
if zap_info:
titles = [str(x.get("title") or "") for x in zap_info[:8]]
secondary = "Additional ZAP informational items: " + "; ".join(titles)
if len(zap_info) > 8:
secondary += f" … (+{len(zap_info) - 8} more)"
return primary, secondary
def extract_article_wrapper(detailed_html_path: Path) -> tuple[str, str]:
"""Return (prefix up to and including , closing suffix)."""
text = detailed_html_path.read_text(encoding="utf-8")
tag = ''
i0 = text.index(tag) + len(tag)
prefix = text[:i0]
suffix = "\n\n\n \n