Create a scheduled scan.
(
name: str = typer.Argument(..., help="Schedule name"),
target: str = typer.Option(..., "--target", "-t", help="Target to scan"),
frequency: str = typer.Option("daily", "--frequency", "-f", help="Frequency (hourly/daily/weekly/monthly)"),
modules: str = typer.Option("dns,headers,tech", "--modules", "-m", help="Comma-separated modules"),
)
| 1286 | provider: str = typer.Option("webhook", "--provider", "-p"), |
| 1287 | url: str = typer.Option(..., "--url", "-u", help="SIEM endpoint URL"), |
| 1288 | token: str | None = typer.Option(None, "--token", "-t"), |
| 1289 | ): |
| 1290 | """Run scan and export results to SIEM.""" |
| 1291 | setup_logging() |
| 1292 | |
| 1293 | console.print(f"[bold]Scan and Export:[/bold] {target} -> {provider}") |
| 1294 | |
| 1295 | results = _collect_scan_results(target, ["dns", "headers", "tech", "ssl"]) |
| 1296 | |
| 1297 | async def run(): |
| 1298 | from modules.siem import ElasticsearchExporter, SplunkExporter, WebhookExporter |
| 1299 | |
| 1300 | if provider == "splunk": |
| 1301 | exporter = SplunkExporter(hec_url=url, hec_token=token or "") |
| 1302 | elif provider == "elasticsearch": |
| 1303 | exporter = ElasticsearchExporter(hosts=[url]) |
| 1304 | else: |
| 1305 | exporter = WebhookExporter(url=url, format="slack" if "slack" in url else "json") |
| 1306 | |
| 1307 | for result in results: |
| 1308 | success, failed = await exporter.export_scan_result(result) |
| 1309 | console.print(f"[green]Exported {success} events[/green]") |
| 1310 | |
| 1311 | run_async(run()) |
| 1312 | |
| 1313 | |
| 1314 | # ============== Schedule Commands ============== |
| 1315 | |
| 1316 | @schedule_app.command("create") |
| 1317 | def schedule_create( |
nothing calls this directly
no test coverage detected