Добавить docs/services/lxc208-manage/alert/crowdsec.py

docs/services/lxc208-manage/alert/crowdsec.py

@@ -0,0 +1,49 @@
+import logging
+from typing import Dict, Optional
+
+from alerts.alerts import ALERTS_CONFIG
+from alerts.utils import escape_html
+from alerts.formatters import get_formatter
+
+logger = logging.getLogger(__name__)
+
+SECURITY_CHANNEL = "security"
+
+
+def process_crowdsec_alert(obj: Dict) -> None:
+    alerts = obj.get("crowdsec_alerts", [])
+    if not alerts:
+        logger.warning("No crowdsec_alerts in payload")
+        return
+
+    from alerts.matrix import send_matrix
+    
+    for alert in alerts:
+        scenario = alert.get("scenario", "Unknown")
+        
+        config = ALERTS_CONFIG.get(scenario)
+        if not config:
+            logger.warning("Unknown CrowdSec scenario: %s", scenario)
+            continue
+        
+        room = config.get("room", SECURITY_CHANNEL)
+        title = config.get("title", f"CrowdSec Alert: {scenario}")
+        
+        formatter_name = config.get("formatter")
+        formatter = get_formatter(formatter_name) if formatter_name else None
+        
+        lines = [f"<b>{title}</b>", ""]
+        
+        if formatter:
+            lines.append(formatter(alert))
+        else:
+            lines.append(f"\U0001F5A5 Host: {escape_html(alert.get('host', 'unknown'))}")
+            lines.append(f"\U0001F4CD Target: {escape_html(alert.get('scope', ''))}:{escape_html(alert.get('value', ''))}")
+        
+        if "summary" in config:
+            lines.append("")
+            lines.append(config["summary"])
+        if "info" in config:
+            lines.append(config["info"])
+        
+        send_matrix("\n".join(lines), room)