add admin stuff
This commit is contained in:
344
tests/test_admin_groups_tokens.py
Normal file
344
tests/test_admin_groups_tokens.py
Normal file
@ -0,0 +1,344 @@
|
||||
"""
|
||||
Tests fuer die Erweiterung des RBAC-Modells um Benutzergruppen (Teams) mit
|
||||
voller Rollen-Vererbung ("Verbindungen mit einer Gruppe teilen", siehe
|
||||
app/rbac.py, app/db/migrations/0004_user_groups.sql) sowie fuer API-Tokens
|
||||
mit granularem ro/rw-Scope je Ressource (app/security/api_tokens.py,
|
||||
app/db/migrations/0005_api_tokens.sql).
|
||||
|
||||
Deckt ab:
|
||||
1) Gruppenmitgliedschaft gewaehrt/entzieht Katalog-Zugriff ohne
|
||||
individuellen Grant.
|
||||
2) Nur Admins duerfen Gruppen/Rollen/Tokens verwalten.
|
||||
3) Token-Scope wird durchgesetzt (read/write je Ressource, write impliziert
|
||||
read), abgelaufene/widerrufene Tokens werden abgelehnt.
|
||||
4) Tokens koennen sich NICHT selbst weitere Tokens ausstellen
|
||||
(Privilege-Escalation-Schutz) -- Token-Verwaltung bleibt reine
|
||||
Session-Admin-Aktion.
|
||||
5) /docs und /openapi.json sind admin-only (kein oeffentliches Swagger-UI).
|
||||
"""
|
||||
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
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# 1) Benutzergruppen: volle Rollen-Vererbung
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_group_role_grant_gives_catalog_access_without_individual_grant(client):
|
||||
from app.db import get_db
|
||||
|
||||
conn = get_db()
|
||||
admin_id = await _create_user(conn, "grp_admin", "Correct-Horse-Battery-Staple-G1", is_admin=True)
|
||||
member_id = await _create_user(conn, "grp_member", "Correct-Horse-Battery-Staple-G2")
|
||||
hg_id, host_id = await _setup_hostgroup_and_host(conn)
|
||||
|
||||
await _login_full(client, "grp_admin", "Correct-Horse-Battery-Staple-G1")
|
||||
|
||||
resp = await client.post("/admin/user-groups", json={"name": "netzwerk-team", "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_name": "ssh_connect"},
|
||||
)
|
||||
assert resp.status_code == 200, resp.text
|
||||
|
||||
# Als Gruppenmitglied einloggen: KEIN individueller user_hostgroup_roles-Eintrag existiert.
|
||||
client.cookies.clear()
|
||||
await _login_full(client, "grp_member", "Correct-Horse-Battery-Staple-G2")
|
||||
|
||||
resp = await client.get("/catalog/hosts")
|
||||
assert resp.status_code == 200, resp.text
|
||||
hosts = resp.json()
|
||||
assert any(h["id"] == host_id for h in hosts), "Gruppenmitglied sollte den geteilten Host sehen"
|
||||
assert next(h for h in hosts if h["id"] == host_id)["can_connect"] is True
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_removing_group_member_revokes_inherited_catalog_access(client):
|
||||
from app.db import get_db
|
||||
|
||||
conn = get_db()
|
||||
await _create_user(conn, "grp_admin2", "Correct-Horse-Battery-Staple-G3", is_admin=True)
|
||||
member_id = await _create_user(conn, "grp_member2", "Correct-Horse-Battery-Staple-G4")
|
||||
hg_id, host_id = await _setup_hostgroup_and_host(conn, group_name="team-b", hostname="srv-b")
|
||||
|
||||
await _login_full(client, "grp_admin2", "Correct-Horse-Battery-Staple-G3")
|
||||
resp = await client.post("/admin/user-groups", json={"name": "devops-team", "description": None})
|
||||
group_id = resp.json()["id"]
|
||||
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"},
|
||||
)
|
||||
|
||||
# Mitgliedschaft wieder entfernen, BEVOR sich der User einloggt.
|
||||
resp = await client.delete(f"/admin/user-groups/{group_id}/members/{member_id}")
|
||||
assert resp.status_code == 200, resp.text
|
||||
|
||||
client.cookies.clear()
|
||||
await _login_full(client, "grp_member2", "Correct-Horse-Battery-Staple-G4")
|
||||
resp = await client.get("/catalog/hosts")
|
||||
assert resp.status_code == 200, resp.text
|
||||
assert all(h["id"] != host_id for h in resp.json()), "Entferntes Mitglied darf Host nicht mehr sehen"
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_non_admin_cannot_manage_user_groups_or_group_roles(client):
|
||||
from app.db import get_db
|
||||
|
||||
conn = get_db()
|
||||
await _create_user(conn, "plain_user", "Correct-Horse-Battery-Staple-P1", is_admin=False)
|
||||
await _login_full(client, "plain_user", "Correct-Horse-Battery-Staple-P1")
|
||||
|
||||
resp = await client.post("/admin/user-groups", json={"name": "sollte-nicht-klappen"})
|
||||
assert resp.status_code == 403
|
||||
resp = await client.get("/admin/user-groups")
|
||||
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"},
|
||||
)
|
||||
assert resp.status_code == 403
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# 2) API-Tokens: Scope-Durchsetzung
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
async def _create_admin_and_token(client, conn, *, scopes, username="tok_admin", pw="Correct-Horse-Battery-Staple-T1", expires_at=None):
|
||||
admin_id = await _create_user(conn, username, pw, is_admin=True)
|
||||
await _login_full(client, username, pw)
|
||||
resp = await client.post(
|
||||
"/admin/tokens",
|
||||
json={"label": "ci-token", "user_id": admin_id, "scopes": scopes, "expires_at": expires_at},
|
||||
)
|
||||
assert resp.status_code == 201, resp.text
|
||||
return resp.json()["token"], resp.json()["id"], admin_id
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_token_with_read_scope_allows_get_but_not_write(client):
|
||||
from app.db import get_db
|
||||
|
||||
conn = get_db()
|
||||
token, _token_id, _admin_id = await _create_admin_and_token(client, conn, scopes=["hosts:read"])
|
||||
|
||||
resp = await client.get("/admin/hosts", headers={"Authorization": f"Bearer {token}"})
|
||||
assert resp.status_code == 200, resp.text
|
||||
|
||||
resp = await client.post(
|
||||
"/admin/host-groups",
|
||||
json={"name": "sollte-nicht-klappen"},
|
||||
headers={"Authorization": f"Bearer {token}"},
|
||||
)
|
||||
# host_groups:write ist nicht im Scope enthalten -> abgelehnt.
|
||||
assert resp.status_code == 401, resp.text
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_token_with_write_scope_implies_read(client):
|
||||
from app.db import get_db
|
||||
|
||||
conn = get_db()
|
||||
token, _token_id, _admin_id = await _create_admin_and_token(
|
||||
client, conn, scopes=["host_groups:write"], username="tok_admin2", pw="Correct-Horse-Battery-Staple-T2"
|
||||
)
|
||||
|
||||
resp = await client.get("/admin/host-groups", headers={"Authorization": f"Bearer {token}"})
|
||||
assert resp.status_code == 200, resp.text
|
||||
|
||||
resp = await client.post(
|
||||
"/admin/host-groups", json={"name": "per-token-angelegt"},
|
||||
headers={"Authorization": f"Bearer {token}"},
|
||||
)
|
||||
assert resp.status_code == 201, resp.text
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_revoked_token_rejected(client):
|
||||
from app.db import get_db
|
||||
|
||||
conn = get_db()
|
||||
token, token_id, _admin_id = await _create_admin_and_token(
|
||||
client, conn, scopes=["hosts:read"], username="tok_admin3", pw="Correct-Horse-Battery-Staple-T3"
|
||||
)
|
||||
|
||||
resp = await client.post(f"/admin/tokens/{token_id}/revoke")
|
||||
assert resp.status_code == 200, resp.text
|
||||
|
||||
resp = await client.get("/admin/hosts", headers={"Authorization": f"Bearer {token}"})
|
||||
assert resp.status_code == 401, resp.text
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_expired_token_rejected(client):
|
||||
from app.db import get_db
|
||||
|
||||
conn = get_db()
|
||||
past = "2000-01-01T00:00:00.000000Z"
|
||||
token, _token_id, _admin_id = await _create_admin_and_token(
|
||||
client, conn, scopes=["hosts:read"], username="tok_admin4",
|
||||
pw="Correct-Horse-Battery-Staple-T4", expires_at=past,
|
||||
)
|
||||
|
||||
resp = await client.get("/admin/hosts", headers={"Authorization": f"Bearer {token}"})
|
||||
assert resp.status_code == 401, resp.text
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_unknown_scope_rejected_on_token_creation(client):
|
||||
from app.db import get_db
|
||||
|
||||
conn = get_db()
|
||||
admin_id = await _create_user(conn, "tok_admin5", "Correct-Horse-Battery-Staple-T5", is_admin=True)
|
||||
await _login_full(client, "tok_admin5", "Correct-Horse-Battery-Staple-T5")
|
||||
|
||||
resp = await client.post(
|
||||
"/admin/tokens",
|
||||
json={"label": "bad-scope", "user_id": admin_id, "scopes": ["hosts:delete-everything"]},
|
||||
)
|
||||
assert resp.status_code == 400, resp.text
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_token_cannot_be_used_to_create_or_list_other_tokens(client):
|
||||
"""Privilege-Escalation-Schutz: /admin/tokens/* laeuft ausschliesslich
|
||||
ueber require_global_admin (Session), niemals ueber require_admin_or_scope
|
||||
-- ein geleaktes Token darf sich selbst keine weiteren/staerkeren Tokens
|
||||
ausstellen, selbst wenn ein (in diesem Scope-Modell gar nicht existierender)
|
||||
'tokens'-Scope faelschlich mitgeschickt wuerde."""
|
||||
from app.db import get_db
|
||||
|
||||
conn = get_db()
|
||||
token, _token_id, admin_id = await _create_admin_and_token(
|
||||
client, conn, scopes=["users:read", "users:write", "hosts:read", "hosts:write"],
|
||||
username="tok_admin6", pw="Correct-Horse-Battery-Staple-T6",
|
||||
)
|
||||
# Session-Cookie bewusst entfernen: es soll ausschliesslich das Token als
|
||||
# Credential vorliegen (sonst wuerde die noch gueltige Admin-Session der
|
||||
# Token-Erzeugung die /admin/tokens-Endpunkte -- die NUR Cookies pruefen
|
||||
# -- unabhaengig vom Token-Test durchwinken und der Test taeuschend gruen sein).
|
||||
client.cookies.clear()
|
||||
|
||||
resp = await client.get("/admin/tokens", headers={"Authorization": f"Bearer {token}"})
|
||||
assert resp.status_code == 401, resp.text
|
||||
|
||||
resp = await client.post(
|
||||
"/admin/tokens",
|
||||
json={"label": "self-issued", "user_id": admin_id, "scopes": ["users:read"]},
|
||||
headers={"Authorization": f"Bearer {token}"},
|
||||
)
|
||||
assert resp.status_code == 401, resp.text
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_non_admin_cannot_create_tokens(client):
|
||||
from app.db import get_db
|
||||
|
||||
conn = get_db()
|
||||
user_id = await _create_user(conn, "tok_plain", "Correct-Horse-Battery-Staple-T7")
|
||||
await _login_full(client, "tok_plain", "Correct-Horse-Battery-Staple-T7")
|
||||
|
||||
resp = await client.post(
|
||||
"/admin/tokens", json={"label": "x", "user_id": user_id, "scopes": ["hosts:read"]}
|
||||
)
|
||||
assert resp.status_code == 403
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# 3) API-Dokumentation: admin-only, kein oeffentliches Swagger-UI
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_docs_and_openapi_require_admin(client):
|
||||
resp = await client.get("/docs")
|
||||
assert resp.status_code == 401
|
||||
resp = await client.get("/openapi.json")
|
||||
assert resp.status_code == 401
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_docs_and_openapi_reject_non_admin(client):
|
||||
from app.db import get_db
|
||||
|
||||
conn = get_db()
|
||||
await _create_user(conn, "docs_plain", "Correct-Horse-Battery-Staple-D1", is_admin=False)
|
||||
await _login_full(client, "docs_plain", "Correct-Horse-Battery-Staple-D1")
|
||||
|
||||
resp = await client.get("/docs")
|
||||
assert resp.status_code == 403
|
||||
resp = await client.get("/openapi.json")
|
||||
assert resp.status_code == 403
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_docs_and_openapi_accessible_to_admin(client):
|
||||
from app.db import get_db
|
||||
|
||||
conn = get_db()
|
||||
await _create_user(conn, "docs_admin", "Correct-Horse-Battery-Staple-D2", is_admin=True)
|
||||
await _login_full(client, "docs_admin", "Correct-Horse-Battery-Staple-D2")
|
||||
|
||||
resp = await client.get("/openapi.json")
|
||||
assert resp.status_code == 200, resp.text
|
||||
spec = resp.json()
|
||||
assert "/admin/tokens" in spec["paths"]
|
||||
|
||||
resp = await client.get("/docs")
|
||||
assert resp.status_code == 200, resp.text
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_admin_page_serves_static_markup_without_serverside_gate(client):
|
||||
"""Wie /dashboard: die Seite selbst enthaelt keine Secrets, die
|
||||
Admin-Pruefung erfolgt clientseitig via /auth/me + serverseitig hart auf
|
||||
jedem einzelnen /admin/*-API-Call."""
|
||||
resp = await client.get("/admin")
|
||||
assert resp.status_code == 200, resp.text
|
||||
assert "Jumphost" in resp.text
|
||||
@ -67,7 +67,7 @@ def _assert_no_inline_script(html: str, page: str) -> None:
|
||||
@pytest.mark.asyncio
|
||||
@pytest.mark.parametrize(
|
||||
"path",
|
||||
["/", "/dashboard", "/terminal/1", "/rdp/1"],
|
||||
["/", "/dashboard", "/terminal/1", "/rdp/1", "/admin"],
|
||||
)
|
||||
async def test_rendered_pages_contain_no_inline_style_or_script(client, path):
|
||||
resp = await client.get(path)
|
||||
@ -77,6 +77,40 @@ async def test_rendered_pages_contain_no_inline_style_or_script(client, path):
|
||||
_assert_no_inline_script(html, path)
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_api_docs_page_contains_no_inline_style_or_script(client):
|
||||
"""/docs (siehe test_admin_groups_tokens.py fuer den Admin-Zugriffsschutz
|
||||
selbst) ist bewusst KEIN vendored/CDN-bezogenes Swagger-UI-Bundle, sondern
|
||||
eine selbstgebaute, CSP-konforme Ansicht -- muss also denselben
|
||||
Inline-Regeln genuegen wie alle anderen Seiten."""
|
||||
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, is_admin, must_change_password) "
|
||||
"VALUES ('docsadmin', ?, 1, 0)",
|
||||
(hash_password("Correct-Horse-Battery-Staple-Docs"),),
|
||||
)
|
||||
await conn.commit()
|
||||
resp = await client.post(
|
||||
"/auth/login", json={"username": "docsadmin", "password": "Correct-Horse-Battery-Staple-Docs"}
|
||||
)
|
||||
pending = resp.json()["pending_token"]
|
||||
import pyotp
|
||||
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()
|
||||
await client.post("/auth/totp/enroll/confirm", json={"pending_token": pending, "code": code})
|
||||
|
||||
resp = await client.get("/docs")
|
||||
assert resp.status_code == 200, resp.text
|
||||
html = resp.text
|
||||
_assert_no_inline_style(html, "/docs")
|
||||
_assert_no_inline_script(html, "/docs")
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_login_page_hidden_sections_use_css_class_not_inline_style(client):
|
||||
"""Die anfangs versteckten Login-Bereiche muessen ueber die `.hidden`
|
||||
|
||||
Reference in New Issue
Block a user