"""Tests fuer Teil D.6 Schritt 6 ("Konsolidierung"): 1) GET /admin/reports/personal-groups gruppiert persoenliche Gruppen nach identischem Rechteprofil -- zwei Gruppen mit identischen Rechten landen im selben Cluster (consolidation_candidate=True), eine Gruppe mit abweichenden Rechten bleibt allein in ihrem Cluster. 2) GET /admin/reports/unconfirmed-credential-grants zeigt nur Achse-B- Freigaben mit granted_by IS NULL (Vorbefuellungs-Herkunft) -- eine nachtraeglich manuell bestaetigte Freigabe (granted_by gesetzt) verschwindet aus dem Report. Beide Endpunkte sind rein lesend; kein Test fuehrt eine automatische Zusammenlegung oder einen automatischen Entzug durch (den gibt es bewusst nicht, siehe Docstrings in app/admin/routes.py). """ from __future__ import annotations import pyotp 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: 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 _make_personal_group(conn, *, name: str, user_id: int) -> int: cursor = await conn.execute( "INSERT INTO user_groups (name, description, is_personal) VALUES (?, NULL, 1)", (name,), ) group_id = cursor.lastrowid await conn.execute( "INSERT INTO user_group_members (user_group_id, user_id, added_by) VALUES (?, ?, NULL)", (group_id, user_id), ) await conn.commit() return group_id @pytest.mark.asyncio async def test_personal_groups_report_clusters_identical_rights(client): from app.db import get_db conn = get_db() admin_id = await _create_user(conn, "s6_admin1", "Correct-Horse-Battery-Staple-1", is_admin=True) u1 = await _create_user(conn, "s6_u1", "Correct-Horse-Battery-Staple-2") u2 = await _create_user(conn, "s6_u2", "Correct-Horse-Battery-Staple-3") u3 = await _create_user(conn, "s6_u3", "Correct-Horse-Battery-Staple-4") await conn.execute("INSERT INTO host_groups (id, name) VALUES (5001, 's6-hg')") role_row = await (await conn.execute("SELECT id FROM roles WHERE name = 'ssh_connect'")).fetchone() role_id = role_row[0] await conn.commit() pg1 = await _make_personal_group(conn, name="s6-pg-1", user_id=u1) pg2 = await _make_personal_group(conn, name="s6-pg-2", user_id=u2) pg3 = await _make_personal_group(conn, name="s6-pg-3", user_id=u3) # pg1 und pg2 bekommen identische Rechte (ssh_connect auf derselben # Hostgruppe) -- Konsolidierungs-Kandidat. pg3 bleibt ohne jede Rolle # und muss daher in einem eigenen (ebenfalls "identischen", aber # einelementigen) Cluster landen. for group_id in (pg1, pg2): await conn.execute( "INSERT INTO group_hostgroup_roles (user_group_id, host_group_id, role_id, granted_by) " "VALUES (?, 5001, ?, ?)", (group_id, role_id, admin_id), ) await conn.commit() await _login_full(client, "s6_admin1", "Correct-Horse-Battery-Staple-1") resp = await client.get("/admin/reports/personal-groups") assert resp.status_code == 200, resp.text clusters = resp.json() def cluster_containing(group_id): for c in clusters: if any(m["user_group_id"] == group_id for m in c["members"]): return c return None c1 = cluster_containing(pg1) c2 = cluster_containing(pg2) c3 = cluster_containing(pg3) assert c1 is c2, "pg1 und pg2 muessen im selben Cluster landen (identisches Rechteprofil)" assert c1["consolidation_candidate"] is True assert c1["group_count"] == 2 assert {m["user_group_id"] for m in c1["members"]} == {pg1, pg2} assert c3 is not c1 assert c3["consolidation_candidate"] is False assert c3["group_count"] == 1 @pytest.mark.asyncio async def test_unconfirmed_credential_grants_report(client): from app.db import get_db conn = get_db() admin_id = await _create_user(conn, "s6_admin2", "Correct-Horse-Battery-Staple-5", is_admin=True) await _login_full(client, "s6_admin2", "Correct-Horse-Battery-Staple-5") resp = await client.post("/admin/user-groups", json={"name": "s6-cred-gruppe", "description": None}) group_id = resp.json()["id"] resp = await client.post("/admin/ssh-keys/generate", json={"key_type": "ed25519"}) km = resp.json() resp = await client.post( "/admin/ssh-keys", json={ "label": "s6-key", "key_type": "ed25519", "private_key_pem": km["private_key_pem"], "public_key": km["public_key"], "passphrase": None, "username": "root", }, ) assert resp.status_code == 201, resp.text key_id = resp.json()["id"] # Simuliert eine Vorbefuellungs-Zeile aus Migration 0018: granted_by IS NULL. await conn.execute( "INSERT INTO group_ssh_key_grants (user_group_id, ssh_key_id, granted_by) VALUES (?, ?, NULL)", (group_id, key_id), ) await conn.commit() resp = await client.get("/admin/reports/unconfirmed-credential-grants") assert resp.status_code == 200, resp.text rows = resp.json() match = [r for r in rows if r["user_group_id"] == group_id and r["credential_id"] == key_id] assert len(match) == 1, rows assert match[0]["kind"] == "ssh_key" # Jetzt manuell ueber die API bestaetigen (granted_by wird auf admin.id # gesetzt) -- die Zeile muss aus dem Report verschwinden. resp = await client.post( "/admin/group-credentials/ssh_key/grant", json={"user_group_id": group_id, "credential_id": key_id, "expires_at": None}, ) assert resp.status_code == 200, resp.text resp = await client.get("/admin/reports/unconfirmed-credential-grants") rows = resp.json() match = [r for r in rows if r["user_group_id"] == group_id and r["credential_id"] == key_id] assert len(match) == 0, rows