feat(reporting): Phase 4 grounded AI reporting, SMB remediation UX & encrypted secrets

app/services/ai_reporter.py

@@ -1,12 +1,17 @@
 import json
 import logging
+import re
 
 from app.config import Settings
 from app.schemas import Finding
-from app.services.owasp import risk_grade
+from app.services.owasp import explain_risk_rule, get_remediation_card, risk_grade
+from app.services.secrets import resolve_anthropic_api_key
 
 logger = logging.getLogger(__name__)
 
+CVE_PATTERN = re.compile(r"\bCVE-\d{4}-\d{4,7}\b", re.I)
+CITATION_PATTERN = re.compile(r"\[ID:\s*([^\]]+)\]")
+
 
 def _count_by_risk(findings: list[Finding]) -> dict[str, int]:
     counts = {"Critical": 0, "High": 0, "Medium": 0, "Low": 0, "Informational": 0}
@@ -24,17 +29,41 @@ def friendly_category_label(owasp: str) -> str:
     return owasp
 
 
+def sanitize_grounded_report(report_text: str, findings: list[Finding]) -> str:
+    """Ensure the report cites only valid finding IDs and contains no hallucinated CVE numbers."""
+    valid_ids = {f.id for f in findings}
+    allowed_cves: set[str] = set()
+    for f in findings:
+        blob = f"{f.name} {f.description} {f.evidence}"
+        for m in CVE_PATTERN.findall(blob):
+            allowed_cves.add(m.upper())
+
+    def _replace_cve(match: re.Match[str]) -> str:
+        cve = match.group(0).upper()
+        return match.group(0) if cve in allowed_cves else "a known security advisory"
+
+    cleaned = CVE_PATTERN.sub(_replace_cve, report_text)
+
+    def _replace_citation(match: re.Match[str]) -> str:
+        cited_id = match.group(1).strip()
+        return f"[ID: {cited_id}]" if cited_id in valid_ids else ""
+
+    cleaned = CITATION_PATTERN.sub(_replace_citation, cleaned)
+    return cleaned
+
+
 def generate_template_report(target_url: str, findings: list[Finding]) -> tuple[str, str, str]:
-    """Fallback report when no AI API key is configured."""
+    """Grounded fallback report when no AI API key is configured (no hallucinated CVEs)."""
     counts = _count_by_risk(findings)
     grade = risk_grade(counts["Critical"], counts["High"], counts["Medium"])
+    score_info = explain_risk_rule(counts["Critical"], counts["High"], counts["Medium"], counts["Low"])
 
     executive = (
         f"We checked {target_url} and gave it a safety score of **{grade}** "
-        f"(A is best, F needs urgent attention). "
+        f"({score_info['numeric_score']}/100 — A is best, F needs urgent attention). "
         f"We found {len(findings)} issue(s): "
-        f"{counts['High']} serious, {counts['Medium']} moderate, {counts['Low']} minor, "
-        f"and {counts['Informational']} low-priority notes. "
+        f"{counts['Critical']} critical, {counts['High']} serious, {counts['Medium']} moderate, "
+        f"{counts['Low']} minor, and {counts['Informational']} low-priority notes. "
     )
     if counts["High"] or counts["Critical"]:
         executive += (
@@ -50,7 +79,7 @@ def generate_template_report(target_url: str, findings: list[Finding]) -> tuple[
     lines = [
         "# ShieldScan Security Report\n",
         f"**Target:** {target_url}\n",
-        f"**Risk Grade:** {grade}\n",
+        f"**Risk Grade:** {grade} ({score_info['numeric_score']}/100 — {score_info['explanation']})\n",
         f"**Total Findings:** {len(findings)}\n",
         "## Executive Summary\n",
         executive,
@@ -60,18 +89,31 @@ def generate_template_report(target_url: str, findings: list[Finding]) -> tuple[
     order = {"Critical": 0, "High": 1, "Medium": 2, "Low": 3, "Informational": 4}
     sorted_findings = sorted(findings, key=lambda f: order.get(f.risk, 5))
 
+    seen_categories: list[str] = []
     for i, f in enumerate(sorted_findings[:20], 1):
-        lines.append(f"### {i}. [{f.risk}] {f.name}\n")
+        card = get_remediation_card(f.owasp_category)
+        if card["code"] not in seen_categories:
+            seen_categories.append(card["code"])
+        lines.append(f"### {i}. [{f.risk}] {f.name} [ID: {f.id}]\n")
+        lines.append(f"- **Finding ID:** `{f.id}`\n")
         lines.append(f"- **URL:** {f.url}\n")
         if f.parameter:
             lines.append(f"- **Parameter:** {f.parameter}\n")
         lines.append(f"- **Category:** {friendly_category_label(f.owasp_category)}\n")
-        lines.append(f"- **What it means:** {f.description or 'Something on your site could be safer.'}\n")
-        lines.append(f"- **How to fix:** {f.solution or 'Ask your web developer or hosting provider to review this.'}\n")
+        lines.append(f"- **What it means:** {f.description or card['what_it_means']}\n")
+        lines.append(f"- **How to fix:** {f.solution or card['how_to_fix']}\n")
         if f.evidence:
             lines.append(f"- **Evidence:** `{f.evidence[:200]}`\n")
         lines.append("\n")
 
+    if seen_categories:
+        lines.append("## Plain-English Remediation Guide by Category\n")
+        for code in seen_categories:
+            card = get_remediation_card(code)
+            lines.append(f"### {card['code']} — {card['title']}\n")
+            lines.append(f"- **Plain-English Meaning:** {card['what_it_means']}\n")
+            lines.append(f"- **Recommended Fix:** {card['how_to_fix']}\n\n")
+
     lines.append("## Tips for small business owners\n")
     lines.append(
         "1. Fix serious problems within 48 hours if you handle customer data.\n"
@@ -81,7 +123,7 @@ def generate_template_report(target_url: str, findings: list[Finding]) -> tuple[
         "5. Before launching payments or personal data, consider a professional security review.\n"
     )
 
-    full_report = "".join(lines)
+    full_report = sanitize_grounded_report("".join(lines), findings)
     return executive, grade, full_report
 
 
@@ -90,13 +132,14 @@ async def generate_ai_report(
     target_url: str,
     findings: list[Finding],
 ) -> tuple[str, str, str]:
-    if not settings.anthropic_api_key:
+    api_key = resolve_anthropic_api_key(settings)
+    if not api_key:
         return generate_template_report(target_url, findings)
 
     try:
         import anthropic
 
-        client = anthropic.Anthropic(api_key=settings.anthropic_api_key)
+        client = anthropic.Anthropic(api_key=api_key)
         findings_payload = [f.model_dump() for f in findings]
         counts = _count_by_risk(findings)
         grade = risk_grade(counts["Critical"], counts["High"], counts["Medium"])
@@ -109,12 +152,12 @@ async def generate_ai_report(
 
 Write a report with these sections:
 1. EXECUTIVE SUMMARY (3-4 short sentences, everyday language, no acronyms unless you explain them)
-2. TOP PRIORITIES (numbered list, most urgent first, max 5 items — say what to fix and why it matters to the business)
-3. DETAILED FINDINGS (for each finding: plain explanation, business impact, simple fix steps)
+2. TOP PRIORITIES (numbered list, most urgent first, max 5 items — cite each finding with [ID: <finding.id>])
+3. DETAILED FINDINGS (for each finding: cite [ID: <finding.id>], plain explanation, business impact, simple fix steps)
 4. NEXT STEPS (short checklist the owner can give their developer or host)
 
 Rules:
-- Only discuss findings in the JSON. Do NOT invent problems.
+- Only discuss findings in the JSON and cite their exact [ID: <id>]. Do NOT invent problems or hallucinate CVE numbers.
 - Write like you're explaining to a smart non-technical person.
 - Give practical fixes, not vague advice.
 - Mention actual page URLs from the data."""
@@ -124,7 +167,8 @@ async def generate_ai_report(
             max_tokens=4000,
             messages=[{"role": "user", "content": prompt}],
         )
-        full_report = message.content[0].text
+        raw_report = message.content[0].text
+        full_report = sanitize_grounded_report(raw_report, findings)
         executive = full_report.split("TOP PRIORITIES")[0].replace("EXECUTIVE SUMMARY", "").strip()
         if len(executive) > 600:
             executive = executive[:600] + "..."

app/services/owasp.py

@@ -58,6 +58,14 @@ def classify_owasp(cwe: str = "", alert_name: str = "") -> str:
 
 
 def risk_grade(critical: int, high: int, medium: int) -> str:
+    """Calculate the letter safety grade (A-F) from severity counts.
+
+    Algorithm:
+    - Critical > 2 -> F; Critical >= 1 -> D
+    - High > 3 -> D; High >= 1 -> C
+    - Medium > 5 -> C; Medium >= 1 -> B
+    - 0 Critical, 0 High, 0 Medium -> A
+    """
     if critical > 0:
         return "F" if critical > 2 else "D"
     if high > 3:
@@ -71,6 +79,94 @@ def risk_grade(critical: int, high: int, medium: int) -> str:
     return "A"
 
 
+def explain_risk_rule(critical: int, high: int, medium: int, low: int = 0) -> dict[str, str | int]:
+    """Return the letter grade, numeric safety score (0-100), and plain-English explanation."""
+    grade = risk_grade(critical, high, medium)
+    penalty = (critical * 25) + (high * 12) + (medium * 4) + (low * 1)
+    numeric_score = max(0, 100 - penalty)
+    reasons = {
+        "A": "No critical, high, or medium risks detected.",
+        "B": "Only moderate configuration issues found (1-5 medium findings, 0 high/critical).",
+        "C": "Serious or numerous moderate issues require developer attention (1-3 high or >5 medium).",
+        "D": "High-risk vulnerabilities present (1-2 critical or >3 high findings).",
+        "F": "Urgent intervention required (>2 critical vulnerabilities).",
+    }
+    return {
+        "grade": grade,
+        "numeric_score": numeric_score,
+        "explanation": reasons[grade],
+    }
+
+
+OWASP_REMEDIATION_CARDS: dict[str, dict[str, str]] = {
+    "A01:2021": {
+        "code": "A01:2021",
+        "title": "Access Control & Form Protection",
+        "what_it_means": "Pages or actions that should be private can be opened or triggered without proper checks.",
+        "how_to_fix": "Require login on admin/user pages, verify user ownership for every record ID, and add CSRF tokens to all POST forms.",
+    },
+    "A02:2021": {
+        "code": "A02:2021",
+        "title": "Data & Connection Encryption",
+        "what_it_means": "Passwords or cookies may travel over unencrypted connections where attackers on Wi-Fi can read them.",
+        "how_to_fix": "Enable HTTPS across your entire website, redirect HTTP to HTTPS, and mark login cookies with Secure and HttpOnly.",
+    },
+    "A03:2021": {
+        "code": "A03:2021",
+        "title": "Input & Database Injection",
+        "what_it_means": "Search boxes, URL parameters, or forms accept untrusted commands that can read your database or run scripts.",
+        "how_to_fix": "Use prepared/parameterised database queries and escape all user text before displaying it on a page.",
+    },
+    "A04:2021": {
+        "code": "A04:2021",
+        "title": "Safe Workflow Design",
+        "what_it_means": "Important business forms (like password resets or public submissions) lack verification steps or spam protection.",
+        "how_to_fix": "Require one-time reset tokens for password recovery and add CAPTCHA or rate limits to public forms.",
+    },
+    "A05:2021": {
+        "code": "A05:2021",
+        "title": "Website Security Settings",
+        "what_it_means": "Browser safety headers are missing or internal folders/files (.env, backups) are publicly visible.",
+        "how_to_fix": "Add standard security headers (CSP, HSTS, X-Frame-Options), disable directory listing, and block access to dotfiles and backups.",
+    },
+    "A06:2021": {
+        "code": "A06:2021",
+        "title": "Outdated Website Libraries",
+        "what_it_means": "Your website loads older JavaScript or framework versions that have publicly known security bugs.",
+        "how_to_fix": "Update jQuery, Bootstrap, WordPress plugins, and frontend packages to their current stable versions.",
+    },
+    "A07:2021": {
+        "code": "A07:2021",
+        "title": "Login & Account Protection",
+        "what_it_means": "Login forms allow unlimited password guessing or expose credentials in browser history.",
+        "how_to_fix": "Serve login forms over HTTPS POST only, limit failed login attempts, and enable two-factor authentication for admins.",
+    },
+    "A08:2021": {
+        "code": "A08:2021",
+        "title": "Software & Script Integrity",
+        "what_it_means": "External scripts loaded from third-party CDNs are not verified with cryptographic hashes.",
+        "how_to_fix": "Add Subresource Integrity (integrity='sha384-...') attributes to third-party script tags.",
+    },
+    "A09:2021": {
+        "code": "A09:2021",
+        "title": "Error Messages & Monitoring",
+        "what_it_means": "Detailed technical error messages or stack traces are shown to visitors when something breaks.",
+        "how_to_fix": "Show friendly generic error pages to visitors and keep technical stack traces in private server logs only.",
+    },
+    "A10:2021": {
+        "code": "A10:2021",
+        "title": "Outbound Server Requests (SSRF)",
+        "what_it_means": "URL parameters accept external web addresses that could trick your server into fetching internal resources.",
+        "how_to_fix": "Allowlist trusted domains and block requests to internal/private IP addresses (127.0.0.1, 169.254.x.x).",
+    },
+}
+
+
+def get_remediation_card(owasp_category: str) -> dict[str, str]:
+    code = (owasp_category or "").split(" ")[0].strip()
+    return dict(OWASP_REMEDIATION_CARDS.get(code, OWASP_REMEDIATION_CARDS["A05:2021"]))
+
+
 def progress_for_status(status: str) -> int:
     return {
         "queued": 5,

app/services/secrets.py

@@ -1,14 +1,100 @@
-"""Startup secrets validation (Drips Wave #29 pattern)."""
+"""Startup secrets validation and passphrase-encrypted API key storage (Drips Wave #29 pattern)."""
 
+import base64
+import hashlib
+import hmac
 import logging
 import os
+import secrets as std_secrets
 
 from app.config import Settings
 
 logger = logging.getLogger(__name__)
 
 DEFAULT_ZAP_KEY = "changeme"
 INSECURE_ZAP_KEYS = frozenset({"changeme", "zap", "password", "admin", ""})
+ENCRYPTED_PREFIX = "enc:v1:"
+PBKDF2_ITERATIONS = 100_000
+
+
+def is_encrypted_secret(value: str) -> bool:
+    """Return True if the secret value uses the enc:v1: encrypted envelope."""
+    return bool(value and value.strip().startswith(ENCRYPTED_PREFIX))
+
+
+def _derive_keys(passphrase: str, salt: bytes) -> tuple[bytes, bytes]:
+    if not passphrase:
+        raise ValueError("Passphrase must not be empty")
+    key_material = hashlib.pbkdf2_hmac(
+        "sha256",
+        passphrase.encode("utf-8"),
+        salt,
+        PBKDF2_ITERATIONS,
+        dklen=64,
+    )
+    return key_material[:32], key_material[32:]
+
+
+def _keystream_xor(data: bytes, enc_key: bytes, salt: bytes) -> bytes:
+    out = bytearray(len(data))
+    block_idx = 0
+    offset = 0
+    while offset < len(data):
+        block = hashlib.sha256(enc_key + salt + block_idx.to_bytes(4, "big")).digest()
+        chunk = data[offset : offset + 32]
+        for i, b in enumerate(chunk):
+            out[offset + i] = b ^ block[i]
+        offset += 32
+        block_idx += 1
+    return bytes(out)
+
+
+def encrypt_secret(plaintext: str, passphrase: str) -> str:
+    """Encrypt an API key using PBKDF2-HMAC-SHA256 + authenticated keystream (enc:v1:...)."""
+    if not plaintext:
+        raise ValueError("Plaintext secret must not be empty")
+    salt = std_secrets.token_bytes(16)
+    enc_key, mac_key = _derive_keys(passphrase, salt)
+    pt_bytes = plaintext.encode("utf-8")
+    ct_bytes = _keystream_xor(pt_bytes, enc_key, salt)
+    mac = hmac.new(mac_key, salt + ct_bytes, hashlib.sha256).digest()
+    salt_b64 = base64.urlsafe_b64encode(salt).decode("ascii")
+    ct_b64 = base64.urlsafe_b64encode(ct_bytes).decode("ascii")
+    mac_b64 = base64.urlsafe_b64encode(mac).decode("ascii")
+    return f"{ENCRYPTED_PREFIX}{salt_b64}:{ct_b64}:{mac_b64}"
+
+
+def decrypt_secret(token: str, passphrase: str) -> str:
+    """Decrypt an enc:v1:... encrypted secret and verify its HMAC integrity tag."""
+    if not is_encrypted_secret(token):
+        raise ValueError("Secret does not have enc:v1: prefix")
+    body = token.strip()[len(ENCRYPTED_PREFIX) :]
+    parts = body.split(":")
+    if len(parts) != 3:
+        raise ValueError("Malformed enc:v1: secret envelope")
+    salt = base64.urlsafe_b64decode(parts[0].encode("ascii"))
+    ct_bytes = base64.urlsafe_b64decode(parts[1].encode("ascii"))
+    expected_mac = base64.urlsafe_b64decode(parts[2].encode("ascii"))
+    enc_key, mac_key = _derive_keys(passphrase, salt)
+    actual_mac = hmac.new(mac_key, salt + ct_bytes, hashlib.sha256).digest()
+    if not hmac.compare_digest(actual_mac, expected_mac):
+        raise ValueError("Invalid passphrase or corrupted encrypted secret")
+    pt_bytes = _keystream_xor(ct_bytes, enc_key, salt)
+    return pt_bytes.decode("utf-8")
+
+
+def resolve_anthropic_api_key(settings: Settings, passphrase: str | None = None) -> str:
+    """Resolve ANTHROPIC_API_KEY, decrypting enc:v1: tokens when a passphrase is provided."""
+    raw = (settings.anthropic_api_key or "").strip()
+    if not raw:
+        return ""
+    if not is_encrypted_secret(raw):
+        return raw
+    effective_passphrase = passphrase if passphrase is not None else os.getenv("SHIELDSCAN_SECRET_PASSPHRASE", "")
+    if not effective_passphrase:
+        logger.warning("[secrets] ANTHROPIC_API_KEY is encrypted (enc:v1:) but SHIELDSCAN_SECRET_PASSPHRASE is not set")
+        return ""
+    return decrypt_secret(raw, effective_passphrase)
 
 
 def validate_settings(settings: Settings) -> list[str]:
@@ -26,9 +112,19 @@ def validate_settings(settings: Settings) -> list[str]:
     if settings.scanner_mode == "zap" and not settings.zap_api_url:
         warnings.append("WARNING: SCANNER_MODE=zap but ZAP_API_URL is empty")
 
-    if env == "production" and settings.anthropic_api_key.startswith("sk-ant-"):
-        if len(settings.anthropic_api_key) < 20:
-            warnings.append("WARNING: ANTHROPIC_API_KEY appears truncated")
+    if settings.anthropic_api_key:
+        if is_encrypted_secret(settings.anthropic_api_key):
+            if not os.getenv("SHIELDSCAN_SECRET_PASSPHRASE"):
+                warnings.append(
+                    "WARNING: ANTHROPIC_API_KEY is encrypted (enc:v1:) but SHIELDSCAN_SECRET_PASSPHRASE is missing"
+                )
+        elif env == "production":
+            warnings.append(
+                "WARNING: ANTHROPIC_API_KEY is stored in plaintext without encryption in production — "
+                "use passphrase-encrypted storage (enc:v1:...)"
+            )
+            if settings.anthropic_api_key.startswith("sk-ant-") and len(settings.anthropic_api_key) < 20:
+                warnings.append("WARNING: ANTHROPIC_API_KEY appears truncated")
 
     for w in warnings:
         logger.warning("[secrets] %s", w)