""" Tests fuer die in "Phase 10" behobenen Ursachen dafuer, dass weder SSH- noch RDP-Sitzungen zustande kamen: 1) SSH: der hinterlegte private Schluessel war passphrasegeschuetzt. asyncssh.import_private_key() wurde ohne Passphrase aufgerufen und warf KeyImportError("Passphrase must be specified to import encrypted private keys") -- ein ValueError, KEIN asyncssh.Error, der deshalb an allen Fehlerbehandlungen vorbei aus der WS-Route hinauslief. Jetzt: Passphrase wird KEK-verschluesselt mitgespeichert (Migration 0009), beim Laden uebergeben und beim Anlegen/Rotieren/Nachtragen sofort geprueft (400). 2) RDP: load_host() selektierte rdp_username/rdp_domain/rdp_require_nla/ clipboard_enabled gar nicht -- build_rdp_params() baute daraus eine connect-Instruktion ohne Benutzernamen. Zusaetzlich war ignore-cert hart auf "false" verdrahtet (jetzt hosts.rdp_ignore_cert, Default an). 3) Guacamole-Protokoll: Laengenangaben zaehlen ZEICHEN, nicht Bytes -- ein Umlaut (etwa im RDP-Passwort) verschob sonst den gesamten Datenstrom. Die WebSocket-Routen selbst lassen sich mit diesem Testsetup (httpx ASGITransport, kein WS-Support) nicht end-to-end fahren; geprueft werden hier die Admin-Endpunkte, die DB-Schicht und die reine Protokoll-/Parameterlogik. """ from __future__ import annotations import pytest from cryptography.hazmat.primitives import serialization from cryptography.hazmat.primitives.asymmetric import ed25519 PASSPHRASE = "Testpassphrase-2026" def _plain_key_pem() -> str: key = ed25519.Ed25519PrivateKey.generate() return key.private_bytes( serialization.Encoding.PEM, serialization.PrivateFormat.OpenSSH, serialization.NoEncryption(), ).decode() def _encrypted_key_pem(passphrase: str = PASSPHRASE) -> str: """Passphrasegeschuetzter Schluessel. Bevorzugt das OpenSSH-Format (das, was `ssh-keygen` erzeugt und was der gemeldete Fehler betraf). Dessen Verschluesselung setzt das optionale bcrypt-Modul voraus; fehlt es, wird auf PKCS#8 ausgewichen -- asyncssh liest beide, und der hier gepruefte Codepfad (fehlende bzw. falsche Passphrase) ist fuer beide identisch.""" from cryptography.exceptions import UnsupportedAlgorithm key = ed25519.Ed25519PrivateKey.generate() enc = serialization.BestAvailableEncryption(passphrase.encode()) try: return key.private_bytes( serialization.Encoding.PEM, serialization.PrivateFormat.OpenSSH, enc ).decode() except UnsupportedAlgorithm: # pragma: no cover - haengt an der Umgebung return key.private_bytes( serialization.Encoding.PEM, serialization.PrivateFormat.PKCS8, enc ).decode() 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") # --------------------------------------------------------------------------- # 1) SSH-Schluessel mit Passphrase # --------------------------------------------------------------------------- @pytest.mark.asyncio async def test_create_ssh_key_rejects_encrypted_key_without_passphrase(client): from app.db import get_db conn = get_db() await _create_user(conn, "p10_admin1", "Correct-Horse-Battery-Staple-A1", is_admin=True) await _login_full(client, "p10_admin1", "Correct-Horse-Battery-Staple-A1") resp = await client.post("/admin/ssh-keys", json={ "label": "verschluesselt-ohne-passphrase", "private_key_pem": _encrypted_key_pem(), "public_key": "ssh-ed25519 AAAAC3NzaC1lZDI1NTE5 test", "key_type": "ed25519", }) # Vorher: 201 -- der Fehler fiel erst beim ersten Verbindungsversuch auf. assert resp.status_code == 400, resp.text assert "passphrasegeschuetzt" in resp.json()["detail"].lower() @pytest.mark.asyncio async def test_create_ssh_key_with_correct_passphrase_succeeds_and_is_stored_encrypted(client): from app.db import get_db from app.security.crypto import decrypt_secret conn = get_db() await _create_user(conn, "p10_admin2", "Correct-Horse-Battery-Staple-A2", is_admin=True) await _login_full(client, "p10_admin2", "Correct-Horse-Battery-Staple-A2") resp = await client.post("/admin/ssh-keys", json={ "label": "verschluesselt-mit-passphrase", "private_key_pem": _encrypted_key_pem(), "public_key": "ssh-ed25519 AAAAC3NzaC1lZDI1NTE5 test", "key_type": "ed25519", "passphrase": PASSPHRASE, }) assert resp.status_code == 201, resp.text key_id = resp.json()["id"] row = await (await conn.execute( "SELECT passphrase_enc FROM ssh_keys WHERE id = ?", (key_id,) )).fetchone() assert row[0] is not None, "Passphrase wurde nicht gespeichert" assert PASSPHRASE.encode() not in row[0], "Passphrase liegt im Klartext in der DB" assert decrypt_secret(row[0], associated_data=b"ssh_key_passphrase") == PASSPHRASE.encode() # Die Liste verraet nur, DASS eine Passphrase hinterlegt ist. listing = (await client.get("/admin/ssh-keys")).json() entry = next(k for k in listing if k["id"] == key_id) assert entry["has_passphrase"] is True assert "passphrase" not in entry, "Endpunkt darf die Passphrase selbst nie zurueckgeben" @pytest.mark.asyncio async def test_wrong_passphrase_is_rejected_and_correct_one_can_be_added_later(client): from app.db import get_db conn = get_db() await _create_user(conn, "p10_admin3", "Correct-Horse-Battery-Staple-A3", is_admin=True) await _login_full(client, "p10_admin3", "Correct-Horse-Battery-Staple-A3") pem = _encrypted_key_pem() resp = await client.post("/admin/ssh-keys", json={ "label": "nachtragen", "private_key_pem": pem, "public_key": "ssh-ed25519 AAAAC3NzaC1lZDI1NTE5 test", "key_type": "ed25519", "passphrase": PASSPHRASE, }) assert resp.status_code == 201, resp.text key_id = resp.json()["id"] # Falsche Passphrase nachtragen -> abgelehnt, nichts geaendert resp = await client.put(f"/admin/ssh-keys/{key_id}", json={"passphrase": "falsch"}) assert resp.status_code == 400, resp.text # Korrekte Passphrase nachtragen -> akzeptiert resp = await client.put(f"/admin/ssh-keys/{key_id}", json={"passphrase": PASSPHRASE}) assert resp.status_code == 200, resp.text # Ein Update ohne das Feld darf die Passphrase NICHT loeschen resp = await client.put(f"/admin/ssh-keys/{key_id}", json={"label": "umbenannt"}) assert resp.status_code == 200, resp.text row = await (await conn.execute( "SELECT passphrase_enc FROM ssh_keys WHERE id = ?", (key_id,) )).fetchone() assert row[0] is not None, "Label-Update hat die Passphrase geloescht" @pytest.mark.asyncio async def test_load_private_key_for_host_uses_stored_passphrase(client): """Der konkret gemeldete Fehlerfall, komplett ueber die DB-Schicht.""" from app.db import get_db from app.ssh_proxy.proxy import PrivateKeyUnusableError, load_private_key_for_host from app.security.crypto import encrypt_secret conn = get_db() cursor = await conn.execute("INSERT INTO host_groups (name) VALUES ('p10-group')") hg_id = cursor.lastrowid cursor = await conn.execute( "INSERT INTO hosts (host_group_id, hostname, address, protocol, port, os_type, ssh_username) " "VALUES (?, 'p10-host', '10.10.0.1', 'ssh', 22, 'linux', 'root')", (hg_id,), ) host_id = cursor.lastrowid pem = _encrypted_key_pem() cursor = await conn.execute( "INSERT INTO ssh_keys (label, private_key_enc, public_key, key_type, passphrase_enc) " "VALUES ('k', ?, 'pub', 'ed25519', ?)", (encrypt_secret(pem.encode(), associated_data=b"ssh_private_key"), encrypt_secret(PASSPHRASE.encode(), associated_data=b"ssh_key_passphrase")), ) key_id = cursor.lastrowid await conn.execute( "INSERT INTO host_ssh_key_map (host_id, ssh_key_id) VALUES (?, ?)", (host_id, key_id) ) await conn.commit() # Mit hinterlegter Passphrase laedt der Schluessel. assert await load_private_key_for_host(conn, host_id) is not None # Ohne sie: klare Meldung statt eines nach aussen durchschlagenden # KeyImportError (das war der gemeldete Abbruch ohne Fehlermeldung). await conn.execute("UPDATE ssh_keys SET passphrase_enc = NULL WHERE id = ?", (key_id,)) await conn.commit() with pytest.raises(PrivateKeyUnusableError) as excinfo: await load_private_key_for_host(conn, host_id) assert "passphrasegeschuetzt" in str(excinfo.value).lower() # --------------------------------------------------------------------------- # 2) RDP-Verbindungsparameter # --------------------------------------------------------------------------- @pytest.mark.asyncio async def test_load_host_returns_rdp_columns(client): """load_host() muss genau die Felder liefern, die build_rdp_params() liest.""" from app.db import get_db from app.ssh_proxy.proxy import load_host conn = get_db() cursor = await conn.execute("INSERT INTO host_groups (name) VALUES ('p10-rdp-group')") hg_id = cursor.lastrowid cursor = await conn.execute( "INSERT INTO hosts (host_group_id, hostname, address, protocol, port, os_type, " "rdp_username, rdp_domain) VALUES (?, 'win-ts01', '10.10.0.5', 'rdp', 3389, 'windows', " "'Administrator', 'CORP')", (hg_id,), ) await conn.commit() host = await load_host(conn, cursor.lastrowid) for field in ("rdp_username", "rdp_domain", "rdp_require_nla", "clipboard_enabled", "rdp_ignore_cert"): assert field in host, f"load_host() liefert '{field}' nicht" assert host["rdp_username"] == "Administrator" assert host["rdp_domain"] == "CORP" assert host["rdp_ignore_cert"] is True # Default aus Migration 0009 def test_build_rdp_params_passes_username_and_cert_policy(): from app.rdp_proxy.guacd_client import GuacamoleProtocolError, build_rdp_params host = { "id": 1, "hostname": "win-ts01", "address": "10.10.0.5", "port": 3389, "file_transfer_enabled": True, "rdp_username": "Administrator", "rdp_domain": "CORP", "rdp_require_nla": True, "clipboard_enabled": True, "rdp_ignore_cert": True, } params = build_rdp_params(host, "geheim") assert params["username"] == "Administrator" assert params["domain"] == "CORP" assert params["security"] == "nla" assert params["ignore-cert"] == "true" assert params["disable-copy"] == "false" strict = build_rdp_params(dict(host, rdp_ignore_cert=False, clipboard_enabled=False), "geheim") assert strict["ignore-cert"] == "false" assert strict["disable-copy"] == "true" and strict["disable-paste"] == "true" # Kein Benutzername -> klare Meldung statt stiller Fehlanmeldung am Ziel with pytest.raises(GuacamoleProtocolError): build_rdp_params(dict(host, rdp_username=""), "geheim") # Unvollstaendiger Hostdatensatz (der alte load_host()-Zustand) with pytest.raises(GuacamoleProtocolError): build_rdp_params({k: host[k] for k in ("id", "hostname", "address", "port", "file_transfer_enabled")}, "geheim") # --------------------------------------------------------------------------- # 3) Guacamole-Protokoll: Laengen zaehlen Zeichen, nicht Bytes # --------------------------------------------------------------------------- def test_instruction_length_counts_characters_not_bytes(): from app.rdp_proxy.guacd_client import encode_instruction, parse_instruction_text # "Grüße" sind 5 Zeichen, aber 7 Bytes in UTF-8. text = encode_instruction("clipboard", "Grüße") assert text == "9.clipboard,5.Grüße;", text assert parse_instruction_text(text) == ["clipboard", "Grüße"] @pytest.mark.asyncio async def test_read_instruction_reads_multibyte_characters(): import asyncio from app.rdp_proxy.guacd_client import encode_instruction, read_instruction reader = asyncio.StreamReader() reader.feed_data(encode_instruction("name", "Bürö-Süd").encode("utf-8")) reader.feed_eof() assert await read_instruction(reader) == ["name", "Bürö-Süd"] def test_internal_opcode_instruction_is_parseable(): """guacamole-common-js sendet ping/UUID mit leerem Opcode ('0.,...').""" from app.rdp_proxy.guacd_client import parse_instruction_text from app.rdp_proxy.ws_tunnel import INTERNAL_DATA_OPCODE parsed = parse_instruction_text("0.,4.ping,13.1755721410123;") assert parsed[0] == INTERNAL_DATA_OPCODE assert parsed[1] == "ping"