Sanitized mirror from private repository - 2026-04-19 15:28:05 UTC

alerting/signal-bridge/app.py

@@ -0,0 +1,130 @@
+#!/usr/bin/env python3
+"""
+Signal Bridge for Alertmanager
+Receives webhooks from Alertmanager and forwards to Signal API
+"""
+
+import os
+import json
+import requests
+from flask import Flask, request, jsonify
+
+app = Flask(__name__)
+
+# Configuration from environment variables
+SIGNAL_API_URL = os.environ.get('SIGNAL_API_URL', 'http://signal-api:8080')
+SIGNAL_SENDER = os.environ.get('SIGNAL_SENDER', '')  # Your Signal number
+SIGNAL_RECIPIENTS = os.environ.get('SIGNAL_RECIPIENTS', '').split(',')  # Comma-separated
+
+def format_alert_message(alert_data):
+    """Format Alertmanager webhook payload into a readable message"""
+    messages = []
+
+    status = alert_data.get('status', 'unknown')
+
+    for alert in alert_data.get('alerts', []):
+        alert_status = alert.get('status', status)
+        labels = alert.get('labels', {})
+        annotations = alert.get('annotations', {})
+
+        severity = labels.get('severity', 'unknown')
+        alertname = labels.get('alertname', 'Unknown Alert')
+        instance = labels.get('instance', 'unknown')
+
+        summary = annotations.get('summary', alertname)
+        description = annotations.get('description', '')
+
+        # Status emoji
+        if alert_status == 'resolved':
+            status_emoji = '✅'
+            status_text = 'RESOLVED'
+        elif severity == 'critical':
+            status_emoji = '🚨'
+            status_text = 'CRITICAL'
+        else:
+            status_emoji = '⚠️'
+            status_text = 'WARNING'
+
+        msg = f"{status_emoji} [{status_text}] {summary}"
+        if description:
+            msg += f"\n{description}"
+
+        messages.append(msg)
+
+    return "\n\n".join(messages)
+
+def send_signal_message(message):
+    """Send message via Signal API"""
+    if not SIGNAL_SENDER or not SIGNAL_RECIPIENTS:
+        app.logger.error("Signal sender or recipients not configured")
+        return False
+
+    success = True
+    for recipient in SIGNAL_RECIPIENTS:
+        recipient = recipient.strip()
+        if not recipient:
+            continue
+
+        try:
+            payload = {
+                "message": message,
+                "number": SIGNAL_SENDER,
+                "recipients": [recipient]
+            }
+
+            response = requests.post(
+                f"{SIGNAL_API_URL}/v2/send",
+                json=payload,
+                timeout=30
+            )
+
+            if response.status_code in [200, 201]:
+                app.logger.info(f"Message sent to {recipient}")
+            else:
+                app.logger.error(f"Failed to send to {recipient}: {response.status_code} - {response.text}")
+                success = False
+
+        except Exception as e:
+            app.logger.error(f"Error sending to {recipient}: {e}")
+            success = False
+
+    return success
+
+@app.route('/health', methods=['GET'])
+def health():
+    return jsonify({"status": "healthy"}), 200
+
+@app.route('/alert', methods=['POST'])
+def receive_alert():
+    """Receive alert from Alertmanager and forward to Signal"""
+    try:
+        alert_data = request.get_json()
+
+        if not alert_data:
+            return jsonify({"error": "No data received"}), 400
+
+        app.logger.info(f"Received alert: {json.dumps(alert_data, indent=2)}")
+
+        message = format_alert_message(alert_data)
+
+        if send_signal_message(message):
+            return jsonify({"status": "sent"}), 200
+        else:
+            return jsonify({"status": "partial_failure"}), 207
+
+    except Exception as e:
+        app.logger.error(f"Error processing alert: {e}")
+        return jsonify({"error": str(e)}), 500
+
+@app.route('/test', methods=['POST'])
+def test_message():
+    """Send a test message"""
+    message = request.json.get('message', '🧪 Test alert from Signal Bridge')
+
+    if send_signal_message(message):
+        return jsonify({"status": "sent"}), 200
+    else:
+        return jsonify({"status": "failed"}), 500
+
+if __name__ == '__main__':
+    app.run(host='0.0.0.0', port=5000)