#!/usr/bin/env python3
"""Dashboard de control pentru puntea Discord -> Claude Code (LXC 171).
Model: `echo-core/dashboard` de pe LXC 110 (server stdlib + handler-ul `eco.py`).
Diferentele deliberate:
- **o singura unitate controlata**: `claude-discord.service`. Nu exista endpoint
care sa primeasca un nume de unit din exterior — altfel dashboard-ul ar deveni
un `systemctl` remote fara parola.
- **bind pe 127.0.0.1 implicit**: butonul "restart" opreste un agent care ruleaza
cu `bypassPermissions` si chei SSH catre tot clusterul. Accesul se face prin
tunel SSH (vezi README), nu expus in LAN.
- **restart protejat de tururi in zbor**: `stop`/`restart` intorc 409 daca exista
fire cu tur in desfasurare, pana cand se cere explicit `force`.
Fara dependinte in afara stdlib: ruleaza cu acelasi python ca botul, dar nu are
nevoie de venv-ul lui.
"""
from __future__ import annotations
import json
import os
import secrets
import shutil
import subprocess
import sys
import threading
import time
from datetime import datetime
from http.server import SimpleHTTPRequestHandler, ThreadingHTTPServer
from pathlib import Path
from urllib.parse import parse_qs, urlparse
# Radacina punti (parintele lui dashboard/) trebuie sa fie importabila: de acolo
# vin config.py, cleanup.py si security/approvals.py.
_DASH = Path(__file__).resolve().parent
_BRIDGE = _DASH.parent
for _p in (str(_BRIDGE), str(_DASH)):
if _p not in sys.path:
sys.path.insert(0, _p)
import config # noqa: E402
from limits import parse_cap # noqa: E402
# ── constante ───────────────────────────────────────────────────────────
SERVICE = "claude-discord.service"
SELF_SERVICE = "claude-discord-dashboard.service"
# Caile se recalculeaza la fiecare apel, nu se ingheata la import: `config.reload()`
# muta STATE_DIR (testele o folosesc ca sa scoata totul din ~/.claude-discord).
def bot_log() -> Path:
return config.LOG_DIR / "bot.log"
def infra_log() -> Path:
return config.LOG_DIR / "infra.log"
COOKIE_NAME = "dashboard"
COOKIE_MAX_AGE = 60 * 60 * 24 * 30
def mount_prefix() -> str:
"""Prefixul sub care e montat panoul (`DASHBOARD_PREFIX`, ex. `/claude`).
`tailscale serve --set-path /claude` TAIE prefixul inainte de a proxa, deci in
mod normal aici ajunge `/api/status`. Prefixul e acceptat totusi si intact,
pentru cazul unui proxy care nu taie si pentru `curl` direct pe localhost.
Paginile nu depind de el: toate URL-urile din HTML sunt relative.
"""
pfx = (config.get("DASHBOARD_PREFIX") or "").strip().rstrip("/")
if pfx and not pfx.startswith("/"):
pfx = "/" + pfx
return pfx
def strip_prefix(path: str) -> str:
pfx = mount_prefix()
if pfx and (path == pfx or path.startswith(pfx + "/")):
return path[len(pfx):] or "/"
return path
_TOKEN: str | None = None
def reset_token_cache() -> None:
"""Uita tokenul memorat (folosit de teste dupa `config.reload`)."""
global _TOKEN
_TOKEN = None
def auth_disabled() -> bool:
"""`DASHBOARD_AUTH=off` in `~/.claude-discord/env` scoate complet login-ul.
Alegere constienta a operatorului, nu implicit: panoul poate opri un agent cu
`bypassPermissions` si chei SSH catre tot clusterul, deci fara token oricine
ajunge la port il poate folosi. Are sens doar pentru ca serviciul e legat de
`127.0.0.1` si publicat exclusiv in tailnet (`tailscale serve`, tainet only),
unde identitatea o face deja Tailscale.
"""
return (config.get("DASHBOARD_AUTH") or "").strip().lower() in ("off", "none", "0", "false")
def dashboard_token() -> str:
"""Tokenul de acces, din `~/.claude-discord/env` (`DASHBOARD_TOKEN`).
Lipsa lui NU deschide dashboard-ul: se genereaza unul aleator per proces si se
tipareste in log, deci ramane accesibil doar cui poate citi logul. Ca sa fie
deschis intentionat, se pune `DASHBOARD_AUTH=off`.
"""
global _TOKEN
if _TOKEN is None:
tok = (config.get("DASHBOARD_TOKEN") or "").strip()
if not tok:
tok = secrets.token_urlsafe(32)
print(
f"[auth] DASHBOARD_TOKEN nesetat in {config.ENV_FILE} — token efemer "
f"pentru acest proces: {tok}",
file=sys.stderr, flush=True,
)
_TOKEN = tok
return _TOKEN
def tailnet_user(headers) -> str:
"""Cine e, dupa Tailscale. `tailscale serve` pune antetul pe cererile din
tailnet; pe localhost lipseste. Doar pentru jurnal — nu e folosit ca decizie
de acces, fiindca un proces local ar putea sa-l fabrice."""
return (headers.get("Tailscale-User-Login") or "").strip()
# ── systemd ─────────────────────────────────────────────────────────────
def _sysctl(*args: str, timeout: float = 30.0) -> subprocess.CompletedProcess:
return subprocess.run(
["systemctl", "--user", *args],
capture_output=True, text=True, timeout=timeout,
)
def _show(unit: str, prop: str) -> str:
try:
return _sysctl("show", "-p", prop, "--value", unit, timeout=5).stdout.strip()
except Exception:
return ""
def _uptime_s(unit: str) -> int | None:
"""Secunde de la ultima pornire a unitatii.
Sursa principala e `ActiveEnterTimestampMonotonic` (microsecunde pe
CLOCK_MONOTONIC), comparat cu `time.monotonic()` — ACELASI ceas. **Nu** se
foloseste `/proc/uptime`: intr-un LXC acela e virtualizat de lxcfs si arata
uptime-ul containerului, mai mic decat monotonic-ul gazdei pe care il
raporteaza systemd, deci diferenta iese negativa si uptime-ul apare 0.
Rezerva e `ActiveEnterTimestamp`, ora de perete in fusul local al masinii cu
numele fusului la coada; il taiem si lasam `.timestamp()` sa-l interpreteze
ca ora locala (`%Z` in strptime nu produce un offset utilizabil).
"""
mono = _show(unit, "ActiveEnterTimestampMonotonic")
if mono.isdigit() and int(mono) > 0:
return max(0, int(time.monotonic() - int(mono) / 1_000_000))
ts = _show(unit, "ActiveEnterTimestamp")
if not ts:
return None
try:
parts = ts.split()
# "Sun 2026-08-30 12:47:20 UTC" -> data + ora, fara ziua si fusul
stamp = datetime.strptime(f"{parts[1]} {parts[2]}", "%Y-%m-%d %H:%M:%S")
except (ValueError, IndexError):
return None
return max(0, int(time.time() - stamp.timestamp()))
def unit_info(unit: str) -> dict:
active = _show(unit, "ActiveState")
pid = _show(unit, "MainPID")
mem = _show(unit, "MemoryCurrent")
restarts = _show(unit, "NRestarts")
return {
"unit": unit,
"active": active == "active",
"state": active or "unknown",
"sub": _show(unit, "SubState"),
"enabled": _show(unit, "UnitFileState"),
"pid": int(pid) if pid.isdigit() and pid != "0" else None,
"memory_bytes": int(mem) if mem.isdigit() else None,
"restarts": int(restarts) if restarts.isdigit() else 0,
"uptime_s": _uptime_s(unit) if active == "active" else None,
}
# ── starea punti ────────────────────────────────────────────────────────
def read_state() -> dict:
"""state.json fara lock: dashboard-ul doar citeste si nu are voie sa blocheze
botul. Un JSON prins la mijlocul unei scrieri intoarce {} — se reincarca la
urmatorul poll."""
try:
data = json.loads(config.STATE_FILE.read_text(encoding="utf-8"))
return data if isinstance(data, dict) else {}
except (OSError, ValueError):
return {}
def _pid_alive(pid) -> bool:
try:
return pid is not None and Path(f"/proc/{int(pid)}").exists()
except (TypeError, ValueError):
return False
def threads_view(state: dict) -> list[dict]:
out = []
for tid, rec in (state.get("threads") or {}).items():
if not isinstance(rec, dict):
continue
inflight = rec.get("inflight") or None
out.append({
"thread_id": tid,
"cwd": rec.get("cwd"),
"model": rec.get("model"),
"pid": rec.get("pid"),
"alive": _pid_alive(rec.get("pid")),
"inflight": bool(inflight),
"inflight_since": (inflight or {}).get("started_at"),
"cost_usd": round(float(rec.get("cost_usd_total") or 0), 4),
"last_active": rec.get("last_active"),
})
out.sort(key=lambda t: t.get("last_active") or 0, reverse=True)
return out
def inflight_threads(state: dict) -> list[str]:
return [t["thread_id"] for t in threads_view(state) if t["inflight"]]
def pending_approvals() -> list[dict]:
"""Cererile de confirmare in asteptare, citite direct din director.
Nu importam `security.approvals` (API-ul lui e async si porneste un watcher);
formatul fisierului e fixat in security/README.md.
"""
out = []
d = config.APPROVALS_DIR
try:
files = sorted(d.glob("*.json"))
except OSError:
return out
now = time.time()
for f in files:
try:
req = json.loads(f.read_text(encoding="utf-8"))
except (OSError, ValueError):
continue
if not isinstance(req, dict) or req.get("status") != "pending":
continue
out.append({
"request_id": req.get("request_id"),
"thread_id": req.get("thread_id"),
"tool_name": req.get("tool_name"),
"command": (req.get("command") or "")[:500],
"rule": req.get("rule"),
"reason": req.get("reason"),
"created_at": req.get("created_at"),
"expires_in": (
round(req["expires_at"] - now, 1)
if isinstance(req.get("expires_at"), (int, float)) else None
),
})
return out
def doctor() -> list[dict]:
checks: list[dict] = []
info = unit_info(SERVICE)
checks.append({
"name": "Serviciu claude-discord",
"pass": info["active"],
"detail": f'{info["state"]}/{info["sub"]}, {info["restarts"]} restarturi',
})
st = read_state()
checks.append({
"name": "state.json",
"pass": bool(st),
"detail": f'{len(st.get("threads") or {})} fire' if st else "ilizibil sau gol",
})
cap = parse_cap(config.get("COST_CAP_USD_DAY"), 0.0)
spent = float((st.get("cost") or {}).get("usd") or 0)
checks.append({
"name": "Plafon de cost pe zi",
"pass": cap <= 0 or spent < cap,
"detail": f"{spent:.2f} / {cap:.2f} USD" if cap > 0
else f"{spent:.2f} USD cheltuiti, fara plafon (abonament)",
})
try:
du = shutil.disk_usage("/")
pct = du.free / du.total * 100
checks.append({
"name": "Spatiu pe disc",
"pass": pct > 10,
"detail": f"{pct:.1f}% liber ({du.free // 1024**3} GB)",
})
except OSError as exc:
checks.append({"name": "Spatiu pe disc", "pass": False, "detail": str(exc)})
try:
log = bot_log()
size_mb = log.stat().st_size / 1024**2 if log.exists() else 0
checks.append({
"name": "bot.log",
"pass": log.exists() and size_mb < 100,
"detail": f"{size_mb:.1f} MB" if log.exists() else "lipseste",
})
except OSError as exc:
checks.append({"name": "bot.log", "pass": False, "detail": str(exc)})
# Regula deny(ssh) a mai taiat o data accesul la infrastructura — vezi
# security/README.md. Verificam sa nu reapara la o editare viitoare.
try:
settings = json.loads(config.SETTINGS_FILE.read_text(encoding="utf-8"))
deny = (settings.get("permissions") or {}).get("deny") or []
bad = [d for d in deny if d.startswith(("Bash(ssh", "Bash(scp"))]
checks.append({
"name": "Reguli deny in bot-settings.json",
"pass": not bad,
"detail": f"blocheaza infrastructura: {bad}" if bad else f"{len(deny)} reguli, ssh liber",
})
except (OSError, ValueError) as exc:
checks.append({"name": "Reguli deny in bot-settings.json", "pass": False, "detail": str(exc)})
hook = _BRIDGE / "security" / "confirm_hook.py"
checks.append({
"name": "Hook de confirmare",
"pass": hook.exists(),
"detail": str(hook) if hook.exists() else "lipseste",
})
claude_bin = shutil.which(config.get("CLAUDE_BIN") or "claude")
checks.append({
"name": "CLI claude",
"pass": bool(claude_bin),
"detail": claude_bin or "nu e in PATH",
})
return checks
def orphans_report(dry_run: bool = True) -> dict:
"""`/cleanup` din Discord, expus si aici. Importul e lenes fiindca modulul
citeste /proc la import-time in unele cai."""
import cleanup # noqa: PLC0415
found = cleanup.find_orphans(read_state())
results = None
if not dry_run and found:
results = cleanup.kill_orphans(found, dry_run=False)
return {
"orphans": found,
"killed": results,
"report": cleanup.format_report(found, results),
}
# ── HTTP ────────────────────────────────────────────────────────────────
def _parse_cookies(raw: str) -> dict[str, str]:
out: dict[str, str] = {}
for chunk in (raw or "").split(";"):
chunk = chunk.strip()
if "=" in chunk:
k, v = chunk.split("=", 1)
out[k.strip()] = v.strip()
return out
class Handler(SimpleHTTPRequestHandler):
server_version = "claude-discord-dashboard"
protocol_version = "HTTP/1.1"
def __init__(self, *a, **kw):
super().__init__(*a, directory=str(_DASH), **kw)
# --- utilitare -----------------------------------------------------
def log_message(self, format, *args): # jurnal compact, o linie # noqa: A002
sys.stderr.write("%s %s\n" % (self.address_string(), format % args))
def send_json(self, payload, status: int = 200, extra_headers: dict | None = None):
body = json.dumps(payload, ensure_ascii=False).encode("utf-8")
self.send_response(status)
self.send_header("Content-Type", "application/json; charset=utf-8")
self.send_header("Content-Length", str(len(body)))
self.send_header("Cache-Control", "no-store")
for k, v in (extra_headers or {}).items():
self.send_header(k, v)
self.end_headers()
try:
self.wfile.write(body)
except (BrokenPipeError, ConnectionResetError):
pass
def read_json(self) -> dict:
try:
n = int(self.headers.get("Content-Length") or 0)
raw = self.rfile.read(n).decode("utf-8") if n > 0 else ""
data = json.loads(raw) if raw else {}
return data if isinstance(data, dict) else {}
except (ValueError, OSError, UnicodeDecodeError):
return {}
def authed(self) -> bool:
if auth_disabled():
return True
got = _parse_cookies(self.headers.get("Cookie", "")).get(COOKIE_NAME, "")
return bool(got) and secrets.compare_digest(got, dashboard_token())
def deny(self):
self.send_json({"error": "neautentificat"}, 401)
# --- GET -----------------------------------------------------------
def do_GET(self):
raw = urlparse(self.path).path
path = strip_prefix(raw)
if path == "/" and not raw.endswith("/"):
# `/claude` fara slash final: URL-urile relative din pagina s-ar
# rezolva la radacina hostului (`/api/status`), unde proxy-ul nu mai
# trimite nimic incoace. Fortam forma cu slash.
self.send_response(301)
self.send_header("Location", raw + "/")
self.send_header("Content-Length", "0")
self.end_headers()
return
if path.startswith("/api/"):
if not self.authed():
return self.deny()
return self.route_get(path)
if path in ("/", "/index.html") and not self.authed():
# Prefixul reintra AICI in mod deliberat. Un "/login.html" absolut ar
# arunca browserul in radacina hostului (alt serviciu), iar un
# "login.html" relativ se rezolva gresit cand adresa vine fara slash
# final (`/claude` -> `/login.html`). Cu prefixul reatasat, ambele
# forme ajung unde trebuie, si direct pe localhost la fel: calea de
# intrare e curatata oricum de `strip_prefix`.
self.send_response(302)
self.send_header("Location", mount_prefix() + "/login.html")
self.send_header("Content-Length", "0")
self.end_headers()
return
if path in ("/", "/index.html"):
return self.send_html("index.html")
if path == "/login.html":
if auth_disabled():
self.send_response(302)
self.send_header("Location", mount_prefix() + "/")
self.send_header("Content-Length", "0")
self.end_headers()
return None
return self.send_html("login.html")
self.path = path
return super().do_GET()
def send_html(self, name: str):
"""Trimite o pagina, cu `` pus la servire.
De ce e nevoie: `tailscale serve --set-path /claude` TAIE prefixul, deci
serverul nu poate sti daca browserul e la `/claude` sau la `/claude/` —
ambele ajung aici ca `/`. Fara slash final, un URL relativ (`api/status`)
se rezolva la radacina hostului, unde proxy-ul nu mai trimite nimic
incoace: pagina se incarca si ramane goala, fara nicio eroare vizibila.
`` fixeaza rezolvarea indiferent de forma adresei.
"""
try:
body = (_DASH / name).read_text(encoding="utf-8")
except OSError:
return self.send_error(404)
pfx = mount_prefix()
if pfx:
body = body.replace("", f'\n', 1)
data = body.encode("utf-8")
self.send_response(200)
self.send_header("Content-Type", "text/html; charset=utf-8")
self.send_header("Content-Length", str(len(data)))
# no-store: paginile poarta acum si configuratia (base href), deci o
# copie veche din cache ar trimite cererile in alta parte.
self.send_header("Cache-Control", "no-store")
self.end_headers()
try:
self.wfile.write(data)
except (BrokenPipeError, ConnectionResetError):
pass
def route_get(self, path: str):
qs = parse_qs(urlparse(self.path).query)
if path == "/api/status":
state = read_state()
return self.send_json({
"service": unit_info(SERVICE),
"dashboard": unit_info(SELF_SERVICE),
"threads": threads_view(state),
"cost": state.get("cost") or {},
"cost_cap": parse_cap(config.get("COST_CAP_USD_DAY"), 0.0),
"pending_approvals": len(pending_approvals()),
"auth": not auth_disabled(),
"user": tailnet_user(self.headers),
"now": time.time(),
})
if path == "/api/logs":
try:
n = min(max(int(qs.get("lines", ["200"])[0]), 1), 2000)
except ValueError:
n = 200
which = qs.get("file", ["bot"])[0]
target = infra_log() if which == "infra" else bot_log()
if not target.exists():
return self.send_json({"lines": [f"({target} nu exista)"]})
r = subprocess.run(["tail", "-n", str(n), str(target)],
capture_output=True, text=True, timeout=15)
return self.send_json({"file": target.name, "lines": r.stdout.splitlines()})
if path == "/api/approvals":
return self.send_json({"approvals": pending_approvals()})
if path == "/api/doctor":
return self.send_json({"checks": doctor()})
if path == "/api/cleanup":
return self.send_json(orphans_report(dry_run=True))
return self.send_json({"error": "ruta necunoscuta"}, 404)
# --- POST ----------------------------------------------------------
def do_POST(self):
path = strip_prefix(urlparse(self.path).path)
if path == "/api/auth/login":
return self.handle_login()
if path == "/api/auth/logout":
return self.send_json(
{"ok": True},
extra_headers={"Set-Cookie": f"{COOKIE_NAME}=; HttpOnly; SameSite=Strict; Path=/; Max-Age=0"},
)
if not self.authed():
return self.deny()
if path == "/api/service":
return self.handle_service()
if path == "/api/cleanup":
data = self.read_json()
return self.send_json(orphans_report(dry_run=bool(data.get("dry_run", True))))
if path == "/api/approvals/decide":
return self.handle_decide()
if path == "/api/restart-self":
self.send_json({"ok": True, "message": "dashboard-ul reporneste in 1s"})
threading.Thread(target=lambda: (time.sleep(1), os._exit(0)), daemon=True).start()
return None
return self.send_json({"error": "ruta necunoscuta"}, 404)
def handle_login(self):
data = self.read_json()
provided = (data.get("token") or "").strip()
if not provided or not secrets.compare_digest(provided, dashboard_token()):
time.sleep(0.5) # incetineste ghicitul
return self.send_json({"error": "token invalid"}, 401)
cookie = (f"{COOKIE_NAME}={dashboard_token()}; HttpOnly; SameSite=Strict; "
f"Path=/; Max-Age={COOKIE_MAX_AGE}")
return self.send_json({"ok": True}, extra_headers={"Set-Cookie": cookie})
def handle_service(self):
"""start / stop / restart pe UNITATEA FIXA. Numele nu vine din request."""
data = self.read_json()
action = str(data.get("action") or "")
if action not in ("start", "stop", "restart"):
return self.send_json({"ok": False, "error": f"actiune necunoscuta: {action}"}, 400)
if action in ("stop", "restart") and not data.get("force"):
busy = inflight_threads(read_state())
if busy:
return self.send_json({
"ok": False,
"error": "tururi in desfasurare",
"inflight": busy,
"hint": "retrimite cu force=true ca sa le intrerupi",
}, 409)
who = tailnet_user(self.headers) or "local"
log_line = f"[actiune] {action} pe {SERVICE}, cerut de {who}"
print(log_line, file=sys.stderr, flush=True)
try:
r = _sysctl(action, SERVICE)
except subprocess.TimeoutExpired:
return self.send_json({"ok": False, "error": "systemctl a depasit timpul"}, 504)
if r.returncode != 0:
return self.send_json({"ok": False, "error": (r.stderr or r.stdout).strip()}, 500)
time.sleep(1.0) # lasa systemd sa actualizeze starea inainte de raspuns
return self.send_json({"ok": True, "action": action, "service": unit_info(SERVICE)})
def handle_decide(self):
data = self.read_json()
rid = str(data.get("request_id") or "")
decision = str(data.get("decision") or "")
if decision not in ("allow", "deny"):
return self.send_json({"ok": False, "error": "decizie invalida"}, 400)
if not rid or "/" in rid or ".." in rid:
return self.send_json({"ok": False, "error": "request_id invalid"}, 400)
import importlib # noqa: PLC0415
approvals = importlib.import_module("security.approvals")
ok = approvals.submit_decision(rid, decision)
return self.send_json({"ok": bool(ok), "request_id": rid, "decision": decision},
200 if ok else 404)
def main() -> None:
bind = config.get("DASHBOARD_BIND") or "127.0.0.1"
try:
port = int(config.get("DASHBOARD_PORT") or 18790)
except ValueError:
port = 18790
dashboard_token() # forteaza avertismentul de token la pornire, nu la primul GET
srv = ThreadingHTTPServer((bind, port), Handler)
srv.daemon_threads = True
print(f"dashboard pe http://{bind}:{port} (unitate controlata: {SERVICE})",
file=sys.stderr, flush=True)
try:
srv.serve_forever()
except KeyboardInterrupt:
pass
finally:
srv.server_close()
if __name__ == "__main__":
main()