backend/scanners/attack_surface_scanner.py

"""

attack_surface_scanner.py — Attack Surface Mapper

==================================================

Enumerates and maps the full attack surface of a web application:

  - All discovered endpoints / routes

  - Query parameters and form fields across all pages

  - External domains and third-party scripts

  - API endpoints (REST / GraphQL hints)

  - Admin / sensitive paths

  - File upload endpoints

  - Technology stack identified

  - Email addresses / internal references exposed

"""
import re, urllib.parse, urllib.request
from scanners.base_scanner import BaseScanner

SENSITIVE_PATH_RE = re.compile(
    r"(admin|administrator|manager|console|panel|dashboard|config|backup|"
    r"phpmyadmin|wp-admin|wp-login|cpanel|webmail|api/v\d|swagger|graphql|"
    r"actuator|debug|trace|health|metrics|env|info|beans|heapdump)", re.I
)

FILE_UPLOAD_RE = re.compile(
    r'<input[^>]+type=["\']file["\']', re.I
)

EMAIL_RE = re.compile(
    r'\b[A-Za-z0-9._%+\-]+@[A-Za-z0-9.\-]+\.[A-Za-z]{2,}\b'
)

INTERNAL_IP_RE = re.compile(
    r'\b(10\.\d+\.\d+\.\d+|172\.(1[6-9]|2\d|3[01])\.\d+\.\d+|192\.168\.\d+\.\d+)\b'
)

API_PATTERNS = re.compile(
    r'["\']/(api|v\d+|rest|graphql|query|endpoint)[^"\']*["\']', re.I
)


class AttackSurfaceScanner(BaseScanner):
    SCANNER_NAME = "Attack Surface Mapper"
    _SCANNER_KEY = "attack_surface"

    def __init__(self, scan_id, target, domain, **kwargs):
        super().__init__(scan_id, target, domain, **kwargs)

    # ------------------------------------------------------------------
    def run(self) -> list:
        self.log("INFO", f"[AttackSurface] Mapping attack surface of {self.target}...")
        try:
            urls, forms, html, all_html = self._crawl_deep()
            self._analyze_surface(urls, forms, html, all_html)
        except Exception as e:
            self.log("WARNING", f"[AttackSurface] Error: {e}")

        self.log("SUCCESS",
            f"[AttackSurface] Mapping complete. {len(self.vulns)} exposure(s) documented.")
        return self.vulns

    # ------------------------------------------------------------------
    def _crawl_deep(self):
        try:
            results = self.discovery_context or {}
            urls = [u["url"] if isinstance(u, dict) else u for u in results.get("urls", [])]
            if self.target not in urls:
                urls.insert(0, self.target)
            forms = results.get("forms", [])
            all_html = results.get("page_contents", {})
            combined = " ".join(all_html.values()) if all_html else ""
            return urls, forms, combined, all_html
        except Exception as e:
            self.log("ERROR", f"[AttackSurface] _crawl_deep error: {e}")
            body = self._fetch(self.target)
            return [self.target], [], body or "", {}

    def _fetch(self, url):
        try:
            req = urllib.request.Request(url,
                headers={"User-Agent": "LarShield/2.0 AttackSurface"})
            with urllib.request.urlopen(req, timeout=8, context=self.get_ssl_context()) as r:
                return r.read().decode("utf-8", errors="ignore")
        except Exception as e:
            self.log("ERROR", f"[AttackSurface] _fetch error: {e}")
            return ""

    # ------------------------------------------------------------------
    def _analyze_surface(self, urls, forms, combined_html, all_html):
        # ── 1. Endpoint inventory ─────────────────────────────────────
        unique_paths = sorted({urllib.parse.urlparse(u).path for u in urls})
        self.log("INFO",
            f"[AttackSurface] Discovered {len(unique_paths)} unique path(s), "
            f"{len(forms)} form(s)")

        # ── 2. Sensitive / admin paths ────────────────────────────────
        sensitive = [p for p in unique_paths if SENSITIVE_PATH_RE.search(p)]
        if sensitive:
            self.add_vuln(
                title=f"Sensitive/Admin Paths Discovered ({len(sensitive)} paths)",
                severity="Medium",
                category="Attack Surface",
                cvss_score=5.3,
                description="The following sensitive endpoints were found accessible:\n\n" +
                    "\n".join(f"- `{self.target.rstrip('/')}{p}`" for p in sensitive[:20]),
                remediation="1. Restrict access to admin/management paths by IP allowlist.\n"
                    "2. Enforce strong authentication on all /admin, /api, /console paths.\n"
                    "3. Remove or disable unused endpoints in production.",
            )

        # ── 3. API endpoints ──────────────────────────────────────────
        api_paths = [p for p in unique_paths if re.search(r'/api|/v\d+|/graphql|/rest', p, re.I)]
        if api_paths:
            self.log("INFO", f"[AttackSurface] API surface: {len(api_paths)} endpoint(s)")
            self.add_vuln(
                title=f"API Attack Surface: {len(api_paths)} Endpoint(s) Discovered",
                severity="Low",
                category="Attack Surface",
                cvss_score=0.0,
                description="The following API endpoints are part of the attack surface "
                    "and should be audited for authentication, authorization, and input validation:\n\n"
                    + "\n".join(f"- `{self.target.rstrip('/')}{p}`" for p in api_paths[:20]),
                remediation="Ensure all API endpoints require authentication, validate input, "
                    "implement rate limiting, and are covered by the security test suite.",
            )

        # ── 4. File upload endpoints ──────────────────────────────────
        upload_forms = [f for f in forms
                        if FILE_UPLOAD_RE.search(str(f.get("raw_html","")))]
        if upload_forms:
            self.add_vuln(
                title=f"File Upload Endpoint(s) Detected ({len(upload_forms)} forms)",
                severity="High",
                category="Attack Surface",
                cvss_score=7.5,
                description=f"Found {len(upload_forms)} file upload form(s). "
                    "File upload functionality is a high-risk attack surface and must be "
                    "protected against:\n"
                    "- Unrestricted file type uploads (webshell upload)\n"
                    "- Malicious filename attacks (path traversal)\n"
                    "- Oversized file DoS\n"
                    "- MIME-type spoofing",
                remediation="1. Validate file types by magic bytes, not extension/MIME.\n"
                    "2. Store uploads outside the web root.\n"
                    "3. Rename uploaded files to random UUIDs.\n"
                    "4. Enforce maximum file size limits.\n"
                    "5. Scan uploads with antivirus before serving.",
            )

        # ── 5. Exposed email addresses ────────────────────────────────
        emails = list(set(EMAIL_RE.findall(combined_html)))
        emails = [e for e in emails if not e.endswith((".png",".jpg",".gif",".css",".js"))]
        if emails:
            self.add_vuln(
                title=f"Email Addresses Exposed in HTML ({len(emails)} address(es))",
                severity="Low",
                category="Information Disclosure",
                cvss_score=3.1,
                description="The following email addresses were found in page source:\n\n"
                    + "\n".join(f"- `{e}`" for e in emails[:15]) + "\n\n"
                    "Exposed emails enable targeted phishing, spam, and social engineering.",
                remediation="1. Obfuscate contact emails using JavaScript or server-side rendering.\n"
                    "2. Use contact forms instead of direct email links.\n"
                    "3. Consider a generic contact@domain.com address.",
            )

        # ── 6. Internal IP / hostname leaks ───────────────────────────
        internal_ips = list(set(INTERNAL_IP_RE.findall(combined_html)))
        if internal_ips:
            self.add_vuln(
                title=f"Internal IP Address(es) Leaked in Page Source ({len(internal_ips)})",
                severity="Medium",
                category="Information Disclosure",
                cvss_score=5.3,
                description="Internal RFC-1918 IP addresses were found in the page source:\n\n"
                    + "\n".join(f"- `{ip}`" for ip in internal_ips[:10]) + "\n\n"
                    "Leaking internal IPs aids network reconnaissance.",
                remediation="Remove all internal hostnames and IPs from HTML output. "
                    "Use public-facing domain names or proxied URLs.",
            )

        # ── 7. Third-party script domains ─────────────────────────────
        ext_domains = set()
        for m in re.finditer(r'src=["\']https?://([^/"\']+)', combined_html, re.I):
            d = m.group(1)
            if self.domain not in d:
                ext_domains.add(d)

        if len(ext_domains) > 5:
            self.add_vuln(
                title=f"Large Third-Party Script Surface ({len(ext_domains)} external domains)",
                severity="Medium",
                category="Supply Chain Security",
                cvss_score=5.9,
                description=f"Found {len(ext_domains)} distinct external domains providing "
                    "scripts or resources. Each represents a supply-chain trust boundary:\n\n"
                    + "\n".join(f"- `{d}`" for d in sorted(ext_domains)[:20]),
                remediation="1. Audit all external dependencies for necessity.\n"
                    "2. Self-host critical scripts where possible.\n"
                    "3. Add SRI (Subresource Integrity) to all external scripts.\n"
                    "4. Monitor CDNs and third-party scripts for tampering.",
            )

        # ── 8. Parameters summary ─────────────────────────────────────
        all_params: set = set()
        for url in urls:
            qs = urllib.parse.parse_qs(urllib.parse.urlparse(url).query)
            all_params.update(qs.keys())
        for form in forms:
            all_params.update(f.get("name","") for f in form.get("fields",[]) if f.get("name"))

        if all_params:
            self.log("INFO",
                f"[AttackSurface] {len(all_params)} unique input parameter(s) identified: "
                f"{', '.join(sorted(all_params)[:20])}")
            self.add_vuln(
                title=f"Input Parameter Inventory ({len(all_params)} parameter(s))",
                severity="Low",
                category="Attack Surface",
                cvss_score=0.0,
                description="The following input parameters were identified across all crawled pages "
                    "and should be tested for injection vulnerabilities:\n\n"
                    + ", ".join(f"`{p}`" for p in sorted(all_params)[:40]),
                remediation="Ensure all listed parameters are covered by:\n"
                    "- Input validation and sanitization\n"
                    "- SQL/NoSQL injection testing\n"
                    "- XSS testing\n"
                    "- Business logic testing",
            )