import concurrent.futures import json import os import sqlite3 import stat import time import pytest from aim_webgui.auth.service import Auth, HASHER, digest from aim_webgui.errors import WebError from aim_webgui.config import Settings @pytest.fixture def settings(tmp_path): state=tmp_path/'state';state.mkdir(mode=0o700) return Settings(state_dir=state) @pytest.fixture def auth(settings): auth=Auth(settings);auth.bootstrap();return auth @pytest.fixture def password(auth):return json.loads(auth.settings.credentials.read_text())['password'] NEW = 'Independent-test-passphrase-2026' def test_bootstrap_secure_and_idempotent(auth, password): assert stat.S_IMODE(auth.settings.credentials.stat().st_mode) == 0o600 assert stat.S_IMODE(auth.settings.database.stat().st_mode) == 0o600 assert len(password) >= 32 with auth.store.read() as db: row = db.execute('SELECT * FROM users').fetchone() assert row['password_hash'].startswith('$argon2id$') assert HASHER.verify(row['password_hash'], password) assert row['must_change_password'] == 1 auth.settings.credentials.unlink() assert auth.bootstrap() is False assert not auth.settings.credentials.exists() assert len(auth.users()) == 1 def test_bootstrap_concurrent(settings): with concurrent.futures.ThreadPoolExecutor(2) as pool: results = list(pool.map(lambda _: Auth(settings).bootstrap(), range(2))) assert sorted(results) == [False, True] assert len(Auth(settings).users()) == 1 def test_missing_database_not_rebootstrapped(auth): auth.settings.database.unlink() with pytest.raises(ValueError, match='missing'): auth.bootstrap() def test_password_change_revokes_and_consumes(auth, password): anon, _ = auth.new_session() token, s = auth.login('admin', password, anon, 'test-peer') assert s['must_change_password'] assert token != anon assert auth.session(anon) is None with auth.store.read() as db: assert db.execute('SELECT token_hash FROM sessions').fetchone()[0] == digest(token) assert token.encode() not in auth.settings.database.read_bytes() auth.change_password(s['user_id'], password, NEW) assert auth.session(token) is None assert not auth.settings.credentials.exists() assert password not in (auth.settings.state_dir / '.credentials.used').read_text() anon, _ = auth.new_session() _, active = auth.login('admin', NEW, anon, 'test-peer') assert not active['must_change_password'] def test_last_admin_guard(auth): for action in ('disable', 'demote'): with pytest.raises(WebError, match='last enabled'): auth.manage_user('admin', action) auth.create_user('second-admin', NEW, 'admin') auth.manage_user('admin', 'disable') with pytest.raises(WebError): auth.manage_user('second-admin', 'disable') def test_role_authorization_rechecked(auth): auth.create_user('viewer', NEW) viewer = next(u for u in auth.users() if u['username'] == 'viewer') with pytest.raises(WebError) as result: auth.manage_user('admin', 'revoke', actor_id=viewer['id']) assert result.value.status == 403 def test_disabled_and_reset_sessions(auth, password): auth.create_user('reader', NEW) reader = next(u for u in auth.users() if u['username'] == 'reader') token, _ = auth.new_session(reader['id']) auth.manage_user('reader', 'disable') assert auth.session(token) is None with pytest.raises(WebError, match='Invalid username'): auth.login('reader', NEW, '', 'peer') auth.manage_user('reader', 'enable') token, _ = auth.new_session(reader['id']) auth.manage_user('reader', 'reset-password', password=NEW + '-reset') assert auth.session(token) is None def test_expired_sessions(auth): token, _ = auth.new_session() with auth.store.transaction() as db: db.execute('UPDATE sessions SET expires_at=?', (int(time.time()) - 1,)) assert auth.session(token) is None def test_throttle(auth): for _ in range(10): with pytest.raises(WebError) as e: auth.login('admin', 'bad', '', 'test-peer') assert e.value.status == 401 with pytest.raises(WebError) as e: auth.login('admin', 'bad', '', 'test-peer') assert e.value.status == 429 def test_migration_backup_and_future_schema(auth, tmp_path): auth.store.migrate() out = tmp_path / 'auth-backup.sqlite3' auth.store.backup(out) assert stat.S_IMODE(out.stat().st_mode) == 0o600 db = sqlite3.connect(out) assert db.execute('PRAGMA integrity_check').fetchone()[0] == 'ok' assert db.execute('SELECT COUNT(*) FROM users').fetchone()[0] == 1 db.close() with auth.store.transaction() as db: db.execute('PRAGMA user_version=999') with pytest.raises(ValueError, match='newer'): auth.store.migrate() def test_credentials_symlink_rejected(settings, tmp_path): other = tmp_path / 'other' other.write_text('do not overwrite') settings.credentials.symlink_to(other) with pytest.raises(ValueError): Auth(settings).bootstrap() assert other.read_text() == 'do not overwrite' def test_short_window_does_not_clear_long_window(auth, monkeypatch): import aim_webgui.auth.service as module monkeypatch.setattr(module.time, 'time', lambda:10000) auth.throttle([('long',1)],window=300) monkeypatch.setattr(module.time, 'time', lambda:10070) auth.throttle([('short',1)],window=60) with pytest.raises(WebError) as e: auth.throttle([('long',1)],window=300) assert e.value.status == 429 def test_bootstrap_repairs_missing_post_commit_marker_without_reset(auth, password): marker = auth.settings.state_dir / '.initialized' marker.unlink() before = auth.settings.credentials.read_bytes() assert auth.bootstrap() is False assert marker.is_file() assert auth.settings.credentials.read_bytes() == before auth.settings.database.unlink() with pytest.raises(ValueError, match='missing'): auth.bootstrap() def test_local_recovery_can_reenable_out_of_band_disabled_admin(auth): with auth.store.transaction() as db: db.execute("UPDATE users SET enabled=0 WHERE username='admin'") with pytest.raises(ValueError, match='administrator'): auth.store.check() auth.store.check(require_admin=False) auth.manage_user('admin', 'enable') auth.store.check()