Run scan and export results to SIEM.
(
target: str = typer.Argument(..., help="Target to scan and export"),
provider: str = typer.Option("webhook", "--provider", "-p"),
url: str = typer.Option(..., "--url", "-u", help="SIEM endpoint URL"),
token: Optional[str] = typer.Option(None, "--token", "-t"),
)
| 1253 | |
| 1254 | @siem_app.command("test") |
| 1255 | def siem_test( |
| 1256 | provider: str = typer.Argument(..., help="SIEM provider (splunk/elasticsearch/syslog/webhook)"), |
| 1257 | url: str = typer.Option(..., "--url", "-u", help="SIEM endpoint URL"), |
| 1258 | token: str | None = typer.Option(None, "--token", "-t", help="Auth token"), |
| 1259 | ): |
| 1260 | """Test SIEM connection.""" |
| 1261 | setup_logging() |
| 1262 | |
| 1263 | console.print(f"[bold]Testing SIEM Connection:[/bold] {provider}") |
| 1264 | |
| 1265 | async def run(): |
| 1266 | from modules.siem import ElasticsearchExporter, SplunkExporter, WebhookExporter |
| 1267 | |
| 1268 | if provider == "splunk": |
| 1269 | exporter = SplunkExporter(hec_url=url, hec_token=token or "") |
| 1270 | elif provider == "elasticsearch": |
| 1271 | exporter = ElasticsearchExporter(hosts=[url]) |
| 1272 | elif provider == "webhook": |
| 1273 | exporter = WebhookExporter(url=url) |
| 1274 | else: |
| 1275 | console.print(f"[red]Unknown provider: {provider}[/red]") |
| 1276 | return |
| 1277 | |
| 1278 | if await exporter.test_connection(): |
| 1279 | console.print("[green]✓ Connection successful[/green]") |
| 1280 | else: |
| 1281 | console.print("[red]✗ Connection failed[/red]") |
| 1282 | |
| 1283 | run_async(run()) |
| 1284 | |
| 1285 |
nothing calls this directly
no test coverage detected