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._status_page_list: dict | 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 []

        @self.sio.on("statusPageList")
        def _on_status_page_list(data):
            self._status_page_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 status_pages(self, timeout: int = 10) -> list:
        deadline = time.time() + timeout
        while self._status_page_list is None and time.time() < deadline:
            self.sio.sleep(0.2)
        return list((self._status_page_list or {}).values())

    def ensure_status_page(self, slug: str, title: str, groups: list):
        """Create the page if absent, then save its config and membership.

        The Homepage `uptimekuma` widget calls /api/status-page/<slug> and
        /api/status-page/heartbeat/<slug>; without a PUBLISHED page at that slug
        the widget polls a 404 forever. That is why the widget labels were held
        back when this service was rebuilt -- a dashboard widget pointed at a
        dead target is the suspected cause of both of Homepage's unkillable
        wedges, so shipping one deliberately would have been daft.
        """
        have = {sp.get("slug") for sp in self.status_pages()}
        if slug not in have:
            self.call("addStatusPage", title, slug)
        cfg = {
            "slug": slug, "title": title,
            "description": "Fleet service layer — is the service actually serving.",
            "logo": None, "theme": "dark", "published": True,
            "showTags": False, "footerText": None, "customCSS": "",
            "showPoweredBy": False, "rssTitle": title,
            "showOnlyLastHeartbeat": False, "showCertificateExpiry": False,
            "autoRefreshInterval": 300, "domainNameList": [],
            "googleAnalyticsId": None, "analyticsId": None,
            "analyticsScriptUrl": None, "analyticsType": None,
        }
        # ⚠ imgDataUrl must be a STRING, not None. The handler calls
        # imgDataUrl.startsWith("data:") unconditionally, so null throws
        # "Cannot read properties of null" and the page is created but never
        # populated -- which looks like success from /api/status-page (200,
        # correct title) while the group list is empty.
        return self.call("saveStatusPage", (slug, cfg, "", groups), timeout=45)

    def monitors_by_name(self) -> dict:
        """Fresh read of the board, keyed by name."""
        return {m.get("name"): m for m in self.monitors().values()}

    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")
        # Case-INSENSITIVE: an ASCII-ordinal sort buries every lowercase
        # service name (talk, vor, task-board) below every capitalised one,
        # which reads as a messy board when it is really a messy sort.
        for m in sorted(mons.values(), key=lambda x: (x.get("name") or "").lower()):
            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 = renamed = 0
        for m in wanted:
            spec_m = {kk: vv for kk, vv in m.items() if kk != "rename_from"}
            target = existing.get(m["name"])

            # RENAME SUPPORT, and it is load-bearing rather than a nicety.
            # This seed is keyed on NAME, so editing a name in the spec would
            # otherwise read as a brand-new monitor: the tool would ADD it and
            # leave the old row orphaned, still checking, still alerting, and
            # carrying all the history. `rename_from` names the old row for one
            # run; drop the line once the rename has landed.
            if target is None and m.get("rename_from"):
                target = existing.get(m["rename_from"])
                if target is not None:
                    k.edit({**target, **spec_m})
                    renamed += 1
                    print(f"  > {m['rename_from']} -> {m['name']}")
                    continue

            if target is not None:
                k.edit({**target, **spec_m})
                updated += 1
                print(f"  ~ {m['name']}")
            else:
                k.add(spec_m)
                added += 1
                print(f"  + {m['name']}")

        # A monitor on the board that the spec no longer names is not silently
        # fine -- a forgotten row keeps checking and keeps alerting.
        #
        # ⚠ RE-READ THE BOARD FIRST. The first cut diffed against `existing`,
        # the snapshot taken BEFORE the edits, so every row this run had just
        # renamed was still in it under its old name and got reported as an
        # orphan that no longer existed. A warning that cries wolf on its own
        # successful work is worse than no warning.
        spec_names = {m["name"] for m in wanted}
        orphans = sorted(n for n in k.monitors_by_name() if n not in spec_names)
        if orphans:
            print("\n  ⚠ on the board but NOT in the spec (left alone, still alerting):")
            for o in orphans:
                print(f"      {o}")

        print(f"\n{added} added, {updated} updated, {renamed} renamed, {len(wanted)} in spec")

        # LAST, because the page references monitor IDs and therefore needs the
        # monitors to exist first.
        sp = spec.get("status_page") if isinstance(spec, dict) else None
        if sp:
            board = k.monitors_by_name()
            members = [{"id": board[n]["id"]} for n in sorted(board, key=str.lower) if n in board]
            k.ensure_status_page(sp["slug"], sp.get("title", "Fleet"),
                                 [{"name": sp.get("group", "Services"), "monitorList": members}])
            print(f"  status page /status/{sp['slug']} -> {len(members)} monitors")
    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()