crowdsec-notify/app.py

#!/usr/bin/env python3
"""
CrowdSec Notification Webhook
Groups alerts by IP and sends formatted notifications via Apprise
Author: Maxime Killinger
"""

import os
import logging
from flask import Flask, request, jsonify
import requests

# Configure logging
logging.basicConfig(
    level=logging.INFO,
    format='%(asctime)s - %(levelname)s - %(message)s',
    datefmt='%Y-%m-%d %H:%M:%S'
)
logger = logging.getLogger(__name__)

app = Flask(__name__)

import pycountry

# Configure logging
logging.basicConfig(
    level=logging.INFO,
    format='%(asctime)s - %(levelname)s - %(message)s',
    datefmt='%Y-%m-%d %H:%M:%S'
)
logger = logging.getLogger(__name__)

app = Flask(__name__)

APPRISE_URL = os.environ.get("APPRISE_URL")

def get_country_name(code):
    if not code or code.lower() == 'unknown':
        logger.warning(f"Country code is unknown or empty: {code}")
        return "Unknown"
    try:
        # Standardize code to alpha_2
        # Clean string just in case
        clean_code = code.strip().upper()
        if len(clean_code) != 2:
            logger.warning(f"Invalid country code format: {code}")
            return clean_code
            
        country = pycountry.countries.get(alpha_2=clean_code)
        if country:
            return country.name
            
        logger.warning(f"Country code not found in pycountry: {clean_code}")
        return clean_code
    except Exception as e:
        logger.warning(f"Error resolving country code {code}: {e}")
        return code

def format_scenario(scenario):
    if not scenario:
        return "unknown"
    return scenario.replace("crowdsecurity/", "").replace("_", " ")

# One icon per alert, mirroring the per-IP icon used by the CrowdSec feed.
# Severity is carried by the icon instead of a "[critical]" suffix, which
# keeps the identity line short enough to read at a glance on a phone.
SEVERITY_ICONS = {
    "critical": "🔴",
    "warning": "🟠",
    "info": "🔵",
}

def format_alert_block(alert, resolved=False):
    """Render one alert the way the CrowdSec feed renders one IP: an identity
    line (icon, name, context in parentheses), then the details indented under
    a tree marker. Returns the block as a single string."""
    labels = alert.get("labels", {})
    annotations = alert.get("annotations", {})

    name = labels.get("alertname", "alerte")
    severity = labels.get("severity", "")
    # "name" is what cAdvisor-based rules use to carry the container name
    context = (labels.get("instance") or labels.get("container")
               or labels.get("name") or labels.get("job") or "")

    icon = "✅" if resolved else SEVERITY_ICONS.get(severity, "🔥")
    head = f"{icon} {name}"
    if context:
        head += f" ({context})"

    summary = annotations.get("summary")
    description = annotations.get("description")
    details = [summary] if summary else []
    # description often just restates summary; only keep it when it adds something
    if description and description != summary:
        details.append(description)

    lines = [head] + [f"   └ {d}" for d in details]
    return "\n".join(lines)

@app.route('/crowdsec', methods=['POST'])
def handle_crowdsec():
    try:
        alerts = request.json
        if not alerts:
            logger.debug("Received empty alert payload")
            return jsonify({"status": "no data"}), 200
        
        # Group by IP
        ip_groups = {}
        for alert in alerts:
            ip = alert.get("source", {}).get("ip", "unknown")
            country = alert.get("source", {}).get("cn", "")
            scenario = alert.get("scenario", "unknown")
            
            if ip not in ip_groups:
                ip_groups[ip] = {"country": country, "scenarios": set()}
            ip_groups[ip]["scenarios"].add(format_scenario(scenario))
        
        # Format message
        num_ips = len(ip_groups)
        lines = []
        
        for ip, data in ip_groups.items():
            country_name = get_country_name(data["country"])
            whois_link = f"https://who.is/whois-ip/ip-address/{ip}"
            lines.append(f"🚫 {ip} ({country_name})")
            lines.append(f"   🔗 {whois_link}")
            for scenario in sorted(data["scenarios"]):
                lines.append(f"   └ {scenario}")
            lines.append("")
        
        body = "\n".join(lines).strip()
        title = f"🛡️ CrowdSec - {num_ips} IP{'s' if num_ips > 1 else ''} bannie{'s' if num_ips > 1 else ''}"
        
        # Send to Apprise
        response = requests.post(APPRISE_URL, json={"title": title, "body": body}, timeout=10)
        
        if response.ok:
            ips_list = ", ".join(ip_groups.keys())
            logger.info(f"Notification sent: {num_ips} IP(s) banned [{ips_list}]")
        else:
            logger.error(f"Apprise error: {response.status_code} - {response.text}")
        
        return jsonify({"status": "sent", "ips": num_ips}), 200
        
    except Exception as e:
        logger.exception(f"Error processing CrowdSec alert: {e}")
        return jsonify({"status": "error", "message": str(e)}), 500

@app.route('/alertmanager', methods=['POST'])
def handle_alertmanager():
    try:
        data = request.json or {}
        alerts = data.get("alerts", [])
        if not alerts:
            logger.debug("Received empty Alertmanager payload")
            return jsonify({"status": "no data"}), 200

        firing = [a for a in alerts if a.get("status") == "firing"]
        resolved = [a for a in alerts if a.get("status") == "resolved"]

        # One block per alert, blank-line separated, like the CrowdSec feed
        blocks = [format_alert_block(a) for a in firing]
        blocks += [format_alert_block(a, resolved=True) for a in resolved]

        # Title carries the counts, so the body never has to repeat them
        counts = []
        if firing:
            plural = "s" if len(firing) > 1 else ""
            counts.append(f"{len(firing)} alerte{plural} active{plural}")
        if resolved:
            counts.append(f"{len(resolved)} résolue{'s' if len(resolved) > 1 else ''}")

        title = ("🔥" if firing else "✅") + f" Alertmanager - {', '.join(counts)}"
        body = "\n\n".join(blocks)

        # Send to Apprise
        response = requests.post(APPRISE_URL, json={"title": title, "body": body}, timeout=10)

        if response.ok:
            logger.info(f"Alertmanager notification sent: {len(firing)} firing, {len(resolved)} resolved")
        else:
            logger.error(f"Apprise error: {response.status_code} - {response.text}")
            return jsonify({"status": "error", "code": response.status_code}), 502

        return jsonify({"status": "sent", "firing": len(firing), "resolved": len(resolved)}), 200

    except Exception as e:
        logger.exception(f"Error processing Alertmanager alert: {e}")
        return jsonify({"status": "error", "message": str(e)}), 500


@app.route('/health', methods=['GET'])
def health():
    return jsonify({"status": "ok"}), 200

if __name__ == '__main__':
    from waitress import serve
    import sys
    import signal

    if not APPRISE_URL:
        logger.critical("APPRISE_URL environment variable is not set")
        sys.exit(1)

    def handle_signal(signum, frame):
        logger.info(f"Received signal {signum}. Shutting down gracefully...")
        # Waitress does not have a public stop method when run this way easily,
        # but sys.exit(0) from a signal handler usually works to terminate.
        # However, waitress catches signals if passed to serve? 
        # Actually waitress handles signals by itself if running in main.
        # But we want to LOG it.
        sys.exit(0)

    signal.signal(signal.SIGTERM, handle_signal)
    signal.signal(signal.SIGINT, handle_signal)

    logger.info(f"Starting CrowdSec Notify - Apprise URL: {APPRISE_URL}")
    serve(app, host='0.0.0.0', port=5000)