| 1 | import httpx |
| 2 | from fastapi import HTTPException |
| 3 | from loguru import logger |
| 4 | |
| 5 | from app.integrations.nuclei.schema.nuclei import DeleteNucleiReportResponse |
| 6 | from app.integrations.nuclei.schema.nuclei import NucleiReportCollectionResponse |
| 7 | from app.integrations.nuclei.schema.nuclei import NucleiReportsAvailableResponse |
| 8 | from app.integrations.nuclei.schema.nuclei import NucleiScanRequest |
| 9 | from app.integrations.nuclei.schema.nuclei import NucleiScanResponse |
| 10 | |
| 11 | |
| 12 | async def get_nuclei_reports_available() -> NucleiReportsAvailableResponse: |
| 13 | """ |
| 14 | Retrieve the list of available Nuclei reports. |
| 15 | |
| 16 | Returns: |
| 17 | NucleiReportsAvailableResponse: The response containing the list of available Nuclei reports. |
| 18 | """ |
| 19 | logger.info("Sending GET request to http://copilot-nuclei-module/all_reports") |
| 20 | async with httpx.AsyncClient() as client: |
| 21 | response = await client.get("http://copilot-nuclei-module/all_reports") |
| 22 | if response.status_code == 404: |
| 23 | raise HTTPException(status_code=404, detail="Nuclei reports not found") |
| 24 | response.raise_for_status() # Raise an exception if the request was unsuccessful |
| 25 | return NucleiReportsAvailableResponse(**response.json()) |
| 26 | |
| 27 | |
| 28 | async def get_nuclei_report(host: str, report: str = "index.md") -> NucleiReportCollectionResponse: |
| 29 | """ |
| 30 | Retrieve a specific Nuclei report. |
| 31 | |
| 32 | Args: |
| 33 | host (str): The host to retrieve the report for. |
| 34 | report (str): The report to retrieve. Defaults to "index.md". |
| 35 | |
| 36 | Returns: |
| 37 | NucleiReportCollectionResponse: The response containing the Nuclei report. |
| 38 | """ |
| 39 | logger.info(f"Sending GET request to http://copilot-nuclei-module/report/{host}/{report}") |
| 40 | async with httpx.AsyncClient() as client: |
| 41 | response = await client.get(f"http://copilot-nuclei-module/report/{host}/{report}") |
| 42 | |
| 43 | if response.status_code == 404: |
| 44 | raise HTTPException(status_code=404, detail="Nuclei report not found") |
| 45 | |
| 46 | response.raise_for_status() # Raise an exception if the request was unsuccessful |
| 47 | return NucleiReportCollectionResponse(**response.json()) |
| 48 | |
| 49 | |
| 50 | async def delete_nuclei_report(host: str) -> DeleteNucleiReportResponse: |
| 51 | """ |
| 52 | Delete a specific Nuclei report. |
| 53 | |
| 54 | Args: |
| 55 | host (str): The host to delete the report for. |
| 56 | |
| 57 | Returns: |
| 58 | DeleteNucleiReportResponse: The response containing the result of the deletion. |
| 59 | """ |
| 60 | logger.info(f"Sending DELETE request to http://copilot-nuclei-module/report/{host}") |
| 61 | async with httpx.AsyncClient() as client: |
| 62 | response = await client.delete(f"http://copilot-nuclei-module/report/{host}") |
| 63 | |
| 64 | if response.status_code == 404: |
| 65 | raise HTTPException(status_code=404, detail="Nuclei report not found") |
| 66 | return DeleteNucleiReportResponse(**response.json()) |
| 67 | |
| 68 | |
| 69 | async def post_to_copilot_nuclei_module(data: NucleiScanRequest) -> NucleiScanResponse: |
| 70 | """ |
| 71 | Send a POST request to the copilot-nuclei-module Docker container. |
| 72 | |
| 73 | Args: |
| 74 | data (NucleiScanRequest): The data to send to the copilot-nuclei-module Docker container. |
| 75 | """ |
| 76 | logger.info(f"Sending POST request to http://copilot-nuclei-module/scan with data: {data.model_dump()}") |
| 77 | # raise HTTPException(status_code=501, detail="Not Implemented Yet") |
| 78 | async with httpx.AsyncClient() as client: |
| 79 | data = await client.post( |
| 80 | "http://copilot-nuclei-module/scan", |
| 81 | json=data.model_dump(), |
| 82 | timeout=120, |
| 83 | ) |
| 84 | return NucleiScanResponse(**data.json()) |