181 lines
7.5 KiB
Python
181 lines
7.5 KiB
Python
"""Tests fuer Teil F.3.6 (Stufe F2, Umsetzungsauftrag_Sonnet5.md): eigene
|
|
Sitzungs-API im Katalog-Router.
|
|
|
|
1) GET /catalog/sessions liefert AUSSCHLIESSLICH die eigenen Sitzungen des
|
|
angemeldeten Benutzers -- eine fremde, gleichzeitig laufende Sitzung
|
|
taucht nicht auf.
|
|
2) active_only=true (Standard) blendet bereits beendete eigene Sitzungen
|
|
aus; active_only=false zeigt sie.
|
|
3) POST /catalog/sessions/{id}/terminate auf eine FREMDE session_id
|
|
liefert 404 (F.5-Risikotabelle: 'Rechteumgehung ueber die neue
|
|
Sitzungs-API' -- explizit geforderter Testfall), nicht etwa 403 (das
|
|
wuerde die Existenz einer fremden Sitzung verraten) und nicht 200.
|
|
4) POST .../terminate auf die EIGENE, tatsaechlich laufende Sitzung
|
|
funktioniert: der zugehoerige asyncio.Task wird abgebrochen, die
|
|
Sitzung landet als beendet in der DB, ein Audit-Ereignis
|
|
'session_terminated_by_owner' wird geschrieben.
|
|
5) POST .../terminate auf eine eigene, bereits beendete Sitzung liefert
|
|
409 (nicht 200 -- kein stiller Erfolg auf einer Sitzung, die gar nicht
|
|
mehr laeuft).
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
import asyncio
|
|
import contextlib
|
|
|
|
import pyotp
|
|
import pytest
|
|
|
|
|
|
async def _create_user(conn, username: str, password: str) -> 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, 0)",
|
|
(username, hash_password(password)),
|
|
)
|
|
await conn.commit()
|
|
return cursor.lastrowid
|
|
|
|
|
|
async def _login_full(client, username: str, password: str) -> None:
|
|
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
|
|
|
|
|
|
async def _make_host(conn, *, hostname: str) -> int:
|
|
cursor = await conn.execute("INSERT INTO host_groups (name) VALUES (?)", (f"hg-{hostname}",))
|
|
group_id = cursor.lastrowid
|
|
cursor = await conn.execute(
|
|
"INSERT INTO hosts (hostname, address, port, protocol, os_type, host_group_id) "
|
|
"VALUES (?, '10.0.0.1', 22, 'ssh', 'linux', ?)",
|
|
(hostname, group_id),
|
|
)
|
|
await conn.commit()
|
|
return cursor.lastrowid
|
|
|
|
|
|
async def _make_session(conn, *, user_id: int, host_id: int, ended: bool) -> int:
|
|
cursor = await conn.execute(
|
|
"INSERT INTO sessions (user_id, host_id, protocol, client_ip, ended_at, end_reason) "
|
|
"VALUES (?, ?, 'ssh', '127.0.0.1', ?, ?)",
|
|
(user_id, host_id, "2026-01-01T00:00:00.000000Z" if ended else None, "logout" if ended else None),
|
|
)
|
|
await conn.commit()
|
|
return cursor.lastrowid
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_my_sessions_shows_only_own_and_respects_active_only(client):
|
|
from app.db import get_db
|
|
|
|
conn = get_db()
|
|
u1 = await _create_user(conn, "f2_u1", "Correct-Horse-Battery-Staple-1")
|
|
u2 = await _create_user(conn, "f2_u2", "Correct-Horse-Battery-Staple-2")
|
|
host_id = await _make_host(conn, hostname="f2-host-1")
|
|
|
|
my_open = await _make_session(conn, user_id=u1, host_id=host_id, ended=False)
|
|
my_closed = await _make_session(conn, user_id=u1, host_id=host_id, ended=True)
|
|
other_open = await _make_session(conn, user_id=u2, host_id=host_id, ended=False)
|
|
|
|
await _login_full(client, "f2_u1", "Correct-Horse-Battery-Staple-1")
|
|
|
|
resp = await client.get("/catalog/sessions")
|
|
assert resp.status_code == 200, resp.text
|
|
ids = {row["id"] for row in resp.json()}
|
|
assert my_open in ids
|
|
assert my_closed not in ids, "aktive-only (Standard) muss beendete eigene Sitzungen ausblenden"
|
|
assert other_open not in ids, "fremde Sitzung darf NIE auftauchen"
|
|
|
|
resp = await client.get("/catalog/sessions", params={"active_only": "false"})
|
|
assert resp.status_code == 200, resp.text
|
|
ids = {row["id"] for row in resp.json()}
|
|
assert my_open in ids
|
|
assert my_closed in ids, "active_only=false muss auch beendete eigene Sitzungen zeigen"
|
|
assert other_open not in ids, "fremde Sitzung darf auch mit active_only=false nie auftauchen"
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_terminate_foreign_session_returns_404_not_403(client):
|
|
from app.db import get_db
|
|
|
|
conn = get_db()
|
|
u1 = await _create_user(conn, "f2_u3", "Correct-Horse-Battery-Staple-3")
|
|
u2 = await _create_user(conn, "f2_u4", "Correct-Horse-Battery-Staple-4")
|
|
host_id = await _make_host(conn, hostname="f2-host-2")
|
|
foreign_session = await _make_session(conn, user_id=u2, host_id=host_id, ended=False)
|
|
|
|
await _login_full(client, "f2_u3", "Correct-Horse-Battery-Staple-3")
|
|
resp = await client.post(f"/catalog/sessions/{foreign_session}/terminate")
|
|
assert resp.status_code == 404, resp.text
|
|
|
|
resp = await client.post("/catalog/sessions/999999/terminate")
|
|
assert resp.status_code == 404, resp.text
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_terminate_own_running_session_cancels_task_and_audits(client):
|
|
from app.db import get_db
|
|
from app.security import active_sessions
|
|
from app.security.audit import verify_chain
|
|
|
|
conn = get_db()
|
|
u1 = await _create_user(conn, "f2_u5", "Correct-Horse-Battery-Staple-5")
|
|
host_id = await _make_host(conn, hostname="f2-host-3")
|
|
session_id = await _make_session(conn, user_id=u1, host_id=host_id, ended=False)
|
|
|
|
async def _fake_long_running():
|
|
await asyncio.sleep(3600)
|
|
|
|
task = asyncio.create_task(_fake_long_running())
|
|
active_sessions.register(session_id=session_id, task=task, user_id=u1)
|
|
try:
|
|
await _login_full(client, "f2_u5", "Correct-Horse-Battery-Staple-5")
|
|
resp = await client.post(f"/catalog/sessions/{session_id}/terminate")
|
|
assert resp.status_code == 200, resp.text
|
|
|
|
# asyncio.Task.cancel() wirkt asynchron -- auf das tatsaechliche
|
|
# Ende der Task warten, statt nur einmal nachzugeben (Python 3.10:
|
|
# kein Task.cancelling(), daher ueber den Ausgang selbst pruefen).
|
|
with contextlib.suppress(asyncio.CancelledError):
|
|
await asyncio.wait_for(task, timeout=1)
|
|
assert task.cancelled()
|
|
|
|
row = await (await conn.execute(
|
|
"SELECT event_type, details_json FROM audit_log WHERE user_id = ? ORDER BY id DESC LIMIT 1",
|
|
(u1,),
|
|
)).fetchone()
|
|
assert row is not None
|
|
assert row[0] == "session_terminated_by_owner"
|
|
assert str(session_id) in row[1]
|
|
|
|
intact, _ = await verify_chain(conn)
|
|
assert intact
|
|
finally:
|
|
if not task.done():
|
|
task.cancel()
|
|
active_sessions.unregister(session_id)
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_terminate_already_ended_own_session_returns_409(client):
|
|
from app.db import get_db
|
|
|
|
conn = get_db()
|
|
u1 = await _create_user(conn, "f2_u6", "Correct-Horse-Battery-Staple-6")
|
|
host_id = await _make_host(conn, hostname="f2-host-4")
|
|
ended_session = await _make_session(conn, user_id=u1, host_id=host_id, ended=True)
|
|
|
|
await _login_full(client, "f2_u6", "Correct-Horse-Battery-Staple-6")
|
|
resp = await client.post(f"/catalog/sessions/{ended_session}/terminate")
|
|
assert resp.status_code == 409, resp.text
|