backend/scanners/api_scanner.py

import urllib.request, urllib.error, urllib.parse, ssl, json
from scanners.base_scanner import BaseScanner
from utils.fuzzer_engine import ContextAwareFuzzer

class ApiScanner(BaseScanner):
    SCANNER_NAME = "API & GraphQL Introspection Scanner"

    def __init__(self, scan_id, target, domain, **kwargs):
        super().__init__(scan_id, target, domain, **kwargs)
        self._ctx = ssl.create_default_context()
        self._ctx.check_hostname = False
        self._ctx.verify_mode = ssl.CERT_NONE
        self._headers = {"User-Agent": "LarShield/2.0 API-Analyzer", "Content-Type": "application/json"}
        if self.auth_headers:
            self._headers.update(self.auth_headers)
        self.base_url = target.rstrip("/")
        self._fuzzer = ContextAwareFuzzer(self._api_fuzzer_req)

    def _api_fuzzer_req(self, url, params, headers=None):
        data = urllib.parse.urlencode(params).encode("utf-8") if params else None
        merged = {"Content-Type": "application/x-www-form-urlencoded"}
        if headers:
            merged.update(headers)
        body, status = self._make_request(url, method="POST", data=data, headers=merged, timeout=8)
        return body or "", status

    def _get(self, path):
        url = f"{self.base_url}{path}"
        try:
            req = urllib.request.Request(url, headers=self._headers)
            with urllib.request.urlopen(req, timeout=5, context=self._ctx) as resp:
                return resp.read().decode("utf-8", errors="ignore"), resp.status
        except urllib.error.HTTPError as e:
            return e.read().decode("utf-8", errors="ignore") if e.fp else "", e.code
        except Exception as e:
            self.log("ERROR", f"[API] GET error: {e}")
            return "", 0

    def _post(self, path, payload):
        url = f"{self.base_url}{path}"
        try:
            data = json.dumps(payload).encode('utf-8')
            req = urllib.request.Request(url, data=data, headers=self._headers, method='POST')
            with urllib.request.urlopen(req, timeout=5, context=self._ctx) as resp:
                return resp.read().decode("utf-8", errors="ignore"), resp.status
        except urllib.error.HTTPError as e:
            return e.read().decode("utf-8", errors="ignore") if e.fp else "", e.code
        except Exception as e:
            self.log("ERROR", f"[API] POST error: {e}")
            return "", 0

    def check_swagger(self):
        self.log("INFO", "[API] Hunting for exposed Swagger/OpenAPI documentation...")
        paths = ["/swagger-ui.html", "/api-docs", "/v2/api-docs", "/openapi.json", "/api/swagger.json", "/docs"]
        for path in paths:
            body, status = self._get(path)
            if status == 200 and ("swagger" in body.lower() or "openapi" in body.lower()):
                self.log("CRITICAL", f"[API] Exposed API documentation found at {path}")
                self.add_vuln(
                    title="Exposed API Documentation (Swagger/OpenAPI)",
                    severity="High",
                    category="Information Disclosure",
                    cvss_score=7.5,
                    description=f"Unauthenticated API documentation was discovered at `{path}`. Attackers can use this to map out the entire backend infrastructure, discover hidden endpoints, and find injection vectors.",
                    remediation="Restrict access to API documentation endpoints in production environments using IP whitelisting or robust authentication."
                )
                break

    def check_graphql(self):
        self.log("INFO", "[API] Testing GraphQL endpoints for Introspection vulnerabilities...")
        endpoints = ["/graphql", "/api/graphql", "/v1/graphql"]
        introspection_query = {
            "query": "{ __schema { types { name fields { name } } } }"
        }
        
        for path in endpoints:
            body, status = self._post(path, introspection_query)
            if status == 200 and "__schema" in body:
                self.log("CRITICAL", f"[API] GraphQL Introspection enabled at {path}")
                self.add_vuln(
                    title="GraphQL Introspection Query Enabled",
                    severity="Critical",
                    category="API Security",
                    cvss_score=9.1,
                    description=f"The GraphQL endpoint at `{path}` allows Introspection queries. An attacker dumped the entire database schema, including all types, mutations, and hidden fields. This completely exposes the application's internal data structures.",
                    remediation="Disable GraphQL introspection in your production environment. In Apollo Server, set `introspection: false`."
                )
                break

    def _fuzz_api_params(self):
        swagger_paths = ["/api-docs", "/v2/api-docs", "/openapi.json", "/api/swagger.json"]
        swagger_spec = None
        for path in swagger_paths:
            body, status = self._get(path)
            if status == 200 and body:
                try:
                    swagger_spec = json.loads(body)
                    break
                except json.JSONDecodeError:
                    continue

        if swagger_spec:
            paths = swagger_spec.get("paths", {})
            for endpoint, methods in paths.items():
                url = f"{self.base_url}{endpoint}"
                for method, details in methods.items():
                    if method.upper() not in ("GET", "POST", "PUT", "DELETE", "PATCH"):
                        continue
                    params = {}
                    for param in details.get("parameters", []):
                        if param.get("in") in ("query", "formData"):
                            params[param["name"]] = str(param.get("default", "test"))
                    if not params:
                        continue
                    self.log("INFO", f"[API] Context-aware fuzzing {endpoint} ({len(params)} params)")
                    self._fuzzer.fuzz(url, params)
                    baseline_body, _ = self._make_request(url, timeout=8)
                    baseline_length = len(baseline_body or "")
                    anomalies = self._fuzzer.anomalies(baseline_length)
                    for anom in anomalies:
                        self.log("WARNING", f"[API] Fuzzer anomaly at {endpoint}: {anom['param']} mutation={anom['mutation']} status={anom['status']}")
                        self.add_vuln(
                            title=f"API Injection — {anom['param']} ({anom['mutation']})",
                            severity="High",
                            category="Injection",
                            cvss_score=7.5,
                            description=(
                                f"API endpoint {endpoint} parameter '{anom['param']}' "
                                f"(classified as '{anom['type']}') returned an anomalous response "
                                f"when mutated with '{anom['mutation']}' (value: {anom['value']}). "
                                f"HTTP {anom['status']}, response length {anom['length']}."
                            ),
                            remediation="Validate and sanitize all API input parameters. Use parameterized queries, input type enforcement, and proper output encoding.",
                            cwe_ids=["CWE-20"],
                            owasp_category="A03:2021 – Injection",
                        )
        else:
            common_api_params = ["id", "q", "search", "query", "page", "limit", "offset", "sort", "filter",
                                 "token", "key", "secret", "user", "email", "name", "status", "type", "role"]
            params = {p: "test" for p in common_api_params}
            self.log("INFO", "[API] No swagger spec found; fuzzing common API parameters on base URL")
            self._fuzzer.fuzz(self.base_url, params)
            baseline_body, _ = self._make_request(self.target, timeout=8)
            baseline_length = len(baseline_body or "")
            anomalies = self._fuzzer.anomalies(baseline_length)
            for anom in anomalies:
                self.log("WARNING", f"[API] Fuzzer anomaly: {anom['param']} mutation={anom['mutation']} status={anom['status']}")
                self.add_vuln(
                    title=f"API Injection — {anom['param']} ({anom['mutation']})",
                    severity="High",
                    category="Injection",
                    cvss_score=7.5,
                    description=(
                        f"Common API parameter '{anom['param']}' (classified as '{anom['type']}') "
                        f"returned an anomalous response when mutated with '{anom['mutation']}' "
                        f"(value: {anom['value']}). HTTP {anom['status']}, response length {anom['length']}."
                    ),
                    remediation="Validate and sanitize all API input parameters. Use parameterized queries, input type enforcement, and proper output encoding.",
                    cwe_ids=["CWE-20"],
                    owasp_category="A03:2021 – Injection",
                )

    def run(self):
        self.log("INFO", f"[API] Starting Advanced API analysis on {self.target}...")
        self.check_swagger()
        self.check_graphql()
        self._fuzz_api_params()
        self.log("SUCCESS" if not self.vulns else "WARNING", "[API] Analysis complete.")
        return self.vulns