|
| 1 | +""" |
| 2 | +Active API Reconnaissance & Discovery Service. |
| 3 | +Probes for hidden or undocumented endpoints and parameters before AI analysis. |
| 4 | +""" |
| 5 | + |
| 6 | +from typing import List, Dict, Set |
| 7 | +import structlog |
| 8 | + |
| 9 | +from ..vulnerability_models import APIEndpoint, SchemaStructure, TestCase |
| 10 | +from .pipeline import PipelineInput |
| 11 | +from ..test_executor import execute_proactive_tests_detailed |
| 12 | + |
| 13 | +logger = structlog.get_logger(__name__) |
| 14 | + |
| 15 | + |
| 16 | +def _mutate_path(path: str) -> List[str]: |
| 17 | + """Generate potential hidden paths based on existing routes.""" |
| 18 | + mutations: Set[str] = set() |
| 19 | + stripped = path.rstrip("/") |
| 20 | + if "/v1/" in stripped: |
| 21 | + mutations.add(stripped.replace("/v1/", "/v2/", 1)) |
| 22 | + mutations.add(stripped.replace("/v1/", "/internal/", 1)) |
| 23 | + if not stripped.endswith("/admin"): |
| 24 | + mutations.add(f"{stripped}/admin") |
| 25 | + if "{id}" in stripped: |
| 26 | + mutations.add(stripped.replace("{id}", "1")) |
| 27 | + mutations.add(stripped.replace("{id}", "2")) |
| 28 | + return [m for m in mutations if m and m != path] |
| 29 | + |
| 30 | + |
| 31 | +def _build_discovery_tests(endpoints: List[APIEndpoint]) -> List[TestCase]: |
| 32 | + """Generate exploratory test cases for undocumented methods and mutated paths.""" |
| 33 | + discovery_tests: List[TestCase] = [] |
| 34 | + |
| 35 | + # Track which methods we already know exist for each path |
| 36 | + existing_methods_by_path: Dict[str, Set[str]] = {} |
| 37 | + for endpoint in endpoints: |
| 38 | + if endpoint.path not in existing_methods_by_path: |
| 39 | + existing_methods_by_path[endpoint.path] = set() |
| 40 | + existing_methods_by_path[endpoint.path].add(endpoint.method.upper()) |
| 41 | + |
| 42 | + # Generate Method Probes and Path Mutations |
| 43 | + idx = 0 |
| 44 | + for path, known_methods in existing_methods_by_path.items(): |
| 45 | + # Probe for undocumented HTTP methods |
| 46 | + for probe_method in ("GET", "POST", "PUT", "DELETE", "PATCH"): |
| 47 | + if probe_method in known_methods: |
| 48 | + continue |
| 49 | + discovery_tests.append( |
| 50 | + TestCase( |
| 51 | + id=f"DISCOVER-{idx}-{probe_method}", |
| 52 | + name="Method probing", |
| 53 | + description="Probe undocumented method support", |
| 54 | + owasp_category="API9: Improper Inventory Management", |
| 55 | + endpoint=path, |
| 56 | + method=probe_method, |
| 57 | + ) |
| 58 | + ) |
| 59 | + idx += 1 |
| 60 | + |
| 61 | + # Probe for path mutations (e.g. /v2/ instead of /v1/) |
| 62 | + for mutation_idx, mutated_path in enumerate(_mutate_path(path)): |
| 63 | + discovery_tests.append( |
| 64 | + TestCase( |
| 65 | + id=f"MUTATE-{idx}-{mutation_idx}", |
| 66 | + name="Path mutation", |
| 67 | + description="Probe possible shadow or alternate API path", |
| 68 | + owasp_category="API9: Improper Inventory Management", |
| 69 | + endpoint=mutated_path, |
| 70 | + method="GET", |
| 71 | + ) |
| 72 | + ) |
| 73 | + idx += 1 |
| 74 | + |
| 75 | + return discovery_tests |
| 76 | + |
| 77 | + |
| 78 | +async def perform_active_recon( |
| 79 | + api_structure: SchemaStructure, pipeline_input: PipelineInput |
| 80 | +) -> List[APIEndpoint]: |
| 81 | + """ |
| 82 | + Actively scan the application for undocumented endpoints and parameters. |
| 83 | + Returns a list of newly discovered APIEndpoints. |
| 84 | + """ |
| 85 | + logger.info("Starting active reconnaissance for hidden endpoints") |
| 86 | + |
| 87 | + discovery_tests = _build_discovery_tests(api_structure.endpoints) |
| 88 | + if not discovery_tests: |
| 89 | + return [] |
| 90 | + |
| 91 | + results, _ = await execute_proactive_tests_detailed( |
| 92 | + test_cases=discovery_tests, |
| 93 | + base_url=api_structure.base_url, |
| 94 | + concurrency=pipeline_input.concurrency, |
| 95 | + auth_headers=pipeline_input.auth_headers, |
| 96 | + proxy=pipeline_input.proxy, |
| 97 | + verify_ssl=pipeline_input.verify_ssl, |
| 98 | + max_requests=len(discovery_tests), |
| 99 | + ) |
| 100 | + |
| 101 | + discovered_endpoints: List[APIEndpoint] = [] |
| 102 | + |
| 103 | + for result in results: |
| 104 | + status = result.status_code |
| 105 | + # A 2xx indicates the endpoint definitely exists. |
| 106 | + # A 401/403/405 also indicates the endpoint exists, just restricted or wrong method logic. |
| 107 | + # 404 indicates it doesn't exist. |
| 108 | + if (200 <= status < 300) or status in (401, 403, 405): |
| 109 | + new_endpoint = APIEndpoint( |
| 110 | + path=result.test_case.endpoint, |
| 111 | + method=result.test_case.method.upper(), |
| 112 | + operation_id=f"recon_discovered_{result.test_case.method.lower()}", |
| 113 | + parameters=[], # We could try to infer parameters from errors, but keep basic for now |
| 114 | + auth_required=(status in (401, 403)), |
| 115 | + request_body_schema={"present": False} # Hard to infer from recon probes without explicit fuzzing |
| 116 | + ) |
| 117 | + discovered_endpoints.append(new_endpoint) |
| 118 | + logger.info("Discovered undocumented endpoint", method=new_endpoint.method, path=new_endpoint.path, status=status) |
| 119 | + |
| 120 | + return discovered_endpoints |
0 commit comments