8779569587
All checks were successful
Build and Push Container / build (push) Successful in 30s
The login method was returning success even when no token was received, which happens when 2FA is required. Now properly detects tokenRequired response and validates token before claiming success. 🤖 Generated with [Claude Code](https://claude.com/claude-code) Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>
286 lines
9.1 KiB
Python
286 lines
9.1 KiB
Python
import os
|
|
import json
|
|
from typing import Optional
|
|
from dataclasses import dataclass
|
|
from uptime_kuma_api import UptimeKumaApi, MonitorType
|
|
|
|
from config import get_config
|
|
|
|
|
|
# Map our monitor types to Uptime Kuma types
|
|
TYPE_MAP = {
|
|
"http": MonitorType.HTTP,
|
|
"tcp": MonitorType.PORT,
|
|
"ping": MonitorType.PING,
|
|
"docker": MonitorType.DOCKER,
|
|
"keyword": MonitorType.KEYWORD,
|
|
"push": MonitorType.PUSH,
|
|
}
|
|
|
|
# Token storage path
|
|
TOKEN_FILE = os.environ.get("KUMA_TOKEN_FILE", "/tmp/kuma_token.json")
|
|
|
|
|
|
@dataclass
|
|
class Monitor:
|
|
"""Uptime Kuma monitor configuration."""
|
|
|
|
type: str # http, tcp, ping, docker, keyword, push
|
|
name: str
|
|
url: Optional[str] = None # For HTTP monitors
|
|
hostname: Optional[str] = None # For TCP/Ping monitors
|
|
port: Optional[int] = None # For TCP monitors
|
|
interval: int = 60
|
|
keyword: Optional[str] = None # For keyword monitors
|
|
docker_container: Optional[str] = None # For Docker monitors
|
|
docker_host: Optional[str] = None # For Docker monitors
|
|
push_token: Optional[str] = None # For Push monitors
|
|
retries: int = 3
|
|
retry_interval: int = 60
|
|
max_redirects: int = 10
|
|
accepted_statuscodes: list[str] = None
|
|
|
|
def __post_init__(self):
|
|
if self.accepted_statuscodes is None:
|
|
self.accepted_statuscodes = ["200-299"]
|
|
|
|
|
|
def load_saved_token() -> Optional[str]:
|
|
"""Load saved token from file."""
|
|
try:
|
|
if os.path.exists(TOKEN_FILE):
|
|
with open(TOKEN_FILE, "r") as f:
|
|
data = json.load(f)
|
|
return data.get("token")
|
|
except Exception:
|
|
pass
|
|
return None
|
|
|
|
|
|
def save_token(token: str) -> None:
|
|
"""Save token to file."""
|
|
try:
|
|
with open(TOKEN_FILE, "w") as f:
|
|
json.dump({"token": token}, f)
|
|
except Exception as e:
|
|
print(f"Warning: Could not save token: {e}")
|
|
|
|
|
|
def clear_saved_token() -> None:
|
|
"""Clear saved token."""
|
|
try:
|
|
if os.path.exists(TOKEN_FILE):
|
|
os.remove(TOKEN_FILE)
|
|
except Exception:
|
|
pass
|
|
|
|
|
|
class UptimeKumaClient:
|
|
"""Client for Uptime Kuma using Socket.io API."""
|
|
|
|
def __init__(self):
|
|
config = get_config()
|
|
self.base_url = config.uptime_kuma_url.rstrip("/")
|
|
self.api_key = config.uptime_kuma_api_key # Can be empty if using saved token
|
|
self._api = None
|
|
self._token = None
|
|
|
|
def _get_token(self) -> Optional[str]:
|
|
"""Get token from config or saved file."""
|
|
if self._token:
|
|
return self._token
|
|
if self.api_key:
|
|
return self.api_key
|
|
return load_saved_token()
|
|
|
|
def _get_api(self) -> UptimeKumaApi:
|
|
"""Get connected API instance."""
|
|
if self._api is None:
|
|
token = self._get_token()
|
|
if not token:
|
|
raise Exception("Not authenticated. Please login to Uptime Kuma first.")
|
|
|
|
self._api = UptimeKumaApi(self.base_url)
|
|
try:
|
|
self._api.login_by_token(token)
|
|
except Exception as e:
|
|
self._api = None
|
|
clear_saved_token()
|
|
raise Exception(f"Authentication failed. Please login again: {e}")
|
|
return self._api
|
|
|
|
def _disconnect(self):
|
|
"""Disconnect API."""
|
|
if self._api:
|
|
try:
|
|
self._api.disconnect()
|
|
except Exception:
|
|
pass
|
|
self._api = None
|
|
|
|
def login(self, username: str, password: str, totp: Optional[str] = None) -> dict:
|
|
"""Login with username/password and optional TOTP."""
|
|
self._disconnect()
|
|
try:
|
|
api = UptimeKumaApi(self.base_url)
|
|
if totp:
|
|
result = api.login(username, password, totp)
|
|
else:
|
|
result = api.login(username, password)
|
|
|
|
# Check if 2FA is required
|
|
if result.get("tokenRequired"):
|
|
api.disconnect()
|
|
raise Exception("2FA token required. Please enter your TOTP code.")
|
|
|
|
token = result.get("token")
|
|
if not token:
|
|
api.disconnect()
|
|
raise Exception("Login failed: No authentication token received")
|
|
|
|
save_token(token)
|
|
self._token = token
|
|
|
|
api.disconnect()
|
|
return {"success": True, "message": "Login successful"}
|
|
except Exception as e:
|
|
error_msg = str(e)
|
|
# Re-raise with clear message for 2FA requirement
|
|
if "tokenRequired" in error_msg or "2fa" in error_msg.lower():
|
|
raise Exception("2FA token required. Please enter your TOTP code.")
|
|
raise Exception(f"Login failed: {error_msg}")
|
|
|
|
def is_authenticated(self) -> bool:
|
|
"""Check if we have valid authentication."""
|
|
token = self._get_token()
|
|
if not token:
|
|
return False
|
|
try:
|
|
api = UptimeKumaApi(self.base_url)
|
|
api.login_by_token(token)
|
|
api.disconnect()
|
|
return True
|
|
except Exception:
|
|
clear_saved_token()
|
|
return False
|
|
|
|
def logout(self) -> None:
|
|
"""Clear saved authentication."""
|
|
self._disconnect()
|
|
self._token = None
|
|
clear_saved_token()
|
|
|
|
def get_monitors(self) -> list[dict]:
|
|
"""Get all monitors."""
|
|
try:
|
|
api = self._get_api()
|
|
return api.get_monitors()
|
|
except Exception as e:
|
|
self._disconnect()
|
|
raise Exception(f"Failed to get monitors: {str(e)}")
|
|
|
|
def get_monitor(self, monitor_id: int) -> dict:
|
|
"""Get a specific monitor."""
|
|
try:
|
|
api = self._get_api()
|
|
return api.get_monitor(monitor_id)
|
|
except Exception as e:
|
|
self._disconnect()
|
|
raise Exception(f"Failed to get monitor {monitor_id}: {str(e)}")
|
|
|
|
def create_monitor(self, monitor: Monitor) -> dict:
|
|
"""Create a new monitor."""
|
|
try:
|
|
api = self._get_api()
|
|
|
|
monitor_type = TYPE_MAP.get(monitor.type)
|
|
if not monitor_type:
|
|
raise ValueError(f"Unknown monitor type: {monitor.type}")
|
|
|
|
kwargs = {
|
|
"type": monitor_type,
|
|
"name": monitor.name,
|
|
"interval": monitor.interval,
|
|
"maxretries": monitor.retries,
|
|
"retryInterval": monitor.retry_interval,
|
|
}
|
|
|
|
# Add type-specific fields
|
|
if monitor.type == "http":
|
|
kwargs["url"] = monitor.url
|
|
kwargs["maxredirects"] = monitor.max_redirects
|
|
kwargs["accepted_statuscodes"] = monitor.accepted_statuscodes
|
|
elif monitor.type == "keyword":
|
|
kwargs["url"] = monitor.url
|
|
kwargs["keyword"] = monitor.keyword
|
|
elif monitor.type == "tcp":
|
|
kwargs["hostname"] = monitor.hostname
|
|
kwargs["port"] = monitor.port
|
|
elif monitor.type == "ping":
|
|
kwargs["hostname"] = monitor.hostname
|
|
elif monitor.type == "docker":
|
|
kwargs["docker_container"] = monitor.docker_container
|
|
if monitor.docker_host:
|
|
kwargs["docker_host"] = monitor.docker_host
|
|
elif monitor.type == "push":
|
|
# Push monitors are created and get a token back
|
|
pass
|
|
|
|
result = api.add_monitor(**kwargs)
|
|
return result
|
|
except Exception as e:
|
|
self._disconnect()
|
|
raise Exception(f"Failed to create monitor: {str(e)}")
|
|
|
|
def delete_monitor(self, monitor_id: int) -> dict:
|
|
"""Delete a monitor."""
|
|
try:
|
|
api = self._get_api()
|
|
result = api.delete_monitor(monitor_id)
|
|
return result
|
|
except Exception as e:
|
|
self._disconnect()
|
|
raise Exception(f"Failed to delete monitor {monitor_id}: {str(e)}")
|
|
|
|
def pause_monitor(self, monitor_id: int) -> dict:
|
|
"""Pause a monitor."""
|
|
try:
|
|
api = self._get_api()
|
|
result = api.pause_monitor(monitor_id)
|
|
return result
|
|
except Exception as e:
|
|
self._disconnect()
|
|
raise Exception(f"Failed to pause monitor {monitor_id}: {str(e)}")
|
|
|
|
def resume_monitor(self, monitor_id: int) -> dict:
|
|
"""Resume a paused monitor."""
|
|
try:
|
|
api = self._get_api()
|
|
result = api.resume_monitor(monitor_id)
|
|
return result
|
|
except Exception as e:
|
|
self._disconnect()
|
|
raise Exception(f"Failed to resume monitor {monitor_id}: {str(e)}")
|
|
|
|
def test_connection(self) -> bool:
|
|
"""Test connection to Uptime Kuma."""
|
|
try:
|
|
api = self._get_api()
|
|
api.get_monitors()
|
|
return True
|
|
except Exception:
|
|
self._disconnect()
|
|
return False
|
|
|
|
|
|
# Global client instance
|
|
_kuma_client: Optional[UptimeKumaClient] = None
|
|
|
|
|
|
def get_kuma_client() -> UptimeKumaClient:
|
|
"""Get the global Uptime Kuma client instance."""
|
|
global _kuma_client
|
|
if _kuma_client is None:
|
|
_kuma_client = UptimeKumaClient()
|
|
return _kuma_client
|