"""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