398 lines
23 KiB
Python
398 lines
23 KiB
Python
import re
|
|
import os
|
|
from sqlalchemy import create_engine
|
|
from sqlalchemy.engine import make_url
|
|
from sqlalchemy.orm import sessionmaker
|
|
from app.db import Base
|
|
from datetime import timedelta
|
|
|
|
import pytest
|
|
from bs4 import BeautifulSoup
|
|
from fastapi.testclient import TestClient
|
|
from sqlalchemy import select
|
|
|
|
from app.config import Settings, get_settings
|
|
from app.db import get_db
|
|
from app.main import app
|
|
from app.models import SystemUser, UserAction, Employee, EmployeePublication, CouncilCorrections, utcnow
|
|
from app.security import (SESSION_COOKIE, _session_data, ensure_bootstrap_admin, hash_password,
|
|
sign_session, temporary_password, verify_password)
|
|
from app.services import council
|
|
|
|
|
|
@pytest.fixture()
|
|
def accounts(db_session, monkeypatch):
|
|
test_engine = None
|
|
native_url = os.getenv('USERS_TEST_DATABASE_URL')
|
|
if native_url:
|
|
assert make_url(native_url).database == 'miem_users_test', 'Use a disposable miem_users_test database only'
|
|
test_engine = create_engine(native_url)
|
|
Base.metadata.create_all(test_engine)
|
|
db_session = sessionmaker(bind=test_engine)()
|
|
settings = Settings(_env_file=None, admin_username='admin', admin_password='initial-admin-test', session_secret='users-test-secret')
|
|
admin = ensure_bootstrap_admin(db_session, settings)
|
|
user = SystemUser(username='reader', full_name='Тестовый пользователь', role='user',
|
|
password_hash=hash_password('Initial user password'), must_change_password=False,
|
|
first_login_at=utcnow())
|
|
db_session.add(user)
|
|
employee = Employee(profile_key='users:employee', canonical_url='https://www.hse.ru/staff/test', full_name='Тестовый сотрудник',
|
|
current_data={'sections': [{'title': 'Учёные степени', 'items': ['Доктор технических наук']}]})
|
|
db_session.add(employee)
|
|
db_session.flush()
|
|
db_session.add(EmployeePublication(employee_id=employee.id, title='Исходная статья', year=2024,
|
|
publication_type='ARTICLE', source_hash='users-publication', raw_data={'journal': 'Журнал'}))
|
|
db_session.commit()
|
|
monkeypatch.setattr(council, '_articles_by_author', lambda: {})
|
|
app.dependency_overrides[get_db] = lambda: db_session
|
|
app.dependency_overrides[get_settings] = lambda: settings
|
|
client = TestClient(app)
|
|
try:
|
|
yield client, db_session, settings, admin, user, employee
|
|
finally:
|
|
client.close()
|
|
app.dependency_overrides.clear()
|
|
if test_engine is not None:
|
|
db_session.close()
|
|
Base.metadata.drop_all(test_engine)
|
|
test_engine.dispose()
|
|
|
|
|
|
def act_as(client, user, settings):
|
|
client.cookies.set(SESSION_COOKIE, sign_session(user.username, settings, user.session_version))
|
|
csrf = _session_data(client.cookies.get(SESSION_COOKIE), settings)['csrf']
|
|
return {'X-CSRF-Token': csrf}
|
|
|
|
|
|
def login(client, username, password):
|
|
page = client.get('/admin/login')
|
|
csrf = BeautifulSoup(page.text, 'html.parser').select_one('[name=csrf_token]')['value']
|
|
return client.post('/admin/login', data={'username': username, 'password': password, 'csrf_token': csrf}, follow_redirects=False)
|
|
|
|
|
|
def test_password_hash_and_eight_character_generator():
|
|
for _ in range(100):
|
|
password = temporary_password()
|
|
assert len(password) == 8
|
|
assert all(re.search(pattern, password) for pattern in (r'[A-Z]', r'[a-z]', r'[0-9]', r'[!@#$%+_\-]'))
|
|
encoded = hash_password('Секретная фраза с пробелами')
|
|
assert verify_password('Секретная фраза с пробелами', encoded)
|
|
assert not verify_password('Другой пароль', encoded)
|
|
assert not verify_password('anything', 'broken hash')
|
|
assert hash_password('same password') != hash_password('same password')
|
|
|
|
|
|
def test_create_user_first_login_forces_password_change(accounts):
|
|
client, db, settings, admin, _, _ = accounts
|
|
headers = act_as(client, admin, settings)
|
|
response = client.post('/admin/users', data={'username': 'new.reader', 'full_name': 'Новый пользователь', 'role': 'user'}, headers=headers)
|
|
assert response.status_code == 200 and response.headers['cache-control'] == 'no-store'
|
|
password = BeautifulSoup(response.text, 'html.parser').select_one('[data-temporary-password]')['value']
|
|
user = db.scalar(select(SystemUser).where(SystemUser.username == 'new.reader'))
|
|
assert verify_password(password, user.password_hash) and user.must_change_password
|
|
assert password not in str(db.scalars(select(UserAction.changes)).all())
|
|
client.cookies.clear()
|
|
signed_in = login(client, user.username, password)
|
|
assert signed_in.status_code == 303 and signed_in.headers['location'] == '/admin/account'
|
|
db.refresh(user)
|
|
assert user.first_login_at and user.last_login_at and user.password_changed_at is None
|
|
first_login = user.first_login_at
|
|
assert 'data-required-password' in client.get('/admin/account').text
|
|
assert client.get('/admin/directory', follow_redirects=False).headers['location'] == '/admin/account'
|
|
assert client.get('/api/employees', follow_redirects=False).status_code == 403
|
|
old_token = client.cookies.get(SESSION_COOKIE)
|
|
csrf = _session_data(old_token, settings)['csrf']
|
|
changed = client.post('/admin/account/password', data={'current_password': password,
|
|
'new_password': 'A new personal passphrase', 'confirmation': 'A new personal passphrase'}, headers={'X-CSRF-Token': csrf}, follow_redirects=False)
|
|
assert changed.status_code == 303
|
|
db.refresh(user)
|
|
assert not user.must_change_password and user.password_changed_at and user.session_version == 1
|
|
assert user.first_login_at == first_login
|
|
assert 'data-required-password' not in client.get('/admin/account').text
|
|
assert 'data-guide-pending="true"' in client.get('/admin/account').text
|
|
assert client.get('/admin/directory').status_code == 200
|
|
assert client.get('/admin/users').status_code == 403
|
|
client.cookies.clear()
|
|
client.cookies.set(SESSION_COOKIE, old_token)
|
|
assert client.get('/api/employees', follow_redirects=False).status_code == 401
|
|
|
|
|
|
@pytest.mark.parametrize('path', ['/admin/crawl-now', '/admin/runs', '/admin/dismissed/refresh', '/admin/employees/1/refresh', '/api/crawl-runs'])
|
|
def test_regular_user_cannot_start_scraping(accounts, path):
|
|
client, db, settings, admin, user, _ = accounts
|
|
response = client.post(path, headers=act_as(client, user, settings), follow_redirects=False)
|
|
assert response.status_code == 403
|
|
assert 'Запустить парсинг' not in client.get('/admin').text
|
|
assert 'Обновить данные' not in client.get('/admin/employees/1').text
|
|
assert client.get('/api/stats').status_code == 200
|
|
|
|
|
|
def test_regular_user_edit_audit_and_personal_history_isolation(accounts):
|
|
client, db, settings, admin, user, employee = accounts
|
|
act_as(client, user, settings)
|
|
candidate = council.council_candidates(db)['candidates'][0]
|
|
article = candidate['articles'][0]
|
|
data = {'employee_id': employee.id, 'edit_key': article['edit_key'], 'revision': 0,
|
|
'title': 'Изменённая статья', 'journal': 'Новый журнал', 'quartile': 'Q2', 'specialty_codes': ['1.2.2']}
|
|
response = client.post('/admin/council/publications', json=data, headers={'X-Council-Edit': '1'})
|
|
assert response.status_code == 200
|
|
action = db.scalar(select(UserAction).where(UserAction.action == 'card_edit'))
|
|
assert action.user_id == user.id and action.employee_id == employee.id and action.publication_year == 2024
|
|
assert action.changes['title'] == {'before': 'Исходная статья', 'after': 'Изменённая статья'}
|
|
assert action.changes['quartile'] == {'before': '', 'after': 'Q2'}
|
|
assert action.changes['specialty_codes']['after']['codes'] == ['1.2.2']
|
|
db.add(UserAction(user_id=admin.id, action='card_edit', article_title='Чужое действие', changes={}))
|
|
db.commit()
|
|
profile = client.get(f'/admin/account?user_id={admin.id}')
|
|
assert 'Изменённая статья' in profile.text and 'Исходная статья' in profile.text
|
|
assert 'Чужое действие' not in profile.text
|
|
assert client.get(f'/admin/users?user_id={admin.id}').status_code == 403
|
|
assert client.post('/admin/council/publications', json=data, headers={'X-Council-Edit': '1'}).status_code == 409
|
|
assert db.scalars(select(UserAction).where(UserAction.user_id == user.id)).all() == [action]
|
|
|
|
|
|
def test_admin_reset_invalidates_sessions_and_preserves_password_change_date(accounts):
|
|
client, db, settings, admin, user, _ = accounts
|
|
user.password_changed_at = utcnow()
|
|
db.commit()
|
|
previous_changed = user.password_changed_at
|
|
old_token = sign_session(user.username, settings, user.session_version)
|
|
response = client.post(f'/admin/users/{user.id}/reset-password', headers=act_as(client, admin, settings))
|
|
assert response.status_code == 200
|
|
password = BeautifulSoup(response.text, 'html.parser').select_one('[data-temporary-password]')['value']
|
|
db.refresh(user)
|
|
assert user.must_change_password and user.password_reset_at and verify_password(password, user.password_hash)
|
|
assert user.password_changed_at == previous_changed
|
|
assert password not in client.get('/admin/users').text
|
|
client.cookies.clear()
|
|
client.cookies.set(SESSION_COOKIE, old_token)
|
|
assert client.get('/api/employees', follow_redirects=False).status_code == 401
|
|
assert login(client, user.username, password).headers['location'] == '/admin/account'
|
|
assert db.scalar(select(UserAction).where(UserAction.action == 'password_reset')).target_user_id == user.id
|
|
|
|
|
|
def test_role_changes_disable_and_protection_against_locking_out_self(accounts):
|
|
client, db, settings, admin, user, _ = accounts
|
|
old_token = sign_session(user.username, settings, user.session_version)
|
|
headers = act_as(client, admin, settings)
|
|
assert client.post(f'/admin/users/{admin.id}', data={'role': 'user', 'is_active': 'false'}, headers=headers).status_code == 422
|
|
assert client.post(f'/admin/users/{admin.id}/reset-password', headers=headers).status_code == 422
|
|
assert client.post(f'/admin/users/{user.id}', data={'role': 'admin', 'is_active': 'true'}, headers=headers, follow_redirects=False).status_code == 303
|
|
db.refresh(user)
|
|
assert user.role == 'admin'
|
|
client.cookies.clear()
|
|
client.cookies.set(SESSION_COOKIE, old_token)
|
|
assert client.get('/admin/users', follow_redirects=False).status_code == 303
|
|
headers = act_as(client, admin, settings)
|
|
assert client.post(f'/admin/users/{user.id}', data={'role': 'user', 'is_active': 'false'}, headers=headers, follow_redirects=False).status_code == 303
|
|
db.refresh(user)
|
|
client.cookies.clear()
|
|
assert login(client, user.username, 'Initial user password').status_code == 401
|
|
assert not user.is_active
|
|
|
|
|
|
def test_csrf_duplicate_user_validation_and_login_throttling(accounts):
|
|
client, db, settings, admin, user, _ = accounts
|
|
headers = act_as(client, admin, settings)
|
|
assert client.post('/admin/users', data={'username': 'unprotected'}).status_code == 403
|
|
assert client.post('/admin/account/password', data={'current_password': 'anything'}).status_code == 403
|
|
assert client.post('/admin/login', data={'username': user.username, 'password': 'anything'}).status_code == 403
|
|
for username, role, code in [('reader', 'user', 409), ('bad login', 'user', 422), ('newuser', 'superuser', 422)]:
|
|
assert client.post('/admin/users', data={'username': username, 'role': role}, headers=headers).status_code == code
|
|
client.cookies.clear()
|
|
for _ in range(5):
|
|
assert login(client, user.username, 'wrong password').status_code == 401
|
|
db.refresh(user)
|
|
assert user.locked_until and user.failed_login_attempts == 5
|
|
assert login(client, user.username, 'Initial user password').status_code == 401
|
|
user.locked_until = utcnow() - timedelta(minutes=1)
|
|
db.commit()
|
|
for attempt in range(1, 6):
|
|
assert login(client, user.username, 'wrong password').status_code == 401
|
|
db.refresh(user)
|
|
assert user.failed_login_attempts == attempt
|
|
assert bool(user.locked_until) == (attempt == 5)
|
|
assert login(client, user.username, 'Initial user password').status_code == 401
|
|
db.refresh(user)
|
|
assert user.failed_login_attempts == 5
|
|
user.locked_until = utcnow() - timedelta(minutes=1)
|
|
db.commit()
|
|
assert login(client, user.username, 'Initial user password').status_code == 303
|
|
db.refresh(user)
|
|
assert user.failed_login_attempts == 0 and user.locked_until is None
|
|
|
|
|
|
def test_bootstrap_admin_first_login_and_changed_password_no_env_fallback(accounts):
|
|
client, db, settings, admin, _, _ = accounts
|
|
assert login(client, admin.username, 'initial-admin-test').headers['location'] == '/admin/account'
|
|
csrf = _session_data(client.cookies.get(SESSION_COOKIE), settings)['csrf']
|
|
assert client.post('/admin/account/password', data={'current_password': 'initial-admin-test',
|
|
'new_password': 'New administrator passphrase', 'confirmation': 'New administrator passphrase'}, headers={'X-CSRF-Token': csrf}, follow_redirects=False).status_code == 303
|
|
client.cookies.clear()
|
|
assert login(client, admin.username, 'initial-admin-test').status_code == 401
|
|
assert login(client, admin.username, 'New administrator passphrase').status_code == 303
|
|
|
|
|
|
def test_cross_origin_and_session_expiry_fail_closed(accounts, monkeypatch):
|
|
import time
|
|
from app import security
|
|
client, db, settings, admin, user, _ = accounts
|
|
headers = act_as(client, admin, settings)
|
|
assert client.post('/admin/users', data={'username': 'attacker'}, headers={**headers, 'Origin': 'https://evil.example'}).status_code == 403
|
|
assert client.post('/admin/users', data={'username': 'attacker'}, headers={**headers, 'Origin': 'http://['}).status_code == 403
|
|
assert db.scalar(select(SystemUser).where(SystemUser.username == 'attacker')) is None
|
|
token = sign_session(user.username, settings, user.session_version)
|
|
now = int(time.time())
|
|
monkeypatch.setattr(security.time, 'time', lambda: now + security.SESSION_SECONDS + 1)
|
|
assert security.read_session(token, settings) is None
|
|
|
|
|
|
def test_password_errors_and_noop_card_edit_are_not_audited(accounts):
|
|
client, db, settings, _, user, employee = accounts
|
|
headers = act_as(client, user, settings)
|
|
before = user.password_hash
|
|
for current, new, confirmation in [('wrong', 'New valid personal password', 'New valid personal password'),
|
|
('Initial user password', 'short', 'short'),
|
|
('Initial user password', 'New valid personal password', 'Different valid password')]:
|
|
response = client.post('/admin/account/password', headers=headers,
|
|
data={'current_password': current, 'new_password': new, 'confirmation': confirmation})
|
|
assert response.status_code == 422 and user.password_hash == before
|
|
article = council.council_candidates(db)['candidates'][0]['articles'][0]
|
|
response = client.post('/admin/council/publications', headers={'X-Council-Edit': '1'}, json={
|
|
'employee_id': employee.id, 'edit_key': article['edit_key'], 'revision': 0, 'title': article['title'],
|
|
'journal': article['edit_journal'], 'quartile': article['edit_quartile'], 'specialty_codes': None})
|
|
assert response.status_code == 200
|
|
assert db.scalars(select(UserAction).where(UserAction.user_id == user.id)).all() == []
|
|
|
|
|
|
def test_logout_revokes_replayed_cookie(accounts):
|
|
client, db, settings, _, user, _ = accounts
|
|
headers = act_as(client, user, settings)
|
|
old_token = client.cookies.get(SESSION_COOKIE)
|
|
assert client.post('/admin/logout', headers=headers, follow_redirects=False).status_code == 303
|
|
client.cookies.set(SESSION_COOKIE, old_token)
|
|
assert client.get('/api/employees', follow_redirects=False).status_code == 401
|
|
|
|
|
|
def test_password_change_and_audit_rollback_atomically(accounts, monkeypatch):
|
|
from sqlalchemy.exc import OperationalError
|
|
client, db, settings, _, user, _ = accounts
|
|
headers = act_as(client, user, settings)
|
|
before = user.password_hash
|
|
def unavailable():
|
|
raise OperationalError('UPDATE', {}, Exception('private database details'))
|
|
monkeypatch.setattr(db, 'commit', unavailable)
|
|
response = client.post('/admin/account/password', data={'current_password': 'Initial user password',
|
|
'new_password': 'A different personal passphrase', 'confirmation': 'A different personal passphrase'}, headers=headers)
|
|
assert response.status_code == 503 and 'private database details' not in response.text
|
|
db.refresh(user)
|
|
assert user.password_hash == before and user.session_version == 0
|
|
assert db.scalar(select(UserAction).where(UserAction.action == 'password_change')) is None
|
|
assert db.scalar(select(UserAction).where(UserAction.action == 'guide_start')) is None
|
|
|
|
|
|
@pytest.mark.parametrize('role', ['user', 'admin'])
|
|
def test_guide_first_password_dismissal_and_repeat_change(accounts, role):
|
|
client, db, settings, _, user, _ = accounts
|
|
user.role = role
|
|
db.commit()
|
|
headers = act_as(client, user, settings)
|
|
assert 'data-guide-pending="false"' in client.get('/admin/account').text
|
|
data = {'current_password': 'Initial user password', 'new_password': 'New onboarding passphrase',
|
|
'confirmation': 'New onboarding passphrase'}
|
|
assert client.post('/admin/account/password', data=data, headers=headers, follow_redirects=False).status_code == 303
|
|
page = client.get('/admin/account')
|
|
assert 'data-guide-pending="true"' in page.text
|
|
assert 'data-required-password' not in page.text
|
|
assert page.headers['Cache-Control'] == 'no-store'
|
|
guide = client.get('/admin/guide')
|
|
parsed = BeautifulSoup(guide.text, 'html.parser')
|
|
assert len(parsed.select('[data-guide-step]')) == 6
|
|
assert not parsed.select('[data-guide-dialog]')
|
|
assert 'Прочитайте показатели сотрудника' in guide.text and 'Проверьте строку публикации' in guide.text
|
|
assert ('управлять доступом' in guide.text) == (role == 'admin')
|
|
assert 'не ограничивает «ВСЕГО»' in guide.text
|
|
csrf = BeautifulSoup(page.text, 'html.parser').select_one('[data-guide-dialog]')['data-guide-csrf']
|
|
assert client.post('/admin/guide/dismiss', headers={'X-CSRF-Token': 'wrong'}, follow_redirects=False).status_code == 403
|
|
assert 'data-guide-pending="true"' in client.get('/admin/directory').text
|
|
for _ in range(2):
|
|
response = client.post('/admin/guide/dismiss', headers={'X-CSRF-Token': csrf, 'Accept': 'application/json'}, follow_redirects=False)
|
|
assert response.status_code == 204
|
|
assert len(db.scalars(select(UserAction).where(UserAction.user_id == user.id, UserAction.action == 'guide_finish')).all()) == 1
|
|
client.cookies.clear()
|
|
headers = act_as(client, user, settings)
|
|
assert 'data-guide-pending="false"' in client.get('/admin').text
|
|
data = {'current_password': data['new_password'], 'new_password': 'Second onboarding passphrase',
|
|
'confirmation': 'Second onboarding passphrase'}
|
|
assert client.post('/admin/account/password', data=data, headers=headers, follow_redirects=False).status_code == 303
|
|
assert 'data-guide-pending="false"' in client.get('/admin/account').text
|
|
assert client.get('/admin/guide').status_code == 200
|
|
assert len(db.scalars(select(UserAction).where(UserAction.user_id == user.id, UserAction.action == 'guide_start')).all()) == 1
|
|
|
|
|
|
def test_guide_access_and_no_javascript_completion(accounts):
|
|
client, db, settings, _, user, _ = accounts
|
|
assert 'data-guide-dialog' not in client.get('/admin/council').text
|
|
assert client.get('/admin/guide', follow_redirects=False).headers['location'] == '/admin/login'
|
|
user.must_change_password = True
|
|
db.commit()
|
|
headers = act_as(client, user, settings)
|
|
assert client.get('/admin/guide', follow_redirects=False).headers['location'] == '/admin/account'
|
|
assert client.post('/admin/guide/dismiss', headers=headers, follow_redirects=False).headers['location'] == '/admin/account'
|
|
user.must_change_password = False
|
|
db.add(UserAction(user_id=user.id, action='guide_start'))
|
|
db.commit()
|
|
page = BeautifulSoup(client.get('/admin').text, 'html.parser')
|
|
assert page.select_one('noscript a[href="/admin/guide"]')
|
|
response = client.post('/admin/guide/dismiss', headers=headers, follow_redirects=False)
|
|
assert response.status_code == 303 and response.headers['location'] == '/admin'
|
|
assert 'data-guide-pending="false"' in client.get('/admin/account').text
|
|
|
|
|
|
def test_guide_dismissal_failure_does_not_mark_it_finished(accounts, monkeypatch):
|
|
from sqlalchemy.exc import OperationalError
|
|
client, db, settings, _, user, _ = accounts
|
|
db.add(UserAction(user_id=user.id, action='guide_start'))
|
|
db.commit()
|
|
headers = act_as(client, user, settings)
|
|
def unavailable():
|
|
raise OperationalError('INSERT', {}, Exception('private database details'))
|
|
monkeypatch.setattr(db, 'commit', unavailable)
|
|
response = client.post('/admin/guide/dismiss', headers={**headers, 'Accept': 'application/json'})
|
|
assert response.status_code == 503 and 'private database details' not in response.text
|
|
assert db.scalar(select(UserAction).where(UserAction.action == 'guide_finish')) is None
|
|
assert 'data-guide-pending="true"' in client.get('/admin/account').text
|
|
|
|
|
|
def test_postgres_concurrent_admin_demotions_keep_an_admin(accounts):
|
|
from concurrent.futures import ThreadPoolExecutor
|
|
client, db, settings, admin, _, _ = accounts
|
|
if db.get_bind().dialect.name != 'postgresql':
|
|
pytest.skip('Requires the disposable PostgreSQL test database')
|
|
second = SystemUser(username='admin-two', role='admin', password_hash=hash_password('Second administrator passphrase'),
|
|
must_change_password=False, first_login_at=utcnow())
|
|
db.add(second)
|
|
db.commit()
|
|
factory = sessionmaker(bind=db.get_bind())
|
|
def sessions():
|
|
with factory() as session:
|
|
yield session
|
|
app.dependency_overrides[get_db] = sessions
|
|
clients = [TestClient(app), TestClient(app)]
|
|
actors = [admin, second]
|
|
targets = [second.id, admin.id]
|
|
def demote(index):
|
|
headers = act_as(clients[index], actors[index], settings)
|
|
response = clients[index].post(f'/admin/users/{targets[index]}',
|
|
data={'role': 'user', 'is_active': 'true'}, headers=headers, follow_redirects=False)
|
|
return response.status_code, response.headers.get('location')
|
|
try:
|
|
with ThreadPoolExecutor(max_workers=2) as executor:
|
|
responses = list(executor.map(demote, [0, 1]))
|
|
assert sum(location == '/admin/users' for status, location in responses) == 1
|
|
assert all(status in {303, 403} for status, location in responses)
|
|
remaining = db.scalars(select(SystemUser).where(SystemUser.role == 'admin', SystemUser.is_active.is_(True))).all()
|
|
assert len(remaining) == 1
|
|
finally:
|
|
for other in clients:
|
|
other.close()
|