7abefa2b46
Panou web pe 127.0.0.1:18790, unit systemd separat de al puntii. Server stdlib
(fara dependinte noi), tokenii de design si tiparul de endpoint-uri preluate din
/home/moltbot/echo-core/dashboard (handlers/eco.py) de pe LXC 110.
Arata: starea unitatii (uptime, PID, memoria cgroup, restarturi), firele din
state.json cu tur in zbor si cost, costul zilei fata de plafon, confirmarile
PreToolUse in asteptare (aprobabile direct din pagina), bot.log / infra.log si
opt verificari de diagnostic.
Face: start / stop / restart pe punte, cautarea si curatarea orfanilor prin
cleanup.py, repornirea propriului serviciu.
Garantii, cu teste:
- unitatea controlata e fixa in cod; un {"unit": "ssh.service"} in cerere nu
schimba nimic, altfel panoul ar fi systemctl remote fara parola;
- stop/restart intorc 409 cu lista firelor active si cer force explicit, fiindca
KillMode=control-group taie tururile in desfasurare;
- state.json se citeste fara lock: panoul nu are voie sa blocheze botul;
- diagnosticul pica daca reapare Bash(ssh:*) in deny (regresia de azi).
Uptime-ul se calculeaza din time.monotonic(), nu din /proc/uptime: in LXC acela
e virtualizat de lxcfs si da diferenta negativa fata de monotonic-ul systemd.
Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01B29CApsP1JkSdjYaGaHpE7
337 lines
12 KiB
Python
337 lines
12 KiB
Python
"""Teste pentru dashboard-ul de control (dashboard/api.py).
|
|
|
|
Zero retea catre exterior si zero systemctl real: `_sysctl` e inlocuit in fiecare
|
|
test cu un dublu care inregistreaza argumentele. Serverul HTTP porneste pe un port
|
|
efemer legat de 127.0.0.1, exact cum ruleaza in productie.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
import pathlib
|
|
import subprocess
|
|
import sys
|
|
import threading
|
|
import time
|
|
import urllib.error
|
|
import urllib.request
|
|
from http.server import ThreadingHTTPServer
|
|
|
|
import pytest
|
|
|
|
ROOT = pathlib.Path(__file__).resolve().parent.parent
|
|
sys.path.insert(0, str(ROOT))
|
|
sys.path.insert(0, str(ROOT / "dashboard"))
|
|
|
|
import config # noqa: E402
|
|
|
|
api = pytest.importorskip("api", reason="dashboard/api.py")
|
|
|
|
|
|
# --- ajutoare ---------------------------------------------------------------
|
|
|
|
def _cp(stdout: str = "", rc: int = 0, stderr: str = "") -> subprocess.CompletedProcess:
|
|
return subprocess.CompletedProcess(args=[], returncode=rc, stdout=stdout, stderr=stderr)
|
|
|
|
|
|
@pytest.fixture()
|
|
def systemctl_fals(monkeypatch):
|
|
"""Inlocuieste systemctl. `apeluri` retine ce s-ar fi executat."""
|
|
apeluri: list[tuple[str, ...]] = []
|
|
props = {
|
|
"ActiveState": "active",
|
|
"SubState": "running",
|
|
"UnitFileState": "enabled",
|
|
"MainPID": "4242",
|
|
"MemoryCurrent": "1048576",
|
|
"NRestarts": "3",
|
|
"ActiveEnterTimestampMonotonic": "0",
|
|
"ActiveEnterTimestamp": "",
|
|
}
|
|
|
|
def fake(*args, timeout=30.0):
|
|
apeluri.append(tuple(args))
|
|
if args[0] == "show":
|
|
return _cp(props.get(args[2], ""))
|
|
return _cp("")
|
|
|
|
monkeypatch.setattr(api, "_sysctl", fake)
|
|
fake.apeluri = apeluri # type: ignore[attr-defined]
|
|
fake.props = props # type: ignore[attr-defined]
|
|
return fake
|
|
|
|
|
|
@pytest.fixture()
|
|
def server(state_dir, systemctl_fals, monkeypatch):
|
|
"""Dashboard-ul pe un port efemer + un client mic cu cookie."""
|
|
(state_dir / "env").write_text("DASHBOARD_TOKEN=secret-de-test\n", encoding="utf-8")
|
|
config.reload(state_dir)
|
|
api.reset_token_cache()
|
|
|
|
srv = ThreadingHTTPServer(("127.0.0.1", 0), api.Handler)
|
|
srv.daemon_threads = True
|
|
threading.Thread(target=srv.serve_forever, daemon=True).start()
|
|
base = f"http://127.0.0.1:{srv.server_address[1]}"
|
|
|
|
class Client:
|
|
def __init__(self):
|
|
self.cookie = ""
|
|
|
|
def call(self, path, data=None, method=None):
|
|
req = urllib.request.Request(
|
|
base + path,
|
|
data=json.dumps(data).encode() if data is not None else None,
|
|
method=method or ("POST" if data is not None else "GET"),
|
|
)
|
|
req.add_header("Content-Type", "application/json")
|
|
if self.cookie:
|
|
req.add_header("Cookie", self.cookie)
|
|
def _json(body: str):
|
|
"""Rutele /api/ intorc JSON; paginile intorc HTML — nu esuam pe ele."""
|
|
try:
|
|
return json.loads(body) if body else {}
|
|
except ValueError:
|
|
return {"_html": body}
|
|
|
|
try:
|
|
with urllib.request.urlopen(req, timeout=10) as r:
|
|
sc = r.headers.get("Set-Cookie")
|
|
if sc:
|
|
self.cookie = sc.split(";", 1)[0]
|
|
return r.status, _json(r.read().decode()), r
|
|
except urllib.error.HTTPError as e:
|
|
return e.code, _json(e.read().decode()), e
|
|
|
|
def login(self, token="secret-de-test"):
|
|
return self.call("/api/auth/login", {"token": token})
|
|
|
|
try:
|
|
yield Client()
|
|
finally:
|
|
srv.shutdown()
|
|
srv.server_close()
|
|
api.reset_token_cache()
|
|
|
|
|
|
def _scrie_stare(state_dir, threads: dict, cost: float = 1.5):
|
|
(state_dir / "state.json").write_text(json.dumps({
|
|
"version": 1,
|
|
"threads": threads,
|
|
"cost": {"day": "2026-08-30", "usd": cost},
|
|
}), encoding="utf-8")
|
|
|
|
|
|
# --- autentificare ----------------------------------------------------------
|
|
|
|
def test_api_fara_cookie_da_401(server):
|
|
assert server.call("/api/status")[0] == 401
|
|
|
|
|
|
def test_index_fara_cookie_redirectioneaza_la_login(server):
|
|
status, data, resp = server.call("/")
|
|
# urllib urmareste redirectul singur: ajungem pe pagina de login, nu pe index
|
|
assert status == 200
|
|
assert resp.url.endswith("/login.html")
|
|
assert "DASHBOARD_TOKEN" in data["_html"]
|
|
|
|
|
|
def test_token_gresit_e_refuzat(server):
|
|
assert server.login("gresit")[0] == 401
|
|
assert server.call("/api/status")[0] == 401
|
|
|
|
|
|
def test_login_apoi_status(server):
|
|
assert server.login()[0] == 200
|
|
status, data, _ = server.call("/api/status")
|
|
assert status == 200
|
|
assert data["service"]["unit"] == api.SERVICE
|
|
assert data["service"]["restarts"] == 3
|
|
assert data["service"]["memory_bytes"] == 1048576
|
|
|
|
|
|
def test_logout_invalideaza_cookie(server):
|
|
server.login()
|
|
server.call("/api/auth/logout", {})
|
|
server.cookie = "dashboard="
|
|
assert server.call("/api/status")[0] == 401
|
|
|
|
|
|
def test_token_lipsa_din_env_nu_deschide_dashboardul(state_dir, monkeypatch):
|
|
"""Fara DASHBOARD_TOKEN se genereaza unul aleator, nu se sare peste auth."""
|
|
(state_dir / "env").write_text("", encoding="utf-8")
|
|
config.reload(state_dir)
|
|
api.reset_token_cache()
|
|
try:
|
|
tok = api.dashboard_token()
|
|
assert len(tok) >= 20
|
|
assert api.dashboard_token() == tok # stabil in cadrul procesului
|
|
finally:
|
|
api.reset_token_cache()
|
|
|
|
|
|
# --- control de serviciu ----------------------------------------------------
|
|
|
|
def test_actiune_necunoscuta_e_respinsa(server):
|
|
server.login()
|
|
status, data, _ = server.call("/api/service", {"action": "mask"})
|
|
assert status == 400
|
|
assert not data["ok"]
|
|
|
|
|
|
def test_nu_se_poate_alege_unitatea_din_request(server, systemctl_fals):
|
|
"""Chiar daca cererea cere alt unit, se actioneaza tot pe puntea Discord."""
|
|
server.login()
|
|
server.call("/api/service", {"action": "restart", "service": "ssh.service",
|
|
"unit": "ssh.service"})
|
|
actiuni = [a for a in systemctl_fals.apeluri if a[0] == "restart"]
|
|
assert actiuni == [("restart", api.SERVICE)]
|
|
|
|
|
|
def test_restart_blocat_de_tur_in_zbor(server, state_dir, systemctl_fals):
|
|
server.login()
|
|
_scrie_stare(state_dir, {"111": {"inflight": {"turn_id": "t1", "started_at": 1.0}}})
|
|
status, data, _ = server.call("/api/service", {"action": "restart"})
|
|
assert status == 409
|
|
assert data["inflight"] == ["111"]
|
|
assert not [a for a in systemctl_fals.apeluri if a[0] == "restart"]
|
|
|
|
|
|
def test_restart_cu_force_trece_peste_tur(server, state_dir, systemctl_fals):
|
|
server.login()
|
|
_scrie_stare(state_dir, {"111": {"inflight": {"turn_id": "t1", "started_at": 1.0}}})
|
|
status, data, _ = server.call("/api/service", {"action": "restart", "force": True})
|
|
assert status == 200 and data["ok"]
|
|
assert ("restart", api.SERVICE) in systemctl_fals.apeluri
|
|
|
|
|
|
def test_start_nu_cere_force(server, state_dir, systemctl_fals):
|
|
"""`start` nu poate intrerupe nimic, deci nu are de ce sa fie blocat."""
|
|
server.login()
|
|
_scrie_stare(state_dir, {"111": {"inflight": {"turn_id": "t1", "started_at": 1.0}}})
|
|
assert server.call("/api/service", {"action": "start"})[0] == 200
|
|
|
|
|
|
def test_esecul_systemctl_ajunge_la_client(server, monkeypatch):
|
|
server.login()
|
|
monkeypatch.setattr(api, "_sysctl",
|
|
lambda *a, timeout=30.0: _cp(rc=1, stderr="Unit not found"))
|
|
status, data, _ = server.call("/api/service", {"action": "restart"})
|
|
assert status == 500 and "Unit not found" in data["error"]
|
|
|
|
|
|
# --- stare, jurnale, diagnostic --------------------------------------------
|
|
|
|
def test_threads_view_marcheaza_tur_in_zbor(state_dir):
|
|
_scrie_stare(state_dir, {
|
|
"a": {"cwd": "/workspace/x", "model": "sonnet", "cost_usd_total": 2.5,
|
|
"last_active": 10, "inflight": {"turn_id": "t"}},
|
|
"b": {"cwd": "/workspace/y", "model": "opus", "last_active": 20},
|
|
})
|
|
view = api.threads_view(api.read_state())
|
|
assert [t["thread_id"] for t in view] == ["b", "a"] # sortare desc dupa activitate
|
|
assert view[1]["inflight"] and not view[0]["inflight"]
|
|
assert view[1]["cost_usd"] == 2.5
|
|
|
|
|
|
def test_state_corupt_nu_arunca(state_dir):
|
|
(state_dir / "state.json").write_text("{ nu e json", encoding="utf-8")
|
|
assert api.read_state() == {}
|
|
assert api.threads_view(api.read_state()) == []
|
|
|
|
|
|
def test_logs_taie_si_plafoneaza(server, state_dir):
|
|
server.login()
|
|
(config.LOG_DIR).mkdir(parents=True, exist_ok=True)
|
|
api.bot_log().write_text("\n".join(f"linia {i}" for i in range(500)), encoding="utf-8")
|
|
_, data, _ = server.call("/api/logs?lines=5")
|
|
assert data["lines"] == [f"linia {i}" for i in range(495, 500)]
|
|
|
|
|
|
def test_logs_fisier_inexistent(server):
|
|
server.login()
|
|
_, data, _ = server.call("/api/logs?file=infra")
|
|
assert "nu exista" in data["lines"][0]
|
|
|
|
|
|
def test_doctor_prinde_deny_ul_care_taie_ssh(server, state_dir):
|
|
"""Regresia din 2026-08-30: Bash(ssh:*) in deny a taiat accesul la infra."""
|
|
server.login()
|
|
(state_dir / "bot-settings.json").write_text(json.dumps(
|
|
{"permissions": {"deny": ["Bash(ssh:*)", "Bash(qm destroy:*)"]}}), encoding="utf-8")
|
|
_, data, _ = server.call("/api/doctor")
|
|
check = next(c for c in data["checks"] if "deny" in c["name"])
|
|
assert not check["pass"] and "Bash(ssh:*)" in check["detail"]
|
|
|
|
|
|
def test_doctor_accepta_deny_ul_curatat(server, state_dir):
|
|
server.login()
|
|
(state_dir / "bot-settings.json").write_text(json.dumps(
|
|
{"permissions": {"deny": ["Bash(qm destroy:*)"]}}), encoding="utf-8")
|
|
_, data, _ = server.call("/api/doctor")
|
|
check = next(c for c in data["checks"] if "deny" in c["name"])
|
|
assert check["pass"]
|
|
|
|
|
|
# --- aprobari ---------------------------------------------------------------
|
|
|
|
def _cerere(state_dir, rid="abc123", status="pending"):
|
|
config.APPROVALS_DIR.mkdir(parents=True, exist_ok=True)
|
|
(config.APPROVALS_DIR / f"{rid}.json").write_text(json.dumps({
|
|
"request_id": rid, "thread_id": "111", "tool_name": "Bash",
|
|
"command": "rm -rf /var/tmp/x", "rule": "rm_recursiv",
|
|
"reason": "stergere recursiva", "created_at": time.time(),
|
|
"expires_at": time.time() + 300, "status": status,
|
|
}), encoding="utf-8")
|
|
|
|
|
|
def test_approvals_arata_doar_pending(server, state_dir):
|
|
server.login()
|
|
_cerere(state_dir, "aaa", "pending")
|
|
_cerere(state_dir, "bbb", "allow")
|
|
_, data, _ = server.call("/api/approvals")
|
|
assert [a["request_id"] for a in data["approvals"]] == ["aaa"]
|
|
assert data["approvals"][0]["expires_in"] > 0
|
|
|
|
|
|
def test_status_numara_aprobarile(server, state_dir):
|
|
server.login()
|
|
_cerere(state_dir, "aaa")
|
|
_, data, _ = server.call("/api/status")
|
|
assert data["pending_approvals"] == 1
|
|
|
|
|
|
def test_decizie_invalida_e_respinsa(server):
|
|
server.login()
|
|
assert server.call("/api/approvals/decide",
|
|
{"request_id": "aaa", "decision": "poate"})[0] == 400
|
|
|
|
|
|
def test_request_id_cu_traversare_e_respins(server):
|
|
server.login()
|
|
assert server.call("/api/approvals/decide",
|
|
{"request_id": "../../etc/passwd", "decision": "allow"})[0] == 400
|
|
|
|
|
|
def test_decizia_ajunge_in_fisierul_cererii(server, state_dir):
|
|
server.login()
|
|
_cerere(state_dir, "aaa")
|
|
status, data, _ = server.call("/api/approvals/decide",
|
|
{"request_id": "aaa", "decision": "allow"})
|
|
assert status == 200 and data["ok"]
|
|
scris = json.loads((config.APPROVALS_DIR / "aaa.json").read_text())
|
|
assert scris["status"] == "allow"
|
|
|
|
|
|
def test_cerere_inexistenta_da_404(server):
|
|
server.login()
|
|
assert server.call("/api/approvals/decide",
|
|
{"request_id": "nuexista", "decision": "allow"})[0] == 404
|
|
|
|
|
|
# --- rute -------------------------------------------------------------------
|
|
|
|
def test_ruta_necunoscuta(server):
|
|
server.login()
|
|
assert server.call("/api/nope")[0] == 404
|
|
assert server.call("/api/nope", {})[0] == 404
|