signal-webhook-handler/signal_webhook/app.py

77 lines
2.2 KiB
Python
Raw Normal View History

2022-03-22 14:15:39 +01:00
import typing as t
2022-03-23 17:45:08 +01:00
from collections import defaultdict
2022-03-22 14:15:39 +01:00
from flask import Flask, request
from . import signal
2022-03-23 17:45:08 +01:00
from . import configuration
2022-03-22 14:15:39 +01:00
app = Flask(__name__)
@app.route("/")
def root() -> t.Tuple[str, int]:
return "Not supported.", 400
@app.route("/alertmanager", methods=["POST"])
@app.route("/alertmanager/<recipient>", methods=["POST"])
def alertmanager(recipient=None):
2022-03-22 14:15:39 +01:00
data = request.get_json(cache=False)
2022-03-23 17:45:08 +01:00
2022-03-22 14:15:39 +01:00
if "version" not in data or int(data["version"]) != 4:
return "Bad API version", 400
if "alerts" not in data:
return "No alerts", 400
if "status" not in data:
return "No status", 400
if data["status"] != "firing":
# Ignore
return "OK", 200
2022-03-23 17:45:08 +01:00
severities = defaultdict(int)
alert_lines = []
2022-03-22 14:15:39 +01:00
for alert in data["alerts"]:
2022-03-23 17:45:08 +01:00
try:
severities[alert["labels"]["severity"]] += 1
alert_lines.append(
f"<{alert['labels']['alertname']}> {alert['annotations']['description']}"
)
except KeyError as exn:
alert_lines.append(f"(malformed alert — {exn})")
msg = f"[PROMETHEUS] {len(data['alerts'])} firing: "
msg += ", ".join([f"{count} {typ}" for typ, count in severities.items()])
msg += "\n\n"
msg += "\n".join(["* " + line for line in alert_lines])
if recipient is None:
recipient = configuration.DEFAULT_RECIPIENT
try:
recipient_nums = configuration.RECIPIENTS[recipient]
except KeyError:
return "Bad recipient", 400
signal.signal_send(recipient_nums, msg)
2022-03-23 17:45:08 +01:00
return "OK", 200
2023-08-14 09:21:49 +02:00
@app.route("/plain", methods=["POST"])
def plain():
"""Send a message through Signal, plain POST args.
Expects:
* message: plaintext message to be sent;
* recipient: ID of recipient. If absent, default recipient is used.
"""
if "message" not in request.form:
return "Missing message", 400
recipient = request.form.get("recipient", configuration.DEFAULT_RECIPIENT)
try:
recipient_num = configuration.RECIPIENTS[recipient]
except KeyError:
return "Unauthorized recipient", 400
signal.signal_send(recipient_num, request.form["message"])
return "OK", 200