342 lines
14 KiB
Python
342 lines
14 KiB
Python
"""Tests fuer Teil D.6 Schritt 5 ("Schreibpfade abschalten"):
|
|
|
|
1) POST /admin/roles/grant und /revoke antworten mit HTTP 410.
|
|
2) GET /admin/roles liefert die abgeleitete Effektiv-Rechte-Sicht
|
|
(via_group_name) statt einer Vergabe-Tabelle.
|
|
3) Die neuen Achse-B-Endpunkte (/admin/group-credentials/{kind}/grant|
|
|
revoke, GET .../{kind}) funktionieren fuer alle drei Credential-Arten.
|
|
4) GET /admin/ssh-password-credentials liefert eine globale Liste.
|
|
5) S4: ein Nicht-Admin mit credentials_manage darf NUR einen SSH-Key/eine
|
|
RDP-Zugangsdaten-Ressource an einen Host haengen, wenn seine Gruppe sie
|
|
ueber Achse B freigegeben hat -- ein fremder, nicht freigegebener
|
|
Datensatz wird mit 403 abgelehnt. Ein API-Token mit passendem Scope
|
|
bleibt davon unberuehrt (admin.is_token).
|
|
6) S10: der Audit-Event bei Gruppenbeitritt/-austritt enthaelt
|
|
gained_rights/lost_rights.
|
|
"""
|
|
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 _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
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_direct_role_grant_endpoints_are_gone(client):
|
|
from app.db import get_db
|
|
|
|
conn = get_db()
|
|
await _create_user(conn, "s5_admin1", "Correct-Horse-Battery-Staple-1", is_admin=True)
|
|
await _login_full(client, "s5_admin1", "Correct-Horse-Battery-Staple-1")
|
|
|
|
resp = await client.post("/admin/roles/grant", json={"foo": "bar"})
|
|
assert resp.status_code == 410, resp.text
|
|
|
|
resp = await client.post("/admin/roles/revoke", json={"foo": "bar"})
|
|
assert resp.status_code == 410, resp.text
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_effective_roles_view_shows_via_group(client):
|
|
from app.db import get_db
|
|
|
|
conn = get_db()
|
|
admin_id = await _create_user(conn, "s5_admin2", "Correct-Horse-Battery-Staple-2", is_admin=True)
|
|
member_id = await _create_user(conn, "s5_member2", "Correct-Horse-Battery-Staple-3")
|
|
hg_id, host_id = await _setup_hostgroup_and_host(conn, group_name="s5-hg", hostname="s5-host")
|
|
|
|
await _login_full(client, "s5_admin2", "Correct-Horse-Battery-Staple-2")
|
|
|
|
resp = await client.post("/admin/user-groups", json={"name": "s5-gruppe", "description": None})
|
|
assert resp.status_code == 201, resp.text
|
|
group_id = resp.json()["id"]
|
|
|
|
resp = await client.post(f"/admin/user-groups/{group_id}/members", json={"user_id": member_id})
|
|
assert resp.status_code == 201, resp.text
|
|
|
|
resp = await client.post(
|
|
"/admin/group-roles/grant",
|
|
json={"user_group_id": group_id, "host_group_id": hg_id, "role_names": ["ssh_connect"]},
|
|
)
|
|
assert resp.status_code == 200, resp.text
|
|
|
|
resp = await client.get("/admin/roles")
|
|
assert resp.status_code == 200, resp.text
|
|
rows = resp.json()
|
|
match = [r for r in rows if r["user_id"] == member_id and r["host_group_id"] == hg_id]
|
|
assert len(match) == 1, rows
|
|
row = match[0]
|
|
assert row["role_name"] == "ssh_connect"
|
|
assert row["via_group_id"] == group_id
|
|
assert row["via_group_name"] == "s5-gruppe"
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_group_credential_grant_revoke_list_all_kinds(client):
|
|
from app.db import get_db
|
|
from app.security.crypto import encrypt_secret
|
|
|
|
conn = get_db()
|
|
await _create_user(conn, "s5_admin3", "Correct-Horse-Battery-Staple-4", is_admin=True)
|
|
await _login_full(client, "s5_admin3", "Correct-Horse-Battery-Staple-4")
|
|
|
|
resp = await client.post("/admin/user-groups", json={"name": "s5-cred-gruppe", "description": None})
|
|
group_id = resp.json()["id"]
|
|
|
|
# SSH-Key anlegen
|
|
resp = await client.post(
|
|
"/admin/ssh-keys/generate", json={"key_type": "ed25519"},
|
|
)
|
|
assert resp.status_code == 200, resp.text
|
|
key_material = resp.json()
|
|
resp = await client.post(
|
|
"/admin/ssh-keys",
|
|
json={
|
|
"label": "s5-key", "key_type": "ed25519",
|
|
"private_key_pem": key_material["private_key_pem"],
|
|
"public_key": key_material["public_key"],
|
|
"passphrase": None, "username": "root",
|
|
},
|
|
)
|
|
assert resp.status_code == 201, resp.text
|
|
ssh_key_id = resp.json()["id"]
|
|
|
|
# RDP-Zugangsdaten anlegen
|
|
resp = await client.post(
|
|
"/admin/rdp-credentials",
|
|
json={"label": "s5-rdp", "username": "administrator", "domain": None, "password": "Sup3rSecret!"},
|
|
)
|
|
assert resp.status_code == 201, resp.text
|
|
rdp_id = resp.json()["id"]
|
|
|
|
# SSH-Passwort-Objekt direkt in der DB anlegen (kein eigener globaler
|
|
# Create-Endpunkt -- entsteht normalerweise ueber PUT .../ssh-password).
|
|
encrypted = encrypt_secret(b"hunter2", associated_data=b"ssh_password")
|
|
cursor = await conn.execute(
|
|
"INSERT INTO ssh_password_credentials (label, username, password_enc) VALUES (?, ?, ?)",
|
|
("s5-pw", "root", encrypted),
|
|
)
|
|
await conn.commit()
|
|
pw_id = cursor.lastrowid
|
|
|
|
cases = [
|
|
("ssh_key", ssh_key_id),
|
|
("rdp_credential", rdp_id),
|
|
("ssh_password_credential", pw_id),
|
|
]
|
|
for kind, credential_id in cases:
|
|
resp = await client.post(
|
|
f"/admin/group-credentials/{kind}/grant",
|
|
json={"user_group_id": group_id, "credential_id": credential_id, "expires_at": None},
|
|
)
|
|
assert resp.status_code == 200, (kind, resp.text)
|
|
|
|
resp = await client.get(f"/admin/group-credentials/{kind}")
|
|
assert resp.status_code == 200, (kind, resp.text)
|
|
rows = resp.json()
|
|
assert any(
|
|
r["user_group_id"] == group_id and r["credential_id"] == credential_id for r in rows
|
|
), (kind, rows)
|
|
|
|
resp = await client.post(
|
|
f"/admin/group-credentials/{kind}/revoke",
|
|
json={"user_group_id": group_id, "credential_id": credential_id},
|
|
)
|
|
assert resp.status_code == 200, (kind, resp.text)
|
|
|
|
resp = await client.get(f"/admin/group-credentials/{kind}")
|
|
rows = resp.json()
|
|
assert not any(
|
|
r["user_group_id"] == group_id and r["credential_id"] == credential_id for r in rows
|
|
), (kind, rows)
|
|
|
|
# Unbekannte Credential-Art im Pfad -> 422 (Literal-Validierung)
|
|
resp = await client.post(
|
|
f"/admin/group-credentials/telepathy/grant",
|
|
json={"user_group_id": group_id, "credential_id": ssh_key_id, "expires_at": None},
|
|
)
|
|
assert resp.status_code == 422, resp.text
|
|
|
|
# Nicht existierender Zugangsdatensatz -> 404
|
|
resp = await client.post(
|
|
"/admin/group-credentials/ssh_key/grant",
|
|
json={"user_group_id": group_id, "credential_id": 999999, "expires_at": None},
|
|
)
|
|
assert resp.status_code == 404, resp.text
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_list_ssh_password_credentials_global(client):
|
|
from app.db import get_db
|
|
from app.security.crypto import encrypt_secret
|
|
|
|
conn = get_db()
|
|
await _create_user(conn, "s5_admin4", "Correct-Horse-Battery-Staple-5", is_admin=True)
|
|
await _login_full(client, "s5_admin4", "Correct-Horse-Battery-Staple-5")
|
|
|
|
hg_id, host_id = await _setup_hostgroup_and_host(conn, group_name="s5-pw-hg", hostname="s5-pw-host")
|
|
encrypted = encrypt_secret(b"hunter2", associated_data=b"ssh_password")
|
|
cursor = await conn.execute(
|
|
"INSERT INTO ssh_password_credentials (label, username, password_enc) VALUES (?, ?, ?)",
|
|
("s5-pw-list", "root", encrypted),
|
|
)
|
|
pw_id = cursor.lastrowid
|
|
await conn.execute(
|
|
"INSERT INTO host_ssh_password_credential_map (host_id, ssh_password_credential_id) VALUES (?, ?)",
|
|
(host_id, pw_id),
|
|
)
|
|
await conn.commit()
|
|
|
|
resp = await client.get("/admin/ssh-password-credentials")
|
|
assert resp.status_code == 200, resp.text
|
|
rows = resp.json()
|
|
match = [r for r in rows if r["id"] == pw_id]
|
|
assert len(match) == 1, rows
|
|
assert match[0]["assigned_hosts"] == [{"id": host_id, "hostname": "s5-pw-host"}]
|
|
assert "password" not in match[0] and "password_enc" not in match[0]
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_s4_non_admin_needs_group_grant_to_attach_credential(client):
|
|
from app.db import get_db
|
|
|
|
conn = get_db()
|
|
admin_id = await _create_user(conn, "s5_admin5", "Correct-Horse-Battery-Staple-6", is_admin=True)
|
|
member_id = await _create_user(conn, "s5_member5", "Correct-Horse-Battery-Staple-7")
|
|
hg_id, host_id = await _setup_hostgroup_and_host(conn, group_name="s5-s4-hg", hostname="s5-s4-host")
|
|
|
|
await _login_full(client, "s5_admin5", "Correct-Horse-Battery-Staple-6")
|
|
|
|
resp = await client.post("/admin/user-groups", json={"name": "s5-s4-gruppe", "description": None})
|
|
group_id = resp.json()["id"]
|
|
resp = await client.post(f"/admin/user-groups/{group_id}/members", json={"user_id": member_id})
|
|
assert resp.status_code == 201, resp.text
|
|
resp = await client.post(
|
|
"/admin/group-roles/grant",
|
|
json={
|
|
"user_group_id": group_id, "host_group_id": hg_id,
|
|
"role_names": ["ssh_connect", "credentials_manage"],
|
|
},
|
|
)
|
|
assert resp.status_code == 200, resp.text
|
|
|
|
# Zwei SSH-Keys anlegen: einer wird der Gruppe freigegeben, der andere nicht.
|
|
key_ids = {}
|
|
for label in ("granted", "not-granted"):
|
|
resp = await client.post("/admin/ssh-keys/generate", json={"key_type": "ed25519"})
|
|
km = resp.json()
|
|
resp = await client.post(
|
|
"/admin/ssh-keys",
|
|
json={
|
|
"label": f"s5-s4-{label}", "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_ids[label] = resp.json()["id"]
|
|
|
|
resp = await client.post(
|
|
"/admin/group-credentials/ssh_key/grant",
|
|
json={"user_group_id": group_id, "credential_id": key_ids["granted"], "expires_at": None},
|
|
)
|
|
assert resp.status_code == 200, resp.text
|
|
|
|
client.cookies.clear()
|
|
await _login_full(client, "s5_member5", "Correct-Horse-Battery-Staple-7")
|
|
|
|
# Freigegebener Key: darf angehaengt werden.
|
|
resp = await client.post(f"/admin/hosts/{host_id}/ssh-keys/{key_ids['granted']}")
|
|
assert resp.status_code == 200, resp.text
|
|
|
|
# Nicht freigegebener Key (existiert im System, aber der Gruppe nicht
|
|
# freigegeben) -- S4: muss abgelehnt werden.
|
|
resp = await client.post(f"/admin/hosts/{host_id}/ssh-keys/{key_ids['not-granted']}")
|
|
assert resp.status_code == 403, resp.text
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_s10_audit_event_carries_gained_and_lost_rights(client):
|
|
from app.db import get_db
|
|
|
|
conn = get_db()
|
|
admin_id = await _create_user(conn, "s5_admin6", "Correct-Horse-Battery-Staple-8", is_admin=True)
|
|
member_id = await _create_user(conn, "s5_member6", "Correct-Horse-Battery-Staple-9")
|
|
hg_id, host_id = await _setup_hostgroup_and_host(conn, group_name="s5-s10-hg", hostname="s5-s10-host")
|
|
|
|
await _login_full(client, "s5_admin6", "Correct-Horse-Battery-Staple-8")
|
|
|
|
resp = await client.post("/admin/user-groups", json={"name": "s5-s10-gruppe", "description": None})
|
|
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"]},
|
|
)
|
|
assert resp.status_code == 200, resp.text
|
|
|
|
resp = await client.post(f"/admin/user-groups/{group_id}/members", json={"user_id": member_id})
|
|
assert resp.status_code == 201, resp.text
|
|
|
|
row = await (await conn.execute(
|
|
"SELECT details_json FROM audit_log WHERE event_type = 'user_group_member_added' "
|
|
"ORDER BY id DESC LIMIT 1"
|
|
)).fetchone()
|
|
import json as _json
|
|
details = _json.loads(row[0])
|
|
assert "gained_rights" in details, details
|
|
assert any(
|
|
r["host_group"] == "s5-s10-hg" and r["role"] == "ssh_connect"
|
|
for r in details["gained_rights"]["roles"]
|
|
), details
|
|
|
|
resp = await client.delete(f"/admin/user-groups/{group_id}/members/{member_id}")
|
|
assert resp.status_code == 200, resp.text
|
|
|
|
row = await (await conn.execute(
|
|
"SELECT details_json FROM audit_log WHERE event_type = 'user_group_member_removed' "
|
|
"ORDER BY id DESC LIMIT 1"
|
|
)).fetchone()
|
|
details = _json.loads(row[0])
|
|
assert "lost_rights" in details, details
|
|
assert any(
|
|
r["host_group"] == "s5-s10-hg" and r["role"] == "ssh_connect"
|
|
for r in details["lost_rights"]["roles"]
|
|
), details
|