midas/ssh-jumphost
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

second commit

Browse Source
This commit is contained in:
Midas Wollinger
2026-08-19 22:33:19 +02:00
parent 411812e954
commit 199f306993
107 changed files with 5984 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files

57
tests/conftest.py Normal file
View File

@ -0,0 +1,57 @@
import os
import secrets
import sys
from pathlib import Path
os.environ.setdefault("JUMPHOST_ENV", "development")
os.environ.setdefault("JUMPHOST_DEV_KEK", secrets.token_hex(32))
os.environ.setdefault("JUMPHOST_DEV_SESSION_SECRET", secrets.token_hex(32))
sys.path.insert(0, str(Path(__file__).resolve().parent.parent))
import pytest # noqa: E402
from httpx import ASGITransport, AsyncClient # noqa: E402
@pytest.fixture()
def tmp_data_dir(tmp_path, monkeypatch):
monkeypatch.setenv("JUMPHOST_DATA_DIR", str(tmp_path))
return tmp_path
@pytest.fixture(autouse=True)
def _reset_rate_limiter():
"""login_rate_limiter (app/security/rate_limit.py) ist bewusst ein
Prozess-weiter In-Memory-Singleton (siehe Konzept/Docstring dort) --
genau das wird hier durch die Tests bestaetigt (test_source_ip_rate_limit_*).
Damit sich einzelne Tests nicht gegenseitig ueber ihr Kontingent
beeinflussen, wird der Zustand vor JEDEM Test zurueckgesetzt."""
from app.security.rate_limit import login_rate_limiter
login_rate_limiter._events.clear()
yield
login_rate_limiter._events.clear()
@pytest.fixture()
async def client(tmp_data_dir):
# app.config.settings ist ein Modul-Singleton, der beim ERSTEN Import des
# Prozesses einmalig aus der Umgebung gelesen wird (siehe app/config.py) --
# spaetere Aenderungen von JUMPHOST_DATA_DIR je Test wirken sich darauf
# nicht mehr aus. Fuer Testisolation wird daher die (immer gleiche)
# DB-Datei vor jedem Test explizit zurueckgesetzt statt sich auf einen
# neuen Datenpfad pro Test zu verlassen.
from app.main import app
from app.config import settings
for suffix in ("", "-wal", "-shm"):
p = settings.data_dir / f"jumphost.db{suffix}"
if p.exists():
p.unlink()
transport = ASGITransport(app=app)
# https:// Base-URL, da unsere Session-Cookies "Secure" gesetzt haben
# (Konzept 6.2) und httpx' Cookie-Jar dies wie ein Browser respektiert.
async with AsyncClient(transport=transport, base_url="https://testserver") as ac:
async with app.router.lifespan_context(app):
yield ac

63
tests/test_audit_chain.py Normal file
View File

@ -0,0 +1,63 @@
import pytest
import aiosqlite
from app.security.audit import verify_chain, write_audit_event
from app.db import MIGRATIONS_DIR
async def _fresh_db() -> aiosqlite.Connection:
conn = await aiosqlite.connect(":memory:")
for migration_file in sorted(MIGRATIONS_DIR.glob("*.sql")):
await conn.executescript(migration_file.read_text(encoding="utf-8"))
return conn
@pytest.mark.asyncio
async def test_chain_intact_after_writes():
conn = await _fresh_db()
for i in range(5):
await write_audit_event(
conn, event_type="test_event", user_id=None, client_ip="127.0.0.1", details={"i": i}
)
await conn.commit()
intact, broken_at = await verify_chain(conn)
assert intact is True
assert broken_at is None
await conn.close()
@pytest.mark.asyncio
async def test_tamper_detected_after_direct_update():
conn = await _fresh_db()
for i in range(3):
await write_audit_event(
conn, event_type="test_event", user_id=None, client_ip="127.0.0.1", details={"i": i}
)
await conn.commit()
# Der Append-only-Trigger blockt normale UPDATEs -- simuliert wird hier
# eine Umgehung auf DB-Ebene (z.B. Datei-Manipulation waehrend der App
# gestoppt ist), um zu zeigen, dass verify_chain() dies unabhaengig von
# den Triggern erkennt.
await conn.execute("DROP TRIGGER no_audit_update")
await conn.execute("UPDATE audit_log SET details_json = '{\"i\": 999}' WHERE id = 2")
await conn.commit()
intact, broken_at = await verify_chain(conn)
assert intact is False
assert broken_at == 2
await conn.close()
@pytest.mark.asyncio
async def test_append_only_trigger_blocks_update():
conn = await _fresh_db()
await write_audit_event(
conn, event_type="test_event", user_id=None, client_ip="127.0.0.1", details={}
)
await conn.commit()
with pytest.raises(aiosqlite.Error):
await conn.execute("UPDATE audit_log SET event_type = 'tampered' WHERE id = 1")
await conn.close()

67
tests/test_auth_flow.py Normal file
View File

@ -0,0 +1,67 @@
"""End-to-End-Test des Login-Flows (Passwort -> TOTP-Enrollment -> Session)
gegen die echte FastAPI-App mit einer temporaeren SQLite-DB.
Die `client`-Fixture liegt in conftest.py (von mehreren Testdateien genutzt,
u.a. tests/test_pentest_security.py)."""
import pyotp
import pytest
@pytest.mark.asyncio
async def test_full_login_flow_with_totp_enrollment(client, monkeypatch):
from app.db import get_db
from app.security.passwords import hash_password
conn = get_db()
await conn.execute(
"INSERT INTO users (username, password_hash, must_change_password) VALUES (?, ?, 0)",
("alice", hash_password("Correct-Horse-Battery-Staple-1")),
)
await conn.commit()
resp = await client.post(
"/auth/login", json={"username": "alice", "password": "Correct-Horse-Battery-Staple-1"}
)
assert resp.status_code == 200, resp.text
body = resp.json()
assert body["totp_enrolled"] is False
pending_token = body["pending_token"]
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
assert len(resp.json()["recovery_codes"]) == 10
assert "jh_session" in resp.cookies
resp = await client.get("/auth/me")
assert resp.status_code == 200, resp.text
assert resp.json()["username"] == "alice"
@pytest.mark.asyncio
async def test_wrong_password_rejected(client):
from app.db import get_db
from app.security.passwords import hash_password
conn = get_db()
await conn.execute(
"INSERT INTO users (username, password_hash, must_change_password) VALUES (?, ?, 0)",
("bob", hash_password("Correct-Horse-Battery-Staple-2")),
)
await conn.commit()
resp = await client.post("/auth/login", json={"username": "bob", "password": "wrong"})
assert resp.status_code == 401
@pytest.mark.asyncio
async def test_unauthenticated_admin_access_denied(client):
resp = await client.get("/admin/users")
assert resp.status_code == 401

28
tests/test_passwords_totp.py Normal file
View File

@ -0,0 +1,28 @@
from app.security.passwords import hash_password, verify_password
from app.security.totp import (
decrypt_totp_secret,
encrypt_totp_secret,
generate_totp_secret,
verify_totp_code,
)
import pyotp
def test_password_hash_roundtrip():
h = hash_password("Sup3rSecret!Passphrase")
assert verify_password(h, "Sup3rSecret!Passphrase")
assert not verify_password(h, "wrong-password")
def test_totp_secret_encryption_roundtrip():
secret = generate_totp_secret()
enc = encrypt_totp_secret(secret)
assert enc != secret.encode()
assert decrypt_totp_secret(enc) == secret
def test_totp_verification():
secret = generate_totp_secret()
code = pyotp.TOTP(secret).now()
assert verify_totp_code(secret, code)
assert not verify_totp_code(secret, "000000") or pyotp.TOTP(secret).now() == "000000"

448
tests/test_pentest_security.py Normal file
View File

@ -0,0 +1,448 @@
"""
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}"

43
tests/test_rbac.py Normal file
View File

@ -0,0 +1,43 @@
import pytest
import aiosqlite
from app.db import MIGRATIONS_DIR
from app.rbac import user_has_role, user_has_role_for_host
async def _fresh_db() -> aiosqlite.Connection:
conn = await aiosqlite.connect(":memory:")
for migration_file in sorted(MIGRATIONS_DIR.glob("*.sql")):
await conn.executescript(migration_file.read_text(encoding="utf-8"))
return conn
@pytest.mark.asyncio
async def test_rbac_grants_and_expiry():
conn = await _fresh_db()
await conn.execute("INSERT INTO users (id, username, password_hash) VALUES (1, 'alice', 'x')")
await conn.execute("INSERT INTO host_groups (id, name) VALUES (1, 'linux-prod')")
await conn.execute(
"INSERT INTO hosts (id, host_group_id, hostname, address, protocol, port, os_type) "
"VALUES (1, 1, 'db01', '10.0.0.1', 'ssh', 22, 'linux')"
)
assert not await user_has_role(conn, user_id=1, host_group_id=1, role_name="ssh_connect")
await conn.execute(
"INSERT INTO user_hostgroup_roles (user_id, host_group_id, role_id) VALUES (1, 1, 1)"
)
await conn.commit()
assert await user_has_role(conn, user_id=1, host_group_id=1, role_name="ssh_connect")
assert await user_has_role_for_host(conn, user_id=1, host_id=1, role_name="ssh_connect")
assert not await user_has_role_for_host(conn, user_id=1, host_id=1, role_name="rdp_connect")
# Abgelaufene Freigabe darf nicht mehr gelten.
await conn.execute(
"UPDATE user_hostgroup_roles SET expires_at = '2000-01-01T00:00:00.000000Z' "
"WHERE user_id = 1 AND host_group_id = 1 AND role_id = 1"
)
await conn.commit()
assert not await user_has_role(conn, user_id=1, host_group_id=1, role_name="ssh_connect")
await conn.close()