171 lines
6.9 KiB
Python
171 lines
6.9 KiB
Python
"""Tests fuer Teil F.5 (Umsetzungsauftrag_Sonnet5.md): Obergrenzen sichtbar
|
|
in der Arbeitsflaeche (F.3.8).
|
|
|
|
GET /catalog/session-limits liefert nur Zahlen -- die eigentliche
|
|
Durchsetzung existiert bereits seit Teil E.4 (app/ssh_proxy/terminal_ws.py,
|
|
app/rdp_proxy/ws_tunnel.py, WS-Code 4429) und wird hier NICHT verdoppelt,
|
|
sondern ueber dieselbe Registry (app/security/active_sessions.py) und
|
|
denselben app/config.py-Wert gelesen. Getestet wird deshalb v.a., dass die
|
|
Zahlen exakt der bereits vorhandenen Durchsetzungslogik entsprechen
|
|
(insbesondere die Admin-Ausnahme bei der Je-Nutzer-Grenze, NICHT bei der
|
|
globalen Grenze -- siehe terminal_ws.py Zeile ~120-136) und dass keine
|
|
Angaben zu FREMDEN Sitzungen durchsickern (F.5-Risikotabelle,
|
|
Rechteumgehung ueber die Katalog-API, dasselbe Prinzip wie in
|
|
test_teil_f_schritt2.py)."""
|
|
from __future__ import annotations
|
|
|
|
import asyncio
|
|
|
|
import pytest
|
|
|
|
from app.security import active_sessions
|
|
|
|
|
|
async def _make_user(client, username: str, is_admin: int = 0):
|
|
from app.db import get_db
|
|
from app.security.passwords import hash_password
|
|
import pyotp
|
|
|
|
conn = get_db()
|
|
await conn.execute(
|
|
"INSERT INTO users (username, password_hash, is_admin, must_change_password) "
|
|
"VALUES (?, ?, ?, 0)",
|
|
(username, hash_password("Correct-Horse-Battery-Staple-F5"), is_admin),
|
|
)
|
|
await conn.commit()
|
|
|
|
resp = await client.post("/auth/login", json={"username": username, "password": "Correct-Horse-Battery-Staple-F5"})
|
|
pending = resp.json()["pending_token"]
|
|
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()
|
|
resp = await client.post("/auth/totp/enroll/confirm", json={"pending_token": pending, "code": code})
|
|
assert resp.status_code == 200, resp.text
|
|
row = await (await conn.execute("SELECT id FROM users WHERE username = ?", (username,))).fetchone()
|
|
return row[0]
|
|
|
|
|
|
class _FakeTask:
|
|
"""Ersatz fuer einen echten asyncio.Task in der Registry -- fuer diese
|
|
Tests wird nie cancel()/await ausgefuehrt, nur die Zaehlung geprueft."""
|
|
|
|
def cancel(self):
|
|
pass
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_session_limits_reflects_config_when_no_sessions_open(client):
|
|
"""Frisch angemeldeter Nutzer ohne offene Sitzungen: current_*_count == 0,
|
|
max_* == die konfigurierten Werte aus app/config.py, keine Grenze
|
|
erreicht."""
|
|
from app.config import settings
|
|
|
|
await _make_user(client, "f5_user_empty")
|
|
|
|
resp = await client.get("/catalog/session-limits")
|
|
assert resp.status_code == 200, resp.text
|
|
body = resp.json()
|
|
assert body["max_per_user"] == settings.max_sessions_per_user
|
|
assert body["max_global"] == settings.max_sessions_global
|
|
assert body["current_user_count"] == 0
|
|
assert body["current_global_count"] == 0
|
|
assert body["user_limit_applies"] is True
|
|
assert body["at_user_limit"] is False
|
|
assert body["at_global_limit"] is False
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_session_limits_counts_only_own_sessions_for_user_count(client):
|
|
"""current_user_count zaehlt NUR die eigenen Registry-Eintraege --
|
|
Sitzungen eines anderen Nutzers duerfen den eigenen Zaehler nicht
|
|
beeinflussen (wohl aber current_global_count, das ist bewusst global)."""
|
|
user_a_id = await _make_user(client, "f5_user_a")
|
|
|
|
other_user_id = 9999 # fremd, muss NICHT in der DB existieren fuer die Registry
|
|
active_sessions.register(9001, _FakeTask(), other_user_id)
|
|
try:
|
|
resp = await client.get("/catalog/session-limits")
|
|
assert resp.status_code == 200, resp.text
|
|
body = resp.json()
|
|
assert body["current_user_count"] == 0 # fremde Sitzung zaehlt NICHT
|
|
assert body["current_global_count"] == 1 # global schon
|
|
finally:
|
|
active_sessions.unregister(9001)
|
|
|
|
active_sessions.register(9002, _FakeTask(), user_a_id)
|
|
try:
|
|
resp = await client.get("/catalog/session-limits")
|
|
body = resp.json()
|
|
assert body["current_user_count"] == 1
|
|
assert body["current_global_count"] == 1
|
|
finally:
|
|
active_sessions.unregister(9002)
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_session_limits_at_user_limit_true_when_reached(client):
|
|
"""Erreicht ein normaler Nutzer max_sessions_per_user, meldet der
|
|
Endpunkt at_user_limit=true -- exakt die Schwelle (>=), die auch
|
|
terminal_ws.py/ws_tunnel.py fuer die Ablehnung verwendet."""
|
|
from app.config import settings
|
|
|
|
user_id = await _make_user(client, "f5_user_at_limit")
|
|
|
|
ids = list(range(9100, 9100 + settings.max_sessions_per_user))
|
|
for sid in ids:
|
|
active_sessions.register(sid, _FakeTask(), user_id)
|
|
try:
|
|
resp = await client.get("/catalog/session-limits")
|
|
body = resp.json()
|
|
assert body["current_user_count"] == settings.max_sessions_per_user
|
|
assert body["at_user_limit"] is True
|
|
assert body["at_global_limit"] is False # weit unter max_sessions_global
|
|
finally:
|
|
for sid in ids:
|
|
active_sessions.unregister(sid)
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_session_limits_admin_exempt_from_user_limit_not_from_global(client):
|
|
"""Deckt exakt die Asymmetrie aus terminal_ws.py Zeile ~120-136 ab: Admins
|
|
sind von der Je-Nutzer-Grenze ausgenommen ('not user.is_admin'-Gate),
|
|
aber NICHT von der globalen Grenze. Weicht die Anzeige davon ab, wuerde
|
|
sie einem Admin faelschlich 'gesperrt' oder faelschlich 'frei' zeigen."""
|
|
from app.config import settings
|
|
|
|
admin_id = await _make_user(client, "f5_admin", is_admin=1)
|
|
|
|
# Je-Nutzer-Grenze fuer den Admin selbst "erreicht" -- darf ihn laut
|
|
# Durchsetzungslogik trotzdem nicht ausbremsen.
|
|
ids = list(range(9200, 9200 + settings.max_sessions_per_user))
|
|
for sid in ids:
|
|
active_sessions.register(sid, _FakeTask(), admin_id)
|
|
try:
|
|
resp = await client.get("/catalog/session-limits")
|
|
body = resp.json()
|
|
assert body["user_limit_applies"] is False
|
|
assert body["at_user_limit"] is False # trotz current_user_count == max_per_user
|
|
assert body["current_user_count"] == settings.max_sessions_per_user
|
|
finally:
|
|
for sid in ids:
|
|
active_sessions.unregister(sid)
|
|
|
|
# Globale Grenze gilt hingegen auch fuer Admins.
|
|
global_ids = list(range(9300, 9300 + settings.max_sessions_global))
|
|
for sid in global_ids:
|
|
active_sessions.register(sid, _FakeTask(), 424242) # fremder Platzhalter-Nutzer
|
|
try:
|
|
resp = await client.get("/catalog/session-limits")
|
|
body = resp.json()
|
|
assert body["at_global_limit"] is True
|
|
finally:
|
|
for sid in global_ids:
|
|
active_sessions.unregister(sid)
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_session_limits_requires_login(client):
|
|
resp = await client.get("/catalog/session-limits")
|
|
assert resp.status_code == 401
|