""" CluPilot ⇄ Uptime Kuma bridge. Uptime Kuma has no official REST API for creating or deleting monitors — all write operations go through its Socket.IO interface. This tiny service exposes exactly the REST contract that CluPilot's HttpMonitoringClient already speaks, and translates it to Socket.IO via the uptime-kuma-api library: GET /monitors -> {"monitors": [{"id", "url", "friendly_name"}]} POST /monitors <- {"friendly_name", "url", "type"} -> {"monitor": {"id": }} GET /monitors/{id} -> {"monitor": {"id", "status": "up"|"down"|...}} DELETE /monitors/{id} -> {"ok": true} GET /health -> liveness: 200 while the bridge serves (no auth) GET /ready -> readiness: 503 when Kuma is unreachable (no auth) All monitor routes require `Authorization: Bearer $BRIDGE_TOKEN`. Kuma's Socket.IO API is internal and may change between Kuma releases — pin the Kuma version you tested against. """ import os import threading import urllib.error import urllib.request from typing import Any from fastapi import Depends, FastAPI, Header, HTTPException, Response from pydantic import BaseModel from uptime_kuma_api import MonitorType, UptimeKumaApi, UptimeKumaException KUMA_URL = os.environ.get("KUMA_URL", "").rstrip("/") KUMA_USERNAME = os.environ.get("KUMA_USERNAME", "") KUMA_PASSWORD = os.environ.get("KUMA_PASSWORD", "") KUMA_TOTP = os.environ.get("KUMA_TOTP", "") # optional, if 2FA is enabled BRIDGE_TOKEN = os.environ.get("BRIDGE_TOKEN", "") app = FastAPI(title="CluPilot Uptime Kuma bridge", docs_url=None, redoc_url=None) # One shared, lazily-created Kuma session. Socket.IO connections are stateful, # so guard it with a lock and re-establish it if a call fails. _lock = threading.Lock() _api: UptimeKumaApi | None = None def _connect() -> UptimeKumaApi: global _api if not KUMA_URL: raise HTTPException(status_code=503, detail="KUMA_URL is not configured") if _api is None: api = UptimeKumaApi(KUMA_URL, timeout=15) try: if KUMA_TOTP: api.login(KUMA_USERNAME, KUMA_PASSWORD, KUMA_TOTP) else: api.login(KUMA_USERNAME, KUMA_PASSWORD) except Exception as exc: # noqa: BLE001 — surface any connect/login problem as 503 try: api.disconnect() except Exception: # noqa: BLE001, S110 pass # Distinguish "cannot reach" from "rejected us" — very different fixes. kind = "login rejected" if "username or password" in str(exc).lower() else "unreachable" raise HTTPException(status_code=503, detail=f"Kuma {kind}: {exc}") from exc _api = api return _api def _reset() -> None: """Drop the session so the next call reconnects (Kuma restart, token expiry…).""" global _api if _api is not None: try: _api.disconnect() except Exception: # noqa: BLE001, S110 pass _api = None def _call(fn, *args, **kwargs) -> Any: """ Run a Kuma call under the lock, retrying once on a stale connection. Any Kuma-side failure becomes a 502/503 — never an unhandled 500, so the caller always gets a meaningful status it can act on. """ with _lock: try: return fn(_connect(), *args, **kwargs) except HTTPException: raise except Exception as first: # noqa: BLE001 — includes UptimeKumaException, OSError, socket errors _reset() try: return fn(_connect(), *args, **kwargs) except HTTPException: raise except Exception as exc: # noqa: BLE001 raise HTTPException(status_code=502, detail=f"Kuma call failed: {exc}") from first def require_token(authorization: str = Header(default="")) -> None: if not BRIDGE_TOKEN: raise HTTPException(status_code=503, detail="BRIDGE_TOKEN is not configured") if authorization != f"Bearer {BRIDGE_TOKEN}": raise HTTPException(status_code=401, detail="Invalid bridge token") class MonitorIn(BaseModel): url: str friendly_name: str | None = None type: str | None = "https" # CluPilot sends "https"; Kuma type is "http" interval: int | None = 60 retries: int | None = 1 # mapped to Kuma's `maxretries` def _shape(monitor: dict) -> dict: """Kuma monitor -> the field names CluPilot's client expects.""" return { "id": monitor.get("id"), "url": monitor.get("url"), "friendly_name": monitor.get("name"), "active": monitor.get("active", True), } def _require_live(api: UptimeKumaApi) -> UptimeKumaApi: """ Guard against a STALE session. `get_monitors()` reads the library's cached monitorList event — it happily returns data while Kuma is unreachable, so any read that callers act on must first prove the socket is actually connected. """ if not getattr(api, "sio", None) or not api.sio.connected: _reset() raise HTTPException(status_code=503, detail="Kuma socket not connected") return api def _kuma_state() -> tuple[str, str | None]: """ (state, detail) for Kuma reachability. Never raises. Does a REAL round-trip: an HTTP hit on Kuma plus a live socket, because the cached monitor list would report "up" long after Kuma died. """ if not KUMA_URL: return "unconfigured", "KUMA_URL is not set" try: urllib.request.urlopen(KUMA_URL, timeout=5) # noqa: S310 — operator-configured URL except urllib.error.HTTPError: pass # any HTTP status means Kuma answered except Exception as exc: # noqa: BLE001 — connection refused/DNS/timeout return "down", f"HTTP unreachable: {exc}" # An authenticated, connected session is the real proof: _connect() performs # the login round-trip, so bad credentials / missing 2FA / an incompatible # login API all surface here as "down" rather than a false "up". try: with _lock: api = _connect() if not getattr(api, "sio", None) or not api.sio.connected: _reset() return "down", "Kuma socket not connected" return "up", None except HTTPException as exc: return "down", str(exc.detail) except Exception as exc: # noqa: BLE001 — must never 500 return "down", str(exc) @app.get("/health") def health() -> dict: """ LIVENESS (unauthenticated): is this process serving? Always 200 while it is. Deliberately NOT tied to Kuma: the bridge being restarted because a *dependency* is down would be wrong. Kuma's state is reported in the body, and /ready is the endpoint that fails when Kuma is unreachable. """ state, detail = _kuma_state() body = {"bridge": "up", "kuma": state} if detail: body["detail"] = detail return body @app.get("/ready") def ready(response: Response) -> dict: """ READINESS (unauthenticated): 200 only when Kuma is actually reachable, 503 otherwise — so uptime checks and operators can detect the outage. """ state, detail = _kuma_state() body = {"bridge": "up", "kuma": state} if detail: body["detail"] = detail if state != "up": response.status_code = 503 return body @app.get("/monitors", dependencies=[Depends(require_token)]) def list_monitors() -> dict: # _require_live first: the list is cached and would otherwise be served # (silently stale) while Kuma is unreachable. monitors = _call(lambda api: _require_live(api).get_monitors()) return {"monitors": [_shape(m) for m in monitors]} @app.post("/monitors", dependencies=[Depends(require_token)]) def create_monitor(payload: MonitorIn) -> dict: name = payload.friendly_name or payload.url # Idempotent: reuse an existing monitor for this exact URL. CluPilot also # guards this, but a crash between its POST and its bookkeeping must not # leave duplicates behind. # # The list is CACHED, so a match is confirmed with a LIVE get_monitor before # we hand the id back — otherwise a stale cache could make us return an id # for a monitor that no longer exists (CluPilot would then record a target # that is never checked). existing = next( (m for m in _call(lambda api: _require_live(api).get_monitors()) if m.get("url") == payload.url), None, ) if existing is not None and existing.get("id") is not None: try: _call(lambda api: api.get_monitor(int(existing["id"]))) return {"monitor": {"id": int(existing["id"])}} except HTTPException: pass # gone or unverifiable — fall through and create it created = _call( lambda api: api.add_monitor( type=MonitorType.HTTP, name=name, url=payload.url, interval=payload.interval or 60, maxretries=payload.retries or 1, # Kuma's own field name ) ) # uptime-kuma-api 1.2.x returns {"msg": ..., "monitorID": N} (verified against # Kuma 1.x). Accept a bare id too, so a future library change can't break us. if isinstance(created, dict): monitor_id = created.get("monitorID", created.get("id")) elif isinstance(created, int): monitor_id = created else: monitor_id = None if monitor_id is None: raise HTTPException(status_code=502, detail=f"Kuma returned no monitor id: {created}") return {"monitor": {"id": int(monitor_id)}} @app.get("/monitors/{monitor_id}", dependencies=[Depends(require_token)]) def get_monitor(monitor_id: int) -> dict: monitor = _call(lambda api: api.get_monitor(monitor_id)) # Kuma heartbeat status: 0 down, 1 up, 2 pending, 3 maintenance. status = "unknown" try: beats = _call(lambda api: api.get_monitor_beats(monitor_id, 1)) or [] if beats: status = {0: "down", 1: "up", 2: "pending", 3: "maintenance"}.get( beats[-1].get("status"), "unknown" ) except HTTPException: status = "unknown" # monitor exists but no heartbeat history yet shaped = _shape(monitor) shaped["status"] = status return {"monitor": shaped} @app.delete("/monitors/{monitor_id}", dependencies=[Depends(require_token)]) def delete_monitor(monitor_id: int) -> Response: _call(lambda api: api.delete_monitor(monitor_id)) return Response(status_code=200, content='{"ok":true}', media_type="application/json")