Search for public exploits and PoCs for given CVE IDs or product names.
| 14 | |
| 15 | class ExploitSearch: |
| 16 | """Search for public exploits and PoCs for given CVE IDs or product names.""" |
| 17 | |
| 18 | def __init__(self, max_cves: int = 5, max_github: int = 3): |
| 19 | self.max_cves = max_cves |
| 20 | self.max_github = max_github |
| 21 | |
| 22 | async def search( |
| 23 | self, |
| 24 | product: str, |
| 25 | version: str, |
| 26 | cve_ids: list[str] | None = None, |
| 27 | ) -> list[dict]: |
| 28 | """ |
| 29 | Search for exploits. CVE-based search is tried first (most reliable), |
| 30 | then GitHub PoC search as fallback. |
| 31 | |
| 32 | Returns list of exploit reference dicts. |
| 33 | """ |
| 34 | exploits: list[dict] = [] |
| 35 | |
| 36 | if cve_ids: |
| 37 | tasks = [self._circl_lookup(cve_id) for cve_id in cve_ids[:self.max_cves]] |
| 38 | results = await asyncio.gather(*tasks, return_exceptions=True) |
| 39 | for res in results: |
| 40 | if isinstance(res, list): |
| 41 | exploits.extend(res) |
| 42 | |
| 43 | if not exploits and product: |
| 44 | github_exploits = await self._github_poc_search(product, version) |
| 45 | exploits.extend(github_exploits) |
| 46 | |
| 47 | return exploits |
| 48 | |
| 49 | async def _circl_lookup(self, cve_id: str) -> list[dict]: |
| 50 | """Check cve.circl.lu for exploit references and CAPEC entries.""" |
| 51 | found: list[dict] = [] |
| 52 | try: |
| 53 | url = f"https://cve.circl.lu/api/cve/{cve_id}" |
| 54 | async with httpx.AsyncClient(timeout=10) as client: |
| 55 | resp = await client.get(url) |
| 56 | if resp.status_code != 200: |
| 57 | return [] |
| 58 | data = resp.json() |
| 59 | |
| 60 | for ref in data.get("references", []): |
| 61 | rl = ref.lower() |
| 62 | if any(kw in rl for kw in ("exploit", "poc", "metasploit", "edb")): |
| 63 | found.append({ |
| 64 | "cve_id": cve_id, |
| 65 | "type": "Reference", |
| 66 | "url": ref, |
| 67 | "description": f"Exploit reference for {cve_id}", |
| 68 | }) |
| 69 | |
| 70 | for capec in data.get("capec", []): |
| 71 | found.append({ |
| 72 | "cve_id": cve_id, |
| 73 | "type": "CAPEC", |