admin_rb
179 lines
6.3 KiB
Python
179 lines
6.3 KiB
Python
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()
|