esh-pfi-infrastructure/services/althing-alert-bridge/bridge.py

#!/usr/bin/env python3
"""Fleet alert bridge — HTTP alerts from monitoring tools into the althing inbox.

One listener, one route per SOURCE. Each source declares its own subject prefix,
its own footer link, and how to read its payload, because the tools do not agree
on a shape and the reader needs to know which tool is talking.

  POST /beszel  <- Beszel, via Shoutrrr generic JSON: {title, message}
  POST /kuma    <- Uptime Kuma, via its webhook:      {heartbeat, monitor, msg}

WHY A REGISTRY RATHER THAN A SECOND SERVICE. The prefix and footer were
hardcoded when Beszel was the only caller, so a second tool routed through it
would have arrived labelled "[Beszel]" -- an alert that lies about its own
source is worse than no alert, because it sends you to the wrong dashboard.
Generalising costs a dict; a sibling service costs a second unit, a second port
and a second thing to notice has died.

⚠ `/beszel` IS FROZEN. Its prefix, footer and default title must stay byte-
identical -- Beszel's delivery path has been verified end-to-end in production
(2026-09-10, thread 01M25Z0WFDJM92GPTJQF769HJ7) and this refactor is not
allowed to quietly change what that path emits. `deliver()` still defaults to
the Beszel route precisely so the original tests exercise it unchanged.
"""
import json
import os
import subprocess
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer


def _env(*names, default=None):
    """First env var that is set. Accepts the legacy BESZEL_* names so a
    half-finished deploy (new bridge, old unit) still starts instead of
    crash-looping on a KeyError."""
    for n in names:
        if os.environ.get(n):
            return os.environ[n]
    if default is None:
        raise KeyError(' / '.join(names))
    return default


def parse_generic(payload):
    """Shoutrrr generic JSON — what Beszel sends. {title, message}."""
    return payload.get('title'), payload.get('message')


def parse_kuma(payload):
    """Uptime Kuma's webhook — {heartbeat, monitor, msg}.

    Kuma's own `msg` already reads like a sentence ("[Homepage] [🔴 Down]
    connect ECONNREFUSED"), so it is the body. The title is built from the
    monitor name and the heartbeat status so the SUBJECT alone says which
    service and which direction -- that is what you see in an inbox list
    without opening anything.
    """
    monitor = payload.get('monitor') or {}
    heartbeat = payload.get('heartbeat') or {}
    body = payload.get('msg') or ''
    name = monitor.get('name')
    # Kuma status: 0 down, 1 up, 2 pending, 3 maintenance.
    state = {0: 'DOWN', 1: 'UP', 2: 'PENDING', 3: 'MAINTENANCE'}.get(heartbeat.get('status'))

    if name and state:
        title = f'{name} is {state}'
    else:
        # NOT every Kuma notification is a monitor transition: `testNotification`
        # and certificate-expiry alerts arrive with no monitor and no heartbeat.
        # Fabricating "unknown monitor is ?" from those makes a real alert read
        # like a bug; the message itself is the better subject.
        first = body.strip().splitlines()[0] if body.strip() else ''
        title = (first[:120] or 'notification')
    detail = heartbeat.get('msg')
    if detail and detail not in body:
        body = f'{body}\n\n{detail}'.strip()
    url = monitor.get('url')
    if url:
        body = f'{body}\n\nTarget: {url}'.strip()
    return title, body


SOURCES = {
    # FROZEN — see the module docstring.
    '/beszel': {
        'prefix': '[Beszel] ',
        'default_title': 'Beszel fleet alert',
        'footer': '\n\nHub: http://10.250.50.70:8090\n',
        'parse': parse_generic,
    },
    '/kuma': {
        'prefix': '[Uptime Kuma] ',
        'default_title': 'Uptime Kuma alert',
        'footer': '\n\nBoard: http://10.250.50.70:3001\n',
        'parse': parse_kuma,
    },
}


def deliver(payload, route='/beszel'):
    source = SOURCES[route]
    title, message = source['parse'](payload)
    if title is None:
        title = source['default_title']
    if not isinstance(title, str) or not isinstance(message, str) or not message.strip():
        raise ValueError('Expected a nonempty message and string title')
    result = subprocess.run(
        [_env('POSTBOX'), '--json', 'send',
         '--to', _env('ALERT_RECIPIENT', 'BESZEL_ALERT_RECIPIENT'),
         '--subject', source['prefix'] + title],
        input=message + source['footer'],
        text=True, capture_output=True, timeout=25,
    )
    if result.returncode:
        raise RuntimeError('postbox delivery failed: ' + result.stderr.strip())
    receipt = json.loads(result.stdout)
    print(json.dumps({'event': 'delivered', 'route': route, 'title': title, 'receipt': receipt}), flush=True)
    return receipt


class Handler(BaseHTTPRequestHandler):
    def respond(self, status, body):
        data = json.dumps(body).encode()
        self.send_response(status)
        self.send_header('Content-Type', 'application/json')
        self.send_header('Content-Length', str(len(data)))
        self.end_headers()
        self.wfile.write(data)

    def do_GET(self):
        self.respond(200 if self.path == '/healthz' else 404,
                     {'service': 'althing-alert-bridge',
                      'routes': sorted(SOURCES),
                      'delivery': 'verified per POST'})

    def do_POST(self):
        if self.path not in SOURCES:
            return self.respond(404, {'error': 'Unknown route', 'routes': sorted(SOURCES)})
        allowed = _env('ALERT_ALLOWED_SOURCES', 'BESZEL_ALLOWED_SOURCES').split(',')
        if self.client_address[0] not in allowed:
            return self.respond(403, {'error': 'Source not allowed'})
        try:
            length = int(self.headers.get('Content-Length', '0'))
            if not 0 < length <= 65536:
                raise ValueError('Invalid body size')
            self.connection.settimeout(10)
            payload = json.loads(self.rfile.read(length))
            if not isinstance(payload, dict):
                raise ValueError('Expected JSON object')
            receipt = deliver(payload, self.path)
        except (ValueError, TypeError) as exc:
            return self.respond(400, {'error': str(exc)})
        except (OSError, RuntimeError, subprocess.TimeoutExpired) as exc:
            print(json.dumps({'event': 'delivery_failed', 'route': self.path, 'error': str(exc)}), flush=True)
            return self.respond(502, {'error': 'Althing delivery failed; inspect service journal'})
        self.respond(200, {'delivered': True, 'receipt': receipt})


if __name__ == '__main__':
    ThreadingHTTPServer(
        (_env('ALERT_BIND_HOST', 'BESZEL_BIND_HOST'),
         int(_env('ALERT_BIND_PORT', 'BESZEL_BIND_PORT', default='8096'))),
        Handler,
    ).serve_forever()