109 lines
3.4 KiB
Python
109 lines
3.4 KiB
Python
import aiosqlite
|
|
import pytest
|
|
|
|
from app.db import MIGRATIONS_DIR
|
|
from app.security.audit import verify_chain
|
|
from app.security.session_reaper import reap_orphaned_sessions
|
|
|
|
|
|
async def _fresh_db() -> aiosqlite.Connection:
|
|
conn = await aiosqlite.connect(":memory:")
|
|
for migration_file in sorted(MIGRATIONS_DIR.glob("*.sql")):
|
|
await conn.executescript(migration_file.read_text(encoding="utf-8"))
|
|
await conn.execute(
|
|
"INSERT INTO users (id, username, password_hash) VALUES (1, 'u', 'x')"
|
|
)
|
|
await conn.execute(
|
|
"INSERT INTO host_groups (id, name) VALUES (1, 'hg')"
|
|
)
|
|
await conn.execute(
|
|
"INSERT INTO hosts (id, host_group_id, hostname, address, protocol, port, os_type) "
|
|
"VALUES (1, 1, 'h1', '10.0.0.1', 'ssh', 22, 'linux')"
|
|
)
|
|
await conn.commit()
|
|
return conn
|
|
|
|
|
|
async def _insert_session(conn, *, ended: bool) -> int:
|
|
if ended:
|
|
await conn.execute(
|
|
"INSERT INTO sessions (user_id, host_id, protocol, client_ip, ended_at, end_reason) "
|
|
"VALUES (1, 1, 'ssh', '127.0.0.1', strftime('%Y-%m-%dT%H:%M:%fZ','now'), 'logout')"
|
|
)
|
|
else:
|
|
await conn.execute(
|
|
"INSERT INTO sessions (user_id, host_id, protocol, client_ip) "
|
|
"VALUES (1, 1, 'ssh', '127.0.0.1')"
|
|
)
|
|
await conn.commit()
|
|
cursor = await conn.execute("SELECT last_insert_rowid()")
|
|
row = await cursor.fetchone()
|
|
return row[0]
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_reap_closes_only_open_sessions():
|
|
"""E7 (Umsetzungsauftrag_Sonnet5.md Teil E.1): beim Start muessen alle
|
|
Sitzungen mit ended_at IS NULL geschlossen werden (server_restart),
|
|
bereits beendete Sitzungen bleiben unangetastet."""
|
|
conn = await _fresh_db()
|
|
|
|
already_ended = await _insert_session(conn, ended=True)
|
|
orphan_1 = await _insert_session(conn, ended=False)
|
|
orphan_2 = await _insert_session(conn, ended=False)
|
|
|
|
count = await reap_orphaned_sessions(conn)
|
|
assert count == 2
|
|
|
|
cursor = await conn.execute(
|
|
"SELECT id, ended_at, end_reason FROM sessions ORDER BY id"
|
|
)
|
|
rows = {row[0]: (row[1], row[2]) async for row in cursor}
|
|
|
|
assert rows[already_ended][1] == "logout"
|
|
for sid in (orphan_1, orphan_2):
|
|
ended_at, end_reason = rows[sid]
|
|
assert ended_at is not None
|
|
assert end_reason == "server_restart"
|
|
|
|
await conn.close()
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_reap_is_noop_when_nothing_open():
|
|
conn = await _fresh_db()
|
|
await _insert_session(conn, ended=True)
|
|
|
|
count = await reap_orphaned_sessions(conn)
|
|
assert count == 0
|
|
|
|
await conn.close()
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_reap_writes_single_audit_event_and_keeps_chain_intact():
|
|
conn = await _fresh_db()
|
|
await _insert_session(conn, ended=False)
|
|
await _insert_session(conn, ended=False)
|
|
await _insert_session(conn, ended=False)
|
|
|
|
await reap_orphaned_sessions(conn)
|
|
|
|
cursor = await conn.execute(
|
|
"SELECT COUNT(*) FROM audit_log WHERE event_type = 'session_reaper_server_restart'"
|
|
)
|
|
row = await cursor.fetchone()
|
|
assert row[0] == 1
|
|
|
|
cursor = await conn.execute(
|
|
"SELECT details_json FROM audit_log WHERE event_type = 'session_reaper_server_restart'"
|
|
)
|
|
row = await cursor.fetchone()
|
|
assert '"count": 3' in row[0]
|
|
|
|
intact, broken_at = await verify_chain(conn)
|
|
assert intact is True
|
|
assert broken_at is None
|
|
|
|
await conn.close()
|