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": [],
}


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.sio.on("monitorList")
        def _on_monitor_list(data):
            self._monitor_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 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
    k = connected_and_logged_in()
    try:
        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_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)
    args = p.parse_args()
    try:
        args.fn(args)
    except Exception as exc:
        sys.exit(f"error: {exc}")


if __name__ == "__main__":
    main()