""" Dynamischer Sicherheitstest (Mini-Pentest) gegen die laufende FastAPI-App. Scope/Methodik: Black-/Grey-Box-Angriffstests auf Anwendungsebene gegen die Jumphost-Webanwendung in einer isolierten lokalen Testumgebung (kein echtes Netzwerk, kein echtes Zielsystem). Abgedeckt: Authentifizierungs-Bypass, Session-/Cookie-Manipulation, RBAC-/IDOR-Bypass, Injection, Rate-Limiting/ Lockout, Security-Header, Informationslecks (Enumeration/Fehlermeldungen), Audit-Log-Vollstaendigkeit. NICHT abgedeckt (siehe Pentest-Report): Netzwerk-/Infrastruktur-Pentest, echter RDP/guacd-Pfad, Social Engineering, physische Sicherheit. """ from __future__ import annotations import time import pyotp import pytest from itsdangerous import URLSafeSerializer from app.security.passwords import hash_password async def _create_user(conn, username: str, password: str, *, is_admin: bool = False) -> int: 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 _enroll_totp_and_login(client, pending_token: str) -> tuple[str, str]: """Fuehrt TOTP-Enrollment durch und gibt (secret, session_cookie) zurueck.""" resp = await client.post("/auth/totp/enroll/start", json={"pending_token": pending_token}) 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_token, "code": code}) assert resp.status_code == 200, resp.text return secret, resp.cookies.get("jh_session") async def _login_full(client, username: str, password: str) -> str: """Login + TOTP-Enrollment fuer einen frischen User, gibt Session-Cookie zurueck.""" resp = await client.post("/auth/login", json={"username": username, "password": password}) assert resp.status_code == 200, resp.text pending = resp.json()["pending_token"] _, cookie = await _enroll_totp_and_login(client, pending) return cookie # --------------------------------------------------------------------------- # 1) Authentifizierung: Enumeration, Brute-Force, Lockout # --------------------------------------------------------------------------- @pytest.mark.asyncio async def test_no_username_enumeration_via_error_message(client): from app.db import get_db conn = get_db() await _create_user(conn, "carol", "Correct-Horse-Battery-Staple-3") resp_unknown = await client.post("/auth/login", json={"username": "no_such_user_xyz", "password": "whatever12345"}) resp_wrongpw = await client.post("/auth/login", json={"username": "carol", "password": "wrong-password-123"}) assert resp_unknown.status_code == resp_wrongpw.status_code == 401 assert resp_unknown.json()["detail"] == resp_wrongpw.json()["detail"] @pytest.mark.asyncio async def test_account_lockout_after_repeated_failures(client): from app.db import get_db conn = get_db() await _create_user(conn, "dave", "Correct-Horse-Battery-Staple-4") last_status = None for _ in range(6): resp = await client.post("/auth/login", json={"username": "dave", "password": "wrong"}) last_status = resp.status_code # Nach >=5 Fehlversuchen muss das Konto gesperrt sein (Konzept 6.2), # selbst wenn danach das RICHTIGE Passwort verwendet wird. assert last_status == 423 resp = await client.post( "/auth/login", json={"username": "dave", "password": "Correct-Horse-Battery-Staple-4"} ) assert resp.status_code == 423 @pytest.mark.asyncio async def test_source_ip_rate_limit_blocks_excessive_attempts(client): from app.db import get_db conn = get_db() await _create_user(conn, "erin", "Correct-Horse-Battery-Staple-5") statuses = [] for i in range(15): # Unterschiedliche Usernamen, damit NICHT der Account-Lockout (obiger # Test), sondern ausschliesslich der IP-basierte Rate-Limiter greift. resp = await client.post( "/auth/login", json={"username": f"no_such_user_{i}", "password": "x" * 12} ) statuses.append(resp.status_code) assert 429 in statuses, f"Rate-Limiter hat nicht ausgeloest, Status-Codes: {statuses}" @pytest.mark.asyncio async def test_sql_injection_payload_in_username_rejected_safely(client): """Klassische SQLi-Payloads duerfen weder einen Serverfehler (500) noch einen Auth-Bypass ausloesen -- erwartet wird 401 (falsche Zugangsdaten) oder 422 (Validierungsfehler durch das Username-Regex).""" payloads = [ "admin' OR '1'='1", "admin'--", "' OR 1=1;--", "admin'; DROP TABLE users;--", "\" OR \"\"=\"", ] for payload in payloads: resp = await client.post("/auth/login", json={"username": payload, "password": "irrelevant123"}) assert resp.status_code in (401, 422), f"Unerwarteter Status fuer Payload {payload!r}: {resp.status_code}" # Datenbank muss danach noch normal funktionieren (kein DROP TABLE griff). from app.db import get_db conn = get_db() cursor = await conn.execute("SELECT COUNT(*) FROM users") row = await cursor.fetchone() assert row is not None # wuerde eine sqlite3.OperationalError werfen, waere die Tabelle weg # --------------------------------------------------------------------------- # 2) Session-/Cookie-Sicherheit # --------------------------------------------------------------------------- @pytest.mark.asyncio async def test_session_cookie_security_flags(client): from app.db import get_db conn = get_db() await _create_user(conn, "frank", "Correct-Horse-Battery-Staple-6") resp = await client.post("/auth/login", json={"username": "frank", "password": "Correct-Horse-Battery-Staple-6"}) pending = resp.json()["pending_token"] resp = await client.post("/auth/totp/enroll/start", json={"pending_token": pending}) 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}) set_cookie = resp.headers.get("set-cookie", "") assert "HttpOnly" in set_cookie assert "Secure" in set_cookie assert "SameSite=strict" in set_cookie or "SameSite=Strict" in set_cookie @pytest.mark.asyncio async def test_forged_session_cookie_with_wrong_secret_rejected(client): """Ein Angreifer, der die Cookie-STRUKTUR kennt (z.B. aus diesem Open-Source-Code) aber NICHT den Session-Secret des Zielsystems, darf sich damit keinen gueltigen Admin-Zugang faelschen koennen.""" forged_serializer = URLSafeSerializer("attacker-controlled-wrong-secret", salt="jumphost-session") forged_token = forged_serializer.dumps({"uid": 1, "sv": 1, "iat": time.time(), "seen": time.time()}) client.cookies.set("jh_session", forged_token) resp = await client.get("/auth/me") assert resp.status_code == 401 @pytest.mark.asyncio async def test_password_change_invalidates_old_session_everywhere(client): from app.db import get_db conn = get_db() await _create_user(conn, "grace", "Correct-Horse-Battery-Staple-7") old_cookie = await _login_full(client, "grace", "Correct-Horse-Battery-Staple-7") resp = await client.get("/auth/me") assert resp.status_code == 200 resp = await client.post( "/auth/change-password", json={"current_password": "Correct-Horse-Battery-Staple-7", "new_password": "Even-Str0nger-Passphrase!"}, ) assert resp.status_code == 200, resp.text # Altes Cookie (vor dem Passwortwechsel ausgestellt) erneut einspielen -- # muss durch die session_version-Pruefung invalidiert sein (Konzept 6.2). # # Bewusst NICHT ueber client.cookies.set(...)/cookies=<...> auf dem # GETEILTEN `client`-Fixture: dessen Jar traegt an dieser Stelle bereits # die NEUE, gueltige Sitzung aus der change-password-Antwort oben. # client.cookies.set() kann je nach httpx-Version einen zweiten, # mehrdeutigen Jar-Eintrag statt eines Ersatzes erzeugen: der Request # wuerde dann weiterhin (auch) die noch gueltige NEUE Sitzung # mitschicken. Der per-Request-Parameter `cookies=` ist fuer genau # diesen Fall (Ueberschreiben einer bereits im Jar vorhandenen Cookie) # KEINE zuverlaessige Abhilfe: httpx warnt selbst per # DeprecationWarning, dass das Zusammenspiel mit einem bereits # gesetzten Jar-Eintrag "ambiguous" ist -- ein eigener Testlauf hat das # bestaetigt (der alte Cookie wurde NICHT wirksam, der Request nutzte # weiterhin die neue Sitzung aus dem Jar, Test schlug trotz korrektem # Server faelschlich fehl). Ein voellig frischer, jar-loser Client hat # dagegen zuverlaessig funktioniert (siehe FORTSETZUNG_Teil_C.md # Abschnitt 3 Punkt 6 fuer die vollstaendige Untersuchung) -- deshalb # hier bewusst ein zweiter, eigener AsyncClient nur fuer diesen einen # Replay-Request, unabhaengig vom Jar-Zustand des Fixture-Clients. from httpx import ASGITransport, AsyncClient from app.main import app as fastapi_app async with AsyncClient(transport=ASGITransport(app=fastapi_app), base_url="https://testserver") as replay_client: # .set() auf dem LEEREN Jar dieses frischen Clients ist eindeutig # (kein vorhandener Eintrag, der ueberschrieben werden muesste) -- # folgt damit auch httpx' eigener Empfehlung in der oben erklaerten # DeprecationWarning ("Set cookies directly on the client instance"). replay_client.cookies.set("jh_session", old_cookie) resp = await replay_client.get("/auth/me") assert resp.status_code == 401, resp.text @pytest.mark.asyncio async def test_logout_everywhere_invalidates_session(client): from app.db import get_db conn = get_db() await _create_user(conn, "heidi", "Correct-Horse-Battery-Staple-8") cookie = await _login_full(client, "heidi", "Correct-Horse-Battery-Staple-8") resp = await client.post("/auth/logout-everywhere") assert resp.status_code == 200 client.cookies.set("jh_session", cookie) resp = await client.get("/auth/me") assert resp.status_code == 401 # --------------------------------------------------------------------------- # 3) RBAC / IDOR (Insecure Direct Object Reference) # --------------------------------------------------------------------------- @pytest.mark.asyncio async def test_non_admin_cannot_reach_admin_endpoints(client): from app.db import get_db conn = get_db() await _create_user(conn, "ivan", "Correct-Horse-Battery-Staple-9", is_admin=False) await _login_full(client, "ivan", "Correct-Horse-Battery-Staple-9") admin_endpoints = [ ("GET", "/admin/users"), ("GET", "/admin/host-groups"), ("GET", "/admin/hosts"), ("GET", "/admin/audit-log"), ("GET", "/admin/audit-log/verify"), ] for method, url in admin_endpoints: resp = await client.request(method, url) assert resp.status_code == 403, f"{method} {url} sollte 403 liefern, war {resp.status_code}" # Auch schreibende Admin-Endpunkte muessen blocken. resp = await client.post("/admin/host-groups", json={"name": "sollte-nicht-klappen"}) assert resp.status_code == 403 resp = await client.post( "/admin/users", json={"username": "eviladmin", "initial_password": "Whatever123456!", "is_admin": True} ) assert resp.status_code == 403 @pytest.mark.asyncio async def test_user_cannot_access_host_outside_granted_hostgroup(client): """Kern-RBAC-Test: Nutzer 'judy' bekommt NUR Rechte auf Hostgruppe A; Zugriffsversuche (Dateitransfer-Endpunkt als Stellvertreter fuer RBAC-gepruefte Aktionen) auf einen Host in Hostgruppe B muessen 403 liefern -- unabhaengig davon, dass die Host-ID gueltig/erratbar ist.""" from app.db import get_db from app.security.passwords import hash_password conn = get_db() admin_id = await _create_user(conn, "admin_rbac", "Correct-Horse-Battery-Staple-A", is_admin=True) user_id = await _create_user(conn, "judy", "Correct-Horse-Battery-Staple-J", is_admin=False) await conn.execute("INSERT INTO host_groups (id, name) VALUES (1, 'gruppe-a')") await conn.execute("INSERT INTO host_groups (id, name) VALUES (2, 'gruppe-b')") await conn.execute( "INSERT INTO hosts (id, host_group_id, hostname, address, protocol, port, os_type, " "file_transfer_enabled) VALUES (1, 1, 'a01', '10.0.0.1', 'ssh', 22, 'linux', 1)" ) await conn.execute( "INSERT INTO hosts (id, host_group_id, hostname, address, protocol, port, os_type, " "file_transfer_enabled) VALUES (2, 2, 'b01', '10.0.0.2', 'ssh', 22, 'linux', 1)" ) # judy bekommt file_transfer NUR auf Hostgruppe 1 (role_id 3 = file_transfer, # siehe Migration 0001) -- seit Teil D Schritt 4 ausschliesslich ueber # eine Benutzergruppe moeglich (direkte Vergabe an einzelne Benutzer ist # entfallen, siehe app/rbac.py-Modul-Docstring). await conn.execute("INSERT INTO user_groups (id, name) VALUES (101, 'judy-team')") await conn.execute("INSERT INTO user_group_members (user_group_id, user_id) VALUES (101, ?)", (user_id,)) await conn.execute( "INSERT INTO group_hostgroup_roles (user_group_id, host_group_id, role_id, granted_by) VALUES (101, 1, 3, ?)", (admin_id,), ) await conn.commit() await _login_full(client, "judy", "Correct-Horse-Battery-Staple-J") # Erlaubter Host (Gruppe 1): RBAC-Check muss durchgehen (Fehler danach # ist erwartet, da kein echtes SSH-Ziel existiert -- wichtig ist der # Unterschied 403 vs. "kommt ueberhaupt durch die Berechtigungspruefung"). resp = await client.post( "/ssh/1/files/upload?remote_path=/tmp/test.txt", files={"file": ("test.txt", b"hello", "text/plain")}, ) assert resp.status_code != 403, "Legitimer Zugriff auf eigene Hostgruppe wurde faelschlich blockiert" # Verbotener Host (Gruppe 2, keine Rolle vergeben) -- MUSS 403 sein. resp = await client.post( "/ssh/2/files/upload?remote_path=/tmp/test.txt", files={"file": ("test.txt", b"hello", "text/plain")}, ) assert resp.status_code == 403 @pytest.mark.asyncio async def test_role_on_one_hostgroup_does_not_grant_different_permission_type(client): """judy hat NUR ssh_connect auf Gruppe 1, NICHT file_transfer -- der Filetransfer-Endpunkt muss trotz gueltiger ssh_connect-Rolle blocken (Rollen sind pro Aktionstyp granular, nicht pauschal 'Zugriff auf Host').""" from app.db import get_db conn = get_db() admin_id = await _create_user(conn, "admin_rbac2", "Correct-Horse-Battery-Staple-B", is_admin=True) user_id = await _create_user(conn, "mallory", "Correct-Horse-Battery-Staple-M", is_admin=False) await conn.execute("INSERT INTO host_groups (id, name) VALUES (1, 'gruppe-c')") await conn.execute( "INSERT INTO hosts (id, host_group_id, hostname, address, protocol, port, os_type, " "file_transfer_enabled) VALUES (1, 1, 'c01', '10.0.0.3', 'ssh', 22, 'linux', 1)" ) # role_id 1 = ssh_connect (siehe Migration 0001) -- explizit KEIN # file_transfer. Seit Teil D Schritt 4 ausschliesslich ueber eine # Benutzergruppe moeglich (siehe Kommentar in # test_user_cannot_access_host_outside_granted_hostgroup oben). await conn.execute("INSERT INTO user_groups (id, name) VALUES (102, 'mallory-team')") await conn.execute("INSERT INTO user_group_members (user_group_id, user_id) VALUES (102, ?)", (user_id,)) await conn.execute( "INSERT INTO group_hostgroup_roles (user_group_id, host_group_id, role_id, granted_by) VALUES (102, 1, 1, ?)", (admin_id,), ) await conn.commit() await _login_full(client, "mallory", "Correct-Horse-Battery-Staple-M") resp = await client.post( "/ssh/1/files/upload?remote_path=/tmp/test.txt", files={"file": ("test.txt", b"hello", "text/plain")}, ) assert resp.status_code == 403 # --------------------------------------------------------------------------- # 4) Dateitransfer-Haertung # --------------------------------------------------------------------------- @pytest.mark.asyncio async def test_oversized_upload_rejected(client, monkeypatch): from app.db import get_db import app.ssh_proxy.sftp as sftp_module conn = get_db() admin_id = await _create_user(conn, "admin_upload", "Correct-Horse-Battery-Staple-C", is_admin=True) await conn.execute("INSERT INTO host_groups (id, name) VALUES (1, 'gruppe-d')") await conn.execute( "INSERT INTO hosts (id, host_group_id, hostname, address, protocol, port, os_type, " "file_transfer_enabled) VALUES (1, 1, 'd01', '10.0.0.4', 'ssh', 22, 'linux', 1)" ) await conn.commit() # Limit fuer den Test drastisch verkleinern statt 200 MiB echt zu senden. monkeypatch.setattr(sftp_module, "MAX_UPLOAD_BYTES", 10) await _login_full(client, "admin_upload", "Correct-Horse-Battery-Staple-C") resp = await client.post( "/ssh/1/files/upload?remote_path=/tmp/big.bin", files={"file": ("big.bin", b"x" * 1000, "application/octet-stream")}, ) assert resp.status_code == 413 @pytest.mark.asyncio async def test_malware_flagged_upload_is_blocked(client, monkeypatch): from app.db import get_db import app.ssh_proxy.sftp as sftp_module conn = get_db() await _create_user(conn, "admin_av", "Correct-Horse-Battery-Staple-D", is_admin=True) await conn.execute("INSERT INTO host_groups (id, name) VALUES (1, 'gruppe-e')") await conn.execute( "INSERT INTO hosts (id, host_group_id, hostname, address, protocol, port, os_type, " "file_transfer_enabled) VALUES (1, 1, 'e01', '10.0.0.5', 'ssh', 22, 'linux', 1)" ) await conn.commit() monkeypatch.setattr(sftp_module, "scan_bytes", lambda data: "infected:EICAR-Test-Signature") await _login_full(client, "admin_av", "Correct-Horse-Battery-Staple-D") resp = await client.post( "/ssh/1/files/upload?remote_path=/tmp/eicar.txt", files={"file": ("eicar.txt", b"malware-like-content", "text/plain")}, ) assert resp.status_code == 400 assert "AV-Scan" in resp.json()["detail"] or "blockiert" in resp.json()["detail"] # Blockierter Upload darf NICHT als erfolgreicher Filetransfer geloggt sein. cursor = await conn.execute("SELECT COUNT(*) FROM file_transfers") row = await cursor.fetchone() assert row[0] == 0 # --------------------------------------------------------------------------- # 5) HTTP-Security-Header (auf jeder Antwort, nicht nur auf Login-Seiten) # --------------------------------------------------------------------------- @pytest.mark.asyncio async def test_security_headers_present_on_every_response(client): for path in ["/", "/healthz", "/admin/users"]: resp = await client.get(path) headers = resp.headers assert headers.get("x-frame-options") == "DENY" assert headers.get("x-content-type-options") == "nosniff" assert "content-security-policy" in headers assert "script-src 'self'" in headers["content-security-policy"] assert "strict-transport-security" in headers assert headers.get("referrer-policy") == "no-referrer" @pytest.mark.asyncio async def test_no_server_stack_traces_leaked_on_bad_input(client): resp = await client.post("/auth/login", json={"username": "x", "password": ""}) assert resp.status_code in (401, 422) assert "Traceback" not in resp.text assert "File \"/" not in resp.text # --------------------------------------------------------------------------- # 6) Audit-Log-Vollstaendigkeit fuer sicherheitsrelevante Ereignisse # --------------------------------------------------------------------------- @pytest.mark.asyncio async def test_failed_and_successful_logins_are_audited(client): from app.db import get_db conn = get_db() await _create_user(conn, "audituser", "Correct-Horse-Battery-Staple-E") await client.post("/auth/login", json={"username": "audituser", "password": "wrong"}) await _login_full(client, "audituser", "Correct-Horse-Battery-Staple-E") cursor = await conn.execute( "SELECT event_type FROM audit_log WHERE event_type IN " "('login_failed','login_password_ok','login_success','totp_enroll_confirmed') ORDER BY id" ) events = [row[0] for row in await cursor.fetchall()] assert "login_failed" in events assert "login_password_ok" in events assert "totp_enroll_confirmed" in events @pytest.mark.asyncio async def test_audit_chain_stays_intact_after_full_test_scenario(client): """Regressionscheck: nach allen obigen Aktionen (viele Logins, Fehlversuche, Admin-Aktionen) muss die Hash-Chain weiterhin intakt sein -- ein Bug, der Eintraege ausser der Reihe schreibt, wuerde hier auffallen.""" from app.db import get_db from app.security.audit import verify_chain conn = get_db() await _create_user(conn, "chaincheck", "Correct-Horse-Battery-Staple-F") await _login_full(client, "chaincheck", "Correct-Horse-Battery-Staple-F") intact, broken_at = await verify_chain(conn) assert intact is True, f"Audit-Chain gebrochen bei id={broken_at}"