Run apisec modules in the background and store findings in scans_db.
(scan_id: str)
| 57 | async def _execute_apisec_scan(scan_id: str) -> None: |
| 58 | """Run apisec modules in the background and store findings in scans_db.""" |
| 59 | scan = scans_db[scan_id] |
| 60 | scan["status"] = "running" |
| 61 | scan["started_at"] = datetime.now(timezone.utc) |
| 62 | |
| 63 | spec_url: str = scan["options"]["spec_url"] |
| 64 | auth_token: str | None = scan["options"].get("auth_token") |
| 65 | max_fuzz: int = scan["options"].get("max_fuzz_requests", 100) |
| 66 | |
| 67 | try: |
| 68 | from modules.apisec import APIAuthTester, APIEndpointTester, APIFuzzer, OpenAPIParser |
| 69 | |
| 70 | parser = OpenAPIParser() |
| 71 | api = await parser.parse_url(spec_url) |
| 72 | logger.info(f"Scan {scan_id}: parsed {len(api.endpoints)} endpoints from {api.title}") |
| 73 | |
| 74 | for module_name in scan["modules"]: |
| 75 | try: |
| 76 | if module_name == "endpoints": |
| 77 | tester = APIEndpointTester(auth_token=auth_token) |
| 78 | result = await tester.test_api(api) |
| 79 | elif module_name == "auth": |
| 80 | tester = APIAuthTester() |
| 81 | result = await tester.test_api_auth(api) |
| 82 | elif module_name == "fuzzer": |
| 83 | fuzzer = APIFuzzer(max_requests=max_fuzz, auth_token=auth_token) |
| 84 | result = await fuzzer.fuzz_api(api) |
| 85 | else: |
| 86 | logger.warning(f"Unknown apisec module: {module_name}") |
| 87 | continue |
| 88 | |
| 89 | for f in result.findings: |
| 90 | scan["findings"].append({ |
| 91 | "id": str(uuid.uuid4()), |
| 92 | "title": f.title, |
| 93 | "description": f.description, |
| 94 | "severity": f.severity, |
| 95 | "module": f.source, |
| 96 | "data": f.data or {}, |
| 97 | "created_at": datetime.now(timezone.utc), |
| 98 | }) |
| 99 | |
| 100 | logger.info(f"Scan {scan_id}: module '{module_name}' found {len(result.findings)} findings") |
| 101 | |
| 102 | except Exception as e: |
| 103 | logger.error(f"Scan {scan_id}: module '{module_name}' error: {e}") |
| 104 | |
| 105 | scan["status"] = "completed" |
| 106 | scan["completed_at"] = datetime.now(timezone.utc) |
| 107 | logger.info(f"Scan {scan_id} completed with {len(scan['findings'])} total findings") |
| 108 | |
| 109 | except Exception as e: |
| 110 | scan["status"] = "failed" |
| 111 | scan["error"] = str(e) |
| 112 | scan["completed_at"] = datetime.now(timezone.utc) |
| 113 | logger.error(f"Scan {scan_id} failed: {e}") |
| 114 | |
| 115 | |
| 116 | # --------------------------------------------------------------------------- |
nothing calls this directly
no test coverage detected