1347 lines
57 KiB
Python
1347 lines
57 KiB
Python
"""Homelab Butler v2.1 – Unified API proxy for Pfannkuchen homelab.
|
||
Reads service config from butler.yaml, credentials from Vaultwarden cache with flat-file fallback."""
|
||
|
||
import os, json, asyncio, logging, time, base64, re, subprocess, ipaddress
|
||
from datetime import datetime, timezone
|
||
import httpx, yaml
|
||
from typing import Literal
|
||
from pydantic import BaseModel
|
||
from fastapi import FastAPI, Request, HTTPException, Depends, Query
|
||
from fastapi.responses import JSONResponse, RedirectResponse
|
||
from contextlib import asynccontextmanager
|
||
|
||
log = logging.getLogger("butler")
|
||
VERSION = "2.3.2"
|
||
|
||
API_DIR = os.environ.get("API_KEY_DIR", "/data/api")
|
||
VAULT_CACHE_DIR = os.environ.get("VAULT_CACHE_DIR", "/data/vault-cache")
|
||
BUTLER_TOKEN = os.environ.get("BUTLER_TOKEN", "")
|
||
CONFIG_PATH = os.environ.get("BUTLER_CONFIG", "/data/butler.yaml")
|
||
|
||
# --- Config loading ---
|
||
|
||
_config: dict = {}
|
||
|
||
def _load_config():
|
||
global _config, SERVICES, VM_CFG, TTS_CFG
|
||
try:
|
||
with open(CONFIG_PATH) as f:
|
||
_config = yaml.safe_load(f) or {}
|
||
SERVICES = _config.get("services", {})
|
||
VM_CFG = _config.get("vm", {})
|
||
TTS_CFG = _config.get("tts", {})
|
||
log.info(f"Loaded config: {len(SERVICES)} services")
|
||
except FileNotFoundError:
|
||
log.warning(f"No config at {CONFIG_PATH}, using defaults")
|
||
SERVICES = {}
|
||
VM_CFG = {}
|
||
TTS_CFG = {}
|
||
|
||
SERVICES: dict = {}
|
||
VM_CFG: dict = {}
|
||
TTS_CFG: dict = {}
|
||
_load_config()
|
||
|
||
# --- Audit log ---
|
||
|
||
_audit_log: list[dict] = []
|
||
MAX_AUDIT = 500
|
||
|
||
def _audit(endpoint: str, method: str, status: int, detail: str = "", dry_run: bool = False):
|
||
entry = {
|
||
"ts": datetime.now(timezone.utc).isoformat(),
|
||
"endpoint": endpoint,
|
||
"method": method,
|
||
"status": status,
|
||
"detail": detail[:200],
|
||
"dry_run": dry_run,
|
||
}
|
||
_audit_log.append(entry)
|
||
if len(_audit_log) > MAX_AUDIT:
|
||
_audit_log.pop(0)
|
||
|
||
API_DIR = os.environ.get("API_KEY_DIR", "/data/api")
|
||
VAULT_CACHE_DIR = os.environ.get("VAULT_CACHE_DIR", "/data/vault-cache")
|
||
BUTLER_TOKEN = os.environ.get("BUTLER_TOKEN", "")
|
||
|
||
# --- Credential cache ---
|
||
|
||
_vault_cache: dict[str, str] = {}
|
||
|
||
def _load_vault_cache():
|
||
"""Load vault items from disk cache (written by host-side vault-sync.sh)."""
|
||
global _vault_cache
|
||
if not os.path.isdir(VAULT_CACHE_DIR):
|
||
log.info(f"No vault cache at {VAULT_CACHE_DIR}")
|
||
return
|
||
new = {}
|
||
for f in os.listdir(VAULT_CACHE_DIR):
|
||
path = os.path.join(VAULT_CACHE_DIR, f)
|
||
if os.path.isfile(path):
|
||
new[f] = open(path).read().strip()
|
||
_vault_cache = new
|
||
log.info(f"Loaded {len(new)} vault items from cache")
|
||
|
||
async def _periodic_cache_reload():
|
||
"""Reload vault cache every 5 minutes (host cron writes new files)."""
|
||
while True:
|
||
await asyncio.sleep(300)
|
||
_load_vault_cache()
|
||
|
||
@asynccontextmanager
|
||
async def lifespan(app: FastAPI):
|
||
_load_config()
|
||
_load_vault_cache()
|
||
task = asyncio.create_task(_periodic_cache_reload())
|
||
yield
|
||
task.cancel()
|
||
|
||
app = FastAPI(title="Homelab Butler", version=VERSION, lifespan=lifespan,
|
||
description="Unified API proxy + infrastructure management. AI agents: see GET / for self-onboarding.")
|
||
|
||
|
||
OverviewState = Literal["healthy", "warning", "critical"]
|
||
|
||
|
||
class OverviewCounts(BaseModel):
|
||
critical: int
|
||
warning: int
|
||
healthy: int
|
||
|
||
|
||
class OverviewFinding(BaseModel):
|
||
severity: OverviewState
|
||
code: str
|
||
target: str
|
||
message: str
|
||
age_hours: float | None = None
|
||
pct: int | None = None
|
||
|
||
|
||
class OverviewModelContract(BaseModel):
|
||
instruction: str
|
||
severity_order: list[OverviewState]
|
||
|
||
|
||
class OverviewResponse(BaseModel):
|
||
schema_version: Literal[1]
|
||
generated: datetime
|
||
overall_state: OverviewState
|
||
action_required: bool
|
||
summary: OverviewCounts
|
||
components: dict[str, OverviewCounts]
|
||
findings: list[OverviewFinding]
|
||
model_contract: OverviewModelContract
|
||
details: dict | None = None
|
||
|
||
|
||
# --- Credential reading (vault-first, file-fallback) ---
|
||
|
||
def _read(name):
|
||
"""Read credential: vault cache first, then flat file."""
|
||
# Vault cache uses lowercase-hyphenated names
|
||
vault_name = name.lower().replace("_", "-")
|
||
if vault_name in _vault_cache:
|
||
return _vault_cache[vault_name]
|
||
# Try uppercase convention
|
||
upper = name.upper().replace("-", "_").lower().replace("_", "-")
|
||
if upper in _vault_cache:
|
||
return _vault_cache[upper]
|
||
# Fallback to flat file
|
||
try:
|
||
return open(f"{API_DIR}/{name}").read().strip()
|
||
except FileNotFoundError:
|
||
return None
|
||
|
||
def _parse_kv(name):
|
||
raw = _read(name)
|
||
if not raw:
|
||
return {}
|
||
d = {}
|
||
for line in raw.splitlines():
|
||
if ":" in line:
|
||
k, v = line.split(":", 1)
|
||
d[k.strip().lower()] = v.strip()
|
||
return d
|
||
|
||
def _parse_url_key(name):
|
||
raw = _read(name)
|
||
if not raw:
|
||
return None, None
|
||
lines = [l.strip() for l in raw.splitlines() if l.strip()]
|
||
return (lines[0] if lines else None, lines[1] if len(lines) > 1 else None)
|
||
|
||
# --- Service configs loaded from butler.yaml ---
|
||
|
||
# --- Dockhand session ---
|
||
|
||
_dockhand_cookie = None
|
||
|
||
async def _dockhand_login(client):
|
||
global _dockhand_cookie
|
||
r = await client.post(
|
||
f"{SERVICES['dockhand']['url']}/api/auth/login",
|
||
json={"username": "admin", "password": _read("dockhand") or ""},
|
||
)
|
||
if r.status_code == 200:
|
||
_dockhand_cookie = dict(r.cookies)
|
||
return _dockhand_cookie
|
||
|
||
# --- Auth ---
|
||
|
||
def _verify(request: Request):
|
||
if not BUTLER_TOKEN:
|
||
return
|
||
auth = request.headers.get("authorization", "")
|
||
if auth != f"Bearer {BUTLER_TOKEN}":
|
||
raise HTTPException(401, "Invalid token")
|
||
|
||
def _get_key(cfg):
|
||
vault_key = cfg.get("vault_key")
|
||
if vault_key and vault_key in _vault_cache:
|
||
return _vault_cache[vault_key]
|
||
return _read(cfg.get("key_file", ""))
|
||
|
||
|
||
SENSITIVE_RESPONSE_FIELDS = {
|
||
"hawsertoken", "webhooksecret", "accesstoken", "refreshtoken",
|
||
"password", "secret", "apikey", "api_key", "privatekey",
|
||
}
|
||
|
||
|
||
def _redact_response(value, extra_fields=None):
|
||
"""Recursively redact known secret fields in proxied JSON responses."""
|
||
sensitive = set(SENSITIVE_RESPONSE_FIELDS)
|
||
sensitive.update(str(x).lower() for x in (extra_fields or []))
|
||
if isinstance(value, dict):
|
||
return {
|
||
key: "[REDACTED]" if str(key).lower() in sensitive
|
||
else _redact_response(item, sensitive)
|
||
for key, item in value.items()
|
||
}
|
||
if isinstance(value, list):
|
||
return [_redact_response(item, sensitive) for item in value]
|
||
return value
|
||
|
||
|
||
def _inventory_hosts(text: str) -> list[dict]:
|
||
"""Parse Ansible inventory host lines and apply Pfannkuchen SSH defaults."""
|
||
hosts = []
|
||
seen = set()
|
||
for raw in text.splitlines():
|
||
line = raw.strip()
|
||
if not line or line.startswith(("#", "[")) or "ansible_host=" not in line:
|
||
continue
|
||
parts = line.split()
|
||
name = parts[0]
|
||
attrs = {k: v for k, v in (p.split("=", 1) for p in parts[1:] if "=" in p)}
|
||
ip = attrs.get("ansible_host")
|
||
if not ip or name in seen:
|
||
continue
|
||
user = attrs.get("ansible_user")
|
||
if not user:
|
||
if name.startswith("node"):
|
||
user = "root"
|
||
elif ip.startswith("10.7.1."):
|
||
user = "chris"
|
||
else:
|
||
user = "sascha"
|
||
hosts.append({"name": name, "ip": ip, "user": user})
|
||
seen.add(name)
|
||
return hosts
|
||
|
||
# --- Routes ---
|
||
|
||
@app.get("/")
|
||
async def root():
|
||
"""AI self-onboarding: returns all available endpoints and services."""
|
||
svc_list = {}
|
||
for name, cfg in SERVICES.items():
|
||
svc_list[name] = {"url": cfg.get("url", ""), "auth": cfg.get("auth", ""), "description": cfg.get("description", "")}
|
||
return {
|
||
"service": "homelab-butler", "version": VERSION,
|
||
"docs": "/docs",
|
||
"openapi": "/openapi.json",
|
||
"services": svc_list,
|
||
"endpoints": {
|
||
"proxy": "GET/POST/PUT/DELETE /{service}/{path} - proxy to backend with auto-auth",
|
||
"vm_list": "GET /vm/list",
|
||
"vm_create": "POST /vm/create {node, ip, hostname, cores?, memory?, disk?}",
|
||
"vm_status": "GET /vm/status/{vmid}",
|
||
"vm_delete": "DELETE /vm/{vmid} - simple Proxmox delete (legacy)",
|
||
"vm_destroy": "DELETE /vm/destroy/{vmid}?dry_run=false - complete cleanup (VM, Dockhand, Repo, Ansible, Kuma)",
|
||
"inventory_add": "POST /inventory/host {name, ip, group?}",
|
||
"ansible_run": "POST /ansible/run {hostname}",
|
||
"tts_speak": "POST /tts/speak {text, target: speaker|telegram}",
|
||
"tts_voices": "GET /tts/voices",
|
||
"tts_health": "GET /tts/health",
|
||
"status": "GET /status - health of all backends",
|
||
"overview": "GET /overview?details=false - deterministic homelab verdict for small models",
|
||
"audit": "GET /audit - recent API calls",
|
||
},
|
||
"vault_items": len(_vault_cache),
|
||
}
|
||
|
||
@app.get("/health")
|
||
async def health():
|
||
return {"status": "ok", "vault_items": len(_vault_cache), "services": len(SERVICES), "version": VERSION}
|
||
|
||
def _classify_http_status(status_code: int, expected: set[int]) -> str:
|
||
"""Return a deterministic service state suitable for small models."""
|
||
if status_code in expected:
|
||
return "healthy"
|
||
if status_code in (401, 403):
|
||
return "auth_failed"
|
||
if status_code == 404:
|
||
return "misconfigured"
|
||
return "degraded"
|
||
|
||
|
||
def _service_auth(cfg: dict) -> dict:
|
||
"""Build secret-bearing request data without ever returning it from an endpoint."""
|
||
auth_type = cfg.get("auth", "none")
|
||
headers = {}
|
||
cookies = {}
|
||
base_url = cfg.get("url")
|
||
if auth_type == "apikey":
|
||
headers["X-Api-Key"] = _get_key(cfg) or ""
|
||
elif auth_type == "apikey_urlfile":
|
||
base_url, key = _parse_url_key(cfg.get("key_file", ""))
|
||
headers["X-Api-Key"] = key or ""
|
||
elif auth_type == "bearer":
|
||
headers["Authorization"] = f"Bearer {_get_key(cfg) or ''}"
|
||
elif auth_type == "n8n":
|
||
headers["X-N8N-API-KEY"] = _get_key(cfg) or ""
|
||
elif auth_type == "proxmox":
|
||
pv = _parse_kv("proxmox")
|
||
headers["Authorization"] = f"PVEAPIToken={pv.get('tokenid', '')}={pv.get('secret', '')}"
|
||
return {"base_url": base_url, "headers": headers, "cookies": cookies}
|
||
|
||
|
||
async def _collect_service_status() -> dict:
|
||
"""Run authenticated functional probes concurrently and classify their result."""
|
||
started = time.monotonic()
|
||
async with httpx.AsyncClient(verify=False, timeout=5, follow_redirects=True) as client:
|
||
async def probe(name: str, cfg: dict):
|
||
probe_started = time.monotonic()
|
||
try:
|
||
request_data = _service_auth(cfg)
|
||
base_url = request_data["base_url"]
|
||
if not base_url:
|
||
return name, {
|
||
"reachable": False, "status": "misconfigured", "message": "No service URL configured"
|
||
}
|
||
if cfg.get("auth") == "session":
|
||
request_data["cookies"] = await _dockhand_login(client) or {}
|
||
health_path = cfg.get("health_path", "")
|
||
target = f"{base_url.rstrip('/')}/{health_path.lstrip('/')}" if health_path else base_url
|
||
expected = {int(code) for code in cfg.get("health_expected", range(200, 400))}
|
||
response = await client.get(
|
||
target,
|
||
headers=request_data["headers"],
|
||
cookies=request_data["cookies"],
|
||
)
|
||
state = _classify_http_status(response.status_code, expected)
|
||
messages = {
|
||
"healthy": "Functional probe succeeded",
|
||
"auth_failed": "Configured credentials were rejected",
|
||
"misconfigured": "Configured health route was not found",
|
||
"degraded": "Backend returned an unexpected HTTP status",
|
||
}
|
||
return name, {
|
||
"reachable": True,
|
||
"status": state,
|
||
"http": response.status_code,
|
||
"latency_ms": round((time.monotonic() - probe_started) * 1000),
|
||
"message": messages[state],
|
||
}
|
||
except Exception as exc:
|
||
return name, {
|
||
"reachable": False,
|
||
"status": "offline",
|
||
"latency_ms": round((time.monotonic() - probe_started) * 1000),
|
||
"error": type(exc).__name__,
|
||
"message": "Backend could not be reached",
|
||
}
|
||
|
||
pairs = await asyncio.gather(*(probe(name, cfg) for name, cfg in SERVICES.items()))
|
||
results = dict(pairs)
|
||
results["_meta"] = {"duration_ms": round((time.monotonic() - started) * 1000)}
|
||
return results
|
||
|
||
|
||
@app.get("/status")
|
||
async def status(_=Depends(_verify)):
|
||
"""Authenticated functional health check for all configured backends."""
|
||
results = await _collect_service_status()
|
||
_audit("/status", "GET", 200)
|
||
return results
|
||
|
||
@app.get("/audit")
|
||
async def audit(_=Depends(_verify), limit: int = Query(50, le=MAX_AUDIT)):
|
||
"""Recent API calls (newest first)."""
|
||
return list(reversed(_audit_log[-limit:]))
|
||
|
||
@app.post("/config/reload")
|
||
async def config_reload(_=Depends(_verify)):
|
||
"""Reload butler.yaml and vault cache."""
|
||
_load_config()
|
||
_load_vault_cache()
|
||
return {"config_services": len(SERVICES), "vault_items": len(_vault_cache)}
|
||
|
||
|
||
@app.get("/info")
|
||
async def info(_=Depends(_verify)):
|
||
"""Secret-free machine-readable context for AI agents and operators."""
|
||
return {
|
||
"service": "homelab-butler",
|
||
"version": VERSION,
|
||
"generated": datetime.now(timezone.utc).isoformat(),
|
||
"services": {
|
||
name: {
|
||
"url": cfg.get("url"),
|
||
"auth": cfg.get("auth"),
|
||
"description": cfg.get("description", ""),
|
||
}
|
||
for name, cfg in SERVICES.items()
|
||
},
|
||
"endpoints": {
|
||
"status": "/status",
|
||
"overview": "/overview?details=false",
|
||
"audit": "/audit",
|
||
"host_health": "/health/all",
|
||
"backups": "/backup/status",
|
||
"disk": "/disk/usage",
|
||
"logs": "/logs/{host}/{container}?tail=200",
|
||
"inspect": "/docker/inspect/{host}/{container}",
|
||
"docs": "/docs",
|
||
},
|
||
"rules": [
|
||
"Backend services are accessed through Butler or dedicated MCP servers",
|
||
"VMs only; no LXC",
|
||
"Docker Compose is stored in Git; no docker run",
|
||
"Persistent volumes live under /app-config",
|
||
"Node 7 VM SSH user is chris",
|
||
],
|
||
}
|
||
|
||
|
||
def _get_inventory_hosts() -> list[dict]:
|
||
rc, out, _err = _ssh(
|
||
AUTOMATION1,
|
||
"python3 -c \"print(open('/app-config/ansible/pfannkuchen.ini').read())\"",
|
||
timeout=15,
|
||
)
|
||
return _inventory_hosts(out) if rc == 0 else []
|
||
|
||
|
||
def _find_inventory_host(name: str) -> dict | None:
|
||
return next((host for host in _get_inventory_hosts() if host["name"] == name), None)
|
||
|
||
|
||
async def _get_inventory_hosts_async() -> list[dict]:
|
||
return await asyncio.to_thread(_get_inventory_hosts)
|
||
|
||
|
||
async def _collect_health_all(concurrency: int = 10) -> dict:
|
||
"""Collect SSH and container health concurrently with bounded fan-out."""
|
||
semaphore = asyncio.Semaphore(concurrency)
|
||
|
||
async def inspect_host(host: dict):
|
||
async with semaphore:
|
||
rc, out, err = await asyncio.to_thread(
|
||
_ssh,
|
||
f'{host["user"]}@{host["ip"]}',
|
||
"echo __BUTLER_OK__; (sudo -n docker ps --format '{{.Names}}: {{.Status}}' 2>/dev/null || docker ps --format '{{.Names}}: {{.Status}}' 2>/dev/null) | head -30",
|
||
10,
|
||
)
|
||
lines = out.strip().splitlines()
|
||
return host["name"], {
|
||
"ip": host["ip"],
|
||
"user": host["user"],
|
||
"reachable": rc == 0 and bool(lines) and lines[0] == "__BUTLER_OK__",
|
||
"containers": lines[1:] if lines and lines[0] == "__BUTLER_OK__" else [],
|
||
"error": err.strip()[:200] if rc != 0 else None,
|
||
}
|
||
|
||
hosts = await _get_inventory_hosts_async()
|
||
pairs = await asyncio.gather(*(inspect_host(host) for host in hosts))
|
||
return dict(pairs)
|
||
|
||
|
||
@app.get("/health/all")
|
||
async def health_all(_=Depends(_verify)):
|
||
"""SSH reachability and Docker status for all inventory hosts."""
|
||
return await _collect_health_all()
|
||
|
||
|
||
def _parse_backup_time(value: str | None) -> datetime | None:
|
||
if not value:
|
||
return None
|
||
try:
|
||
parsed = datetime.fromisoformat(value.replace("Z", "+00:00"))
|
||
return parsed.replace(tzinfo=timezone.utc) if parsed.tzinfo is None else parsed.astimezone(timezone.utc)
|
||
except (TypeError, ValueError):
|
||
return None
|
||
|
||
|
||
def _backup_item(rc: int, out: str, err: str, now: datetime | None = None) -> dict:
|
||
"""Normalize borgmatic output into one small-model-friendly state object."""
|
||
item = {"state": "unknown", "ok": False, "last_backup": None, "age_hours": None}
|
||
if rc != 0 or not out.strip():
|
||
if err:
|
||
item["error"] = err.strip()[:200]
|
||
return item
|
||
try:
|
||
data = json.loads(out)
|
||
archives = data[0].get("archives", []) if isinstance(data, list) and data else []
|
||
if not archives:
|
||
item["error"] = "no archives returned"
|
||
return item
|
||
last = archives[-1]
|
||
started_at = _parse_backup_time(last.get("start"))
|
||
if not started_at:
|
||
item["error"] = "invalid backup timestamp"
|
||
return item
|
||
age_hours = max(0, ((now or datetime.now(timezone.utc)) - started_at).total_seconds() / 3600)
|
||
state = "healthy" if age_hours <= 30 else "warning" if age_hours <= 48 else "critical"
|
||
return {
|
||
"state": state,
|
||
"ok": state == "healthy",
|
||
"last_backup": last.get("start"),
|
||
"age_hours": round(age_hours, 1),
|
||
"name": last.get("name"),
|
||
}
|
||
except (json.JSONDecodeError, TypeError, IndexError, KeyError):
|
||
item["error"] = "invalid borgmatic JSON"
|
||
return item
|
||
|
||
|
||
async def _collect_backup_status(concurrency: int = 10) -> dict:
|
||
"""Query VM backups concurrently; one slow host no longer blocks all others serially."""
|
||
semaphore = asyncio.Semaphore(concurrency)
|
||
hosts = [host for host in await _get_inventory_hosts_async() if not host["name"].startswith("node")]
|
||
|
||
async def inspect_backup(host: dict):
|
||
async with semaphore:
|
||
rc, out, err = await asyncio.to_thread(
|
||
_ssh,
|
||
f'{host["user"]}@{host["ip"]}',
|
||
"sudo -n borgmatic list --last 1 --json 2>/dev/null",
|
||
12,
|
||
)
|
||
return host["name"], _backup_item(rc, out, err)
|
||
|
||
pairs = await asyncio.gather(*(inspect_backup(host) for host in hosts))
|
||
results = dict(pairs)
|
||
summary = {"total": len(results), "healthy": 0, "warning": 0, "critical": 0, "unknown": 0}
|
||
for item in results.values():
|
||
summary[item["state"]] += 1
|
||
return {"summary": summary, "hosts": results}
|
||
|
||
|
||
@app.get("/backup/status")
|
||
async def backup_status(_=Depends(_verify)):
|
||
"""Latest Borgmatic archive, age and severity for all VM inventory hosts."""
|
||
return await _collect_backup_status()
|
||
|
||
|
||
async def _collect_disk_usage(concurrency: int = 10) -> dict:
|
||
semaphore = asyncio.Semaphore(concurrency)
|
||
|
||
async def inspect_disk(host: dict):
|
||
async with semaphore:
|
||
rc, out, _err = await asyncio.to_thread(
|
||
_ssh, f'{host["user"]}@{host["ip"]}', "df -P / | tail -1", 10
|
||
)
|
||
parts = out.split()
|
||
if rc != 0 or len(parts) < 6:
|
||
return host["name"], None
|
||
return host["name"], {
|
||
"size_kib": int(parts[1]), "used_kib": int(parts[2]),
|
||
"avail_kib": int(parts[3]), "pct": parts[4], "mount": parts[5],
|
||
}
|
||
|
||
hosts = await _get_inventory_hosts_async()
|
||
pairs = await asyncio.gather(*(inspect_disk(host) for host in hosts))
|
||
return {name: item for name, item in pairs if item is not None}
|
||
|
||
|
||
@app.get("/disk/usage")
|
||
async def disk_usage(_=Depends(_verify)):
|
||
"""Root filesystem usage for all reachable inventory hosts."""
|
||
return await _collect_disk_usage()
|
||
|
||
|
||
def _add_component(summary: dict, bucket: dict, state: str):
|
||
normalized = state if state in ("healthy", "warning", "critical") else "warning"
|
||
summary[normalized] += 1
|
||
bucket[normalized] += 1
|
||
|
||
|
||
@app.get("/overview", response_model=OverviewResponse, response_model_exclude_none=True)
|
||
async def overview(details: bool = Query(False), _=Depends(_verify)):
|
||
"""Compact deterministic homelab verdict designed for small language models."""
|
||
services, hosts, backups, disks = await asyncio.gather(
|
||
_collect_service_status(),
|
||
_collect_health_all(),
|
||
_collect_backup_status(),
|
||
_collect_disk_usage(),
|
||
)
|
||
summary = {"critical": 0, "warning": 0, "healthy": 0}
|
||
component_summary = {
|
||
name: {"critical": 0, "warning": 0, "healthy": 0}
|
||
for name in ("services", "hosts", "backups", "disks")
|
||
}
|
||
findings = []
|
||
|
||
for name in sorted(hosts):
|
||
item = hosts[name]
|
||
containers = item.get("containers", [])
|
||
bad_container = next(
|
||
(line for line in containers if "unhealthy" in line.lower() or "restarting" in line.lower()),
|
||
None,
|
||
)
|
||
if not item.get("reachable"):
|
||
state = "critical"
|
||
findings.append({
|
||
"severity": "critical", "code": "host_unreachable", "target": name,
|
||
"message": "Host is not reachable over SSH",
|
||
})
|
||
elif bad_container:
|
||
state = "critical"
|
||
findings.append({
|
||
"severity": "critical", "code": "container_unhealthy", "target": name,
|
||
"message": bad_container[:200],
|
||
})
|
||
else:
|
||
state = "healthy"
|
||
_add_component(summary, component_summary["hosts"], state)
|
||
|
||
service_states = {
|
||
"healthy": "healthy", "degraded": "warning", "offline": "critical",
|
||
"auth_failed": "critical", "misconfigured": "critical",
|
||
}
|
||
service_codes = {
|
||
"degraded": "service_degraded", "offline": "service_offline",
|
||
"auth_failed": "service_auth_failed", "misconfigured": "service_misconfigured",
|
||
}
|
||
for name in sorted(key for key in services if not key.startswith("_")):
|
||
item = services[name]
|
||
raw_state = item.get("status", "degraded")
|
||
state = service_states.get(raw_state, "warning")
|
||
_add_component(summary, component_summary["services"], state)
|
||
if state != "healthy":
|
||
findings.append({
|
||
"severity": state,
|
||
"code": service_codes.get(raw_state, "service_degraded"),
|
||
"target": name,
|
||
"message": item.get("message", "Service health probe failed"),
|
||
})
|
||
|
||
for name in sorted(backups.get("hosts", {})):
|
||
item = backups["hosts"][name]
|
||
raw_state = item.get("state", "unknown")
|
||
state = "healthy" if raw_state == "healthy" else "critical" if raw_state in ("critical", "unknown") else "warning"
|
||
_add_component(summary, component_summary["backups"], state)
|
||
if state != "healthy":
|
||
findings.append({
|
||
"severity": state,
|
||
"code": f"backup_{raw_state}",
|
||
"target": name,
|
||
"message": "Backup is missing, stale or could not be verified",
|
||
"age_hours": item.get("age_hours"),
|
||
})
|
||
|
||
for name in sorted(disks):
|
||
pct = int(str(disks[name].get("pct", "0")).rstrip("%") or 0)
|
||
state = "critical" if pct >= 90 else "warning" if pct >= 80 else "healthy"
|
||
_add_component(summary, component_summary["disks"], state)
|
||
if state != "healthy":
|
||
findings.append({
|
||
"severity": state,
|
||
"code": "disk_critical" if state == "critical" else "disk_high",
|
||
"target": name,
|
||
"message": f"Root filesystem usage is {pct}%",
|
||
"pct": pct,
|
||
})
|
||
|
||
overall_state = "critical" if summary["critical"] else "warning" if summary["warning"] else "healthy"
|
||
response = {
|
||
"schema_version": 1,
|
||
"generated": datetime.now(timezone.utc).isoformat(),
|
||
"overall_state": overall_state,
|
||
"action_required": overall_state != "healthy",
|
||
"summary": summary,
|
||
"components": component_summary,
|
||
"findings": findings,
|
||
"model_contract": {
|
||
"instruction": "Report overall_state, then findings in the returned order. Do not infer missing facts.",
|
||
"severity_order": ["critical", "warning", "healthy"],
|
||
},
|
||
}
|
||
if details:
|
||
response["details"] = {"services": services, "hosts": hosts, "backups": backups, "disks": disks}
|
||
_audit("/overview", "GET", 200, f"state={overall_state} findings={len(findings)}")
|
||
return response
|
||
|
||
|
||
@app.get("/logs/{host}/{container}")
|
||
async def docker_logs(host: str, container: str, tail: int = Query(200, ge=1, le=20000), _=Depends(_verify)):
|
||
"""Read Docker logs from an inventory host."""
|
||
if not re.fullmatch(r"[A-Za-z0-9_.-]+", host) or not re.fullmatch(r"[A-Za-z0-9_.-]+", container):
|
||
raise HTTPException(400, "Invalid host or container name")
|
||
target = _find_inventory_host(host)
|
||
if not target:
|
||
raise HTTPException(404, f"Host {host} not found")
|
||
rc, out, err = _ssh(
|
||
f'{target["user"]}@{target["ip"]}',
|
||
f"sudo -n docker logs {container} --tail {tail} 2>&1 || docker logs {container} --tail {tail} 2>&1",
|
||
timeout=30,
|
||
)
|
||
if rc != 0:
|
||
raise HTTPException(502, (err or out).strip()[:500])
|
||
return {"host": host, "container": container, "tail": tail, "output": out}
|
||
|
||
|
||
@app.get("/docker/inspect/{host}/{container}")
|
||
async def docker_inspect(host: str, container: str, _=Depends(_verify)):
|
||
"""Return a sanitized runtime/resource summary for a Docker container."""
|
||
if not re.fullmatch(r"[A-Za-z0-9_.-]+", host) or not re.fullmatch(r"[A-Za-z0-9_.-]+", container):
|
||
raise HTTPException(400, "Invalid host or container name")
|
||
target = _find_inventory_host(host)
|
||
if not target:
|
||
raise HTTPException(404, f"Host {host} not found")
|
||
rc, out, err = _ssh(
|
||
f'{target["user"]}@{target["ip"]}',
|
||
f"sudo -n docker inspect {container}",
|
||
timeout=20,
|
||
)
|
||
if rc != 0:
|
||
raise HTTPException(502, (err or out).strip()[:500])
|
||
try:
|
||
raw = json.loads(out)[0]
|
||
except (json.JSONDecodeError, IndexError, TypeError):
|
||
raise HTTPException(502, "Invalid docker inspect response")
|
||
host_cfg = raw.get("HostConfig", {})
|
||
cfg = raw.get("Config", {})
|
||
state = raw.get("State", {})
|
||
return {
|
||
"name": raw.get("Name", "").lstrip("/"),
|
||
"image": cfg.get("Image"),
|
||
"state": {
|
||
"status": state.get("Status"), "running": state.get("Running"),
|
||
"started_at": state.get("StartedAt"), "exit_code": state.get("ExitCode"),
|
||
"oom_killed": state.get("OOMKilled"), "restart_count": raw.get("RestartCount"),
|
||
},
|
||
"runtime": host_cfg.get("Runtime"),
|
||
"resources": {
|
||
"memory": host_cfg.get("Memory"), "memory_reservation": host_cfg.get("MemoryReservation"),
|
||
"nano_cpus": host_cfg.get("NanoCpus"), "device_requests": host_cfg.get("DeviceRequests"),
|
||
},
|
||
"restart_policy": host_cfg.get("RestartPolicy"),
|
||
"log_config": host_cfg.get("LogConfig"),
|
||
"environment_keys": sorted(item.split("=", 1)[0] for item in cfg.get("Env", []) if "=" in item),
|
||
"mounts": [
|
||
{"type": mount.get("Type"), "source": mount.get("Source"), "destination": mount.get("Destination"), "rw": mount.get("RW")}
|
||
for mount in raw.get("Mounts", [])
|
||
],
|
||
}
|
||
|
||
|
||
@app.post("/docker/restart/{host}/{container}")
|
||
async def docker_restart(host: str, container: str, _=Depends(_verify), dry_run: bool = Query(False)):
|
||
"""Restart a named Docker container, with optional dry-run."""
|
||
if not re.fullmatch(r"[A-Za-z0-9_.-]+", host) or not re.fullmatch(r"[A-Za-z0-9_.-]+", container):
|
||
raise HTTPException(400, "Invalid host or container name")
|
||
target = _find_inventory_host(host)
|
||
if not target:
|
||
raise HTTPException(404, f"Host {host} not found")
|
||
if dry_run:
|
||
return {"dry_run": True, "host": host, "container": container}
|
||
rc, out, err = _ssh(f'{target["user"]}@{target["ip"]}', f"sudo -n docker restart {container}", timeout=45)
|
||
_audit(f"/docker/restart/{host}/{container}", "POST", 200 if rc == 0 else 502)
|
||
if rc != 0:
|
||
raise HTTPException(502, (err or out).strip()[:500])
|
||
return {"success": True, "output": out.strip()}
|
||
|
||
|
||
@app.post("/vault/reload")
|
||
async def vault_reload(_=Depends(_verify)):
|
||
_load_vault_cache()
|
||
return {"reloaded": True, "items": len(_vault_cache)}
|
||
|
||
|
||
# --- VM Lifecycle Endpoints ---
|
||
import subprocess as _sp
|
||
|
||
AUTOMATION1 = VM_CFG.get("automation_host", "sascha@10.5.85.5") if VM_CFG else "sascha@10.5.85.5"
|
||
ISO_BUILDER = VM_CFG.get("iso_builder_path", "/app-config/ansible/iso-builder/build-iso.sh") if VM_CFG else "/app-config/ansible/iso-builder/build-iso.sh"
|
||
|
||
class VMCreate(BaseModel):
|
||
node: int
|
||
ip: str
|
||
hostname: str
|
||
cores: int = 2
|
||
memory: int = 4096
|
||
disk: int = 32
|
||
|
||
def _ssh(host, cmd, timeout=600):
|
||
try:
|
||
r = _sp.run(["ssh","-o","ConnectTimeout=10","-o","StrictHostKeyChecking=accept-new",
|
||
"-o","UserKnownHostsFile=/tmp/butler_known_hosts",host,cmd],
|
||
capture_output=True, text=True, timeout=timeout)
|
||
return r.returncode, r.stdout, r.stderr
|
||
except _sp.TimeoutExpired:
|
||
return 124, "", f"SSH command timed out after {timeout} seconds"
|
||
|
||
def _pve_auth():
|
||
pv = _parse_kv("proxmox")
|
||
return f"PVEAPIToken={pv.get('tokenid','')}={pv.get('secret','')}"
|
||
|
||
@app.get("/vm/list")
|
||
async def vm_list(_=Depends(_verify)):
|
||
auth = _pve_auth()
|
||
vms = []
|
||
async with httpx.AsyncClient(verify=False, timeout=15) as c:
|
||
nodes = await c.get("https://10.5.85.11:8006/api2/json/nodes", headers={"Authorization": auth})
|
||
for n in nodes.json().get("data", []):
|
||
r = await c.get(f"https://10.5.85.11:8006/api2/json/nodes/{n['node']}/qemu", headers={"Authorization": auth})
|
||
for vm in r.json().get("data", []):
|
||
vm["node"] = n["node"]
|
||
vms.append(vm)
|
||
return vms
|
||
|
||
@app.post("/vm/create")
|
||
async def vm_create(req: VMCreate, _=Depends(_verify), dry_run: bool = Query(False)):
|
||
if dry_run:
|
||
_audit("/vm/create", "POST", 200, f"dry_run: {req.hostname} {req.ip} node{req.node}", dry_run=True)
|
||
return {"dry_run": True, "would_create": {"hostname": req.hostname, "ip": req.ip, "node": req.node,
|
||
"cores": req.cores, "memory": req.memory, "disk": req.disk},
|
||
"steps": ["iso-builder", "wait ssh", "add inventory", "ansible setup"]}
|
||
steps = []
|
||
# Step 1: Build ISO + create VM via iso-builder on automation1
|
||
cmd = f"{ISO_BUILDER} --node {req.node} --ip {req.ip} --hostname {req.hostname} --cores {req.cores} --memory {req.memory} --disk {req.disk} --password '{VM_CFG.get('default_password', 'changeme')}' --create-vm"
|
||
rc, out, err = _ssh(AUTOMATION1, f"cd /app-config/ansible/iso-builder && {cmd}", timeout=300)
|
||
if rc != 0:
|
||
return JSONResponse({"error": "iso-builder failed", "stderr": err[-500:], "stdout": out[-500:]}, status_code=500)
|
||
steps.append("iso-builder: ok")
|
||
|
||
# Step 2: Wait for SSH (up to 6 min)
|
||
ok = False
|
||
for _ in range(36):
|
||
try:
|
||
rc2, out2, _ = _ssh(f"sascha@{req.ip}", "hostname", timeout=10)
|
||
if rc2 == 0:
|
||
ok = True
|
||
steps.append(f"ssh: {out2.strip()} reachable")
|
||
break
|
||
except Exception:
|
||
pass
|
||
await asyncio.sleep(10)
|
||
if not ok:
|
||
return JSONResponse({"error": "SSH timeout", "steps": steps}, status_code=504)
|
||
|
||
# Step 2.5: Add to Ansible inventory
|
||
ini = "/app-config/ansible/pfannkuchen.ini"
|
||
group = getattr(req, 'group', 'auto')
|
||
inv_cmd = f"""python3 -c "
|
||
lines = open('{ini}').readlines()
|
||
if not any('{req.hostname} ' in l for l in lines):
|
||
out = []
|
||
found = False
|
||
for l in lines:
|
||
out.append(l)
|
||
if l.strip() == '[auto]':
|
||
found = True
|
||
elif found and (l.startswith('[') or l.strip() == ''):
|
||
out.insert(-1, '{req.hostname} ansible_host={req.ip}\\n')
|
||
found = False
|
||
if found:
|
||
out.append('{req.hostname} ansible_host={req.ip}\\n')
|
||
open('{ini}','w').writelines(out)
|
||
print('added')
|
||
else:
|
||
print('exists')
|
||
" """
|
||
_ssh(AUTOMATION1, inv_cmd, timeout=30)
|
||
_ssh(AUTOMATION1, f"mkdir -p /app-config/ansible/host_vars/{req.hostname} && printf 'ansible_host: {req.ip}\\nansible_user: sascha\\n' > /app-config/ansible/host_vars/{req.hostname}/vars.yml", timeout=30)
|
||
_ssh(AUTOMATION1, f"ssh-keygen -f /home/sascha/.ssh/known_hosts -R {req.ip} 2>/dev/null; ssh -o StrictHostKeyChecking=accept-new sascha@{req.ip} hostname 2>/dev/null", timeout=30)
|
||
steps.append("inventory: added")
|
||
|
||
# Step 3: Ansible base setup via direct SSH (reliable fallback)
|
||
rc3, _, err3 = _ssh(AUTOMATION1, f"cd /app-config/ansible && bash pfannkuchen.sh setup {req.hostname}", timeout=600)
|
||
steps.append(f"ansible: {'ok' if rc3 == 0 else 'failed (rc=' + str(rc3) + ')'}")
|
||
|
||
_audit("/vm/create", "POST", 200 if rc3 == 0 else 500, f"{req.hostname} {req.ip}")
|
||
return {"status": "ok" if rc3 == 0 else "partial", "hostname": req.hostname, "ip": req.ip, "node": req.node, "steps": steps}
|
||
|
||
@app.get("/vm/status/{vmid}")
|
||
async def vm_status(vmid: int, _=Depends(_verify)):
|
||
auth = _pve_auth()
|
||
async with httpx.AsyncClient(verify=False, timeout=10) as c:
|
||
nodes = await c.get("https://10.5.85.11:8006/api2/json/nodes", headers={"Authorization": auth})
|
||
for n in nodes.json().get("data", []):
|
||
r = await c.get(f"https://10.5.85.11:8006/api2/json/nodes/{n['node']}/qemu/{vmid}/status/current", headers={"Authorization": auth})
|
||
if r.status_code == 200:
|
||
return r.json().get("data", {})
|
||
return JSONResponse({"error": "VM not found"}, status_code=404)
|
||
|
||
@app.delete("/vm/{vmid}")
|
||
async def vm_delete(vmid: int, _=Depends(_verify)):
|
||
"""Simple VM delete - Proxmox only (legacy)."""
|
||
auth = _pve_auth()
|
||
async with httpx.AsyncClient(verify=False, timeout=30) as c:
|
||
nodes = await c.get("https://10.5.85.11:8006/api2/json/nodes", headers={"Authorization": auth})
|
||
for n in nodes.json().get("data", []):
|
||
r = await c.delete(f"https://10.5.85.11:8006/api2/json/nodes/{n['node']}/qemu/{vmid}", headers={"Authorization": auth})
|
||
if r.status_code == 200:
|
||
return r.json()
|
||
return JSONResponse({"error": "VM not found"}, status_code=404)
|
||
|
||
@app.delete("/vm/destroy/{vmid}")
|
||
async def vm_destroy_full(vmid: int, _=Depends(_verify), dry_run: bool = Query(False)):
|
||
"""
|
||
Complete VM destruction with full cleanup:
|
||
1. Stop VM (required before destroy)
|
||
2. Destroy VM (Proxmox)
|
||
3. Remove from Dockhand (by IP)
|
||
4. Delete Forgejo repo (by hostname)
|
||
5. Remove from Ansible inventory
|
||
6. Remove from Uptime Kuma monitoring
|
||
|
||
Returns detailed cleanup report.
|
||
"""
|
||
auth = _pve_auth()
|
||
results = {"vmid": vmid, "dry_run": dry_run, "steps": {}}
|
||
|
||
async with httpx.AsyncClient(verify=False, timeout=30) as c:
|
||
# Step 1: Find VM and get details from vm/list (more reliable than config endpoint)
|
||
vm_info = None
|
||
node_name = None
|
||
vms = await c.get("https://10.5.85.11:8006/api2/json/nodes", headers={"Authorization": auth})
|
||
for n in vms.json().get("data", []):
|
||
r = await c.get(f"https://10.5.85.11:8006/api2/json/nodes/{n['node']}/qemu", headers={"Authorization": auth})
|
||
for vm in r.json().get("data", []):
|
||
if vm.get("vmid") == vmid:
|
||
vm_info = vm
|
||
node_name = n["node"]
|
||
break
|
||
if vm_info:
|
||
break
|
||
|
||
if not vm_info:
|
||
return JSONResponse({"error": f"VM {vmid} not found"}, status_code=404)
|
||
|
||
hostname = vm_info.get("name", "") or str(vmid)
|
||
# Extract IP from vm list or config
|
||
ip = vm_info.get("ip", "")
|
||
if not ip:
|
||
# Try to get from net0 config
|
||
cfg = await c.get(f"https://10.5.85.11:8006/api2/json/nodes/{node_name}/qemu/{vmid}/config", headers={"Authorization": auth})
|
||
net0 = cfg.json().get("data", {}).get("net0", "")
|
||
if "ip=" in net0:
|
||
ip = net0.split("ip=")[-1].split(",")[0]
|
||
results["vm_info"] = {"hostname": hostname, "ip": ip, "node": node_name, "status": vm_info.get("status", "unknown")}
|
||
|
||
# Step 2: Stop VM (if running)
|
||
if vm_info.get("status") == "running":
|
||
if dry_run:
|
||
results["steps"]["stop_vm"] = {"status": "dry_run", "message": f"Would stop VM {vmid}"}
|
||
else:
|
||
r = await c.post(f"https://10.5.85.11:8006/api2/json/nodes/{node_name}/qemu/{vmid}/status/stop", headers={"Authorization": auth})
|
||
results["steps"]["stop_vm"] = {"status": "ok" if r.status_code == 200 else "failed", "detail": r.json()}
|
||
if r.status_code != 200:
|
||
return JSONResponse({"error": f"Failed to stop VM: {r.json()}"}, status_code=500)
|
||
# Wait for VM to stop
|
||
await asyncio.sleep(5)
|
||
else:
|
||
results["steps"]["stop_vm"] = {"status": "skipped", "message": "VM already stopped"}
|
||
|
||
# Step 3: Destroy VM
|
||
if dry_run:
|
||
results["steps"]["destroy_vm"] = {"status": "dry_run", "message": f"Would destroy VM {vmid}"}
|
||
else:
|
||
r = await c.delete(f"https://10.5.85.11:8006/api2/json/nodes/{node_name}/qemu/{vmid}", headers={"Authorization": auth})
|
||
results["steps"]["destroy_vm"] = {"status": "ok" if r.status_code == 200 else "failed", "detail": r.json()}
|
||
|
||
# Step 4: Remove from Dockhand (by IP)
|
||
if ip:
|
||
if dry_run:
|
||
results["steps"]["dockhand_remove"] = {"status": "dry_run", "message": f"Would remove Dockhand env for IP {ip}"}
|
||
else:
|
||
# Find environment by IP
|
||
envs = await c.get("http://10.4.1.116:3000/api/environments", headers={"Authorization": f"Bearer {BUTLER_TOKEN}"})
|
||
env_id = None
|
||
for env in envs.json():
|
||
if env.get("name", "").lower() == hostname.lower() or env.get("ip") == ip:
|
||
env_id = env.get("id")
|
||
break
|
||
if env_id:
|
||
r = await c.delete(f"http://10.4.1.116:3000/api/environments/{env_id}", headers={"Authorization": f"Bearer {BUTLER_TOKEN}"})
|
||
results["steps"]["dockhand_remove"] = {"status": "ok" if r.status_code in [200, 204] else "failed", "env_id": env_id}
|
||
else:
|
||
results["steps"]["dockhand_remove"] = {"status": "skipped", "message": "No Dockhand environment found"}
|
||
else:
|
||
results["steps"]["dockhand_remove"] = {"status": "skipped", "message": "No IP found"}
|
||
|
||
# Step 5: Delete Forgejo repo (by hostname)
|
||
if hostname:
|
||
if dry_run:
|
||
results["steps"]["forgejo_repo_delete"] = {"status": "dry_run", "message": f"Would delete repo sascha/{hostname}"}
|
||
else:
|
||
r = await c.delete(f"http://10.4.1.116:8888/forgejo/api/v1/repos/sascha/{hostname}", headers={"Authorization": f"Bearer {BUTLER_TOKEN}"})
|
||
results["steps"]["forgejo_repo_delete"] = {"status": "ok" if r.status_code in [200, 204] else "not_found", "detail": r.json() if r.status_code != 204 else "deleted"}
|
||
else:
|
||
results["steps"]["forgejo_repo_delete"] = {"status": "skipped", "message": "No hostname found"}
|
||
|
||
# Step 6: Remove from Ansible inventory
|
||
if hostname:
|
||
if dry_run:
|
||
results["steps"]["ansible_cleanup"] = {"status": "dry_run", "message": f"Would remove {hostname} from pfannkuchen.ini"}
|
||
else:
|
||
# Remove host from inventory
|
||
remove_cmd = f'''python3 -c "
|
||
lines = open('/app-config/ansible/pfannkuchen.ini').readlines()
|
||
out = [l for l in lines if '{hostname}' not in l]
|
||
open('/app-config/ansible/pfannkuchen.ini','w').writelines(out)
|
||
print('removed')
|
||
" '''
|
||
rc, out, err = _ssh(AUTOMATION1, remove_cmd, timeout=30)
|
||
# Also remove host_vars
|
||
_ssh(AUTOMATION1, f"rm -rf /app-config/ansible/host_vars/{hostname}", timeout=30)
|
||
results["steps"]["ansible_cleanup"] = {"status": "ok" if rc == 0 else "failed", "detail": out.strip()}
|
||
else:
|
||
results["steps"]["ansible_cleanup"] = {"status": "skipped", "message": "No hostname found"}
|
||
|
||
# Step 7: Remove from Uptime Kuma (if monitoring exists)
|
||
if hostname:
|
||
if dry_run:
|
||
results["steps"]["uptime_kuma_remove"] = {"status": "dry_run", "message": f"Would remove monitor for {hostname}"}
|
||
else:
|
||
try:
|
||
# Get all monitors from Kuma (no auth needed for local network)
|
||
kuma_monitors = await c.get("http://10.200.200.1:3001/api/monitors", timeout=5)
|
||
for monitor in kuma_monitors.json().get("data", []):
|
||
if hostname.lower() in monitor.get("name", "").lower():
|
||
r = await c.delete(f"http://10.200.200.1:3001/api/monitors/{monitor['id']}", timeout=5)
|
||
results["steps"]["uptime_kuma_remove"] = {"status": "ok" if r.status_code == 200 else "failed", "monitor_id": monitor["id"]}
|
||
break
|
||
else:
|
||
results["steps"]["uptime_kuma_remove"] = {"status": "skipped", "message": "No Kuma monitor found"}
|
||
except Exception as e:
|
||
results["steps"]["uptime_kuma_remove"] = {"status": "error", "detail": str(e)}
|
||
else:
|
||
results["steps"]["uptime_kuma_remove"] = {"status": "skipped", "message": "No hostname found"}
|
||
|
||
_audit(f"/vm/destroy/{vmid}", "DELETE", 200, f"dry_run={dry_run}")
|
||
return results
|
||
|
||
@app.post("/inventory/host")
|
||
async def inventory_host(request: Request, _=Depends(_verify)):
|
||
"""Create or update an Ansible inventory host idempotently."""
|
||
body = await request.json()
|
||
name, ip = body.get("name", ""), body.get("ip", "")
|
||
group = body.get("group", "auto")
|
||
user = body.get("user", "sascha")
|
||
if not all(re.fullmatch(r"[A-Za-z0-9_.-]+", value) for value in (name, group, user)):
|
||
raise HTTPException(400, "Invalid name, group, or user")
|
||
try:
|
||
ipaddress.ip_address(ip)
|
||
except ValueError:
|
||
raise HTTPException(400, "Invalid IP address")
|
||
|
||
ini = "/app-config/ansible/pfannkuchen.ini"
|
||
host_line = f"{name} ansible_host={ip} ansible_user={user}"
|
||
script = f'''lines = open({ini!r}).readlines()
|
||
name = {name!r}
|
||
host_line = {host_line!r}
|
||
group = {group!r}
|
||
updated = False
|
||
for idx, line in enumerate(lines):
|
||
parts = line.split()
|
||
if parts and parts[0] == name and "ansible_host=" in line:
|
||
lines[idx] = host_line + "\\n"
|
||
updated = True
|
||
break
|
||
if not updated:
|
||
insert_at = None
|
||
in_group = False
|
||
for idx, line in enumerate(lines):
|
||
if line.strip() == "[" + group + "]":
|
||
in_group = True
|
||
insert_at = idx + 1
|
||
continue
|
||
if in_group and line.startswith("["):
|
||
break
|
||
if in_group:
|
||
insert_at = idx + 1
|
||
if insert_at is None:
|
||
lines.extend(["\\n[" + group + "]\\n", host_line + "\\n"])
|
||
else:
|
||
lines.insert(insert_at, host_line + "\\n")
|
||
open({ini!r}, "w").writelines(lines)
|
||
print("updated" if updated else "added")'''
|
||
encoded_script = base64.b64encode(script.encode()).decode()
|
||
rc, out, err = _ssh(
|
||
AUTOMATION1,
|
||
f"python3 -c \"import base64;exec(base64.b64decode('{encoded_script}'))\"",
|
||
timeout=30,
|
||
)
|
||
if rc != 0:
|
||
raise HTTPException(502, err.strip()[:500])
|
||
vars_script = (
|
||
f"mkdir -p /app-config/ansible/host_vars/{name} && "
|
||
f"printf 'ansible_host: {ip}\\nansible_user: {user}\\n' > "
|
||
f"/app-config/ansible/host_vars/{name}/vars.yml"
|
||
)
|
||
rc2, _out2, err2 = _ssh(AUTOMATION1, vars_script, timeout=30)
|
||
if rc2 != 0:
|
||
raise HTTPException(502, err2.strip()[:500])
|
||
_audit("/inventory/host", "POST", 200, f"{name} {ip} {user}")
|
||
return {"status": "ok", "name": name, "ip": ip, "group": group, "user": user, "result": out.strip()}
|
||
|
||
@app.post("/ansible/run")
|
||
async def ansible_run(request: Request, _=Depends(_verify)):
|
||
body = await request.json()
|
||
hostname = body.get("limit", body.get("hostname", ""))
|
||
template_id = body.get("template_id", 10)
|
||
if not hostname:
|
||
return JSONResponse({"error": "limit/hostname required"}, status_code=400)
|
||
rc, out, err = _ssh(AUTOMATION1, f"cd /app-config/ansible && bash pfannkuchen.sh setup {hostname}", timeout=600)
|
||
|
||
# After successful ansible run: sync Hawser token to Dockhand
|
||
if rc == 0:
|
||
try:
|
||
# Get VM IP from inventory
|
||
inv_path = "/app-config/ansible/pfannkuchen.ini"
|
||
rc2, ip_out, _ = _ssh(AUTOMATION1, f"grep -E '^{hostname} ' {inv_path} | awk '{{print $2}}' | cut -d= -f2", timeout=10)
|
||
vm_ip = ip_out.strip() if rc2 == 0 else None
|
||
|
||
if vm_ip:
|
||
# Read Hawser token from VM
|
||
rc3, token_out, _ = _ssh(AUTOMATION1, f"ssh -o StrictHostKeyChecking=no sascha@{vm_ip} 'sudo grep ^TOKEN= /etc/hawser/config | cut -d= -f2' 2>/dev/null", timeout=15)
|
||
hawser_token = token_out.strip() if rc3 == 0 and token_out.strip() else None
|
||
|
||
if hawser_token:
|
||
# Find environment in Dockhand by IP and update token
|
||
async with httpx.AsyncClient(verify=False, timeout=10) as c:
|
||
# Login to Dockhand
|
||
login = await c.post(f"{SERVICES['dockhand']['url']}/api/auth/login",
|
||
json={"username": "admin", "password": _read("dockhand") or ""})
|
||
if login.status_code == 200:
|
||
cookie = dict(login.cookies)
|
||
# Get all environments
|
||
envs = await c.get(f"{SERVICES['dockhand']['url']}/api/environments", cookies=cookie)
|
||
for env in envs.json():
|
||
if env.get("host") == vm_ip:
|
||
# Update environment with Hawser token
|
||
await c.put(f"{SERVICES['dockhand']['url']}/api/environments/{env['id']}",
|
||
json={"hawserToken": hawser_token},
|
||
cookies=cookie)
|
||
log.info(f"Updated Hawser token for {hostname} (env {env['id']})")
|
||
break
|
||
except Exception as e:
|
||
log.warning(f"Hawser token sync failed for {hostname}: {e}")
|
||
|
||
# NEW: SOPS + .env handling for Git-centric deployments
|
||
try:
|
||
log.info(f"Checking for SOPS .env setup for {hostname}")
|
||
# Get SOPS age public key from automation1
|
||
rc4, age_pub, _ = _ssh(AUTOMATION1, "cat ~/.config/sops/age/keys.txt | grep '^# public key' | awk '{print $4}'", timeout=10)
|
||
age_pub = age_pub.strip() if rc4 == 0 else None
|
||
|
||
if age_pub:
|
||
# Check if compose.yaml exists in /app-config/github/{hostname}/
|
||
rc5, compose_check, _ = _ssh(AUTOMATION1, f"test -f /app-config/github/{hostname}/compose.yaml && echo 'found' || echo 'missing'", timeout=10)
|
||
|
||
if compose_check.strip() == "found":
|
||
log.info(f"Found compose.yaml for {hostname}, generating .env")
|
||
|
||
# Generate secrets
|
||
import secrets as sec
|
||
secret_key = base64.b64encode(sec.token_bytes(32)).decode()
|
||
admin_pw = sec.token_urlsafe(16)
|
||
db_pw = sec.token_urlsafe(16)
|
||
|
||
# Store in vault cache
|
||
_vault_cache[f"{hostname}_secret_key"] = secret_key
|
||
_vault_cache[f"{hostname}_admin_password"] = admin_pw
|
||
_vault_cache[f"{hostname}_db_password"] = db_pw
|
||
|
||
# Build .env content
|
||
env_lines = ["# Auto-generated by Butler", f"TZ=Europe/Berlin", "PUID=1000", "PGID=1000"]
|
||
|
||
# Detect service type from hostname
|
||
if "paperless" in hostname.lower():
|
||
env_lines.extend([
|
||
f"PAPERLESS_ADMIN_USER=admin",
|
||
f"PAPERLESS_ADMIN_PASSWORD={admin_pw}",
|
||
f"PAPERLESS_SECRET_KEY={secret_key}",
|
||
f"PAPERLESS_URL=http://{vm_ip}:8000",
|
||
"PAPERLESS_TIME_ZONE=Europe/Berlin",
|
||
"PAPERLESS_OCR_LANGUAGE=deu",
|
||
"PAPERLESS_REDIS=redis://redis:6379",
|
||
"PAPERLESS_DBHOST=postgres",
|
||
"PAPERLESS_DBPORT=5432",
|
||
"PAPERLESS_DBNAME=paperless",
|
||
"PAPERLESS_DBUSER=paperless",
|
||
f"PAPERLESS_DBPASS={db_pw}",
|
||
"POSTGRES_DB=paperless",
|
||
"POSTGRES_USER=paperless",
|
||
f"POSTGRES_PASSWORD={db_pw}",
|
||
])
|
||
|
||
env_content = "\\n".join(env_lines) + "\\n"
|
||
|
||
# Write .env to automation1
|
||
env_tmp = f"/tmp/{hostname}.env"
|
||
_ssh(AUTOMATION1, f"printf '%s' '{env_content}' > {env_tmp}", timeout=10)
|
||
|
||
# Encrypt with SOPS
|
||
enc_tmp = f"/tmp/{hostname}.env.enc"
|
||
_ssh(AUTOMATION1, f"cd /tmp && SOPS_AGE_RECIPIENT={age_pub} sops --encrypted-regex 'PASSWORD|SECRET_KEY|_PASS' --encrypt {env_tmp} > {enc_tmp}", timeout=30)
|
||
|
||
# Copy to VM
|
||
_ssh(AUTOMATION1, f"scp -o StrictHostKeyChecking=no {env_tmp} sascha@{vm_ip}:/app-config/github/{hostname}/.env 2>/dev/null || true", timeout=15)
|
||
_ssh(AUTOMATION1, f"scp -o StrictHostKeyChecking=no {enc_tmp} sascha@{vm_ip}:/app-config/github/{hostname}/.env.enc 2>/dev/null || true", timeout=15)
|
||
|
||
log.info(f"SOPS .env setup completed for {hostname}")
|
||
except Exception as e:
|
||
log.warning(f"SOPS .env setup failed for {hostname}: {e}")
|
||
|
||
return {"status": "ok" if rc == 0 else "error", "rc": rc, "output": out[-1000:]}
|
||
|
||
@app.get("/ansible/status/{job_id}")
|
||
async def ansible_status(job_id: int, _=Depends(_verify)):
|
||
return {"info": "direct SSH mode - no async job tracking"}
|
||
|
||
# --- TTS Endpoints ---
|
||
|
||
class TTSRequest(BaseModel):
|
||
text: str
|
||
target: str = "speaker" # "speaker" or "telegram"
|
||
voice: str = "deep_thought.mp3"
|
||
language: str = "de"
|
||
|
||
SPEAKER_URL = TTS_CFG.get("speaker_url", "http://10.10.1.166:10800") if TTS_CFG else "http://10.10.1.166:10800"
|
||
CHATTERBOX_URL = TTS_CFG.get("chatterbox_url", "http://10.2.1.104:8004/tts") if TTS_CFG else "http://10.2.1.104:8004/tts"
|
||
|
||
@app.post("/tts/speak")
|
||
async def tts_speak(req: TTSRequest, _=Depends(_verify)):
|
||
if req.target == "speaker":
|
||
async with httpx.AsyncClient(verify=False, timeout=120) as c:
|
||
r = await c.post(SPEAKER_URL, json={"text": req.text})
|
||
return {"status": "ok" if r.status_code == 200 else "error", "target": "speaker"}
|
||
elif req.target == "telegram":
|
||
# Generate WAV via Chatterbox, save to hermes VM as OGG for Telegram voice
|
||
async with httpx.AsyncClient(verify=False, timeout=120) as c:
|
||
r = await c.post(CHATTERBOX_URL, json={
|
||
"text": req.text, "voice_mode": "clone",
|
||
"reference_audio_filename": req.voice,
|
||
"output_format": "wav", "language": req.language,
|
||
"exaggeration": 0.3, "cfg_weight": 0.7, "temperature": 0.6,
|
||
})
|
||
if r.status_code != 200:
|
||
return JSONResponse({"error": "chatterbox failed"}, status_code=500)
|
||
# Save WAV and convert to OGG on hermes
|
||
import tempfile
|
||
wav_path = tempfile.mktemp(suffix=".wav")
|
||
ogg_path = "/tmp/trulla_voice.ogg"
|
||
with open(wav_path, "wb") as f:
|
||
f.write(r.content)
|
||
rc, _, _ = _ssh("sascha@10.4.1.100", f"rm -f {ogg_path}", timeout=10)
|
||
# Copy WAV to hermes and convert
|
||
_sp.run(["scp", "-o", "ConnectTimeout=5", wav_path, f"sascha@10.4.1.100:/tmp/trulla_voice.wav"], timeout=30)
|
||
_ssh("sascha@10.4.1.100", f"ffmpeg -y -i /tmp/trulla_voice.wav -c:a libopus -b:a 64k {ogg_path} 2>/dev/null", timeout=30)
|
||
os.unlink(wav_path)
|
||
return {"status": "ok", "target": "telegram", "media_path": ogg_path, "hint": "Use MEDIA:/tmp/trulla_voice.ogg in response"}
|
||
else:
|
||
return JSONResponse({"error": f"unknown target: {req.target}"}, status_code=400)
|
||
|
||
@app.get("/tts/voices")
|
||
async def tts_voices(_=Depends(_verify)):
|
||
async with httpx.AsyncClient(verify=False, timeout=10) as c:
|
||
r = await c.get("http://10.2.1.104:8004/get_predefined_voices")
|
||
return r.json()
|
||
|
||
@app.get("/tts/health")
|
||
async def tts_health(_=Depends(_verify)):
|
||
results = {}
|
||
async with httpx.AsyncClient(verify=False, timeout=5) as c:
|
||
try:
|
||
r = await c.get(SPEAKER_URL)
|
||
results["speaker"] = r.json()
|
||
except Exception as e:
|
||
results["speaker"] = {"status": "offline", "error": str(e)}
|
||
try:
|
||
r = await c.get("http://10.2.1.104:8004/api/model-info")
|
||
results["chatterbox"] = "ok"
|
||
except Exception as e:
|
||
results["chatterbox"] = {"status": "offline", "error": str(e)}
|
||
return results
|
||
|
||
|
||
@app.api_route("/{service}/{path:path}", methods=["GET", "POST", "PUT", "DELETE", "PATCH"])
|
||
async def proxy(service: str, path: str, request: Request, _=Depends(_verify)):
|
||
SKIP_SERVICES = {"vm", "inventory", "ansible", "debug", "tts", "status", "audit", "config"}
|
||
if service in SKIP_SERVICES:
|
||
raise HTTPException(404, f"Unknown service: {service}")
|
||
cfg = SERVICES.get(service)
|
||
if not cfg:
|
||
raise HTTPException(404, f"Unknown service: {service}. Available: {list(SERVICES.keys())}")
|
||
|
||
base_url = cfg["url"]
|
||
auth_type = cfg["auth"]
|
||
headers = dict(request.headers)
|
||
cookies = {}
|
||
|
||
for h in ["host", "content-length", "transfer-encoding", "authorization"]:
|
||
headers.pop(h, None)
|
||
|
||
if auth_type == "apikey":
|
||
headers["X-Api-Key"] = _get_key(cfg) or ""
|
||
elif auth_type == "apikey_urlfile":
|
||
url, key = _parse_url_key(cfg["key_file"])
|
||
base_url = url.rstrip("/") if url else ""
|
||
headers["X-Api-Key"] = key or ""
|
||
elif auth_type == "bearer":
|
||
headers["Authorization"] = f"Bearer {_get_key(cfg)}"
|
||
elif auth_type == "n8n":
|
||
headers["X-N8N-API-KEY"] = _get_key(cfg) or ""
|
||
elif auth_type == "proxmox":
|
||
pv = _parse_kv("proxmox")
|
||
headers["Authorization"] = f"PVEAPIToken={pv.get('tokenid', '')}={pv.get('secret', '')}"
|
||
elif auth_type == "session":
|
||
global _dockhand_cookie
|
||
if not _dockhand_cookie:
|
||
async with httpx.AsyncClient(verify=False) as c:
|
||
await _dockhand_login(c)
|
||
cookies = _dockhand_cookie or {}
|
||
|
||
target = f"{base_url}/{path}"
|
||
body = await request.body()
|
||
|
||
timeout = float(cfg.get("timeout", 30))
|
||
async with httpx.AsyncClient(verify=False, timeout=timeout) as client:
|
||
resp = await client.request(method=request.method, url=target,
|
||
headers=headers, cookies=cookies, content=body,
|
||
params=request.query_params)
|
||
if auth_type == "session" and resp.status_code == 401:
|
||
_dockhand_cookie = None
|
||
await _dockhand_login(client)
|
||
resp = await client.request(method=request.method, url=target,
|
||
headers=headers, cookies=_dockhand_cookie or {}, content=body,
|
||
params=request.query_params)
|
||
|
||
try:
|
||
data = _redact_response(resp.json(), cfg.get("redact_response_fields"))
|
||
except Exception:
|
||
data = resp.text
|
||
_audit(f"/{service}/{path}", request.method, resp.status_code)
|
||
return JSONResponse(content=data, status_code=resp.status_code)
|
||
|
||
|
||
|
||
|
||
|