""" Tests fuer die in dieser Session ("Phase 9") umgesetzten Punkte: 1) Bugfix "Host-Key ermitteln" 500 -> discover_and_store_host_key() fehlgeschlagene Verbindungen wurden bisher NICHT abgefangen (siehe app/ssh_proxy/proxy.py); der Endpunkt muss jetzt 502 (echter Verbindungsfehler) statt eines unbehandelten 500 liefern, und bei einem Fingerprint trotz Auth-Fehler NACH dem Key-Exchange weiterhin 200. (Punkt 2 "Login-Verlauf entfernen" und Punkt 5 "Hostgruppen/Server im Menue trennen" sind reine Admin-UI-Aenderungen ohne eigenen Endpunkt -- dafuer siehe templates/admin.html + static/js/admin.js, keine Backend-Tests noetig/moeglich.) 3) Verbindungslog (Live-Tail) -- app/security/log_stream.py ist reine In-Process-Logik (Ring-Buffer + Pub/Sub) ohne DB-/HTTP-Abhaengigkeit und wird NICHT hier, sondern eigenstaendig verifiziert (siehe verify_migrations.py/verify_hostkey_fix.py-Analoga aus der Sandbox-Verifikation dieser Session); ein WebSocket-Test wuerde einen echten ASGI-WS-Client benoetigen, den dieses Testsetup (httpx ASGITransport, kein WS-Support) nicht bietet. 4) Superadmin-'Sessionview': GET /admin/sessions, POST /admin/sessions/{id}/terminate, GET /admin/sessions/{id}/recording -- alle require_global_admin (auch Mandanten-Admins muessen 403 bekommen). 6) "Credentials ins RBAC-Modell": neue Rollen credentials_view/ credentials_manage (Migration 0008) erlauben NICHT-Admins gezielten Zugriff auf Zugangsdaten-Endpunkte fuer Hosts ihrer Hostgruppe. """ from __future__ import annotations import pytest async def _create_user(conn, username: str, password: str, *, is_admin: bool = False) -> int: from app.security.passwords import hash_password cursor = await conn.execute( "INSERT INTO users (username, password_hash, is_admin, must_change_password) " "VALUES (?, ?, ?, 0)", (username, hash_password(password), int(is_admin)), ) await conn.commit() return cursor.lastrowid async def _login_full(client, username: str, password: str) -> str: import pyotp resp = await client.post("/auth/login", json={"username": username, "password": password}) assert resp.status_code == 200, resp.text pending = resp.json()["pending_token"] resp = await client.post("/auth/totp/enroll/start", json={"pending_token": pending}) assert resp.status_code == 200, resp.text provisioning_uri = resp.json()["provisioning_uri"] secret = dict(part.split("=") for part in provisioning_uri.split("?", 1)[1].split("&"))["secret"] code = pyotp.TOTP(secret).now() resp = await client.post("/auth/totp/enroll/confirm", json={"pending_token": pending, "code": code}) assert resp.status_code == 200, resp.text return resp.cookies.get("jh_session") async def _setup_hostgroup_and_host(conn, *, group_name="p9-group", hostname="p9-host"): cursor = await conn.execute("INSERT INTO host_groups (name) VALUES (?)", (group_name,)) hg_id = cursor.lastrowid cursor = await conn.execute( "INSERT INTO hosts (host_group_id, hostname, address, protocol, port, os_type) " "VALUES (?, ?, '10.9.0.1', 'ssh', 22, 'linux')", (hg_id, hostname), ) await conn.commit() return hg_id, cursor.lastrowid # --------------------------------------------------------------------------- # 1) Host-Key ermitteln: kein 500 mehr bei Verbindungsfehlern # --------------------------------------------------------------------------- @pytest.mark.asyncio async def test_discover_host_key_connection_failure_returns_502_not_500(client, monkeypatch): from app.db import get_db import app.admin.routes as admin_routes from app.ssh_proxy.proxy import HostKeyDiscoveryError conn = get_db() await _create_user(conn, "hk_admin", "Correct-Horse-Battery-Staple-K1", is_admin=True) hg_id, host_id = await _setup_hostgroup_and_host(conn, group_name="hk-group", hostname="hk-host") await _login_full(client, "hk_admin", "Correct-Horse-Battery-Staple-K1") async def _boom(conn, host_id, *, admin_user_id): raise HostKeyDiscoveryError(host_id, "Connection refused") monkeypatch.setattr(admin_routes, "discover_and_store_host_key", _boom) resp = await client.post(f"/admin/hosts/{host_id}/discover-host-key", json={}) # Vorher: unbehandelte Exception -> 500. Jetzt: sauber gemappt auf 502. assert resp.status_code == 502, resp.text assert "Connection refused" in resp.json()["detail"] @pytest.mark.asyncio async def test_discover_host_key_success_after_auth_failure_past_kex(client, monkeypatch): """Simuliert den eigentlichen Bug-Fall: Key-Exchange erfolgreich (Fingerprint erfasst), Authentifizierung schlaegt DANACH fehl -- muss trotzdem als Erfolg gemeldet werden (siehe verify_hostkey_fix.py fuer die isolierte Kontrollfluss-Verifikation der proxy.py-Logik selbst).""" from app.db import get_db import app.admin.routes as admin_routes conn = get_db() await _create_user(conn, "hk_admin2", "Correct-Horse-Battery-Staple-K2", is_admin=True) hg_id, host_id = await _setup_hostgroup_and_host(conn, group_name="hk-group2", hostname="hk-host2") await _login_full(client, "hk_admin2", "Correct-Horse-Battery-Staple-K2") async def _fake_discover(conn, host_id, *, admin_user_id): return "SHA256:fake-fingerprint-after-auth-failure" monkeypatch.setattr(admin_routes, "discover_and_store_host_key", _fake_discover) resp = await client.post(f"/admin/hosts/{host_id}/discover-host-key", json={}) assert resp.status_code == 200, resp.text assert resp.json()["fingerprint"] == "SHA256:fake-fingerprint-after-auth-failure" # --------------------------------------------------------------------------- # 4) Sessionview (nur Super-Admin) # --------------------------------------------------------------------------- @pytest.mark.asyncio async def test_sessions_list_requires_global_admin_not_tenant_admin(client): from app.db import get_db conn = get_db() superadmin_id = await _create_user(conn, "sv_super", "Correct-Horse-Battery-Staple-S1", is_admin=True) tenant_admin_id = await _create_user(conn, "sv_tenant", "Correct-Horse-Battery-Staple-S2") await conn.execute( "INSERT INTO tenant_admins (user_id, tenant_id) VALUES (?, 1)", (tenant_admin_id,) ) await conn.commit() await _login_full(client, "sv_tenant", "Correct-Horse-Battery-Staple-S2") resp = await client.get("/admin/sessions") assert resp.status_code == 403, resp.text client.cookies.clear() await _login_full(client, "sv_super", "Correct-Horse-Battery-Staple-S1") resp = await client.get("/admin/sessions") assert resp.status_code == 200, resp.text assert resp.json() == [] @pytest.mark.asyncio async def test_sessions_list_and_terminate_and_recording(client): from app.db import get_db conn = get_db() admin_id = await _create_user(conn, "sv_super2", "Correct-Horse-Battery-Staple-S3", is_admin=True) user_id = await _create_user(conn, "sv_user", "Correct-Horse-Battery-Staple-S4") hg_id, host_id = await _setup_hostgroup_and_host(conn, group_name="sv-group", hostname="sv-host") cursor = await conn.execute( "INSERT INTO sessions (user_id, host_id, protocol, client_ip) VALUES (?, ?, 'ssh', '9.9.9.9')", (user_id, host_id), ) session_id = cursor.lastrowid await conn.commit() await _login_full(client, "sv_super2", "Correct-Horse-Battery-Staple-S3") resp = await client.get("/admin/sessions?active_only=true") assert resp.status_code == 200, resp.text rows = resp.json() assert len(rows) == 1 assert rows[0]["id"] == session_id assert rows[0]["username"] == "sv_user" assert rows[0]["hostname"] == "sv-host" assert rows[0]["is_active"] is True # Diese Sitzung wurde nur direkt in der DB angelegt (kein echter # laufender WS-Task) -> nicht in app/security/active_sessions.py # registriert -> darf NICHT als 'killable' gemeldet werden. assert rows[0]["killable"] is False assert rows[0]["has_recording"] is False # 'Beenden' muss sauber 409 liefern statt eine KeyError/AttributeError zu # werfen, wenn die Sitzung nicht (mehr) auf diesem Prozess laeuft. resp = await client.post(f"/admin/sessions/{session_id}/terminate", json={}) assert resp.status_code == 409, resp.text resp = await client.get(f"/admin/sessions/{session_id}/recording") assert resp.status_code == 404, resp.text resp = await client.post("/admin/sessions/999999/terminate", json={}) assert resp.status_code == 404, resp.text # --------------------------------------------------------------------------- # 6) Credentials ins RBAC-Modell # --------------------------------------------------------------------------- @pytest.mark.asyncio async def test_credentials_manage_role_grants_non_admin_write_access(client): from app.db import get_db conn = get_db() admin_id = await _create_user(conn, "cr_admin", "Correct-Horse-Battery-Staple-C1", is_admin=True) holder_id = await _create_user(conn, "cr_holder", "Correct-Horse-Battery-Staple-C2") other_id = await _create_user(conn, "cr_other", "Correct-Horse-Battery-Staple-C3") hg_id, host_id = await _setup_hostgroup_and_host(conn, group_name="cr-group", hostname="cr-host") await _login_full(client, "cr_admin", "Correct-Horse-Battery-Staple-C1") resp = await client.post( "/admin/roles/grant", json={"user_id": holder_id, "host_group_id": hg_id, "role_names": ["credentials_manage"]}, ) assert resp.status_code == 200, resp.text # Rolleninhaber (kein Admin!) darf Zugangsdaten lesen UND setzen. client.cookies.clear() await _login_full(client, "cr_holder", "Correct-Horse-Battery-Staple-C2") resp = await client.get(f"/admin/hosts/{host_id}/credentials") assert resp.status_code == 200, resp.text assert resp.json()["rdp_credentials_set"] is False resp = await client.put(f"/admin/hosts/{host_id}/rdp-credentials", json={"password": "s3hr-geheim!!"}) assert resp.status_code == 200, resp.text resp = await client.get(f"/admin/hosts/{host_id}/credentials") assert resp.status_code == 200, resp.text assert resp.json()["rdp_credentials_set"] is True # Ein User OHNE diese Rolle bleibt weiterhin ausgesperrt. client.cookies.clear() await _login_full(client, "cr_other", "Correct-Horse-Battery-Staple-C3") resp = await client.get(f"/admin/hosts/{host_id}/credentials") assert resp.status_code == 403, resp.text resp = await client.put(f"/admin/hosts/{host_id}/rdp-credentials", json={"password": "andere-1234"}) assert resp.status_code == 403, resp.text @pytest.mark.asyncio async def test_credentials_view_role_is_read_only(client): from app.db import get_db conn = get_db() await _create_user(conn, "cv_admin", "Correct-Horse-Battery-Staple-C4", is_admin=True) viewer_id = await _create_user(conn, "cv_viewer", "Correct-Horse-Battery-Staple-C5") hg_id, host_id = await _setup_hostgroup_and_host(conn, group_name="cv-group", hostname="cv-host") await _login_full(client, "cv_admin", "Correct-Horse-Battery-Staple-C4") resp = await client.post( "/admin/roles/grant", json={"user_id": viewer_id, "host_group_id": hg_id, "role_names": ["credentials_view"]}, ) assert resp.status_code == 200, resp.text client.cookies.clear() await _login_full(client, "cv_viewer", "Correct-Horse-Battery-Staple-C5") resp = await client.get(f"/admin/hosts/{host_id}/credentials") assert resp.status_code == 200, resp.text # 'credentials_view' allein darf NICHT schreiben. resp = await client.put(f"/admin/hosts/{host_id}/rdp-credentials", json={"password": "nope-12345"}) assert resp.status_code == 403, resp.text @pytest.mark.asyncio async def test_catalog_hosts_reports_can_view_credentials_flag(client): from app.db import get_db conn = get_db() await _create_user(conn, "cc_admin", "Correct-Horse-Battery-Staple-C6", is_admin=True) user_id = await _create_user(conn, "cc_user", "Correct-Horse-Battery-Staple-C7") hg_id, host_id = await _setup_hostgroup_and_host(conn, group_name="cc-group", hostname="cc-host") await _login_full(client, "cc_admin", "Correct-Horse-Battery-Staple-C6") await client.post( "/admin/roles/grant", json={"user_id": user_id, "host_group_id": hg_id, "role_names": ["ssh_connect", "credentials_view"]}, ) client.cookies.clear() await _login_full(client, "cc_user", "Correct-Horse-Battery-Staple-C7") resp = await client.get("/catalog/hosts") assert resp.status_code == 200, resp.text hosts = resp.json() assert len(hosts) == 1 assert hosts[0]["can_view_credentials"] is True