49 lines
1.5 KiB
Python
49 lines
1.5 KiB
Python
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) |