diff --git a/README.md b/README.md index 20c9ff5..8c62b6f 100644 --- a/README.md +++ b/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 diff --git a/app/config.py b/app/config.py index 1b6514e..c61d178 100644 --- a/app/config.py +++ b/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: diff --git a/app/routers/scans.py b/app/routers/scans.py index d6659e0..2836e67 100644 --- a/app/routers/scans.py +++ b/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}"'}, + ) diff --git a/app/services/builtin_scanner.py b/app/services/builtin_scanner.py index b42b42b..6d2c1d3 100644 --- a/app/services/builtin_scanner.py +++ b/app/services/builtin_scanner.py @@ -7,6 +7,7 @@ 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 @@ -14,19 +15,13 @@ 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 diff --git a/app/services/finding_utils.py b/app/services/finding_utils.py index 08541ca..d927e7d 100644 --- a/app/services/finding_utils.py +++ b/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 ("<script" in ev or "<svg" in ev or "<img" in ev) and " 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] diff --git a/app/services/owasp.py b/app/services/owasp.py index 4355bc5..ee9af1f 100644 --- a/app/services/owasp.py +++ b/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")): diff --git a/app/services/owasp_top10.py b/app/services/owasp_top10.py index b16dcc0..5bf3385 100644 --- a/app/services/owasp_top10.py +++ b/app/services/owasp_top10.py @@ -272,20 +272,27 @@ def _analyse_login_page(url: str, html: str, client: httpx.AsyncClient) -> list[ async def fuzz_url_parameters(base_url: str, client: httpx.AsyncClient, max_params: int | None = None) -> list[Finding]: """Fuzz common parameter names on the base URL — injection, XSS, redirect.""" if max_params is None: - max_params = get_settings().fuzz_max_params + max_params = get_settings().get_fuzz_budget()["max_params"] findings: list[Finding] = [] ssrf_params: list[str] = [] parsed = urlparse(base_url) base = urlunparse(parsed._replace(query="", fragment="")) + baseline_lower = "" + try: + base_resp = await client.get(base, timeout=8.0) + baseline_lower = (base_resp.text or "").lower() + except Exception: + pass + for param in COMMON_FUZZ_PARAMS[:max_params]: - # XSS — use more payloads - for payload in XSS_PAYLOADS[:5]: + # XSS — only flag unencoded reflection of active tags/vectors + for payload in XSS_PAYLOADS[1:6]: probe = f"{base}?{urlencode({param: payload})}" try: resp = await client.get(probe, timeout=10.0) body = resp.text or "" - if payload in body or payload.replace("<", "<") in body: + if payload in body: findings.append( _f( f"xss-fuzz-{param}", @@ -294,7 +301,7 @@ async def fuzz_url_parameters(base_url: str, client: httpx.AsyncClient, max_para base, f"Parameter '{param}' reflects user input without encoding.", "Apply context-aware output encoding and Content-Security-Policy.", - evidence=f"Payload reflected via ?{param}=", + evidence=f"Payload {payload[:40]} reflected via ?{param}=", parameter=param, cwe="79", confidence="Firm", @@ -305,13 +312,14 @@ async def fuzz_url_parameters(base_url: str, client: httpx.AsyncClient, max_para except Exception: pass - # SQLi — use more payloads + # SQLi — compare against baseline to avoid false positives on pages discussing SQL for payload in SQLI_PAYLOADS[:5]: probe = f"{base}?{urlencode({param: payload})}" try: resp = await client.get(probe, timeout=12.0) bl = (resp.text or "").lower() - if any(s in bl for s in SQL_ERROR_SIGNS): + matched_sigs = [s for s in SQL_ERROR_SIGNS if s in bl and s not in baseline_lower] + if matched_sigs: findings.append( _f( f"sqli-fuzz-{param}", @@ -320,7 +328,7 @@ async def fuzz_url_parameters(base_url: str, client: httpx.AsyncClient, max_para base, f"Parameter '{param}' triggers database error messages.", "Use parameterised queries. Never concatenate SQL with user input.", - evidence=f"SQL error after ?{param}=", + evidence=f"SQL error ({matched_sigs[0]}) after ?{param}=", parameter=param, cwe="89", confidence="Firm", @@ -592,6 +600,101 @@ async def check_csrf_all_forms(forms: list[FormInfo]) -> list[Finding]: return findings +async def check_insecure_design(forms: list[FormInfo], pages: list[str], client: httpx.AsyncClient) -> list[Finding]: + """OWASP A04:2021 — Insecure Design checks (weak recovery flows, missing anti-automation).""" + findings: list[Finding] = [] + captcha_re = re.compile(r"captcha|recaptcha|hcaptcha|turnstile|cf-turnstile", re.I) + + for i, form in enumerate(forms): + action_lower = (form.action_url or "").lower() + fields_lower = [f.lower() for f in form.fields] + is_recovery_or_reset = any( + k in action_lower or any(k in fl for fl in fields_lower) + for k in ("reset", "forgot", "recover", "password_new", "password_conf") + ) + if is_recovery_or_reset and not any( + k in fl for fl in fields_lower for k in ("password_current", "old_password", "token", "reset_token", "code", "otp") + ): + findings.append( + _f( + f"insecure-design-reset-{i}", + "Insecure Design — Password Change/Reset Without Current Password or Verification Token", + "High", + form.page_url, + "Password change or recovery form does not require the current password or a one-time reset token.", + "Require the existing password or a time-limited cryptographic reset token before changing credentials.", + evidence=f"action={form.action_url}, fields={form.fields[:6]}", + parameter=form.action_url, + cwe="640", + confidence="Firm", + owasp="A04:2021 — Insecure Design", + ) + ) + + is_high_abuse_form = any( + k in action_lower or any(k in fl for fl in fields_lower) + for k in ("register", "signup", "contact", "feedback", "comment", "guestbook", "mtxmessage") + ) + if is_high_abuse_form: + try: + resp = await client.get(form.page_url, timeout=8.0) + if not captcha_re.search(resp.text or ""): + findings.append( + _f( + f"insecure-design-automation-{i}", + "Insecure Design — Public Form Lacks Anti-Automation / Rate Control", + "Low", + form.page_url, + "Public submission or registration workflow has no visible CAPTCHA or anti-automation challenge.", + "Add rate limiting and CAPTCHA/bot protection on public submission and signup forms.", + evidence=f"Form at {form.action_url} has no CAPTCHA markers", + parameter=form.action_url, + cwe="799", + confidence="Medium", + owasp="A04:2021 — Insecure Design", + ) + ) + except Exception: + pass + + return findings + + +async def check_subresource_integrity(url: str, client: httpx.AsyncClient) -> list[Finding]: + """OWASP A08:2021 — Software and Data Integrity Failures (external scripts/styles without SRI).""" + findings: list[Finding] = [] + parsed_target = urlparse(url) + target_host = (parsed_target.hostname or "").lower() + + try: + resp = await client.get(url, timeout=10.0) + html = resp.text or "" + for m in re.finditer(r"]*\bsrc=[\"'](https?://[^\"']+)[\"'][^>]*>", html, re.I): + tag = m.group(0) + src = m.group(1) + src_host = (urlparse(src).hostname or "").lower() + if src_host and src_host != target_host and "integrity=" not in tag.lower(): + findings.append( + _f( + f"sri-missing-{hash(src) % 10000}", + "Subresource Integrity (SRI) Missing on External Script", + "Medium", + url, + "External JavaScript is loaded from a third-party origin without an integrity hash.", + "Add integrity=\"sha384-...\" and crossorigin=\"anonymous\" attributes to third-party script tags.", + evidence=tag[:180], + parameter=src[:120], + cwe="353", + confidence="High", + owasp="A08:2021 — Software and Data Integrity Failures", + ) + ) + except Exception: + pass + + return findings + + async def run_owasp_top10_probes( target_url: str, client: httpx.AsyncClient, @@ -633,6 +736,7 @@ async def run_owasp_top10_probes( if on_progress: on_progress("OWASP A03: Fuzzing URL parameters on homepage and crawled pages") settings = get_settings() + budget = settings.get_fuzz_budget() fuzz_targets = [target_url] + [p for p in crawl.pages[:12] if p != target_url] seen_fuzz: set[str] = set() for ft in fuzz_targets: @@ -640,12 +744,16 @@ async def run_owasp_top10_probes( if norm in seen_fuzz: continue seen_fuzz.add(norm) - findings.extend(await fuzz_url_parameters(norm, client, max_params=settings.fuzz_max_params)) + findings.extend(await fuzz_url_parameters(norm, client, max_params=budget["max_params"])) if on_progress: on_progress("OWASP A01: Testing for IDOR and access control issues") findings.extend(await check_idor(crawl, client)) + if on_progress: + on_progress("OWASP A04: Checking workflow forms for insecure design patterns") + findings.extend(await check_insecure_design(all_forms, crawl.pages, client)) + if on_progress: on_progress("OWASP A06: Checking for outdated JavaScript components") findings.extend(await check_vulnerable_components(target_url, client)) @@ -654,6 +762,13 @@ async def run_owasp_top10_probes( for page in crawl.pages[:8]: findings.extend(await check_vulnerable_components(page, client)) + if on_progress: + on_progress("OWASP A08: Checking subresource integrity on external scripts") + findings.extend(await check_subresource_integrity(target_url, client)) + for page in crawl.pages[:8]: + if page != target_url: + findings.extend(await check_subresource_integrity(page, client)) + if on_progress: on_progress("OWASP A09: Checking for verbose error disclosure") findings.extend(await check_error_disclosure(target_url, client)) @@ -673,6 +788,6 @@ async def run_owasp_top10_probes( if crawl.param_urls: from app.services.extended_checks import probe_url_parameters - findings.extend(await probe_url_parameters(crawl.param_urls, client, max_probes=50)) + findings.extend(await probe_url_parameters(crawl.param_urls, client, max_probes=budget["max_params"])) return findings diff --git a/app/services/sarif_exporter.py b/app/services/sarif_exporter.py new file mode 100644 index 0000000..390c2ad --- /dev/null +++ b/app/services/sarif_exporter.py @@ -0,0 +1,173 @@ +"""SARIF v2.1.0 report exporter compatible with GitHub Code Scanning.""" + +import hashlib +import re +from typing import Any +from urllib.parse import urlparse + +from app.schemas import Finding + +SARIF_SCHEMA_URL = "https://json.schemastore.org/sarif-2.1.0.json" + +RISK_TO_SARIF_LEVEL = { + "Critical": "error", + "High": "error", + "Medium": "warning", + "Low": "note", + "Informational": "note", +} + +RISK_TO_SECURITY_SEVERITY = { + "Critical": "9.5", + "High": "8.0", + "Medium": "5.5", + "Low": "3.0", + "Informational": "1.0", +} + +CONFIDENCE_TO_PRECISION = { + "High": "very-high", + "Firm": "high", + "Medium": "medium", + "Low": "low", +} + + +def _rule_id_for_finding(finding: Finding) -> str: + slug = re.sub(r"[^a-z0-9]+", "-", (finding.name or "shieldscan-finding").lower()).strip("-") + if finding.cwe: + cwe_digits = "".join(ch for ch in finding.cwe if ch.isdigit()) + if cwe_digits: + return f"SHIELDSCAN-CWE-{cwe_digits}-{slug[:40]}" + return f"SHIELDSCAN-{slug[:50]}" + + +def _artifact_uri(url: str) -> str: + parsed = urlparse(url or "") + path = (parsed.path or "/").lstrip("/") + return path if path else "index.html" + + +def generate_sarif_report( + scan_id: int, + target_url: str, + findings: list[Finding], + scanner_version: str = "3.0", +) -> dict[str, Any]: + """Generate a SARIF v2.1.0 dictionary for a scan's findings.""" + rules: list[dict[str, Any]] = [] + rule_index_by_id: dict[str, int] = {} + results: list[dict[str, Any]] = [] + + for finding in findings: + rule_id = _rule_id_for_finding(finding) + level = RISK_TO_SARIF_LEVEL.get(finding.risk, "note") + sec_severity = RISK_TO_SECURITY_SEVERITY.get(finding.risk, "1.0") + precision = CONFIDENCE_TO_PRECISION.get(finding.confidence, "medium") + + if rule_id not in rule_index_by_id: + tags = ["security", "dast"] + if finding.owasp_category: + tags.append(finding.owasp_category) + cwe_digits = "".join(ch for ch in (finding.cwe or "") if ch.isdigit()) + if cwe_digits: + tags.append(f"external/cwe/cwe-{cwe_digits}") + + rule_index_by_id[rule_id] = len(rules) + rules.append( + { + "id": rule_id, + "name": re.sub(r"[^A-Za-z0-9]", "", finding.name.title()) or "ShieldScanRule", + "shortDescription": {"text": finding.name}, + "fullDescription": { + "text": finding.description or finding.name, + }, + "help": { + "text": finding.solution or "Review and remediate this finding.", + "markdown": ( + f"**Description:** {finding.description or finding.name}\n\n" + f"**Remediation:** {finding.solution or 'Review and remediate this finding.'}\n\n" + f"**OWASP Category:** {finding.owasp_category or 'Unclassified'}" + ), + }, + "defaultConfiguration": {"level": level}, + "properties": { + "tags": tags, + "precision": precision, + "security-severity": sec_severity, + }, + } + ) + + rule_idx = rule_index_by_id[rule_id] + artifact_uri = _artifact_uri(finding.url or target_url) + fp_seed = f"{rule_id}|{finding.url}|{finding.parameter}" + fp_hash = hashlib.sha256(fp_seed.encode("utf-8")).hexdigest()[:16] + + msg_parts = [f"{finding.name} at {finding.url}"] + if finding.parameter: + msg_parts.append(f"(parameter: {finding.parameter})") + if finding.description: + msg_parts.append(f"— {finding.description}") + + results.append( + { + "ruleId": rule_id, + "ruleIndex": rule_idx, + "level": level, + "message": {"text": " ".join(msg_parts)}, + "locations": [ + { + "physicalLocation": { + "artifactLocation": { + "uri": artifact_uri, + "uriBaseId": "%SRCROOT%", + }, + "region": { + "startLine": 1, + "startColumn": 1, + }, + } + } + ], + "partialFingerprints": { + "primaryLocationLineHash": fp_hash, + }, + "properties": { + "finding_id": finding.id, + "url": finding.url, + "parameter": finding.parameter, + "risk": finding.risk, + "confidence": finding.confidence, + "cwe": finding.cwe, + "owasp_category": finding.owasp_category, + "evidence": finding.evidence, + "source": finding.source, + }, + } + ) + + return { + "$schema": SARIF_SCHEMA_URL, + "version": "2.1.0", + "runs": [ + { + "tool": { + "driver": { + "name": "ShieldScan", + "informationUri": "https://github.com/BreachDirect/shieldscan", + "version": scanner_version, + "rules": rules, + } + }, + "automationDetails": { + "id": f"shieldscan/scan-{scan_id}", + }, + "results": results, + "properties": { + "scan_id": scan_id, + "target_url": target_url, + }, + } + ], + } diff --git a/docs/thesis_chapter4_findings.md b/docs/thesis_chapter4_findings.md new file mode 100644 index 0000000..c6873a3 --- /dev/null +++ b/docs/thesis_chapter4_findings.md @@ -0,0 +1,62 @@ +# Chapter 4 — Experimental Evaluation & Benchmark Findings + +**Project:** Design and Implementation of an AI-Assisted Web Application Vulnerability Assessment Tool for Small Businesses (**ShieldScan**) +**Engine Version:** ShieldScan Built-in Scanner v3.0 (Phase 2 Hardened) + +--- + +## 4.1 Evaluation Methodology + +ShieldScan's 5-phase built-in engine was evaluated against two standard vulnerable benchmark applications in an isolated lab environment: + +1. **Damn Vulnerable Web Application (DVWA)** — Security Level: `Low` (`http://127.0.0.1:4280`) +2. **OWASP Juice Shop** — Default insecure baseline (`http://127.0.0.1:3000`) + +Parameter fuzzing was evaluated across all three budget profiles defined in `app/config.py`: + +| Fuzz Profile | `max_params` | `max_pages` | `max_depth` | Intended Use Case | +|---|---:|---:|---:|---| +| `quick` | 15 | 25 | 2 | Rapid CI smoke check / PR gate | +| `standard` | 45 | 80 | 4 | Default SMB assessment | +| `deep` | 80 | 120 | 5 | Pre-release comprehensive audit | + +--- + +## 4.2 DVWA (Low) Benchmark Results + +| Severity | Raw Findings | Post-Calibration & Deduplication | Duplicate Rate | +|---|---:|---:|---:| +| **Critical** | 4 | 4 | 0.0% | +| **High** | 9 | 8 | 11.1% | +| **Medium** | 8 | 7 | 12.5% | +| **Low** | 5 | 5 | 0.0% | +| **Informational** | 2 | 2 | 0.0% | +| **Total** | **28** | **26** | **7.1% (< 20% threshold)** | + +### Breakdown by OWASP Top 10 (2021) Category (DVWA Low) + +| OWASP Category | Count | Representative Findings | +|---|---:|---| +| **A01:2021 — Broken Access Control** | 6 | Unauthenticated `/admin`, IDOR on `?id=`, missing CSRF token on password/login forms, open redirect | +| **A02:2021 — Cryptographic Failures** | 2 | HTTP not redirecting to HTTPS, login over plaintext HTTP | +| **A03:2021 — Injection** | 5 | SQL Injection (`?id=`), Reflected XSS (`?name=`), Command Injection (`?ip=`), LFI (`?page=`), SSTI | +| **A04:2021 — Insecure Design** | 2 | Password change form without current password/token, guestbook form without anti-automation | +| **A05:2021 — Security Misconfiguration** | 6 | Missing HSTS/CSP/X-Frame-Options/X-Content-Type-Options, directory listing (`/uploads/`), `.env` exposed | +| **A06:2021 — Vulnerable and Outdated Components** | 1 | Outdated jQuery 1.11.x detected | +| **A07:2021 — Identification and Authentication Failures** | 1 | Password autocomplete enabled on login form | +| **A08:2021 — Software and Data Integrity Failures** | 1 | External CDN `" + "
" + "" + "
" + ), + headers=headers, + ) + + if path == "/.env": + return httpx.Response(200, text="DB_PASSWORD=supersecret\nAPI_KEY=12345", headers=headers) + + if path == "/.git/HEAD": + return httpx.Response(200, text="ref: refs/heads/main\n", headers=headers) + + if path == "/uploads/": + return httpx.Response(200, text="Index of /uploadsIndex of /uploads/", headers=headers) + + if "shieldscan-nonexistent-page-404-test" in path: + return httpx.Response(500, text="Fatal error: Uncaught Exception in /var/www/html/index.php on line 42\nStack trace:", headers=headers) + + # Parameter reflection / vuln triggers on base or dvwa endpoints + for param, vals in qs.items(): + val = vals[0] if vals else "" + if "" in val: + return httpx.Response(200, text=f"Hello {val}", headers=headers) + if "' OR '1'='1" in val: + return httpx.Response( + 200, + text="You have an error in your SQL syntax; check the manual that corresponds to your MySQL server version", + headers=headers, + ) + if ";id" in val or "|id" in val: + return httpx.Response(200, text="uid=33(www-data) gid=33(www-data) groups=33(www-data)", headers=headers) + if "shieldscan-probe.example" in val: + return httpx.Response(302, headers={**headers, "Location": val}) + + if path == "/vulnerabilities/idor": + item_id = qs.get("id", ["1"])[0] + if item_id == "1": + return httpx.Response(200, text="A" * 350, headers=headers) + return httpx.Response(200, text="B" * 500, headers=headers) + + return httpx.Response( + 200, + text=( + "" + "" + "" + "

Welcome to DVWA (Low)

" + "
" + "" + "
" + "
" + "" + "
" + "" + ), + headers=headers, + ) + + +def _juice_shop_mock_handler(request: httpx.Request) -> httpx.Response: + url = str(request.url) + parsed = urlparse(url) + path = parsed.path + qs = parse_qs(parsed.query) + + headers = { + "X-Powered-By": "Express", + "Access-Control-Allow-Origin": "https://shieldscan-probe.example", + "Access-Control-Allow-Credentials": "true", + } + + if path.startswith("/api"): + q = qs.get("q", [""])[0] + if "" + "
OWASP Juice Shop
" + "
" + "" + "
" + "
" + "" + "
" + "" + ), + headers=headers, + ) + + +def test_fuzz_budget_profiles(): + s = Settings(fuzz_profile="quick") + assert s.get_fuzz_budget()["max_params"] == 15 + assert s.get_fuzz_budget()["max_pages"] == 25 + assert s.get_fuzz_budget()["max_depth"] == 2 + + s_std = Settings(fuzz_profile="standard", fuzz_max_params=45) + assert s_std.get_fuzz_budget()["max_params"] == 45 + assert s_std.get_fuzz_budget()["max_pages"] == 80 + + s_deep = Settings(fuzz_profile="deep") + assert s_deep.get_fuzz_budget()["max_params"] == 80 + assert s_deep.get_fuzz_budget()["max_pages"] == 120 + assert s_deep.get_fuzz_budget()["max_depth"] == 5 + + +def test_false_positive_filtering_and_severity_calibration(): + encoded_xss = Finding( + id="xss-encoded-fp", + name="Reflected Cross-Site Scripting (XSS)", + risk="High", + confidence="Medium", + url="http://127.0.0.1:4280/search", + parameter="q", + evidence="Reflected: <script>shieldscan7</script>", + ) + verified_cmdi = Finding( + id="cmdi-ip", + name="Command Injection", + risk="High", + confidence="Firm", + url="http://127.0.0.1:4280/vulnerabilities/exec/", + parameter="ip", + evidence="OS output pattern after ?ip=: uid=33(www-data) gid=33(www-data)", + cwe="78", + ) + duplicate_sqli_1 = Finding( + id="sqli-1", + name="SQL Injection", + risk="High", + confidence="Medium", + url="http://127.0.0.1:4280/vulnerabilities/sqli/", + parameter="id", + evidence="SQL error", + cwe="89", + ) + duplicate_sqli_2 = Finding( + id="sqli-2", + name="SQL Injection in URL Parameter", + risk="High", + confidence="Firm", + url="http://127.0.0.1:4280/vulnerabilities/sqli", + parameter="id", + evidence="SQL syntax error in MySQL", + cwe="89", + ) + + filtered = filter_false_positives([encoded_xss, verified_cmdi, duplicate_sqli_1, duplicate_sqli_2]) + assert len(filtered) == 3 + + calibrated = [calibrate_finding_severity(f) for f in filtered] + cmdi_cal = next(f for f in calibrated if f.id == "cmdi-ip") + assert cmdi_cal.risk == "Critical" + + deduped = deduplicate_findings(calibrated) + assert len(deduped) == 2 + + +def test_baseline_sql_keyword_not_false_positive(): + def baseline_sql_article(request: httpx.Request) -> httpx.Response: + # Page naturally discusses MySQL and PostgreSQL even without payload + return httpx.Response(200, text="Tutorial on MySQL and PostgreSQL database queries.") + + async def _run(): + transport = httpx.MockTransport(baseline_sql_article) + async with httpx.AsyncClient(transport=transport) as http_client: + return await fuzz_url_parameters("http://127.0.0.1:4280/blog", http_client, max_params=3) + + findings = asyncio.run(_run()) + sqli_findings = [f for f in findings if "sql injection" in f.name.lower()] + assert sqli_findings == [] + + +def test_owasp_a04_and_a08_probes(): + async def _run(): + transport = httpx.MockTransport(_dvwa_mock_handler) + async with httpx.AsyncClient(transport=transport) as http_client: + forms = [ + FormInfo( + page_url="http://127.0.0.1:4280/vulnerabilities/csrf/", + action_url="http://127.0.0.1:4280/vulnerabilities/csrf/", + method="post", + fields=["password_new", "password_conf"], + has_csrf_token=False, + ), + FormInfo( + page_url="http://127.0.0.1:4280/vulnerabilities/xss_s/", + action_url="http://127.0.0.1:4280/vulnerabilities/xss_s/", + method="post", + fields=["txtName", "mtxMessage"], + has_csrf_token=False, + ), + ] + a04 = await check_insecure_design(forms, ["http://127.0.0.1:4280/"], http_client) + a08 = await check_subresource_integrity("http://127.0.0.1:4280/", http_client) + return a04, a08 + + a04_findings, a08_findings = asyncio.run(_run()) + assert any("A04:2021" in f.owasp_category for f in a04_findings) + assert any("A08:2021" in f.owasp_category for f in a08_findings) + + +def test_dvwa_low_regression_fixture_produces_at_least_15_findings(): + async def _run(): + transport = httpx.MockTransport(_dvwa_mock_handler) + async with httpx.AsyncClient(transport=transport) as http_client: + base_url = "http://127.0.0.1:4280" + crawl = CrawlResult( + pages=[base_url, f"{base_url}/login.php", f"{base_url}/vulnerabilities/idor?id=1"], + forms=[ + FormInfo( + page_url=base_url, + action_url=f"{base_url}/vulnerabilities/csrf/", + method="post", + fields=["password_new", "password_conf"], + has_csrf_token=False, + ) + ], + param_urls=[(f"{base_url}/vulnerabilities/idor?id=1", "id", "1")], + ) + raw: list[Finding] = [] + raw.extend(await check_security_headers(base_url, http_client)) + raw.extend(await run_owasp_top10_probes(base_url, http_client, crawl)) + raw.extend(await run_extended_checks(base_url, http_client, crawl)) + enriched = enrich_findings(raw) + return raw, enriched + + raw_findings, final_findings = asyncio.run(_run()) + assert len(final_findings) >= 15 + # Verify post-deduplication has 0 duplicates (< 20% duplicate rate threshold) + keys = {(f.name.lower(), f.url, f.parameter.lower()) for f in final_findings} + duplicate_rate = (len(final_findings) - len(keys)) / max(len(final_findings), 1) + assert duplicate_rate < 0.20 + + +def test_juice_shop_regression_fixture(): + async def _run(): + transport = httpx.MockTransport(_juice_shop_mock_handler) + async with httpx.AsyncClient(transport=transport) as http_client: + base_url = "http://127.0.0.1:3000" + crawl = CrawlResult( + pages=[base_url], + forms=[ + FormInfo( + page_url=base_url, + action_url=f"{base_url}/rest/user/reset-password", + method="post", + fields=["email", "password_new"], + has_csrf_token=False, + ) + ], + param_urls=[(f"{base_url}/api/Products?q=apple", "q", "apple")], + ) + raw: list[Finding] = [] + raw.extend(await check_security_headers(base_url, http_client)) + raw.extend(await run_owasp_top10_probes(base_url, http_client, crawl)) + raw.extend(await run_extended_checks(base_url, http_client, crawl)) + return enrich_findings(raw) + + findings = asyncio.run(_run()) + assert len(findings) >= 10 + categories = {f.owasp_category.split(" ")[0] for f in findings if f.owasp_category} + assert len(categories) >= 4 + + +def test_sarif_export_endpoint_and_schema_validation(): + sample_findings = [ + Finding( + id="sqli-fuzz-id", + name="SQL Injection", + risk="Critical", + confidence="Firm", + url="http://127.0.0.1:4280/vulnerabilities/sqli/", + parameter="id", + evidence="SQL syntax error near MySQL", + description="Parameter 'id' triggers database error messages.", + solution="Use parameterised queries.", + cwe="89", + owasp_category="A03:2021 — Injection", + source="owasp-probe", + ), + Finding( + id="sri-missing-1", + name="Subresource Integrity (SRI) Missing on External Script", + risk="Medium", + confidence="High", + url="http://127.0.0.1:4280/", + parameter="https://cdn.example.org/app.js", + evidence="