121 lines
4.8 KiB
Python
121 lines
4.8 KiB
Python
from __future__ import annotations
|
|
|
|
import json
|
|
import queue
|
|
import threading
|
|
import time
|
|
import urllib.error
|
|
import urllib.request
|
|
from typing import Any
|
|
|
|
|
|
class WebhookNotifier:
|
|
"""Bounded asynchronous JSON webhook delivery for high-risk NDR incidents."""
|
|
|
|
def __init__(self, url: str, min_risk: int = 80, timeout: int = 5) -> None:
|
|
self.url = str(url or "").strip()
|
|
self.min_risk = max(1, min(100, int(min_risk)))
|
|
self.timeout = max(1, min(30, int(timeout)))
|
|
self.enabled = self.url.startswith(("http://", "https://"))
|
|
self._queue: queue.Queue[dict[str, Any]] = queue.Queue(maxsize=256)
|
|
self._stop = threading.Event()
|
|
self._thread = threading.Thread(target=self._run, name="ndr-webhook", daemon=True)
|
|
self._sent_state: dict[int, tuple[int, float]] = {}
|
|
self._sent = 0
|
|
self._failed = 0
|
|
self._dropped = 0
|
|
self._last_error = ""
|
|
self._last_sent_at = ""
|
|
|
|
def start(self) -> None:
|
|
if self.enabled and not self._thread.is_alive():
|
|
self._thread.start()
|
|
|
|
def stop(self, timeout: float = 3.0) -> None:
|
|
self._stop.set()
|
|
if self._thread.is_alive():
|
|
self._thread.join(timeout=timeout)
|
|
|
|
def notify(self, incident: dict[str, Any], evidence: dict[str, Any]) -> None:
|
|
if not self.enabled:
|
|
return
|
|
risk = int(incident.get("risk_score") or 0)
|
|
incident_id = int(incident.get("id") or 0)
|
|
if risk < self.min_risk or incident_id <= 0:
|
|
return
|
|
now = time.monotonic()
|
|
previous_risk, previous_at = self._sent_state.get(incident_id, (0, 0.0))
|
|
# Re-notify only if risk meaningfully escalated or 15 minutes passed.
|
|
if previous_at and risk < previous_risk + 10 and now - previous_at < 900:
|
|
return
|
|
payload = {
|
|
"event": "mikrosuricata.ndr.incident",
|
|
"incident": {
|
|
"id": incident_id,
|
|
"subject_ip": incident.get("subject_ip"),
|
|
"risk_score": risk,
|
|
"severity": incident.get("severity"),
|
|
"status": incident.get("status"),
|
|
"title": incident.get("title"),
|
|
"summary": incident.get("summary"),
|
|
"stages": incident.get("stages") or [],
|
|
"destinations": incident.get("destinations") or [],
|
|
"blocked": bool(incident.get("blocked")),
|
|
"block_target": incident.get("block_target"),
|
|
"last_seen": incident.get("last_seen"),
|
|
},
|
|
"evidence": {
|
|
"kind": evidence.get("kind"),
|
|
"stage": evidence.get("stage"),
|
|
"risk": evidence.get("risk"),
|
|
"summary": evidence.get("summary"),
|
|
"src_ip": evidence.get("src_ip"),
|
|
"dest_ip": evidence.get("dest_ip"),
|
|
"signature_id": evidence.get("signature_id"),
|
|
},
|
|
}
|
|
try:
|
|
self._queue.put_nowait(payload)
|
|
self._sent_state[incident_id] = (risk, now)
|
|
except queue.Full:
|
|
self._dropped += 1
|
|
|
|
def status(self) -> dict[str, Any]:
|
|
return {
|
|
"enabled": self.enabled,
|
|
"running": self._thread.is_alive(),
|
|
"min_risk": self.min_risk,
|
|
"queue": self._queue.qsize(),
|
|
"sent": self._sent,
|
|
"failed": self._failed,
|
|
"dropped": self._dropped,
|
|
"last_sent_at": self._last_sent_at,
|
|
"last_error": self._last_error,
|
|
}
|
|
|
|
def _run(self) -> None:
|
|
while not self._stop.is_set() or not self._queue.empty():
|
|
try:
|
|
payload = self._queue.get(timeout=0.25)
|
|
except queue.Empty:
|
|
continue
|
|
try:
|
|
body = json.dumps(payload, ensure_ascii=False, separators=(",", ":")).encode("utf-8")
|
|
request = urllib.request.Request(
|
|
self.url,
|
|
data=body,
|
|
headers={"Content-Type": "application/json", "User-Agent": "MikroSuricata-NDR/0.8"},
|
|
method="POST",
|
|
)
|
|
with urllib.request.urlopen(request, timeout=self.timeout) as response:
|
|
if int(getattr(response, "status", 200)) >= 400:
|
|
raise urllib.error.HTTPError(self.url, response.status, "webhook error", response.headers, None)
|
|
self._sent += 1
|
|
self._last_sent_at = time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime())
|
|
self._last_error = ""
|
|
except (urllib.error.URLError, urllib.error.HTTPError, TimeoutError, OSError) as exc:
|
|
self._failed += 1
|
|
self._last_error = str(exc)[:300]
|
|
finally:
|
|
self._queue.task_done()
|