backend/scanners/graphql_scanner.py

"""
graphql_scanner.py — GraphQL Security Scanner
==============================================
Advanced GraphQL vulnerability detection module.

This scanner:
  1. Identifies GraphQL endpoints
  2. Performs introspection to discover schema
  3. Tests for GraphQL-specific vulnerabilities
  4. Detects information disclosure via introspection
  5. Tests for query depth limiting and DoS vulnerabilities
  6. Checks for authorization bypass in GraphQL queries
  7. Multi-stage detection: probe endpoint, then confirm vulnerabilities
"""
import urllib.request, urllib.error, urllib.parse, re, json
from scanners.base_scanner import BaseScanner
from utils.fuzzer_engine import ContextAwareFuzzer

GRAPHQL_ENDPOINTS = [
    "/graphql", "/api/graphql", "/graphiql", "/api/graphiql",
    "/graphql.php", "/graphql/api", "/v1/graphql", "/v2/graphql",
    "/gql", "/api/gql", "/query", "/api/query",
    "/console/graphql", "/graphql/console", "/playground",
    "/api/v1/graphql", "/api/v2/graphql",
]

GRAPHQL_INDICATORS = [
    r"graphql", r"GraphQL", r"query\s+\w+", r"mutation\s+\w+",
    r"subscription\s+\w+", r"__schema", r"__type", r"__typename",
]

INTROSPECTION_QUERY = """
{
  __schema {
    queryType { name fields { name type { name kind } } }
    mutationType { name fields { name type { name kind } } }
    subscriptionType { name fields { name type { name kind } } }
    types { name kind description fields { name type { name kind } } }
  }
}
"""

BATCHING_QUERY = """
[
  { "query": "{ __typename }" },
  { "query": "{ __typename }" },
  { "query": "{ __typename }" },
  { "query": "{ __typename }" },
  { "query": "{ __typename }" }
]
"""

DEEP_NESTED_QUERY = """
{
  __schema {
    queryType { fields { name type { fields { name type { fields { name type { fields { name type { fields { name } } } } } } } } }
  }
}
"""

ALIAS_QUERY = """
query {
  a1: __typename
  a2: __typename
  a3: __typename
  a4: __typename
  a5: __typename
  a6: __typename
  a7: __typename
  a8: __typename
  a9: __typename
  a10: __typename
}
"""

SQLI_GQL_QUERY = """
query {
  __typename
  search(query: "' OR '1'='1")
}
"""

TEST_QUERIES = {
    "introspection": INTROSPECTION_QUERY,
    "basic_query": "{ __typename }",
    "nested_query": DEEP_NESTED_QUERY,
    "mutation_test": "mutation { __typename }",
    "batching": BATCHING_QUERY,
    "alias_spam": ALIAS_QUERY,
    "sqli_test": SQLI_GQL_QUERY,
}

ERROR_PATTERNS = [
    r"GraphQL error", r"Cannot query field", r"Cannot return null",
    r"Variable", r"Syntax Error", r"Parse error", r"Validation error",
]

SQLI_ERROR_PATTERNS = [
    r"SQL syntax", r"mysql_fetch", r"ORA-[0-9]{5}", r"PostgreSQL",
    r"SQLite", r"unclosed quotation mark", r"quoted string not properly terminated",
    r"division by zero", r"syntax error at or near", r"Unclosed",
]


class GraphqlScanner(BaseScanner):
    SCANNER_NAME = "GraphQL Security Scanner"
    _SCANNER_KEY = "graphql"

    def __init__(self, scan_id, target, domain, **kwargs):
        super().__init__(scan_id, target, domain, **kwargs)
        self._headers = {
            "User-Agent": "LarShield/2.0 GraphQL Scanner",
            "Content-Type": "application/json",
            "Accept": "application/json",
        }
        if self.auth_headers:
            self._headers.update(self.auth_headers)
        self._tested_endpoints = 0
        self._vulns_found = 0
        self._fuzzer = ContextAwareFuzzer(self._gql_fuzzer_req)

    def _gql_fuzzer_req(self, url, params, headers=None):
        variables = {}
        for key, val in params.items():
            variables[key] = val
        payload_dict = {"query": "query($vars: JSON!) { __typename }", "variables": variables}
        encoded = json.dumps(payload_dict).encode("utf-8")
        merged = {"Content-Type": "application/json"}
        if headers:
            merged.update(headers)
        body, status = self._make_request(url, method="POST", data=encoded, headers=merged, timeout=8)
        return body or "", status

    def _make_gql_request(self, url, query, variables=None, timeout=8):
        """Send a GraphQL query and return the response."""
        payload = {"query": query}
        if variables:
            payload["variables"] = variables
        encoded = json.dumps(payload).encode("utf-8")
        body, status = self._make_request(
            url, method="POST", data=encoded,
            headers=self._headers, timeout=timeout,
        )
        return body, status

    def _detect_graphql_endpoint(self, url):
        """Check if a URL is a GraphQL endpoint."""
        try:
            body, status = self._make_request(url)
            if body:
                for indicator in GRAPHQL_INDICATORS:
                    if re.search(indicator, body, re.IGNORECASE):
                        return True, "GET"

            body, status = self._make_gql_request(url, TEST_QUERIES["basic_query"])
            if body:
                try:
                    response_json = json.loads(body)
                    if "data" in response_json or "errors" in response_json:
                        return True, "POST"
                except json.JSONDecodeError:
                    pass

                for indicator in GRAPHQL_INDICATORS:
                    if re.search(indicator, body, re.IGNORECASE):
                        return True, "POST"
        except Exception as e:
            self.log("ERROR", f"[GraphQL] Error detecting endpoint {url}: {e}")

        return False, None

    def _test_introspection(self, url):
        """Test if introspection is enabled (information disclosure)."""
        try:
            body, status = self._make_gql_request(url, TEST_QUERIES["introspection"])
            if not body:
                return False

            try:
                response_json = json.loads(body)
                if "data" in response_json:
                    data = response_json["data"]
                    if "__schema" in data or "__type" in data:
                        self._vulns_found += 1
                        self.log("CRITICAL", "[GraphQL] Introspection is ENABLED — Full schema disclosure!")

                        schema = data.get("__schema", {})
                        types = schema.get("types", [])
                        query_fields = schema.get("queryType", {}).get("fields", [])
                        mutation_fields = schema.get("mutationType", {}).get("fields", [])

                        self.log("INFO", f"[GraphQL] Schema contains {len(types)} types, "
                                         f"{len(query_fields)} query fields, {len(mutation_fields)} mutation fields")

                        self.add_vuln(
                            title="GraphQL — Introspection Enabled",
                            severity="High",
                            category="Information Disclosure",
                            cvss_score=7.5,
                            description=(
                                f"GraphQL introspection is enabled at {url}.\n"
                                f"Discovered schema contains:\n"
                                f"- {len(types)} types\n"
                                f"- {len(query_fields)} query fields\n"
                                f"- {len(mutation_fields)} mutation fields\n\n"
                                f"Introspection exposes the entire GraphQL schema, including "
                                f"all queries, mutations, types, and their relationships. "
                                f"This information can be used by attackers to craft targeted attacks."
                            ),
                            remediation=(
                                "1. DISABLE introspection in production:\n"
                                "   - Apollo Server: introspection: false in config\n"
                                "   - GraphQL Yoga: disableIntrospection: true\n"
                                "   - Graphene: disable_introspection = True\n"
                                "2. Use environment-specific configuration\n"
                                "3. Implement proper authentication and authorization\n"
                                "4. Monitor for introspection queries in logs"
                            ),
                            evidence=json.dumps({k: v for k, v in data.items() if k == "__schema"}, indent=2)[:500],
                            payload="Introspection query (see evidence)",
                            request_details=f"URL: {url}",
                            response_details=f"Schema: {len(types)} types, {len(query_fields)} queries, {len(mutation_fields)} mutations",
                            confidence="Confirmed",
                        )
                        return True
            except json.JSONDecodeError:
                pass
        except Exception as e:
            self.log("ERROR", f"[GraphQL] Error testing introspection: {e}")

        return False

    def _test_query_depth(self, url):
        """Test for query depth limiting (DoS prevention)."""
        try:
            body, status = self._make_gql_request(url, TEST_QUERIES["nested_query"], timeout=15)
            if not body:
                return False

            try:
                response_json = json.loads(body)
                if "data" in response_json and response_json["data"]:
                    self.log("WARNING", "[GraphQL] Query depth limiting may not be configured")

                    self.add_vuln(
                        title="GraphQL — Missing Query Depth Limiting",
                        severity="Medium",
                        category="Denial of Service",
                        cvss_score=5.3,
                        description=(
                            f"GraphQL endpoint at {url} accepted deeply nested queries without depth limiting. "
                            "This can lead to denial of service attacks through complex nested queries."
                        ),
                        remediation=(
                            "1. IMPLEMENT query depth limiting:\n"
                            "   - Apollo Server: maxDepth or validation rules\n"
                            "   - Set reasonable depth limits (e.g., 5-10 levels)\n"
                            "2. Implement query complexity analysis\n"
                            "3. Use query whitelisting for production"
                        ),
                        evidence="Deeply nested query returned data successfully",
                        payload=TEST_QUERIES["nested_query"][:200],
                        request_details=f"URL: {url}",
                        response_details="Query executed without depth limit enforcement",
                        confidence="Confirmed",
                    )
                    return True
            except json.JSONDecodeError:
                pass
        except Exception as e:
            self.log("ERROR", f"[GraphQL] Error testing query depth: {e}")

        return False

    def _test_batching_attack(self, url):
        """Test for batching attack vulnerabilities (batched queries bypassing rate limits)."""
        try:
            body, status = self._make_gql_request(
                url, TEST_QUERIES["batching"],
                timeout=15,
            )
            if not body:
                return False

            try:
                response_json = json.loads(body)
                if isinstance(response_json, list) and len(response_json) >= 5:
                    self._vulns_found += 1
                    self.log("WARNING",
                             f"[GraphQL] Batching is supported — {len(response_json)} batched queries accepted at {url}")

                    self.add_vuln(
                        title="GraphQL — Batching Attack Possible",
                        severity="High",
                        category="Denial of Service",
                        cvss_score=7.5,
                        description=(
                            f"GraphQL endpoint at {url} supports query batching.\n"
                            f"Accepted {len(response_json)} batched queries in a single request.\n\n"
                            f"Impact: Attackers can bypass rate limiting by sending multiple "
                            f"operations in a single request, making brute-force attacks, "
                            f"enumeration, and DoS more effective."
                        ),
                        remediation=(
                            "1. Implement rate limiting per operation, not per request.\n"
                            "2. Limit the number of operations allowed per batch request.\n"
                            "3. Use cost-based analysis to limit batch complexity.\n"
                            "4. Consider disabling batching if not required."
                        ),
                        evidence=f"Batched query returned {len(response_json)} results",
                        payload="Batched query with 5 operations",
                        request_details=f"URL: {url}",
                        response_details=f"{len(response_json)} operations accepted",
                        confidence="Confirmed",
                    )
                    return True
            except (json.JSONDecodeError, TypeError):
                pass
        except Exception as e:
            self.log("ERROR", f"[GraphQL] Error testing batching attack: {e}")

        return False

    def _test_alias_dos(self, url):
        """Test for alias-based DoS attacks (many aliases consuming resources)."""
        try:
            body, status = self._make_gql_request(url, TEST_QUERIES["alias_spam"], timeout=15)
            if not body:
                return False

            try:
                response_json = json.loads(body)
                if "data" in response_json:
                    data = response_json["data"]
                    alias_count = sum(1 for k in data if k.startswith("a"))
                    if alias_count >= 8:
                        self._vulns_found += 1
                        self.log("WARNING",
                                 f"[GraphQL] Alias-based DoS possible at {url} — {alias_count} aliases accepted")

                        self.add_vuln(
                            title="GraphQL — Alias-Based DoS Possible",
                            severity="Medium",
                            category="Denial of Service",
                            cvss_score=5.0,
                            description=(
                                f"GraphQL endpoint at {url} accepted {alias_count} aliases in a single query.\n"
                                f"Attackers can use many aliases to amplify resource consumption.\n\n"
                                f"Aliases allow the same field to be queried multiple times under different names, "
                                f"bypassing query depth limits while consuming significant server resources."
                            ),
                            remediation=(
                                "1. Implement query complexity/cost analysis.\n"
                                "2. Limit the number of aliases allowed per query.\n"
                                "3. Use rate limiting based on field resolution cost.\n"
                                "4. Consider using persisted queries in production."
                            ),
                            evidence=f"{alias_count} aliases accepted",
                            payload="Alias-based amplification query",
                            request_details=f"URL: {url}",
                            response_details=f"Accepted {alias_count} aliases",
                            confidence="Confirmed",
                        )
                        return True
            except json.JSONDecodeError:
                pass
        except Exception as e:
            self.log("ERROR", f"[GraphQL] Error testing alias DoS: {e}")

        return False

    def _test_graphql_sqli(self, url):
        """Test for SQL injection via GraphQL query parameters."""
        try:
            sqli_query = TEST_QUERIES["sqli_test"]
            body, status = self._make_gql_request(url, sqli_query, timeout=15)
            if not body:
                return False

            try:
                response_json = json.loads(body)
                # Check for SQL errors in response
                errors = response_json.get("errors", [])
                error_str = json.dumps(errors)
                for pattern in SQLI_ERROR_PATTERNS:
                    if re.search(pattern, error_str, re.IGNORECASE):
                        self._vulns_found += 1
                        self.log("CRITICAL",
                                 f"[GraphQL] SQL Injection via GraphQL at {url}! "
                                 f"Pattern matched: {pattern}")

                        self.add_vuln(
                            title="GraphQL — SQL Injection via GraphQL Parameters",
                            severity="Critical",
                            category="Injection",
                            cvss_score=9.8,
                            description=(
                                f"A SQL injection vulnerability was detected via GraphQL at {url}.\n"
                                f"The GraphQL endpoint appears to pass user input directly to SQL queries.\n"
                                f"Pattern matched: {pattern}\n\n"
                                f"Impact: Attackers can extract, modify, or delete database contents, "
                                f"potentially leading to complete database compromise."
                            ),
                            remediation=(
                                "1. Use parameterized queries / prepared statements.\n"
                                "2. Implement input validation and sanitization.\n"
                                "3. Use an ORM/ODM with proper injection protections.\n"
                                "4. Implement least-privilege database access.\n"
                                "5. Regularly audit and test for injection vulnerabilities."
                            ),
                            evidence=f"SQL error pattern matched: {pattern}",
                            payload=SQLI_GQL_QUERY,
                            request_details=f"URL: {url}",
                            response_details=f"Error message: {error_str[:200]}",
                            confidence="Confirmed" if any(re.search(p, error_str, re.IGNORECASE) for p in SQLI_ERROR_PATTERNS) else "Medium",
                        )
                        return True
            except json.JSONDecodeError:
                pass
        except Exception as e:
            self.log("ERROR", f"[GraphQL] Error testing SQLi: {e}")

        return False

    def _test_graphql_params_fuzz(self, url):
        try:
            body, status = self._make_gql_request(url, TEST_QUERIES["introspection"])
            if not body:
                return
            data = json.loads(body)
            schema = data.get("data", {}).get("__schema", {})
            query_type = schema.get("queryType", {})
            fields = query_type.get("fields", []) if query_type else []
            params = {}
            for field in fields:
                for arg in field.get("args", []):
                    params[arg["name"]] = "test"
            mutation_type = schema.get("mutationType", {})
            if mutation_type:
                for field in mutation_type.get("fields", []):
                    for arg in field.get("args", []):
                        params[arg["name"]] = "test"
            if not params:
                params = {"query": "test", "id": "1", "filter": "test"}
            self.log("INFO", f"[GraphQL] Context-aware fuzzing {len(params)} parameter(s) at {url}")
            self._fuzzer.fuzz(url, params)
            baseline_body, _ = self._make_gql_request(url, TEST_QUERIES["basic_query"])
            baseline_length = len(baseline_body or "")
            anomalies = self._fuzzer.anomalies(baseline_length)
            for anom in anomalies:
                self._vulns_found += 1
                self.log("WARNING", f"[GraphQL] Fuzzer anomaly: {anom['param']} mutation={anom['mutation']} status={anom['status']}")
                self.add_vuln(
                    title=f"GraphQL — Parameter Injection ({anom['param']})",
                    severity="High",
                    category="Injection",
                    cvss_score=7.5,
                    description=(
                        f"GraphQL parameter '{anom['param']}' (classified as '{anom['type']}') "
                        f"at {url} returned an anomalous response when mutated with "
                        f"'{anom['mutation']}' (value: {anom['value']}). "
                        f"HTTP {anom['status']}, response length {anom['length']}."
                    ),
                    remediation="Validate and sanitize all GraphQL argument inputs. Use parameterized queries, input type enforcement, and proper output encoding.",
                    cwe_ids=["CWE-20"],
                    owasp_category="A03:2021 – Injection",
                )
        except Exception as e:
            self.log("ERROR", f"[GraphQL] Error in param fuzzing: {e}")

    def _test_authorization(self, url):
        """Test for authorization bypass in GraphQL."""
        try:
            test_query = """
            {
              __schema {
                queryType {
                  fields { name args { name type { name } } }
                }
              }
            }
            """

            body, status = self._make_gql_request(url, test_query)
            if not body:
                return False

            try:
                response_json = json.loads(body)
                if "data" in response_json:
                    data = response_json["data"]
                    if "__schema" in data:
                        fields = data["__schema"].get("queryType", {}).get("fields", [])
                        sensitive_fields = []
                        for field in fields:
                            field_name = field.get("name", "").lower()
                            if any(s in field_name for s in ["user", "password", "secret", "key", "token", "admin"]):
                                sensitive_fields.append(field.get("name"))

                        if sensitive_fields:
                            self.log("WARNING", f"[GraphQL] Discovered potentially sensitive fields: {sensitive_fields}")
                            self.add_vuln(
                                title="GraphQL — Sensitive Field Exposure",
                                severity="Medium",
                                category="Information Disclosure",
                                cvss_score=5.5,
                                description=(
                                    f"GraphQL schema exposes potentially sensitive fields: {', '.join(sensitive_fields)}. "
                                    "These fields may contain sensitive information that should be protected."
                                ),
                                remediation=(
                                    "1. Review and restrict field access based on user roles\n"
                                    "2. Implement field-level authorization\n"
                                    "3. Use custom resolvers with proper access checks\n"
                                    "4. Audit schema for sensitive data exposure"
                                ),
                                evidence=f"Sensitive fields: {', '.join(sensitive_fields)}",
                                payload="Schema field discovery query",
                                request_details=f"URL: {url}",
                                response_details=f"Fields found: {[f.get('name') for f in fields[:10]]}",
                                confidence="High",
                            )
                            return True
            except json.JSONDecodeError:
                pass
        except Exception as e:
            self.log("ERROR", f"[GraphQL] Error testing authorization: {e}")

        return False

    def _discover_graphql_endpoints(self):
        """Discover GraphQL endpoints."""
        endpoints = []
        base_url = self.target.rstrip("/")

        for path in GRAPHQL_ENDPOINTS:
            url = f"{base_url}{path}"
            is_graphql, method = self._detect_graphql_endpoint(url)
            if is_graphql:
                endpoints.append((url, method))
                self.log("INFO", f"[GraphQL] Discovered GraphQL endpoint: {url} ({method})")

        is_graphql, method = self._detect_graphql_endpoint(self.target)
        if is_graphql:
            endpoints.append((self.target, method))
            self.log("INFO", f"[GraphQL] Main URL is GraphQL endpoint: {self.target} ({method})")

        return endpoints

    def run(self):
        self.log("INFO", f"[GraphQL] Starting GraphQL security scanning on {self.target}...")

        try:
            # Step 1: Discover GraphQL endpoints
            self.log("INFO", "[GraphQL] Discovering GraphQL endpoints...")
            endpoints = self._discover_graphql_endpoints()
            self.log("INFO", f"[GraphQL] Found {len(endpoints)} GraphQL endpoint(s)")

            if not endpoints:
                self.log("INFO", "[GraphQL] No GraphQL endpoints detected")
                return self.vulns

            # Step 2: Test each endpoint
            for url, method in endpoints:
                self._tested_endpoints += 1
                self.log("INFO", f"[GraphQL] Testing endpoint: {url}")

                if self._test_introspection(url):
                    pass

                self._test_query_depth(url)

                self._test_authorization(url)

                self._test_batching_attack(url)

                self._test_alias_dos(url)

                self._test_graphql_sqli(url)

                self._test_graphql_params_fuzz(url)

        except Exception as e:
            self.log("ERROR", f"[GraphQL] Unexpected error during scan: {e}")

        self.log("SUCCESS" if not self.vulns else "WARNING",
                 f"[GraphQL] Complete — {self._tested_endpoints} endpoint(s) tested | "
                 f"{self._vulns_found} vulnerability/vulnerabilities found")
        return self.vulns