more admin stuff 2
This commit is contained in:
@ -84,7 +84,7 @@ async def test_group_role_grant_gives_catalog_access_without_individual_grant(cl
|
||||
|
||||
resp = await client.post(
|
||||
"/admin/group-roles/grant",
|
||||
json={"user_group_id": group_id, "host_group_id": hg_id, "role_name": "ssh_connect"},
|
||||
json={"user_group_id": group_id, "host_group_id": hg_id, "role_names": ["ssh_connect"]},
|
||||
)
|
||||
assert resp.status_code == 200, resp.text
|
||||
|
||||
@ -114,7 +114,7 @@ async def test_removing_group_member_revokes_inherited_catalog_access(client):
|
||||
await client.post(f"/admin/user-groups/{group_id}/members", json={"user_id": member_id})
|
||||
await client.post(
|
||||
"/admin/group-roles/grant",
|
||||
json={"user_group_id": group_id, "host_group_id": hg_id, "role_name": "ssh_connect"},
|
||||
json={"user_group_id": group_id, "host_group_id": hg_id, "role_names": ["ssh_connect"]},
|
||||
)
|
||||
|
||||
# Mitgliedschaft wieder entfernen, BEVOR sich der User einloggt.
|
||||
@ -142,7 +142,7 @@ async def test_non_admin_cannot_manage_user_groups_or_group_roles(client):
|
||||
assert resp.status_code == 403
|
||||
resp = await client.post(
|
||||
"/admin/group-roles/grant",
|
||||
json={"user_group_id": 1, "host_group_id": 1, "role_name": "ssh_connect"},
|
||||
json={"user_group_id": 1, "host_group_id": 1, "role_names": ["ssh_connect"]},
|
||||
)
|
||||
assert resp.status_code == 403
|
||||
|
||||
|
||||
286
tests/test_phase9.py
Normal file
286
tests/test_phase9.py
Normal file
@ -0,0 +1,286 @@
|
||||
"""
|
||||
Tests fuer die in dieser Session ("Phase 9") umgesetzten Punkte:
|
||||
|
||||
1) Bugfix "Host-Key ermitteln" 500 -> discover_and_store_host_key()
|
||||
fehlgeschlagene Verbindungen wurden bisher NICHT abgefangen (siehe
|
||||
app/ssh_proxy/proxy.py); der Endpunkt muss jetzt 502 (echter
|
||||
Verbindungsfehler) statt eines unbehandelten 500 liefern, und bei einem
|
||||
Fingerprint trotz Auth-Fehler NACH dem Key-Exchange weiterhin 200.
|
||||
(Punkt 2 "Login-Verlauf entfernen" und Punkt 5 "Hostgruppen/Server im
|
||||
Menue trennen" sind reine Admin-UI-Aenderungen ohne eigenen Endpunkt --
|
||||
dafuer siehe templates/admin.html + static/js/admin.js, keine
|
||||
Backend-Tests noetig/moeglich.)
|
||||
3) Verbindungslog (Live-Tail) -- app/security/log_stream.py ist reine
|
||||
In-Process-Logik (Ring-Buffer + Pub/Sub) ohne DB-/HTTP-Abhaengigkeit und
|
||||
wird NICHT hier, sondern eigenstaendig verifiziert (siehe
|
||||
verify_migrations.py/verify_hostkey_fix.py-Analoga aus der
|
||||
Sandbox-Verifikation dieser Session); ein WebSocket-Test wuerde einen
|
||||
echten ASGI-WS-Client benoetigen, den dieses Testsetup (httpx
|
||||
ASGITransport, kein WS-Support) nicht bietet.
|
||||
4) Superadmin-'Sessionview': GET /admin/sessions, POST
|
||||
/admin/sessions/{id}/terminate, GET /admin/sessions/{id}/recording --
|
||||
alle require_global_admin (auch Mandanten-Admins muessen 403 bekommen).
|
||||
6) "Credentials ins RBAC-Modell": neue Rollen credentials_view/
|
||||
credentials_manage (Migration 0008) erlauben NICHT-Admins gezielten
|
||||
Zugriff auf Zugangsdaten-Endpunkte fuer Hosts ihrer Hostgruppe.
|
||||
"""
|
||||
from __future__ import annotations
|
||||
|
||||
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="p9-group", hostname="p9-host"):
|
||||
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.9.0.1', 'ssh', 22, 'linux')",
|
||||
(hg_id, hostname),
|
||||
)
|
||||
await conn.commit()
|
||||
return hg_id, cursor.lastrowid
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# 1) Host-Key ermitteln: kein 500 mehr bei Verbindungsfehlern
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_discover_host_key_connection_failure_returns_502_not_500(client, monkeypatch):
|
||||
from app.db import get_db
|
||||
import app.admin.routes as admin_routes
|
||||
from app.ssh_proxy.proxy import HostKeyDiscoveryError
|
||||
|
||||
conn = get_db()
|
||||
await _create_user(conn, "hk_admin", "Correct-Horse-Battery-Staple-K1", is_admin=True)
|
||||
hg_id, host_id = await _setup_hostgroup_and_host(conn, group_name="hk-group", hostname="hk-host")
|
||||
await _login_full(client, "hk_admin", "Correct-Horse-Battery-Staple-K1")
|
||||
|
||||
async def _boom(conn, host_id, *, admin_user_id):
|
||||
raise HostKeyDiscoveryError(host_id, "Connection refused")
|
||||
|
||||
monkeypatch.setattr(admin_routes, "discover_and_store_host_key", _boom)
|
||||
|
||||
resp = await client.post(f"/admin/hosts/{host_id}/discover-host-key", json={})
|
||||
# Vorher: unbehandelte Exception -> 500. Jetzt: sauber gemappt auf 502.
|
||||
assert resp.status_code == 502, resp.text
|
||||
assert "Connection refused" in resp.json()["detail"]
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_discover_host_key_success_after_auth_failure_past_kex(client, monkeypatch):
|
||||
"""Simuliert den eigentlichen Bug-Fall: Key-Exchange erfolgreich (Fingerprint
|
||||
erfasst), Authentifizierung schlaegt DANACH fehl -- muss trotzdem als
|
||||
Erfolg gemeldet werden (siehe verify_hostkey_fix.py fuer die isolierte
|
||||
Kontrollfluss-Verifikation der proxy.py-Logik selbst)."""
|
||||
from app.db import get_db
|
||||
import app.admin.routes as admin_routes
|
||||
|
||||
conn = get_db()
|
||||
await _create_user(conn, "hk_admin2", "Correct-Horse-Battery-Staple-K2", is_admin=True)
|
||||
hg_id, host_id = await _setup_hostgroup_and_host(conn, group_name="hk-group2", hostname="hk-host2")
|
||||
await _login_full(client, "hk_admin2", "Correct-Horse-Battery-Staple-K2")
|
||||
|
||||
async def _fake_discover(conn, host_id, *, admin_user_id):
|
||||
return "SHA256:fake-fingerprint-after-auth-failure"
|
||||
|
||||
monkeypatch.setattr(admin_routes, "discover_and_store_host_key", _fake_discover)
|
||||
|
||||
resp = await client.post(f"/admin/hosts/{host_id}/discover-host-key", json={})
|
||||
assert resp.status_code == 200, resp.text
|
||||
assert resp.json()["fingerprint"] == "SHA256:fake-fingerprint-after-auth-failure"
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# 4) Sessionview (nur Super-Admin)
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_sessions_list_requires_global_admin_not_tenant_admin(client):
|
||||
from app.db import get_db
|
||||
|
||||
conn = get_db()
|
||||
superadmin_id = await _create_user(conn, "sv_super", "Correct-Horse-Battery-Staple-S1", is_admin=True)
|
||||
tenant_admin_id = await _create_user(conn, "sv_tenant", "Correct-Horse-Battery-Staple-S2")
|
||||
await conn.execute(
|
||||
"INSERT INTO tenant_admins (user_id, tenant_id) VALUES (?, 1)", (tenant_admin_id,)
|
||||
)
|
||||
await conn.commit()
|
||||
|
||||
await _login_full(client, "sv_tenant", "Correct-Horse-Battery-Staple-S2")
|
||||
resp = await client.get("/admin/sessions")
|
||||
assert resp.status_code == 403, resp.text
|
||||
|
||||
client.cookies.clear()
|
||||
await _login_full(client, "sv_super", "Correct-Horse-Battery-Staple-S1")
|
||||
resp = await client.get("/admin/sessions")
|
||||
assert resp.status_code == 200, resp.text
|
||||
assert resp.json() == []
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_sessions_list_and_terminate_and_recording(client):
|
||||
from app.db import get_db
|
||||
|
||||
conn = get_db()
|
||||
admin_id = await _create_user(conn, "sv_super2", "Correct-Horse-Battery-Staple-S3", is_admin=True)
|
||||
user_id = await _create_user(conn, "sv_user", "Correct-Horse-Battery-Staple-S4")
|
||||
hg_id, host_id = await _setup_hostgroup_and_host(conn, group_name="sv-group", hostname="sv-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, "sv_super2", "Correct-Horse-Battery-Staple-S3")
|
||||
|
||||
resp = await client.get("/admin/sessions?active_only=true")
|
||||
assert resp.status_code == 200, resp.text
|
||||
rows = resp.json()
|
||||
assert len(rows) == 1
|
||||
assert rows[0]["id"] == session_id
|
||||
assert rows[0]["username"] == "sv_user"
|
||||
assert rows[0]["hostname"] == "sv-host"
|
||||
assert rows[0]["is_active"] is True
|
||||
# Diese Sitzung wurde nur direkt in der DB angelegt (kein echter
|
||||
# laufender WS-Task) -> nicht in app/security/active_sessions.py
|
||||
# registriert -> darf NICHT als 'killable' gemeldet werden.
|
||||
assert rows[0]["killable"] is False
|
||||
assert rows[0]["has_recording"] is False
|
||||
|
||||
# 'Beenden' muss sauber 409 liefern statt eine KeyError/AttributeError zu
|
||||
# werfen, wenn die Sitzung nicht (mehr) auf diesem Prozess laeuft.
|
||||
resp = await client.post(f"/admin/sessions/{session_id}/terminate", json={})
|
||||
assert resp.status_code == 409, resp.text
|
||||
|
||||
resp = await client.get(f"/admin/sessions/{session_id}/recording")
|
||||
assert resp.status_code == 404, resp.text
|
||||
|
||||
resp = await client.post("/admin/sessions/999999/terminate", json={})
|
||||
assert resp.status_code == 404, resp.text
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# 6) Credentials ins RBAC-Modell
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_credentials_manage_role_grants_non_admin_write_access(client):
|
||||
from app.db import get_db
|
||||
|
||||
conn = get_db()
|
||||
admin_id = await _create_user(conn, "cr_admin", "Correct-Horse-Battery-Staple-C1", is_admin=True)
|
||||
holder_id = await _create_user(conn, "cr_holder", "Correct-Horse-Battery-Staple-C2")
|
||||
other_id = await _create_user(conn, "cr_other", "Correct-Horse-Battery-Staple-C3")
|
||||
hg_id, host_id = await _setup_hostgroup_and_host(conn, group_name="cr-group", hostname="cr-host")
|
||||
|
||||
await _login_full(client, "cr_admin", "Correct-Horse-Battery-Staple-C1")
|
||||
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
|
||||
|
||||
# Rolleninhaber (kein Admin!) darf Zugangsdaten lesen UND setzen.
|
||||
client.cookies.clear()
|
||||
await _login_full(client, "cr_holder", "Correct-Horse-Battery-Staple-C2")
|
||||
|
||||
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
|
||||
|
||||
resp = await client.put(f"/admin/hosts/{host_id}/rdp-credentials", json={"password": "s3hr-geheim!!"})
|
||||
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 True
|
||||
|
||||
# Ein User OHNE diese Rolle bleibt weiterhin ausgesperrt.
|
||||
client.cookies.clear()
|
||||
await _login_full(client, "cr_other", "Correct-Horse-Battery-Staple-C3")
|
||||
resp = await client.get(f"/admin/hosts/{host_id}/credentials")
|
||||
assert resp.status_code == 403, resp.text
|
||||
resp = await client.put(f"/admin/hosts/{host_id}/rdp-credentials", json={"password": "andere-1234"})
|
||||
assert resp.status_code == 403, resp.text
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_credentials_view_role_is_read_only(client):
|
||||
from app.db import get_db
|
||||
|
||||
conn = get_db()
|
||||
await _create_user(conn, "cv_admin", "Correct-Horse-Battery-Staple-C4", is_admin=True)
|
||||
viewer_id = await _create_user(conn, "cv_viewer", "Correct-Horse-Battery-Staple-C5")
|
||||
hg_id, host_id = await _setup_hostgroup_and_host(conn, group_name="cv-group", hostname="cv-host")
|
||||
|
||||
await _login_full(client, "cv_admin", "Correct-Horse-Battery-Staple-C4")
|
||||
resp = await client.post(
|
||||
"/admin/roles/grant",
|
||||
json={"user_id": viewer_id, "host_group_id": hg_id, "role_names": ["credentials_view"]},
|
||||
)
|
||||
assert resp.status_code == 200, resp.text
|
||||
|
||||
client.cookies.clear()
|
||||
await _login_full(client, "cv_viewer", "Correct-Horse-Battery-Staple-C5")
|
||||
|
||||
resp = await client.get(f"/admin/hosts/{host_id}/credentials")
|
||||
assert resp.status_code == 200, resp.text
|
||||
|
||||
# 'credentials_view' allein darf NICHT schreiben.
|
||||
resp = await client.put(f"/admin/hosts/{host_id}/rdp-credentials", json={"password": "nope-12345"})
|
||||
assert resp.status_code == 403, resp.text
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_catalog_hosts_reports_can_view_credentials_flag(client):
|
||||
from app.db import get_db
|
||||
|
||||
conn = get_db()
|
||||
await _create_user(conn, "cc_admin", "Correct-Horse-Battery-Staple-C6", is_admin=True)
|
||||
user_id = await _create_user(conn, "cc_user", "Correct-Horse-Battery-Staple-C7")
|
||||
hg_id, host_id = await _setup_hostgroup_and_host(conn, group_name="cc-group", hostname="cc-host")
|
||||
|
||||
await _login_full(client, "cc_admin", "Correct-Horse-Battery-Staple-C6")
|
||||
await client.post(
|
||||
"/admin/roles/grant",
|
||||
json={"user_id": user_id, "host_group_id": hg_id, "role_names": ["ssh_connect", "credentials_view"]},
|
||||
)
|
||||
|
||||
client.cookies.clear()
|
||||
await _login_full(client, "cc_user", "Correct-Horse-Battery-Staple-C7")
|
||||
resp = await client.get("/catalog/hosts")
|
||||
assert resp.status_code == 200, resp.text
|
||||
hosts = resp.json()
|
||||
assert len(hosts) == 1
|
||||
assert hosts[0]["can_view_credentials"] is True
|
||||
Reference in New Issue
Block a user