diff --git a/docs/services/lxc208-manage/alert/crowdsec.py b/docs/services/lxc208-manage/alert/crowdsec.py new file mode 100644 index 0000000..c8d31b5 --- /dev/null +++ b/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"{title}", ""] + + 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) \ No newline at end of file