diff --git a/app.py b/app.py index dd38041..4c95141 100644 --- a/app.py +++ b/app.py @@ -61,6 +61,43 @@ def format_scenario(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: @@ -120,48 +157,23 @@ def handle_alertmanager(): logger.debug("Received empty Alertmanager payload") return jsonify({"status": "no data"}), 200 - def format_alert(alert): - labels = alert.get("labels", {}) - annotations = alert.get("annotations", {}) - name = labels.get("alertname", "alerte") - severity = labels.get("severity", "") - instance = labels.get("instance") or labels.get("container") or labels.get("job") or "" - - head = name - if severity: - head += f" [{severity}]" - if instance: - head += f" — {instance}" - - out = [head] - summary = annotations.get("summary") - description = annotations.get("description") - if summary: - out.append(f" {summary}") - if description and description != summary: - out.append(f" {description}") - return "\n".join(out) - firing = [a for a in alerts if a.get("status") == "firing"] resolved = [a for a in alerts if a.get("status") == "resolved"] - lines = [] + # 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: - lines.append(f"🔥 {len(firing)} alerte(s) active(s)") - lines.extend(format_alert(a) for a in firing) + plural = "s" if len(firing) > 1 else "" + counts.append(f"{len(firing)} alerte{plural} active{plural}") if resolved: - if lines: - lines.append("") - lines.append(f"✅ {len(resolved)} résolue(s)") - lines.extend(format_alert(a) for a in resolved) + counts.append(f"{len(resolved)} résolue{'s' if len(resolved) > 1 else ''}") - status = data.get("status", "firing") - group_name = data.get("groupLabels", {}).get("alertname", "") - title = ("🔥" if status == "firing" else "✅") + " Alertmanager" - if group_name: - title += f" — {group_name}" - - body = "\n".join(lines).strip() + 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)