esh-pfi-infrastructure/scripts/kuma

#!/usr/bin/env -S uv run --quiet --script
# /// script
# requires-python = ">=3.11"
# dependencies = ["python-socketio[client]>=5.11", "requests>=2.31", "PyYAML>=6"]
# ///
"""kuma — a first-party Socket.IO client for Uptime Kuma 2.x.

WHY THIS EXISTS, since the obvious answers are both wrong:

  * There is NO REST CRUD API for monitors, in either major version. Checked
    against the 2.5.5 source tree, not the docs: `server/routers/` holds exactly
    two files -- api-router.js (Prometheus /metrics, badges, entry page) and
    status-page-router.js. API keys unlock /metrics and badges only. Every
    monitor operation lives on Socket.IO, the same channel the web UI uses.

  * The community wrapper `uptime-kuma-api` is ABANDONED -- last release
    2023-09-26 -- and its stated ceiling is "support for uptime kuma 1.23.0 and
    1.23.1". It does not speak 2.x and never will. Depending on it would trade a
    maintained server for an unmaintained client.

So: a thin client of our own, against events read out of server.js at tag 2.5.5
(`setup`, `login`, `getMonitorList`, `add`, `editMonitor`, `deleteMonitor`).
Small on purpose -- this is four events and a socket, not a framework.

USAGE
    scripts/kuma setup                      # create the admin (first run only)
    scripts/kuma list                       # every monitor, with its state
    scripts/kuma seed <file.yaml>           # idempotent upsert from a file
    scripts/kuma add <name> <url> [--type http|ping|port] [--port N]

CREDENTIALS come from the vault, never from a flag:
    secret get ana-docker/.config/uptime-kuma/admin
Password on the command line lands in shell history and in the process table.
"""

from __future__ import annotations

import argparse
import json
import subprocess
import sys
import time

import socketio
import yaml

URL = "http://10.250.50.70:3001"
VAULT_KEY = "ana-docker/.config/uptime-kuma/admin"

# Defaults the 2.x `add` handler REQUIRES rather than defaults itself. It
# JSON.stringify()s kafkaProducerBrokers, kafkaProducerSaslOptions, conditions
# and rabbitmqNodes unconditionally, so they must be present even for an HTTP
# check -- omit them and you get "undefined" written into the row.
MONITOR_DEFAULTS = {
    "type": "http",
    "method": "GET",
    "interval": 60,
    "retryInterval": 60,
    "resendInterval": 0,
    "maxretries": 2,
    "timeout": 16,
    "maxredirects": 10,
    "accepted_statuscodes": ["200-299"],
    "notificationIDList": {},
    "active": True,
    "expiryNotification": False,
    "ignoreTls": True,      # the fleet is full of self-signed internal certs
    "upsideDown": False,
    "kafkaProducerBrokers": [],
    "kafkaProducerSaslOptions": {},
    "conditions": [],
    "rabbitmqNodes": [],
}


# A channel registered with isDefault + applyExisting attaches to monitors that
# ALREADY exist, not merely to ones created afterwards -- without it, seeding
# monitors before channels silently leaves the existing board unwired.
NOTIFICATION_DEFAULTS = {
    "type": "webhook",
    "isDefault": True,
    "applyExisting": True,
    "webhookContentType": "json",
    "active": True,
}


def vault_get(key: str) -> dict:
    """{'username','password'} from the vault. Loud on failure -- a silent
    empty credential would show up as a confusing auth error much later."""
    out = subprocess.run(
        ["secret", "get", key], capture_output=True, text=True, timeout=60
    )
    if out.returncode != 0:
        sys.exit(f"vault: could not read {key}: {out.stderr.strip() or out.stdout.strip()}")
    body = out.stdout.strip()
    try:
        return json.loads(body)
    except json.JSONDecodeError:
        # tolerate a "user:pass" single line as well
        if ":" in body:
            u, _, p = body.partition(":")
            return {"username": u.strip(), "password": p.strip()}
        sys.exit(f"vault: {key} is neither JSON nor user:pass")


class Kuma:
    def __init__(self, url: str = URL):
        self.url = url
        self.sio = socketio.Client(reconnection=False)
        self._connected = False
        # ⚠ THE TRAP THAT COST A DUPLICATED BOARD (2026-09-21).
        # `getMonitorList`'s callback returns ONLY {ok: true} -- server.js:1001
        # calls server.sendMonitorList(socket) and then acks with a bare ok. The
        # actual list arrives as a SEPARATE pushed "monitorList" event. Reading
        # the callback therefore yields an empty board that looks authoritative,
        # and an idempotent seed built on it duplicates every row instead of
        # updating it. Capture the push; never trust the ack for data.
        self._monitor_list: dict | None = None

        self._notification_list: list | None = None

        @self.sio.on("monitorList")
        def _on_monitor_list(data):
            self._monitor_list = data or {}

        @self.sio.on("notificationList")
        def _on_notification_list(data):
            self._notification_list = data or []

    def __enter__(self):
        self.sio.connect(self.url, transports=["websocket"], wait_timeout=20)
        self._connected = True
        # 2.x emits a burst of state on connect; give it a beat to settle so a
        # call() issued immediately does not race the handshake.
        time.sleep(1.0)
        return self

    def __exit__(self, *exc):
        if self._connected:
            try:
                self.sio.disconnect()
            except Exception:
                pass

    def call(self, event: str, *args, timeout: int = 30):
        """Emit with an ack. Every Kuma event answers {ok, msg, ...}; a falsey
        `ok` is raised rather than returned, so no caller can ignore it."""
        res = self.sio.call(event, args if len(args) != 1 else args[0], timeout=timeout)
        if isinstance(res, dict) and not res.get("ok", True):
            raise RuntimeError(f"{event}: {res.get('msg', res)}")
        return res

    def need_setup(self) -> bool:
        return bool(self.sio.call("needSetup", timeout=20))

    def setup(self, username: str, password: str):
        return self.call("setup", username, password)

    def login(self, username: str, password: str):
        return self.call("login", {"username": username, "password": password, "token": ""})

    def monitors(self, timeout: int = 20) -> dict:
        """Every monitor, keyed by id. Waits for the PUSHED `monitorList` event
        rather than reading the ack -- see the note in __init__."""
        self._monitor_list = None
        self.sio.emit("getMonitorList")
        deadline = time.time() + timeout
        while self._monitor_list is None and time.time() < deadline:
            self.sio.sleep(0.2)
        if self._monitor_list is None:
            raise RuntimeError("no monitorList event within %ss -- not logged in?" % timeout)
        return self._monitor_list

    def delete(self, monitor_id: int):
        return self.call("deleteMonitor", monitor_id, False)

    def notifications(self, timeout: int = 10) -> list:
        """Registered notification channels.

        ⚠ DIFFERENT PUSH RULE FROM monitorList, and the difference bit once.
        There is no `getNotificationList` event to ask with -- the server pushes
        `notificationList` ONCE, at login, and again after any change. So this
        must NOT clear the captured value before waiting: doing that discards
        the only copy it will ever be sent and then blocks for the full timeout
        before reporting an empty list, which reads exactly like "no channels
        are configured" and is a lie.
        """
        deadline = time.time() + timeout
        while self._notification_list is None and time.time() < deadline:
            self.sio.sleep(0.2)
        return self._notification_list or []

    def add_notification(self, notification: dict):
        """Register (or update) a channel. notificationID None => create."""
        return self.call("addNotification", notification, None)

    def test_notification(self, notification: dict):
        return self.call("testNotification", notification)

    def add(self, monitor: dict):
        return self.call("add", {**MONITOR_DEFAULTS, **monitor})

    def edit(self, monitor: dict):
        return self.call("editMonitor", monitor)


def connected_and_logged_in() -> Kuma:
    creds = vault_get(VAULT_KEY)
    k = Kuma().__enter__()
    k.login(creds["username"], creds["password"])
    return k


def cmd_setup(args):
    creds = vault_get(VAULT_KEY)
    with Kuma() as k:
        if not k.need_setup():
            print("already initialised — nothing to do")
            return
        k.setup(creds["username"], creds["password"])
        print(f"admin '{creds['username']}' created")


def cmd_list(args):
    k = connected_and_logged_in()
    try:
        mons = k.monitors()
        if not mons:
            print("no monitors")
            return
        print(f"{'ID':>4}  {'NAME':<28} {'TYPE':<6} {'ACT':<4} TARGET")
        for m in sorted(mons.values(), key=lambda x: (x.get("name") or "")):
            tgt = m.get("url") or f"{m.get('hostname','')}:{m.get('port','')}"
            print(f"{m.get('id'):>4}  {(m.get('name') or '')[:28]:<28} "
                  f"{(m.get('type') or '')[:6]:<6} {str(m.get('active')):<4} {tgt[:52]}")
        print(f"\n{len(mons)} monitor(s)")
    finally:
        k.__exit__()


def cmd_seed(args):
    """Idempotent: a monitor whose NAME already exists is edited, not duplicated.
    Name is the key because a URL legitimately repeats (root and /healthz on the
    same service), and re-running a seed must never fork the board."""
    spec = yaml.safe_load(open(args.file))
    wanted = spec["monitors"] if isinstance(spec, dict) else spec
    channels = spec.get("notifications", []) if isinstance(spec, dict) else []
    k = connected_and_logged_in()
    try:
        # Channels FIRST: a monitor created while a default channel exists
        # inherits it. Seeding monitors first and alerting second is how you get
        # a board that detects and tells nobody -- the exact gap this fleet's
        # service layer was built to close.
        have = {n.get("name") for n in k.notifications()}
        for ch in channels:
            if ch["name"] in have:
                print(f"  = {ch['name']} (channel)")
                continue
            k.add_notification({**NOTIFICATION_DEFAULTS, **ch})
            print(f"  + {ch['name']} (channel)")
        existing = {m.get("name"): m for m in k.monitors().values()}
        added = updated = 0
        for m in wanted:
            if m["name"] in existing:
                merged = {**existing[m["name"]], **m}
                k.edit(merged)
                updated += 1
                print(f"  ~ {m['name']}")
            else:
                k.add(m)
                added += 1
                print(f"  + {m['name']}")
        print(f"\n{added} added, {updated} updated, {len(wanted)} in spec")
    finally:
        k.__exit__()


def cmd_channels(args):
    k = connected_and_logged_in()
    try:
        ns = k.notifications()
        if not ns:
            print("no notification channels — THE BOARD ALERTS NOBODY")
            return
        for n in ns:
            cfg = json.loads(n.get("config") or "{}")
            print("  %-26s %-9s default=%-5s %s" % (
                n.get("name"), cfg.get("type"), n.get("isDefault"), cfg.get("webhookURL", "")))
    finally:
        k.__exit__()


def cmd_dedupe(args):
    """Keep the LOWEST id per name, delete the rest. Needed because the first
    version of this client read the getMonitorList ack instead of the pushed
    event, saw an empty board, and re-added all 13 rows on a second seed."""
    k = connected_and_logged_in()
    try:
        by_name: dict[str, list] = {}
        for m in k.monitors().values():
            by_name.setdefault(m.get("name"), []).append(m)
        removed = 0
        for name, group in sorted(by_name.items()):
            if len(group) < 2:
                continue
            group.sort(key=lambda m: m.get("id"))
            for dup in group[1:]:
                k.delete(dup["id"])
                removed += 1
                print("  - %s (id %s)" % (name, dup["id"]))
        print("\n%d duplicate(s) removed" % removed)
    finally:
        k.__exit__()


def cmd_add(args):
    k = connected_and_logged_in()
    try:
        mon = {"name": args.name, "type": args.type}
        if args.type == "http":
            mon["url"] = args.target
        else:
            mon["hostname"] = args.target
            if args.port:
                mon["port"] = args.port
        k.add(mon)
        print(f"added {args.name}")
    finally:
        k.__exit__()


def main():
    p = argparse.ArgumentParser(prog="kuma", description=__doc__.split("\n")[0])
    sub = p.add_subparsers(dest="cmd", required=True)
    sub.add_parser("setup").set_defaults(fn=cmd_setup)
    sub.add_parser("list").set_defaults(fn=cmd_list)
    s = sub.add_parser("seed"); s.add_argument("file"); s.set_defaults(fn=cmd_seed)
    a = sub.add_parser("add")
    a.add_argument("name"); a.add_argument("target")
    a.add_argument("--type", default="http", choices=["http", "ping", "port"])
    a.add_argument("--port", type=int)
    a.set_defaults(fn=cmd_add)
    sub.add_parser("dedupe").set_defaults(fn=cmd_dedupe)
    sub.add_parser("channels").set_defaults(fn=cmd_channels)
    args = p.parse_args()
    try:
        args.fn(args)
    except Exception as exc:
        sys.exit(f"error: {exc}")


if __name__ == "__main__":
    main()