#!/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)