Export events via webhooks (Slack, Discord, PagerDuty, custom).
| 9 | class WebhookExporter(SIEMExporter): |
| 10 | """Export events via webhooks (Slack, Discord, PagerDuty, custom).""" |
| 11 | |
| 12 | def __init__( |
| 13 | self, |
| 14 | url: str, |
| 15 | method: str = "POST", |
| 16 | headers: dict | None = None, |
| 17 | auth_token: str | None = None, |
| 18 | auth_header: str = "Authorization", |
| 19 | format: str = "json", # json, slack, discord, pagerduty |
| 20 | min_severity: str = "low", # Only send events at or above this severity |
| 21 | timeout: float = 10.0, |
| 22 | ): |
| 23 | """Initialize webhook exporter. |
| 24 | |
| 25 | Args: |
| 26 | url: Webhook URL |
| 27 | method: HTTP method |
| 28 | headers: Custom headers |
| 29 | auth_token: Authentication token |
| 30 | auth_header: Header name for auth token |
| 31 | format: Payload format (json, slack, discord, pagerduty) |
| 32 | min_severity: Minimum severity to trigger webhook |
| 33 | timeout: Request timeout |
| 34 | """ |
| 35 | super().__init__() |
| 36 | self.url = url |
| 37 | self.method = method.upper() |
| 38 | self.custom_headers = headers or {} |
| 39 | self.auth_token = auth_token |
| 40 | self.auth_header = auth_header |
| 41 | self.format = format.lower() |
| 42 | self.min_severity = min_severity.lower() |
| 43 | self.timeout = timeout |
| 44 | |
| 45 | self._severity_order = ["info", "low", "medium", "high", "critical"] |
| 46 | |
| 47 | def _should_send(self, event: SIEMEvent) -> bool: |
| 48 | """Check if event meets severity threshold.""" |
| 49 | event_idx = self._severity_order.index(event.severity.lower()) |
| 50 | min_idx = self._severity_order.index(self.min_severity) |
| 51 | return event_idx >= min_idx |
| 52 | |
| 53 | def _get_headers(self) -> dict: |
| 54 | """Get request headers.""" |
| 55 | headers = { |
| 56 | "Content-Type": "application/json", |
| 57 | **self.custom_headers, |
| 58 | } |
| 59 | |
| 60 | if self.auth_token: |
| 61 | if self.format == "pagerduty": |
| 62 | headers["Authorization"] = f"Token token={self.auth_token}" |
| 63 | else: |
| 64 | headers[self.auth_header] = f"Bearer {self.auth_token}" |
| 65 | |
| 66 | return headers |
| 67 | |
| 68 | def _format_payload(self, event: SIEMEvent) -> dict: |
no outgoing calls
no test coverage detected