CluPilotCloud/docker/kuma-bridge/app.py

"""
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": <int>}}
    GET    /monitors/{id}     -> {"monitor": {"id", "status": "up"|"down"|...}}
    DELETE /monitors/{id}     -> {"ok": true}
    GET    /health            -> bridge + Kuma reachability (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
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
            raise HTTPException(status_code=503, detail=f"Kuma unreachable: {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),
    }


@app.get("/health")
def health() -> dict:
    """Unauthenticated: is the bridge up, and can it reach Kuma?"""
    if not KUMA_URL:
        return {"bridge": "up", "kuma": "unconfigured"}
    try:
        _call(lambda api: api.get_monitors())
        return {"bridge": "up", "kuma": "up"}
    except HTTPException as exc:
        return {"bridge": "up", "kuma": "down", "detail": str(exc.detail)}
    except Exception as exc:  # noqa: BLE001 — health must never 500
        return {"bridge": "up", "kuma": "down", "detail": str(exc)}


@app.get("/monitors", dependencies=[Depends(require_token)])
def list_monitors() -> dict:
    monitors = _call(lambda api: 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.
    existing = next(
        (m for m in _call(lambda api: api.get_monitors()) if m.get("url") == payload.url),
        None,
    )
    if existing is not None:
        return {"monitor": {"id": existing.get("id")}}

    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
        )
    )
    monitor_id = created.get("monitorID") if isinstance(created, dict) else None
    if monitor_id is None:
        raise HTTPException(status_code=502, detail=f"Kuma returned no monitor id: {created}")

    return {"monitor": {"id": 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")