56027e2e6c
Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01KddsXCqEbKMhdFJDYbAsx8
494 lines
17 KiB
Python
494 lines
17 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
|
|
|
|
|
|
# --- montare sub prefix (tailscale serve --set-path) ------------------------
|
|
|
|
@pytest.fixture()
|
|
def server_cu_prefix(server, state_dir):
|
|
"""Acelasi server, dar cu DASHBOARD_PREFIX=/claude in env."""
|
|
(state_dir / "env").write_text(
|
|
"DASHBOARD_TOKEN=secret-de-test\nDASHBOARD_PREFIX=/claude\n", encoding="utf-8")
|
|
config.reload(state_dir)
|
|
return server
|
|
|
|
|
|
def test_prefixul_e_normalizat():
|
|
assert api.strip_prefix("/api/status") == "/api/status"
|
|
|
|
|
|
def test_rutele_merg_si_cu_prefix_si_fara(server_cu_prefix):
|
|
"""Proxy-ul taie prefixul, dar `curl` direct pe localhost nu — merg ambele."""
|
|
server_cu_prefix.login()
|
|
assert server_cu_prefix.call("/api/status")[0] == 200
|
|
assert server_cu_prefix.call("/claude/api/status")[0] == 200
|
|
|
|
|
|
def test_prefixul_gol_nu_taie_nimic(server):
|
|
server.login()
|
|
assert server.call("/api/status")[0] == 200
|
|
assert server.call("/claude/api/status")[0] == 404
|
|
|
|
|
|
def test_redirect_de_login_pastreaza_prefixul(server_cu_prefix):
|
|
"""Un `/login.html` absolut ar arunca browserul in radacina hostului."""
|
|
status, _, resp = server_cu_prefix.call("/claude/")
|
|
assert status == 200
|
|
assert resp.url.endswith("/claude/login.html")
|
|
|
|
|
|
def test_calea_fara_slash_final_e_redirectionata(server_cu_prefix):
|
|
"""`/claude` fara slash ar rezolva `api/status` la radacina hostului."""
|
|
_, _, resp = server_cu_prefix.call("/claude")
|
|
assert resp.url.endswith("/claude/login.html")
|
|
|
|
|
|
def test_static_servit_si_sub_prefix(server_cu_prefix):
|
|
server_cu_prefix.login()
|
|
status, data, _ = server_cu_prefix.call("/claude/static/app.css")
|
|
assert status == 200 and ".card" in data["_html"]
|
|
|
|
|
|
def test_base_href_pus_cand_exista_prefix(server_cu_prefix):
|
|
"""Proxy-ul taie prefixul, deci serverul nu poate sti daca browserul e la
|
|
`/claude` sau `/claude/`. `<base href>` face rezolvarea determinista."""
|
|
server_cu_prefix.login()
|
|
_, data, _ = server_cu_prefix.call("/")
|
|
assert '<base href="/claude/">' in data["_html"]
|
|
|
|
|
|
def test_base_href_si_pe_pagina_de_login(server_cu_prefix):
|
|
_, data, _ = server_cu_prefix.call("/login.html")
|
|
assert '<base href="/claude/">' in data["_html"]
|
|
|
|
|
|
def test_fara_prefix_nu_se_pune_base(server):
|
|
server.login()
|
|
_, data, _ = server.call("/")
|
|
assert "<base" not in data["_html"]
|
|
|
|
|
|
def test_paginile_nu_se_pun_in_cache(server_cu_prefix):
|
|
"""O copie veche ar purta un `base href` gresit si ar trimite cererile aiurea."""
|
|
server_cu_prefix.login()
|
|
_, _, resp = server_cu_prefix.call("/")
|
|
assert resp.headers.get("Cache-Control") == "no-store"
|
|
|
|
|
|
# --- mod fara autentificare (DASHBOARD_AUTH=off) ----------------------------
|
|
|
|
@pytest.fixture()
|
|
def server_fara_auth(server, state_dir):
|
|
(state_dir / "env").write_text(
|
|
"DASHBOARD_TOKEN=secret-de-test\nDASHBOARD_AUTH=off\n", encoding="utf-8")
|
|
config.reload(state_dir)
|
|
return server
|
|
|
|
|
|
def test_fara_auth_apiul_raspunde_fara_cookie(server_fara_auth):
|
|
status, data, _ = server_fara_auth.call("/api/status")
|
|
assert status == 200 and data["auth"] is False
|
|
|
|
|
|
def test_fara_auth_pagina_se_serveste_direct(server_fara_auth):
|
|
status, data, resp = server_fara_auth.call("/")
|
|
assert status == 200
|
|
assert "Punte Discord" in data["_html"]
|
|
assert not resp.url.endswith("login.html")
|
|
|
|
|
|
def test_fara_auth_login_html_duce_inapoi_la_panou(server_fara_auth):
|
|
_, _, resp = server_fara_auth.call("/login.html")
|
|
assert not resp.url.endswith("login.html")
|
|
|
|
|
|
def test_fara_auth_se_poate_actiona_pe_serviciu(server_fara_auth, systemctl_fals):
|
|
assert server_fara_auth.call("/api/service", {"action": "restart"})[0] == 200
|
|
assert ("restart", api.SERVICE) in systemctl_fals.apeluri
|
|
|
|
|
|
def test_valoare_necunoscuta_pentru_auth_nu_deschide_panoul(server, state_dir):
|
|
"""Doar off/none/0/false scot login-ul; orice altceva il pastreaza."""
|
|
(state_dir / "env").write_text(
|
|
"DASHBOARD_TOKEN=secret-de-test\nDASHBOARD_AUTH=da\n", encoding="utf-8")
|
|
config.reload(state_dir)
|
|
assert not api.auth_disabled()
|
|
assert server.call("/api/status")[0] == 401
|
|
|
|
|
|
# --- tab Infrastructura ------------------------------------------------------
|
|
|
|
def test_infra_run_id_inexistent(server):
|
|
server.login()
|
|
status, data, _ = server.call("/api/infra/run", {"id": "nu-exista", "dry_run": False})
|
|
assert status == 400
|
|
assert not data["ok"]
|
|
|
|
|
|
def test_infra_run_ignora_campurile_in_plus(server, monkeypatch):
|
|
"""Payload-ul poate purta cmd/host straine -- api.py trebuie sa apeleze
|
|
infra_actions.run doar cu (id, dry_run), restul se ignora."""
|
|
server.login()
|
|
apeluri = []
|
|
|
|
def fake_run(action_id, dry_run):
|
|
apeluri.append((action_id, dry_run))
|
|
return {"ok": True, "code": 200, "host": "pvemini"}
|
|
|
|
import infra_actions
|
|
monkeypatch.setattr(infra_actions, "run", fake_run)
|
|
status, data, _ = server.call("/api/infra/run", {
|
|
"id": "oprire", "dry_run": True, "cmd": "rm -rf /", "host": "x",
|
|
})
|
|
assert status == 200 and data["ok"]
|
|
assert apeluri == [("oprire", True)]
|
|
|
|
|
|
def test_infra_log_id_invalid(server):
|
|
server.login()
|
|
status, _, _ = server.call("/api/infra/log?id=../x")
|
|
assert status == 400
|
|
|
|
|
|
def test_identitatea_tailscale_e_doar_pentru_jurnal(server_fara_auth):
|
|
"""Antetul pus de `tailscale serve` se raporteaza, dar nu decide accesul:
|
|
pe localhost lipseste, si totusi cererea trece."""
|
|
assert api.tailnet_user({"Tailscale-User-Login": " ana@example.com "}) == "ana@example.com"
|
|
assert api.tailnet_user({}) == ""
|
|
_, data, _ = server_fara_auth.call("/api/status")
|
|
assert data["user"] == ""
|