feat(scanner): Phase 2 built-in scanner hardening & SARIF v2.1.0 export

README.md

@@ -128,6 +128,25 @@ shieldscan/
 | GET | `/api/scans/{id}/progress` | Live progress |
 | GET | `/api/scans/{id}/report/html` | HTML report |
 | GET | `/api/scans/{id}/report/download` | Markdown download |
+| GET | `/api/scans/{id}/report/sarif` | SARIF v2.1.0 export (GitHub Code Scanning) |
+
+## SARIF Export & GitHub Code Scanning
+
+ShieldScan exports scan results in **SARIF v2.1.0** format (`https://json.schemastore.org/sarif-2.1.0.json`) for integration with GitHub Code Scanning and CI security gates:
+
+```bash
+# Export SARIF report for scan #1
+curl -s http://127.0.0.1:8000/api/scans/1/report/sarif -o shieldscan-results.sarif
+
+# Upload to GitHub Code Scanning in CI
+gh api \
+  --method POST \
+  -H "Accept: application/vnd.github+json" \
+  /repos/OWNER/REPO/code-scanning/sarifs \
+  -f commit_sha="$(git rev-parse HEAD)" \
+  -f ref="refs/heads/main" \
+  -f sarif="$(gzip -c shieldscan-results.sarif | base64 -w0)"
+```
 
 ## Development
 

app/config.py

@@ -2,6 +2,24 @@
 
 from pydantic_settings import BaseSettings, SettingsConfigDict
 
+FUZZ_BUDGET_PROFILES: dict[str, dict[str, int]] = {
+    "quick": {
+        "max_params": 15,
+        "max_pages": 25,
+        "max_depth": 2,
+    },
+    "standard": {
+        "max_params": 45,
+        "max_pages": 80,
+        "max_depth": 4,
+    },
+    "deep": {
+        "max_params": 80,
+        "max_pages": 120,
+        "max_depth": 5,
+    },
+}
+
 
 class Settings(BaseSettings):
     model_config = SettingsConfigDict(env_file=".env", env_file_encoding="utf-8", extra="ignore")
@@ -20,10 +38,21 @@ class Settings(BaseSettings):
     scan_spider_timeout: int = 300
     scan_active_timeout: int = 600
     max_scan_duration: int = 900
+    fuzz_profile: str = "standard"  # quick | standard | deep
     crawl_max_pages: int = 80
     crawl_max_depth: int = 4
     fuzz_max_params: int = 45
 
+    def get_fuzz_budget(self, profile: str | None = None) -> dict[str, int]:
+        """Return parameter fuzz and crawl budgets for the selected profile."""
+        selected = (profile or self.fuzz_profile or "standard").strip().lower()
+        base = dict(FUZZ_BUDGET_PROFILES.get(selected, FUZZ_BUDGET_PROFILES["standard"]))
+        if selected == "standard":
+            base["max_params"] = self.fuzz_max_params
+            base["max_pages"] = self.crawl_max_pages
+            base["max_depth"] = self.crawl_max_depth
+        return base
+
 
 @lru_cache
 def get_settings() -> Settings:

app/routers/scans.py

@@ -2,7 +2,7 @@
 import json
 
 from fastapi import APIRouter, BackgroundTasks, Depends
-from fastapi.responses import HTMLResponse, Response
+from fastapi.responses import HTMLResponse, JSONResponse, Response
 from sqlalchemy.orm import Session
 
 from app.config import Settings, get_settings
@@ -12,6 +12,7 @@
 from app.schemas import Finding, ScanCreate, ScanDetail, ScanProgress, ScanSummary
 from app.services.finding_utils import enrich_findings
 from app.services.owasp import progress_for_status
+from app.services.sarif_exporter import generate_sarif_report
 from app.services.scan_orchestrator import ScanOrchestrator
 from app.services.target_guard import validate_scan_request
 
@@ -139,3 +140,26 @@ def download_report(scan_id: int, db: Session = Depends(get_db)):
         media_type="text/markdown",
         headers={"Content-Disposition": f'attachment; filename="{filename}"'},
     )
+
+
+@router.get("/{scan_id}/report/sarif")
+def export_sarif(scan_id: int, db: Session = Depends(get_db)):
+    scan = db.query(Scan).filter(Scan.id == scan_id).first()
+    if not scan:
+        raise_api_error(404, "NOT_FOUND", "Scan not found")
+
+    findings = []
+    if scan.findings_json:
+        findings = enrich_findings([Finding(**f) for f in json.loads(scan.findings_json)])
+
+    sarif = generate_sarif_report(
+        scan_id=scan.id,
+        target_url=scan.target_url,
+        findings=findings,
+    )
+    filename = f"shieldscan-report-{scan_id}.sarif.json"
+    return JSONResponse(
+        content=sarif,
+        media_type="application/sarif+json",
+        headers={"Content-Disposition": f'attachment; filename="{filename}"'},
+    )

app/services/builtin_scanner.py

@@ -7,26 +7,21 @@
 from app.services.crawler import crawl_site
 from app.services.deep_probes import run_deep_probes
 from app.services.extended_checks import run_extended_checks
+from app.services.finding_utils import enrich_findings
 from app.services.owasp_top10 import run_owasp_top10_probes
 from app.services.passive_checks import check_security_headers
 
 logger = logging.getLogger(__name__)
 
 
 def _dedupe_findings(findings: list[Finding]) -> list[Finding]:
-    seen: set[str] = set()
-    unique: list[Finding] = []
-    for f in findings:
-        key = f.id
-        if key not in seen:
-            seen.add(key)
-            unique.append(f)
-    return unique
+    return enrich_findings(findings)
 
 
 async def run_builtin_scan(target_url: str, on_progress=None) -> list[Finding]:
     """Full built-in scan: crawl, OWASP Top 10 probes, headers, extended checks."""
     settings = get_settings()
+    budget = settings.get_fuzz_budget()
     findings: list[Finding] = []
 
     async with httpx.AsyncClient(
@@ -40,8 +35,8 @@ async def run_builtin_scan(target_url: str, on_progress=None) -> list[Finding]:
         crawl = await crawl_site(
             target_url,
             client,
-            max_pages=settings.crawl_max_pages,
-            max_depth=settings.crawl_max_depth,
+            max_pages=budget["max_pages"],
+            max_depth=budget["max_depth"],
         )
         if on_progress:
             on_progress(
@@ -71,11 +66,12 @@ async def run_builtin_scan(target_url: str, on_progress=None) -> list[Finding]:
                 client,
                 crawl,
                 on_progress=on_progress,
-                fuzz_max_params=settings.fuzz_max_params,
+                fuzz_max_params=budget["max_params"],
             )
         )
 
+        deduped = _dedupe_findings(findings)
         if on_progress:
-            on_progress(f"Scan complete — {len(_dedupe_findings(findings))} findings")
+            on_progress(f"Scan complete — {len(deduped)} findings")
 
-    return _dedupe_findings(findings)
+    return deduped

app/services/finding_utils.py

@@ -1,8 +1,120 @@
-"""Normalise findings for display — fill missing parameter labels."""
+"""Normalise findings for display — fill missing parameter labels, filter false positives, and calibrate severity."""
 
 from urllib.parse import urlparse
 
 from app.schemas import Finding
+from app.services.owasp import classify_owasp
+
+VALID_RISKS = ("Critical", "High", "Medium", "Low", "Informational")
+RISK_MAP = {
+    "critical": "Critical",
+    "high": "High",
+    "medium": "Medium",
+    "low": "Low",
+    "informational": "Informational",
+    "info": "Informational",
+}
+
+CONFIDENCE_MAP = {
+    "high": "High",
+    "firm": "Firm",
+    "medium": "Medium",
+    "low": "Low",
+    "confirmed": "High",
+}
+
+
+def is_false_positive(finding: Finding) -> bool:
+    """Detect common scanner false positives before reporting."""
+    name = (finding.name or "").lower()
+    ev = (finding.evidence or "").lower()
+
+    # Reflected XSS where angle brackets were HTML-entity encoded and not executable
+    if "xss" in name or "cross-site scripting" in name:
+        if ("&lt;script" in ev or "&lt;svg" in ev or "&lt;img" in ev) and "<script" not in ev and "<svg" not in ev and "<img" not in ev:
+            return True
+
+    # Noise jQuery 3.x informational pattern without CVE match
+    if "jquery version detected — verify against cve database" in name and finding.confidence.lower() == "low":
+        return True
+
+    return False
+
+
+def filter_false_positives(findings: list[Finding]) -> list[Finding]:
+    """Filter out known false-positive findings."""
+    return [f for f in findings if not is_false_positive(f)]
+
+
+def calibrate_finding_severity(finding: Finding) -> Finding:
+    """Calibrate risk level and confidence based on evidence and vulnerability class."""
+    raw_risk = (finding.risk or "Informational").strip().lower()
+    risk = RISK_MAP.get(raw_risk, "Informational")
+
+    raw_conf = (finding.confidence or "Medium").strip().lower()
+    confidence = CONFIDENCE_MAP.get(raw_conf, "Medium")
+
+    name = (finding.name or "").lower()
+    ev = (finding.evidence or "").lower()
+    owasp_cat = finding.owasp_category or classify_owasp(finding.cwe, finding.name)
+
+    # Upgrade verified RCE / Command Injection / LFI / SQLi with concrete OS/DB proof to Critical
+    if confidence in ("High", "Firm"):
+        if any(k in name for k in ("command injection", "local file inclusion", "path traversal")) and any(
+            sig in ev for sig in ("uid=", "gid=", "root:x:0", "os output pattern", "system file content")
+        ):
+            risk = "Critical"
+        elif "sql injection" in name and any(
+            sig in ev for sig in ("sql syntax", "mysql", "sqlite", "postgresql", "database error", "sql error")
+        ):
+            risk = "Critical" if "union" in ev or "syntax" in ev else max_risk(risk, "High")
+
+    # Downgrade speculative High findings when confidence is Low and evidence is weak
+    if risk == "High" and confidence == "Low" and not ev:
+        risk = "Medium"
+
+    # Passive informational headers should never exceed Medium
+    if finding.source == "passive" and "header" in name and risk in ("Critical", "High"):
+        risk = "Medium"
+
+    return finding.model_copy(
+        update={
+            "risk": risk,
+            "confidence": confidence,
+            "owasp_category": owasp_cat,
+        }
+    )
+
+
+def max_risk(r1: str, r2: str) -> str:
+    order = {"Critical": 4, "High": 3, "Medium": 2, "Low": 1, "Informational": 0}
+    return r1 if order.get(r1, 0) >= order.get(r2, 0) else r2
+
+
+def deduplicate_findings(findings: list[Finding]) -> list[Finding]:
+    """Deduplicate findings across phases by normalised vulnerability class, URL path, and parameter."""
+    seen: dict[tuple[str, str, str], Finding] = {}
+    for f in findings:
+        parsed = urlparse(f.url or "")
+        norm_url = f"{parsed.scheme}://{parsed.netloc}{parsed.path.rstrip('/') or '/'}"
+        norm_name = (f.name or "").strip().lower()
+        # Consolidate Reflected XSS / SQLi across builtin, owasp-probe, and deep-probe on the same parameter
+        if "reflected cross-site scripting" in norm_name or "reflected xss" in norm_name:
+            norm_name = "reflected cross-site scripting (xss)"
+        elif norm_name.startswith("sql injection"):
+            norm_name = "sql injection"
+        elif "missing" in norm_name and "csrf" in norm_name:
+            norm_name = "csrf form missing token"
+
+        norm_param = (f.parameter or "").strip().lower()
+        key = (norm_name, norm_url, norm_param)
+        if key not in seen:
+            seen[key] = f
+        else:
+            existing = seen[key]
+            if max_risk(f.risk, existing.risk) != existing.risk:
+                seen[key] = f
+    return list(seen.values())
 
 
 def enrich_finding_parameter(finding: Finding) -> Finding:
@@ -23,6 +135,10 @@ def enrich_finding_parameter(finding: Finding) -> Finding:
         param = "Access-Control-Allow-Origin"
     elif "csrf" in name:
         param = "HTML form (no CSRF token)"
+    elif "subresource integrity" in name or "sri" in name:
+        param = "External script/stylesheet tag"
+    elif "insecure design" in name:
+        param = "Business workflow form"
     elif "directory listing" in name:
         param = urlparse(finding.url).path or "URL path"
     elif any(k in name for k in ("exposed", "disclosure", "backup", "git", ".env")):
@@ -42,4 +158,8 @@ def enrich_finding_parameter(finding: Finding) -> Finding:
 
 
 def enrich_findings(findings: list[Finding]) -> list[Finding]:
-    return [enrich_finding_parameter(f) for f in findings]
+    """Filter false positives, calibrate severity, deduplicate, and enrich parameter labels."""
+    filtered = filter_false_positives(findings)
+    calibrated = [calibrate_finding_severity(f) for f in filtered]
+    deduped = deduplicate_findings(calibrated)
+    return [enrich_finding_parameter(f) for f in deduped]

app/services/owasp.py

@@ -17,18 +17,33 @@
     "79": "A03:2021",
     "89": "A03:2021",
     "78": "A03:2021",
+    "94": "A03:2021",
+    "98": "A03:2021",
     "22": "A01:2021",
+    "284": "A01:2021",
+    "352": "A01:2021",
+    "601": "A01:2021",
+    "639": "A01:2021",
     "287": "A07:2021",
     "306": "A07:2021",
+    "307": "A07:2021",
     "798": "A07:2021",
     "311": "A02:2021",
     "319": "A02:2021",
     "326": "A02:2021",
     "327": "A02:2021",
+    "640": "A04:2021",
+    "799": "A04:2021",
+    "840": "A04:2021",
     "614": "A05:2021",
     "693": "A05:2021",
     "1021": "A05:2021",
     "829": "A06:2021",
+    "1104": "A06:2021",
+    "353": "A08:2021",
+    "494": "A08:2021",
+    "502": "A08:2021",
+    "209": "A09:2021",
     "918": "A10:2021",
 }
 
@@ -40,8 +55,12 @@ def classify_owasp(cwe: str = "", alert_name: str = "") -> str:
         return f"{code} — {OWASP_MAP[code]}"
 
     name = alert_name.lower()
-    if any(k in name for k in ("sql", "injection", "xss", "script", "command", "ldap", "xpath")):
+    if any(k in name for k in ("sql", "injection", "xss", "script", "command", "ldap", "xpath", "ssti", "lfi")):
         return "A03:2021 — Injection"
+    if any(k in name for k in ("subresource integrity", "sri", "deserialization", "integrity")):
+        return "A08:2021 — Software and Data Integrity Failures"
+    if any(k in name for k in ("insecure design", "business logic", "password reset", "anti-automation")):
+        return "A04:2021 — Insecure Design"
     if any(k in name for k in ("cookie", "ssl", "tls", "crypto", "hash", "password")):
         return "A02:2021 — Cryptographic Failures"
     if any(k in name for k in ("auth", "session", "login", "credential")):