umbau 1.0
This commit is contained in:
397
tests/test_admin_crud.py
Normal file
397
tests/test_admin_crud.py
Normal file
@ -0,0 +1,397 @@
|
||||
"""
|
||||
Tests fuer die CRUD-Vervollstaendigung (Edit/Delete fuer alles, was die
|
||||
Admin-API anlegt, siehe 0007_crud_extras.sql) und die Mehrfachauswahl bei
|
||||
Rollenvergaben (role_names: list statt role_name: str).
|
||||
|
||||
Vormals (bis Teil C des Umsetzungsauftrags) enthielt diese Datei zusaetzlich
|
||||
Tests fuer die inzwischen vollstaendig entfernte Mandantenfaehigkeit
|
||||
(app/tenancy.py, 0006_tenants.sql) -- siehe app/db/migrations/
|
||||
0014_drop_tenants.sql fuer den Rueckbau. Diese Datei heisst deshalb nicht
|
||||
mehr test_tenants.py.
|
||||
|
||||
Deckt ab:
|
||||
1) Mehrfachauswahl bei Rollenvergabe (role_names) funktioniert fuer
|
||||
Einzel-User UND Benutzergruppen.
|
||||
2) Update/Delete-Endpunkte fuer User, Hostgruppen, Hosts, SSH-Keys,
|
||||
Benutzergruppen funktionieren wie vorgesehen (inkl. Anonymisieren statt
|
||||
Hard-Delete bei vorhandener Audit-Historie, Soft-Delete bei Hosts).
|
||||
3) GET /admin/hosts/{id} (Grundlage fuer den "Details"-Fix) liefert den
|
||||
vollstaendigen aktuellen Datensatz inkl. SSH-Key-Zuordnungen.
|
||||
"""
|
||||
from __future__ import annotations
|
||||
|
||||
import pyotp
|
||||
import pytest
|
||||
|
||||
from cryptography.hazmat.primitives import serialization
|
||||
from cryptography.hazmat.primitives.asymmetric import ed25519
|
||||
|
||||
|
||||
def _key_pem() -> str:
|
||||
"""Echtes, unverschluesseltes Schluesselmaterial.
|
||||
|
||||
Seit Phase 10 prueft POST/PUT /admin/ssh-keys das Material sofort gegen
|
||||
asyncssh (damit ein unbrauchbarer Schluessel nicht erst beim ersten
|
||||
Verbindungsversuch auffaellt). Platzhalter wie "PEM" werden deshalb --
|
||||
voellig korrekt -- mit HTTP 400 abgelehnt; diese Tests hier pruefen CRUD
|
||||
und brauchen einen gueltigen Schluessel.
|
||||
"""
|
||||
return ed25519.Ed25519PrivateKey.generate().private_bytes(
|
||||
serialization.Encoding.PEM,
|
||||
serialization.PrivateFormat.OpenSSH,
|
||||
serialization.NoEncryption(),
|
||||
).decode()
|
||||
|
||||
|
||||
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 _setup_hostgroup_and_host(conn, *, group_name="team-a", hostname="srv-a"):
|
||||
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.0.0.9', 'ssh', 22, 'linux')",
|
||||
(hg_id, hostname),
|
||||
)
|
||||
await conn.commit()
|
||||
return hg_id, cursor.lastrowid
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# 1) Mehrfachauswahl bei Rollenvergabe (role_names)
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_direct_multi_role_grant_for_individual_user_is_retired(client):
|
||||
"""Teil D Schritt 5: Direktvergabe an einzelne Benutzer (auch mit
|
||||
Mehrfachauswahl -- vormals test_multi_role_grant_for_individual_user,
|
||||
siehe FORTSETZUNG_Teil_D.md Abschnitt 1c) ist ersatzlos entfallen. Diese
|
||||
Faehigkeit existiert nur noch fuer Benutzergruppen, siehe
|
||||
test_multi_role_grant_for_user_group() unten -- deshalb hier bewusst
|
||||
KEIN Ersatztest fuer den Einzel-User-Pfad, sondern die Bestaetigung,
|
||||
dass er tatsaechlich abgeschaltet ist (statt z.B. nur zu vergessen und
|
||||
fuer immer gruen zu bleiben, obwohl der Pfad gar nicht mehr existiert)."""
|
||||
from app.db import get_db
|
||||
|
||||
conn = get_db()
|
||||
await _create_user(conn, "mr_admin", "Correct-Horse-Battery-Staple-G1", is_admin=True)
|
||||
member_id = await _create_user(conn, "mr_member", "Correct-Horse-Battery-Staple-G2")
|
||||
hg_id, host_id = await _setup_hostgroup_and_host(conn, group_name="mr-team", hostname="mr-srv")
|
||||
|
||||
await _login_full(client, "mr_admin", "Correct-Horse-Battery-Staple-G1")
|
||||
resp = await client.post(
|
||||
"/admin/roles/grant",
|
||||
json={
|
||||
"user_id": member_id, "host_group_id": hg_id,
|
||||
"role_names": ["ssh_connect", "file_transfer"],
|
||||
},
|
||||
)
|
||||
assert resp.status_code == 410, resp.text
|
||||
|
||||
resp = await client.post(
|
||||
"/admin/roles/revoke",
|
||||
json={"user_id": member_id, "host_group_id": hg_id, "role_name": "file_transfer"},
|
||||
)
|
||||
assert resp.status_code == 410, resp.text
|
||||
|
||||
# GET /admin/roles (die abgeleitete Effektiv-Sicht, Schritt 5) zeigt
|
||||
# entsprechend keine Rolle fuer member_id -- der Grant-Versuch oben ist
|
||||
# tatsaechlich folgenlos geblieben, nicht nur mit 410 quittiert worden.
|
||||
resp = await client.get("/admin/roles")
|
||||
assert resp.status_code == 200, resp.text
|
||||
granted_roles = {r["role_name"] for r in resp.json() if r["user_id"] == member_id}
|
||||
assert granted_roles == set()
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_multi_role_grant_for_user_group(client):
|
||||
from app.db import get_db
|
||||
|
||||
conn = get_db()
|
||||
await _create_user(conn, "mrg_admin", "Correct-Horse-Battery-Staple-H1", is_admin=True)
|
||||
hg_id, host_id = await _setup_hostgroup_and_host(conn, group_name="mrg-team", hostname="mrg-srv")
|
||||
|
||||
await _login_full(client, "mrg_admin", "Correct-Horse-Battery-Staple-H1")
|
||||
resp = await client.post("/admin/user-groups", json={"name": "mrg-group"})
|
||||
group_id = resp.json()["id"]
|
||||
|
||||
resp = await client.post(
|
||||
"/admin/group-roles/grant",
|
||||
json={
|
||||
"user_group_id": group_id, "host_group_id": hg_id,
|
||||
"role_names": ["ssh_connect", "file_transfer"],
|
||||
},
|
||||
)
|
||||
assert resp.status_code == 200, resp.text
|
||||
assert set(resp.json()["roles_granted"]) == {"ssh_connect", "file_transfer"}
|
||||
|
||||
resp = await client.get("/admin/group-roles")
|
||||
roles = {r["role_name"] for r in resp.json() if r["user_group_id"] == group_id}
|
||||
assert {"ssh_connect", "file_transfer"} <= roles
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# 2) CRUD-Vervollstaendigung: Edit/Delete ueberall
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_user_update_deactivate_and_delete_semantics(client):
|
||||
from app.db import get_db
|
||||
|
||||
conn = get_db()
|
||||
await _create_user(conn, "crud_admin", "Correct-Horse-Battery-Staple-I1", is_admin=True)
|
||||
await _login_full(client, "crud_admin", "Correct-Horse-Battery-Staple-I1")
|
||||
|
||||
resp = await client.post(
|
||||
"/admin/users",
|
||||
json={"username": "crud_target", "initial_password": "Correct-Horse-Battery-Staple-I2"},
|
||||
)
|
||||
assert resp.status_code == 201, resp.text
|
||||
target_id = resp.json()["id"]
|
||||
|
||||
# Update: Passwort setzen -> must_change_password/session_version bumped.
|
||||
resp = await client.put(f"/admin/users/{target_id}", json={"new_password": "Correct-Horse-Battery-Staple-I3"})
|
||||
assert resp.status_code == 200, resp.text
|
||||
assert resp.json()["changed"] is True
|
||||
|
||||
# Deactivate.
|
||||
resp = await client.post(f"/admin/users/{target_id}/deactivate")
|
||||
assert resp.status_code == 200, resp.text
|
||||
resp = await client.get("/admin/users")
|
||||
row = next(u for u in resp.json() if u["id"] == target_id)
|
||||
assert row["is_active"] is False
|
||||
|
||||
# Reaktivieren ueber update.
|
||||
resp = await client.put(f"/admin/users/{target_id}", json={"is_active": True})
|
||||
assert resp.status_code == 200, resp.text
|
||||
|
||||
# Dieses Konto hat inzwischen Audit-Historie (user_updated/deactivated) ->
|
||||
# Loeschen anonymisiert statt hart zu entfernen.
|
||||
resp = await client.delete(f"/admin/users/{target_id}")
|
||||
assert resp.status_code == 200, resp.text
|
||||
assert resp.json()["hard_deleted"] is False
|
||||
|
||||
resp = await client.get("/admin/users")
|
||||
assert all(u["id"] != target_id for u in resp.json()), "Anonymisierter User darf nicht mehr gelistet werden"
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_user_without_audit_history_is_hard_deleted(client):
|
||||
from app.db import get_db
|
||||
|
||||
conn = get_db()
|
||||
await _create_user(conn, "crud_admin2", "Correct-Horse-Battery-Staple-J1", is_admin=True)
|
||||
await _login_full(client, "crud_admin2", "Correct-Horse-Battery-Staple-J1")
|
||||
|
||||
resp = await client.post(
|
||||
"/admin/users",
|
||||
json={"username": "throwaway", "initial_password": "Correct-Horse-Battery-Staple-J2"},
|
||||
)
|
||||
target_id = resp.json()["id"]
|
||||
|
||||
# Das einzige Audit-Ereignis, das diesen User referenziert, waere sein
|
||||
# eigenes Login -- er hat sich nie eingeloggt, also existiert keine
|
||||
# audit_log-Zeile mit user_id=target_id -> Hard-Delete moeglich.
|
||||
resp = await client.delete(f"/admin/users/{target_id}")
|
||||
assert resp.status_code == 200, resp.text
|
||||
assert resp.json()["hard_deleted"] is True
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_user_cannot_delete_own_account(client):
|
||||
from app.db import get_db
|
||||
|
||||
conn = get_db()
|
||||
await _create_user(conn, "self_admin", "Correct-Horse-Battery-Staple-K1", is_admin=True)
|
||||
await _login_full(client, "self_admin", "Correct-Horse-Battery-Staple-K1")
|
||||
resp = await client.get("/auth/me")
|
||||
my_id = resp.json()["id"]
|
||||
resp = await client.delete(f"/admin/users/{my_id}")
|
||||
assert resp.status_code == 400
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_host_group_and_host_update_delete(client):
|
||||
from app.db import get_db
|
||||
|
||||
conn = get_db()
|
||||
await _create_user(conn, "hcrud_admin", "Correct-Horse-Battery-Staple-L1", is_admin=True)
|
||||
await _login_full(client, "hcrud_admin", "Correct-Horse-Battery-Staple-L1")
|
||||
|
||||
resp = await client.post("/admin/host-groups", json={"name": "hcrud-team"})
|
||||
hg_id = resp.json()["id"]
|
||||
resp = await client.put(f"/admin/host-groups/{hg_id}", json={"description": "Aktualisiert"})
|
||||
assert resp.status_code == 200, resp.text
|
||||
|
||||
resp = await client.post(
|
||||
"/admin/hosts",
|
||||
json={
|
||||
"host_group_id": hg_id, "hostname": "hcrud-srv", "address": "10.2.0.1",
|
||||
"protocol": "ssh", "port": 22, "os_type": "linux",
|
||||
},
|
||||
)
|
||||
host_id = resp.json()["id"]
|
||||
|
||||
resp = await client.put(f"/admin/hosts/{host_id}", json={"hostname": "hcrud-srv-renamed"})
|
||||
assert resp.status_code == 200, resp.text
|
||||
resp = await client.get(f"/admin/hosts/{host_id}")
|
||||
assert resp.json()["hostname"] == "hcrud-srv-renamed"
|
||||
|
||||
# Hostgruppe kann nicht geloescht werden, solange sie noch (auch inaktive) Hosts enthaelt.
|
||||
resp = await client.delete(f"/admin/host-groups/{hg_id}")
|
||||
assert resp.status_code == 409, resp.text
|
||||
|
||||
# Host-Delete ist standardmaessig Soft-Delete.
|
||||
resp = await client.delete(f"/admin/hosts/{host_id}")
|
||||
assert resp.status_code == 200, resp.text
|
||||
assert resp.json()["hard_deleted"] is False
|
||||
resp = await client.get(f"/admin/hosts/{host_id}")
|
||||
assert resp.json()["is_active"] is False
|
||||
|
||||
# ?hard=true entfernt ihn tatsaechlich (keine Sitzungshistorie vorhanden).
|
||||
resp = await client.delete(f"/admin/hosts/{host_id}?hard=true")
|
||||
assert resp.status_code == 200, resp.text
|
||||
assert resp.json()["hard_deleted"] is True
|
||||
resp = await client.get(f"/admin/hosts/{host_id}")
|
||||
assert resp.status_code == 404
|
||||
|
||||
resp = await client.delete(f"/admin/host-groups/{hg_id}")
|
||||
assert resp.status_code == 200, resp.text
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_ssh_key_update_rotate_and_delete(client):
|
||||
from app.db import get_db
|
||||
|
||||
conn = get_db()
|
||||
await _create_user(conn, "kcrud_admin", "Correct-Horse-Battery-Staple-M1", is_admin=True)
|
||||
await _login_full(client, "kcrud_admin", "Correct-Horse-Battery-Staple-M1")
|
||||
|
||||
resp = await client.post(
|
||||
"/admin/ssh-keys",
|
||||
json={"label": "orig-key", "private_key_pem": _key_pem(), "public_key": "PUB", "key_type": "ed25519"},
|
||||
)
|
||||
key_id = resp.json()["id"]
|
||||
|
||||
resp = await client.put(f"/admin/ssh-keys/{key_id}", json={"label": "renamed-key"})
|
||||
assert resp.status_code == 200, resp.text
|
||||
|
||||
# Rotation erfordert alle drei Felder gemeinsam (400 noch vor jeder
|
||||
# Materialpruefung -- der Platzhalter hier ist deshalb Absicht).
|
||||
resp = await client.put(f"/admin/ssh-keys/{key_id}", json={"private_key_pem": "NEWPEM"})
|
||||
assert resp.status_code == 400, resp.text
|
||||
resp = await client.put(
|
||||
f"/admin/ssh-keys/{key_id}",
|
||||
json={"private_key_pem": _key_pem(), "public_key": "NEWPUB", "key_type": "rsa-4096"},
|
||||
)
|
||||
assert resp.status_code == 200, resp.text
|
||||
|
||||
resp = await client.delete(f"/admin/ssh-keys/{key_id}")
|
||||
assert resp.status_code == 200, resp.text
|
||||
resp = await client.get("/admin/ssh-keys")
|
||||
assert all(k["id"] != key_id for k in resp.json())
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_user_group_update_and_delete(client):
|
||||
from app.db import get_db
|
||||
|
||||
conn = get_db()
|
||||
await _create_user(conn, "gcrud_admin", "Correct-Horse-Battery-Staple-N1", is_admin=True)
|
||||
await _login_full(client, "gcrud_admin", "Correct-Horse-Battery-Staple-N1")
|
||||
|
||||
resp = await client.post("/admin/user-groups", json={"name": "gcrud-team"})
|
||||
group_id = resp.json()["id"]
|
||||
resp = await client.put(f"/admin/user-groups/{group_id}", json={"description": "Aktualisiert"})
|
||||
assert resp.status_code == 200, resp.text
|
||||
|
||||
resp = await client.delete(f"/admin/user-groups/{group_id}")
|
||||
assert resp.status_code == 200, resp.text
|
||||
resp = await client.get("/admin/user-groups")
|
||||
assert all(g["id"] != group_id for g in resp.json())
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# 3) GET /admin/hosts/{id} -- Grundlage fuer den "Details"-Fix
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_host_detail_endpoint_includes_ssh_keys_and_rdp_flag(client):
|
||||
from app.db import get_db
|
||||
|
||||
conn = get_db()
|
||||
await _create_user(conn, "detail_admin", "Correct-Horse-Battery-Staple-O1", is_admin=True)
|
||||
await _login_full(client, "detail_admin", "Correct-Horse-Battery-Staple-O1")
|
||||
|
||||
resp = await client.post("/admin/host-groups", json={"name": "detail-team"})
|
||||
hg_id = resp.json()["id"]
|
||||
resp = await client.post(
|
||||
"/admin/hosts",
|
||||
json={
|
||||
"host_group_id": hg_id, "hostname": "win-srv", "address": "10.3.0.1",
|
||||
"protocol": "rdp", "port": 3389, "os_type": "windows",
|
||||
},
|
||||
)
|
||||
host_id = resp.json()["id"]
|
||||
|
||||
resp = await client.post(
|
||||
"/admin/ssh-keys",
|
||||
json={"label": "detail-key", "private_key_pem": _key_pem(), "public_key": "PUB", "key_type": "ed25519"},
|
||||
)
|
||||
key_id = resp.json()["id"]
|
||||
resp = await client.post(f"/admin/hosts/{host_id}/ssh-keys/{key_id}")
|
||||
assert resp.status_code == 200, resp.text
|
||||
|
||||
# Migration 0012: RDP-Zugangsdaten sind ein eigenstaendiges Objekt, das
|
||||
# separat angelegt und dann dem Host zugewiesen wird (wie ein SSH-Key).
|
||||
resp = await client.post(
|
||||
"/admin/rdp-credentials",
|
||||
json={"label": "detail-cred", "username": "Administrator",
|
||||
"password": "Correct-Horse-Battery-Staple-O2"},
|
||||
)
|
||||
assert resp.status_code == 201, resp.text
|
||||
credential_id = resp.json()["id"]
|
||||
resp = await client.post(f"/admin/hosts/{host_id}/rdp-credentials/{credential_id}", json={})
|
||||
assert resp.status_code == 200, resp.text
|
||||
|
||||
resp = await client.get(f"/admin/hosts/{host_id}")
|
||||
assert resp.status_code == 200, resp.text
|
||||
data = resp.json()
|
||||
assert data["hostname"] == "win-srv"
|
||||
assert any(k["id"] == key_id for k in data["ssh_keys"])
|
||||
assert data["rdp_credentials_set"] is True
|
||||
assert data["rdp_credentials_id"] == credential_id
|
||||
|
||||
resp = await client.get("/admin/rdp-credentials")
|
||||
assert resp.status_code == 200, resp.text
|
||||
row = next(r for r in resp.json() if r["id"] == credential_id)
|
||||
assert any(h["id"] == host_id for h in row["assigned_hosts"])
|
||||
|
||||
resp = await client.get("/admin/hosts/424242")
|
||||
assert resp.status_code == 404
|
||||
Reference in New Issue
Block a user