""" Tests fuer Phase 13 (diese Session -- Sammel-Feedback aus dem Live-Test): 1) RDP-Zugangsdaten muessen loeschbar sein: DELETE /admin/hosts/{id}/rdp-credentials (bereits vorhandener Endpunkt -- wird hier nur mitverifiziert, da neu vom Admin-UI aus verlinkt). 2) SSH-Passwort als Alternative zum SSH-Key (Migration 0011): PUT/DELETE /admin/hosts/{id}/ssh-password, Spiegelung in GET /admin/hosts/{id} UND im schlanken GET /admin/hosts/{id}/credentials (RBAC-faehig). 5) "Besitzer (User-ID)"-Feld bei SSH-Keys entfernt: owner_user_id darf in KEINER Antwort von POST/GET/PUT /admin/ssh-keys(...) mehr auftauchen. 7) "Rotieren" -> klareres Wording ist eine reine UI-Textaenderung (templates/admin.html/static/js/admin.js) ohne eigenen Endpunkt -- der Rotations-MECHANISMUS selbst ist bereits durch test_phase12.py und die update_ssh_key-Tests hier mitabgedeckt. 8) SSH-Key automatisch generieren: POST /admin/ssh-keys/generate fuer alle drei unterstuetzten Typen sowie 400 fuer den nicht unterstuetzten Typ ("ca-cert" kann nicht "generiert" werden -- ein CA-Zertifikat setzt eine bereits existierende eigene CA voraus). 9) Troubleshooting bei fehlgeschlagenen Verbindungen ist ein reiner Frontend-/WebSocket-Close-Reason-Fix (static/js/terminal.js, static/js/rdp.js, app/ssh_proxy/terminal_ws.py::_reject) ohne eigenen HTTP-Endpunkt -- dafuer siehe die manuelle Sandbox-Verifikation (HTML/JS-Konsistenzpruefung) statt eines pytest-Tests hier. 10) Strg+Alt+Entf-Button ist eine reine Frontend-Aenderung (templates/rdp.html/static/js/rdp.js) ohne Backend-Anteil. Sitzungs-Wiedergabe (volle grafische RDP-Wiedergabe + SSH-Textwiedergabe): GET /admin/sessions/{id}/recording/entries -- nur require_global_admin (Mandanten-Admins bekommen 403, genau wie beim bereits bestehenden GET /admin/sessions/{id}/recording), liefert die entschluesselten/rohen Eintraege NICHT den Klartext einer Passphrase o.ae., sondern ausschliesslich die bereits im Klartext aufgezeichneten Terminal-/Guacamole-Stroeme (siehe Konzept 6.5: die Aufzeichnung selbst enthaelt nie Anmeldedaten, sondern nur Ein-/Ausgabe der Sitzung) -- UND erzeugt bei jedem Aufruf einen eigenen Audit-Log-Eintrag ('session_recording_viewed'), weil das Ansehen einer Mitschnitt-Wiedergabe selbst ein auditierenswerter Zugriff ist. Wie in test_phase9.py/test_phase12.py: echte SQLite-DB + echte Migrationen ueber den httpx-'client'-Fixture-Pfad (conftest.py), kein laufender SSH-/RDP- Server noetig. Diese Datei setzt (wie test_phase9.py) fastapi/httpx/aiosqlite/ asyncssh voraus und wurde in DIESER Sandbox mangels PyPI-Netzzugang nicht mit echtem pytest ausgefuehrt (siehe Sandbox-Einschraenkung, Projektnotiz) -- Aufbau/Assertions wurden stattdessen von Hand gegen den tatsaechlichen Code in app/admin/routes.py, app/models/schemas.py und app/recordings/recorder.py geprueft. Bitte lokal mit `pytest -q` laufen lassen. """ from __future__ import annotations import json 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="p13-group", hostname="p13-host", protocol="ssh"): 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.13.0.1', ?, ?, ?)", (hg_id, hostname, protocol, 22 if protocol == "ssh" else 3389, "linux" if protocol == "ssh" else "windows"), ) await conn.commit() return hg_id, cursor.lastrowid # --------------------------------------------------------------------------- # 1) RDP-Zugangsdaten: Loeschen # --------------------------------------------------------------------------- @pytest.mark.asyncio async def test_rdp_credentials_delete(client): from app.db import get_db conn = get_db() await _create_user(conn, "rd_admin", "Correct-Horse-Battery-Staple-R1", is_admin=True) hg_id, host_id = await _setup_hostgroup_and_host(conn, group_name="rd-group", hostname="rd-host", protocol="rdp") await _login_full(client, "rd_admin", "Correct-Horse-Battery-Staple-R1") resp = await client.put(f"/admin/hosts/{host_id}/rdp-credentials", json={"username": "Administrator", "password": "s3hr-geheim!!"}) assert resp.status_code == 200, resp.text resp = await client.get(f"/admin/hosts/{host_id}/credentials") assert resp.json()["rdp_credentials_set"] is True resp = await client.delete(f"/admin/hosts/{host_id}/rdp-credentials") 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 False assert resp.json()["rdp_credentials_username"] is None # --------------------------------------------------------------------------- # 2) SSH-Passwort als Alternative zum SSH-Key # --------------------------------------------------------------------------- @pytest.mark.asyncio async def test_ssh_password_credentials_set_and_delete(client): from app.db import get_db conn = get_db() await _create_user(conn, "sp_admin", "Correct-Horse-Battery-Staple-P1", is_admin=True) hg_id, host_id = await _setup_hostgroup_and_host(conn, group_name="sp-group", hostname="sp-host") await _login_full(client, "sp_admin", "Correct-Horse-Battery-Staple-P1") # Leerer Benutzername wird abgelehnt (analog RDP/SSH-Key). resp = await client.put(f"/admin/hosts/{host_id}/ssh-password", json={"username": " ", "password": "geheim123"}) assert resp.status_code == 400, resp.text resp = await client.put(f"/admin/hosts/{host_id}/ssh-password", json={"username": "l4u", "password": "sehr-geheimes-passwort"}) assert resp.status_code == 200, resp.text # Spiegelung im vollen Host-Datensatz (Admin-only)... resp = await client.get(f"/admin/hosts/{host_id}") assert resp.status_code == 200, resp.text body = resp.json() assert body["ssh_password_credentials_set"] is True assert body["ssh_password_credentials_username"] == "l4u" # ...UND im schlanken RBAC-faehigen Endpunkt. resp = await client.get(f"/admin/hosts/{host_id}/credentials") assert resp.status_code == 200, resp.text body = resp.json() assert body["ssh_password_credentials_set"] is True assert body["ssh_password_credentials_username"] == "l4u" # Der Klartext des Passworts wird in KEINER Antwort zurueckgegeben. assert "password" not in json.dumps(body) assert "sehr-geheimes-passwort" not in json.dumps(body) # Erneutes Setzen ist ein Upsert (ON CONFLICT), kein Duplikat/Fehler. resp = await client.put(f"/admin/hosts/{host_id}/ssh-password", json={"username": "root", "password": "anderes-passwort-123"}) assert resp.status_code == 200, resp.text resp = await client.get(f"/admin/hosts/{host_id}/credentials") assert resp.json()["ssh_password_credentials_username"] == "root" resp = await client.delete(f"/admin/hosts/{host_id}/ssh-password") assert resp.status_code == 200, resp.text resp = await client.get(f"/admin/hosts/{host_id}/credentials") assert resp.json()["ssh_password_credentials_set"] is False assert resp.json()["ssh_password_credentials_username"] is None @pytest.mark.asyncio async def test_ssh_password_credentials_via_credentials_manage_role(client): """credentials_manage erlaubt Nicht-Admins auch das SSH-Passwort zu setzen (gleiche RBAC-Erweiterung wie fuer RDP/SSH-Key, siehe test_phase9.py).""" from app.db import get_db conn = get_db() admin_id = await _create_user(conn, "sp2_admin", "Correct-Horse-Battery-Staple-P2", is_admin=True) holder_id = await _create_user(conn, "sp2_holder", "Correct-Horse-Battery-Staple-P3") hg_id, host_id = await _setup_hostgroup_and_host(conn, group_name="sp2-group", hostname="sp2-host") await _login_full(client, "sp2_admin", "Correct-Horse-Battery-Staple-P2") 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 client.cookies.clear() await _login_full(client, "sp2_holder", "Correct-Horse-Battery-Staple-P3") resp = await client.put(f"/admin/hosts/{host_id}/ssh-password", json={"username": "l4u", "password": "abc12345"}) assert resp.status_code == 200, resp.text resp = await client.delete(f"/admin/hosts/{host_id}/ssh-password") assert resp.status_code == 200, resp.text # --------------------------------------------------------------------------- # 5) "Besitzer"-Feld ist vollstaendig entfernt # --------------------------------------------------------------------------- @pytest.mark.asyncio async def test_ssh_key_owner_field_is_gone(client): from app.db import get_db conn = get_db() await _create_user(conn, "ow_admin", "Correct-Horse-Battery-Staple-O1", is_admin=True) await _login_full(client, "ow_admin", "Correct-Horse-Battery-Staple-O1") resp = await client.post( "/admin/ssh-keys", json={ "label": "ohne-besitzer", "key_type": "ed25519", "username": "l4u", "private_key_pem": _ED25519_TEST_KEY_PEM, "public_key": "ssh-ed25519 AAAAtest", }, ) assert resp.status_code == 201, resp.text assert "owner_user_id" not in resp.json() key_id = resp.json()["id"] resp = await client.get("/admin/ssh-keys") assert resp.status_code == 200, resp.text rows = resp.json() assert len(rows) == 1 assert "owner_user_id" not in rows[0] # Die 9 tatsaechlich noch vorhandenen Felder (Admin-Tabelle hat 9 Spalten, # siehe static/js/admin.js::refreshSshKeys()). assert set(rows[0]) == { "id", "label", "key_type", "created_at", "rotated_at", "expires_at", "tenant_id", "tenant_name", "has_passphrase", "username", } | {"username"} # (Mengen sind idempotent -- nur zur Lesbarkeit doppelt genannt) resp = await client.put(f"/admin/ssh-keys/{key_id}", json={"label": "umbenannt", "owner_user_id": 999}) assert resp.status_code == 200, resp.text # Ein unbekanntes Feld im Request wird von Pydantic (extra="ignore" per # Default) klaglos verworfen -- es darf insbesondere KEINEN Fehler geben # und keine owner_user_id-Spalte mehr existieren, in die geschrieben # werden koennte (siehe Migration -- das Feld wurde nie in einer eigenen # Migration entfernt, weil es nie eine eigene Spalte hatte, sondern nur # eine tote FK-Spalte namens owner_user_id in Migration 0001; das UPDATE # in update_ssh_key() adressiert sie nirgends mehr). resp = await client.get("/admin/ssh-keys") assert "owner_user_id" not in resp.json()[0] # --------------------------------------------------------------------------- # 8) SSH-Key automatisch generieren # --------------------------------------------------------------------------- @pytest.mark.asyncio @pytest.mark.parametrize("key_type", ["ed25519", "rsa-3072", "rsa-4096"]) async def test_generate_ssh_key_supported_types(client, key_type): from app.db import get_db conn = get_db() await _create_user(conn, f"gen_admin_{key_type.replace('-', '_')}", "Correct-Horse-Battery-Staple-G1", is_admin=True) await _login_full(client, f"gen_admin_{key_type.replace('-', '_')}", "Correct-Horse-Battery-Staple-G1") resp = await client.post("/admin/ssh-keys/generate", json={"key_type": key_type}) assert resp.status_code == 200, resp.text body = resp.json() assert body["key_type"] == key_type assert "PRIVATE KEY" in body["private_key_pem"] assert body["public_key"].startswith("ssh-") # Erzeugtes Material muss tatsaechlich als SSH-Key uebernehmbar sein -- # end-to-end-Beweis, dass "generieren" und "speichern" zusammenspielen. resp = await client.post( "/admin/ssh-keys", json={ "label": f"generiert-{key_type}", "key_type": key_type, "username": "l4u", "private_key_pem": body["private_key_pem"], "public_key": body["public_key"], }, ) assert resp.status_code == 201, resp.text @pytest.mark.asyncio async def test_generate_ssh_key_rejects_ca_cert(client): from app.db import get_db conn = get_db() await _create_user(conn, "gen_admin_ca", "Correct-Horse-Battery-Staple-G2", is_admin=True) await _login_full(client, "gen_admin_ca", "Correct-Horse-Battery-Staple-G2") resp = await client.post("/admin/ssh-keys/generate", json={"key_type": "ca-cert"}) # Pydantic weist den Literal-Wert schon vor dem Handler zurueck (422), # NICHT das im Handler behandelte generische ValueError->400 -- beides # bedeutet aber "kein 500 / keine stillschweigend falsche Antwort". assert resp.status_code == 422, resp.text # --------------------------------------------------------------------------- # Sitzungs-Wiedergabe: GET /admin/sessions/{id}/recording/entries # --------------------------------------------------------------------------- @pytest.mark.asyncio async def test_session_recording_entries_requires_global_admin(client): from app.db import get_db conn = get_db() # Mandanten-Admin (Tenant-gebunden ueber tenant_admins, kein globaler # Super-Admin) darf NICHT -- dieser Endpunkt liefert vollstaendigen # Klartext-Mitschnitt und ist bewusst auf require_global_admin # beschraenkt (siehe Docstring des Endpunkts in app/admin/routes.py). # Aufbau eines Mandanten-Admins wie in tests/test_tenants.py: ein # Super-Admin legt einen Mandanten an und traegt einen (sonst nicht # privilegierten) Benutzer als dessen Tenant-Admin ein. super_id = await _create_user(conn, "rec_bootstrap_super", "Correct-Horse-Battery-Staple-N0", is_admin=True) await _login_full(client, "rec_bootstrap_super", "Correct-Horse-Battery-Staple-N0") resp = await client.post("/admin/tenants", json={"name": "Kunde Rec"}) assert resp.status_code == 201, resp.text tenant_id = resp.json()["id"] tenant_admin_id = await _create_user(conn, "rec_tenant_admin", "Correct-Horse-Battery-Staple-N1") resp = await client.post(f"/admin/tenants/{tenant_id}/admins", json={"user_id": tenant_admin_id}) assert resp.status_code == 201, resp.text client.cookies.clear() hg_id, host_id = await _setup_hostgroup_and_host(conn, group_name="rec-group", hostname="rec-host") user_id = await _create_user(conn, "rec_user", "Correct-Horse-Battery-Staple-N2") 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, "rec_tenant_admin", "Correct-Horse-Battery-Staple-N1") resp = await client.get(f"/admin/sessions/{session_id}/recording/entries") assert resp.status_code == 403, resp.text @pytest.mark.asyncio async def test_session_recording_entries_ssh_playback_and_audit(client): from app.db import get_db from app.recordings.recorder import SessionRecorder conn = get_db() await _create_user(conn, "rec_super", "Correct-Horse-Battery-Staple-N3", is_admin=True) user_id = await _create_user(conn, "rec_user2", "Correct-Horse-Battery-Staple-N4") hg_id, host_id = await _setup_hostgroup_and_host(conn, group_name="rec2-group", hostname="rec2-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() # Echter SessionRecorder statt Fake -- schreibt dieselbe hash-verkettete # JSONL-Datei, die terminal_ws.py/ws_tunnel.py im Betrieb erzeugen. recorder = SessionRecorder(session_id) recorder.record("input", "ls\n") recorder.record("output", "total 0\n") recorder.record("output", "$ ") recorder.close() await conn.execute("UPDATE sessions SET recording_path = ? WHERE id = ?", (str(recorder.path), session_id)) await conn.commit() await _login_full(client, "rec_super", "Correct-Horse-Battery-Staple-N3") resp = await client.get(f"/admin/sessions/{session_id}/recording/entries") assert resp.status_code == 200, resp.text body = resp.json() assert body["session_id"] == session_id assert body["protocol"] == "ssh" assert body["verified"] is True assert len(body["entries"]) == 3 assert [e["dir"] for e in body["entries"]] == ["input", "output", "output"] assert body["entries"][1]["data"] == "total 0\n" # Jeder Aufruf erzeugt einen eigenen, gesonderten Audit-Log-Eintrag. resp = await client.get("/admin/audit-log") assert resp.status_code == 200, resp.text events = [e["event_type"] for e in resp.json()] assert "session_recording_viewed" in events @pytest.mark.asyncio async def test_session_recording_entries_404_without_recording(client): from app.db import get_db conn = get_db() await _create_user(conn, "rec_super2", "Correct-Horse-Battery-Staple-N5", is_admin=True) user_id = await _create_user(conn, "rec_user3", "Correct-Horse-Battery-Staple-N6") hg_id, host_id = await _setup_hostgroup_and_host(conn, group_name="rec3-group", hostname="rec3-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, "rec_super2", "Correct-Horse-Battery-Staple-N5") resp = await client.get(f"/admin/sessions/{session_id}/recording/entries") assert resp.status_code == 404, resp.text # --------------------------------------------------------------------------- # Testschluessel (nur fuer diese Datei -- kein echtes Geheimnis). # --------------------------------------------------------------------------- _ED25519_TEST_KEY_PEM = None def _generate_test_key_pem() -> str: from cryptography.hazmat.primitives import serialization from cryptography.hazmat.primitives.asymmetric import ed25519 return ed25519.Ed25519PrivateKey.generate().private_bytes( serialization.Encoding.PEM, serialization.PrivateFormat.OpenSSH, serialization.NoEncryption(), ).decode() _ED25519_TEST_KEY_PEM = _generate_test_key_pem()