feat: Add Uptime Kuma version check via Socket.IO

Retrieve the running Uptime Kuma version by authenticating over
Socket.IO and reading the info event. This adds a new login_socket
dataset field, the fetch_socket_info helper, and the python-socketio
and websocket-client dependencies.

check_updates.py

@@ -12,6 +12,7 @@ from urllib.parse import urlencode, quote
 from urllib.request import HTTPCookieProcessor, Request, build_opener, urlopen
 
 import bcrypt
+import socketio
 
 import yaml
 
@@ -34,6 +35,7 @@ class ServiceConfig:
     upstream_latest_headers: Optional[Dict[str, str]]
     notes: Optional[str]
     login: Optional[Dict[str, Any]]
+    login_socket: Optional[Dict[str, Any]]
 
 
 def load_yaml(path: str) -> Dict[str, Any]:
@@ -109,6 +111,10 @@ def load_services(config: Dict[str, Any]) -> Dict[str, ServiceConfig]:
         if login is not None and not isinstance(login, dict):
             raise ValueError("login must be a mapping")
 
+        login_socket = entry.get("login_socket")
+        if login_socket is not None and not isinstance(login_socket, dict):
+            raise ValueError("login_socket must be a mapping")
+
         loaded[name] = ServiceConfig(
             name=name,
             base_url=base_url,
@@ -120,6 +126,7 @@ def load_services(config: Dict[str, Any]) -> Dict[str, ServiceConfig]:
             upstream_latest_headers=upstream_headers,
             notes=entry.get("notes"),
             login=login,
+            login_socket=login_socket,
         )
     return loaded
 
@@ -414,6 +421,53 @@ def authenticate_service(service: ServiceConfig, timeout: float, user_agent: str
     return "; ".join(cookies)
 
 
+def fetch_socket_info(service: ServiceConfig, timeout: float) -> Optional[str]:
+    if not service.login_socket:
+        return None
+    base_url = service.login_socket.get("url")
+    username_env = service.login_socket.get("username_env")
+    password_env = service.login_socket.get("password_env")
+    if not base_url or not username_env or not password_env:
+        raise ValueError(f"Login socket config incomplete for {service.name}")
+
+    username = os.getenv(username_env)
+    password = os.getenv(password_env)
+    if not username or not password:
+        raise ValueError(f"Missing credentials for {service.name}")
+
+    info_payload: Dict[str, Any] = {}
+    sio = socketio.Client(reconnection=False, request_timeout=timeout)
+
+    @sio.on("info")
+    def on_info(data):
+        nonlocal info_payload
+        info_payload = data or {}
+
+    sio.connect(base_url, transports=["websocket"], wait_timeout=timeout)
+
+    response = sio.call(
+        "login",
+        {
+            "username": username,
+            "password": password,
+            "token": "",
+        },
+        timeout=timeout,
+    )
+    if not response or not response.get("ok"):
+        sio.disconnect()
+        raise ValueError(f"Socket login failed for {service.name}")
+
+    if not info_payload.get("version"):
+        try:
+            info_payload = sio.call("info", timeout=timeout) or info_payload
+        except Exception:
+            pass
+
+    sio.disconnect()
+    return info_payload.get("version")
+
+
 def check_service(service: ServiceConfig, timeout: float, user_agent: str) -> Dict[str, Any]:
     result: Dict[str, Any] = {
         "name": service.name,
@@ -444,6 +498,11 @@ def check_service(service: ServiceConfig, timeout: float, user_agent: str) -> Di
             )
         except Exception as exc:
             result["current_error"] = str(exc)
+    elif service.login_socket:
+        try:
+            result["current"] = fetch_socket_info(service, timeout)
+        except Exception as exc:
+            result["current_error"] = str(exc)
 
     if service.upstream_latest_version_url and service.upstream_latest_extract:
         headers = build_headers(service.upstream_latest_headers)