449 lines
19 KiB
Python
449 lines
19 KiB
Python
"""
|
|
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).
|
|
client.cookies.set("jh_session", old_cookie)
|
|
resp = await client.get("/auth/me")
|
|
assert resp.status_code == 401
|
|
|
|
|
|
@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).
|
|
await conn.execute(
|
|
"INSERT INTO user_hostgroup_roles (user_id, host_group_id, role_id, granted_by) VALUES (?, 1, 3, ?)",
|
|
(user_id, 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.
|
|
await conn.execute(
|
|
"INSERT INTO user_hostgroup_roles (user_id, host_group_id, role_id, granted_by) VALUES (?, 1, 1, ?)",
|
|
(user_id, 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}"
|