aim-web2.1.0rc9

scripts/addons/webgui/tests/test_auth.py

@@ -0,0 +1,178 @@
+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()