E1 (B21): RBAC-ядро — подразделения МЧС, роли/права, сессии, auth-журнал
- Модель данных: mchs_units (РЦУ РЧС → ОУМЧС → Г(Р)ОЧС, kind+parent_id), users +8 колонок (unit_id, position, status, lockout, must_change_password, created_by, updated_at), roles/permissions/role_permissions/user_roles (M2M матрица), sessions (server-side, token_hash=sha256(secret)), auth_events, security_settings. - Alembic 007_e1_rbac: схема + сид 4 ролей / 9 прав / матрицы (9/6/3/1) / 10 настроек безопасности; downgrade → upgrade идемпотентен (проверено на pg16-контейнере, сид не дублируется). - auth.py: login с lockout (5 неудач → 15 мин) и auth-журналом; server-side сессии (JWT несёт secret, в БД — sha256-хэш; max_concurrent_sessions); /me с permissions (user_roles → role_permissions, fallback legacy-роль); /change-password (мин. длина из settings, отзыв сессий); /logout. require_permission(code) — атомарные права (fallback legacy-роли). - Обратная совместимость: require_roles не тронут, старые JWT без sid продолжают работать. - Тесты E1 (9): login/logout/lockout/me-права/смена пароля/журнал. SQLite in-memory (StaticPool + компиляторы UUID/JSONB/ARRAY). 202 passed, 5 skipped.
This commit is contained in:
@@ -0,0 +1,198 @@
|
||||
"""E1 (B21): тесты RBAC-ядра — сессии, блокировка, права, журнал.
|
||||
|
||||
SQLite in-memory с RBAC-таблицами; UUID/JSONB подменяются компиляторами.
|
||||
"""
|
||||
from __future__ import annotations
|
||||
|
||||
import sys
|
||||
from pathlib import Path
|
||||
|
||||
import bcrypt
|
||||
import pytest
|
||||
|
||||
REPO_ROOT = Path(__file__).resolve().parents[2]
|
||||
if str(REPO_ROOT) not in sys.path:
|
||||
sys.path.insert(0, str(REPO_ROOT))
|
||||
|
||||
import sqlalchemy
|
||||
from sqlalchemy import create_engine
|
||||
from sqlalchemy.dialects.postgresql import JSONB, UUID as PG_UUID
|
||||
from sqlalchemy.ext.compiler import compiles
|
||||
from sqlalchemy.orm import sessionmaker
|
||||
from sqlalchemy.pool import StaticPool
|
||||
|
||||
|
||||
@compiles(PG_UUID, 'sqlite')
|
||||
def _uuid_sqlite(type_, compiler, **kw):
|
||||
return 'CHAR(36)'
|
||||
|
||||
|
||||
@compiles(JSONB, 'sqlite')
|
||||
def _jsonb_sqlite(type_, compiler, **kw):
|
||||
return 'JSON'
|
||||
|
||||
|
||||
@compiles(sqlalchemy.ARRAY, 'sqlite')
|
||||
def _array_sqlite(type_, compiler, **kw):
|
||||
return 'TEXT'
|
||||
|
||||
|
||||
@compiles(sqlalchemy.ARRAY, 'sqlite')
|
||||
def _array_sqlite(type_, compiler, **kw):
|
||||
return 'TEXT'
|
||||
|
||||
|
||||
@pytest.fixture()
|
||||
def rbac_env():
|
||||
from backend import models as m
|
||||
from backend.database import Base
|
||||
from backend.models import (
|
||||
MchsUnit,
|
||||
Permission,
|
||||
Role,
|
||||
RolePermission,
|
||||
SecuritySetting,
|
||||
UserRole,
|
||||
)
|
||||
|
||||
engine = create_engine(
|
||||
'sqlite:///:memory:',
|
||||
connect_args={'check_same_thread': False},
|
||||
poolclass=StaticPool, # одна connection на все threads (иначе per-thread БД)
|
||||
)
|
||||
Base.metadata.create_all(engine)
|
||||
TestingSession = sessionmaker(bind=engine)
|
||||
db = TestingSession()
|
||||
|
||||
unit = MchsUnit(name='Минское Г(Р)ОЧС', region='Минская', kind='gor_rayon')
|
||||
rcu = MchsUnit(name='РЦУ РЧС', region='Республика', kind='rcu')
|
||||
db.add_all([unit, rcu])
|
||||
db.flush()
|
||||
|
||||
def _hash(pw: str) -> str:
|
||||
return bcrypt.hashpw(pw.encode(), bcrypt.gensalt()).decode()
|
||||
|
||||
user = m.User(username='op1', email='op1@test.by', hashed_password=_hash('oldpass123'),
|
||||
full_name='Оператор Г(Р)ОЧС', role='operator', is_active=True,
|
||||
unit_id=unit.id)
|
||||
admin = m.User(username='root_rcu', email='root@test.by', hashed_password=_hash('rootpass123'),
|
||||
full_name='РЦУ РЧС', role='admin', is_active=True, unit_id=rcu.id)
|
||||
db.add_all([user, admin])
|
||||
db.commit()
|
||||
|
||||
roles = {}
|
||||
for name, desc in [('admin', ''), ('coordinator', ''), ('operator', ''), ('observer', '')]:
|
||||
r = Role(name=name, description=desc, is_system=True)
|
||||
db.add(r)
|
||||
roles[name] = r
|
||||
perms = {}
|
||||
for code in ('view', 'create', 'update', 'delete', 'export', 'manage_users',
|
||||
'manage_roles', 'view_audit', 'manage_security'):
|
||||
p = Permission(code=code)
|
||||
db.add(p)
|
||||
perms[code] = p
|
||||
db.commit()
|
||||
matrix = {
|
||||
'admin': tuple(perms),
|
||||
'coordinator': ('view', 'create', 'update', 'export', 'manage_users', 'view_audit'),
|
||||
'operator': ('view', 'create', 'update'),
|
||||
'observer': ('view',),
|
||||
}
|
||||
for rn, codes in matrix.items():
|
||||
for c in codes:
|
||||
db.add(RolePermission(role_id=roles[rn].id, permission_id=perms[c].id))
|
||||
db.add(UserRole(user_id=user.id, role_id=roles['operator'].id))
|
||||
db.commit()
|
||||
|
||||
from backend.rbac import SECURITY_DEFAULTS
|
||||
for k, v in SECURITY_DEFAULTS.items():
|
||||
db.add(SecuritySetting(key=k, value=v))
|
||||
db.commit()
|
||||
|
||||
from backend.main import app
|
||||
from backend.routers import auth as auth_router
|
||||
from fastapi.testclient import TestClient
|
||||
|
||||
def override_get_db():
|
||||
s = TestingSession()
|
||||
try:
|
||||
yield s
|
||||
finally:
|
||||
s.close()
|
||||
|
||||
app.dependency_overrides[auth_router.get_db] = override_get_db
|
||||
client = TestClient(app)
|
||||
|
||||
yield {'db': db, 'client': client, 'user': user, 'admin': admin}
|
||||
|
||||
app.dependency_overrides.pop(auth_router.get_db, None)
|
||||
|
||||
|
||||
def _login(client, username: str, password: str):
|
||||
return client.post('/api/v1/auth/login', data={'username': username, 'password': password})
|
||||
|
||||
|
||||
class TestLogin:
|
||||
def test_login_creates_session_and_token(self, rbac_env):
|
||||
r = _login(rbac_env['client'], 'op1', 'oldpass123')
|
||||
assert r.status_code == 200
|
||||
assert 'access_token' in r.json()
|
||||
|
||||
def test_wrong_password_fails(self, rbac_env):
|
||||
assert _login(rbac_env['client'], 'op1', 'wrong').status_code == 401
|
||||
|
||||
def test_lockout_after_max_failures(self, rbac_env):
|
||||
c = rbac_env['client']
|
||||
for _ in range(5):
|
||||
c.post('/api/v1/auth/login', data={'username': 'op1', 'password': 'wrong'})
|
||||
r = _login(c, 'op1', 'oldpass123')
|
||||
assert r.status_code == 403
|
||||
assert 'locked' in r.json()['detail'].lower()
|
||||
|
||||
def test_me_has_permissions(self, rbac_env):
|
||||
c = rbac_env['client']
|
||||
tok = _login(c, 'op1', 'oldpass123').json()['access_token']
|
||||
me = c.get('/api/v1/auth/me', headers={'Authorization': f'Bearer {tok}'}).json()
|
||||
assert me['permissions'] == ['create', 'update', 'view']
|
||||
assert me['unit_id'] is not None
|
||||
|
||||
def test_admin_permissions_full(self, rbac_env):
|
||||
c = rbac_env['client']
|
||||
tok = _login(c, 'root_rcu', 'rootpass123').json()['access_token']
|
||||
me = c.get('/api/v1/auth/me', headers={'Authorization': f'Bearer {tok}'}).json()
|
||||
assert 'manage_users' in me['permissions']
|
||||
assert len(me['permissions']) == 9
|
||||
|
||||
def test_logout_revokes_session(self, rbac_env):
|
||||
c = rbac_env['client']
|
||||
tok = _login(c, 'op1', 'oldpass123').json()['access_token']
|
||||
h = {'Authorization': f'Bearer {tok}'}
|
||||
assert c.get('/api/v1/auth/me', headers=h).status_code == 200
|
||||
c.post('/api/v1/auth/logout', headers=h)
|
||||
assert c.get('/api/v1/auth/me', headers=h).status_code == 401
|
||||
|
||||
def test_change_password_revokes_sessions(self, rbac_env):
|
||||
c = rbac_env['client']
|
||||
tok = _login(c, 'op1', 'oldpass123').json()['access_token']
|
||||
h = {'Authorization': f'Bearer {tok}'}
|
||||
r = c.post('/api/v1/auth/change-password', headers=h,
|
||||
json={'current_password': 'oldpass123', 'new_password': 'newpass456'})
|
||||
assert r.status_code == 200
|
||||
assert c.get('/api/v1/auth/me', headers=h).status_code == 401
|
||||
assert _login(c, 'op1', 'newpass456').status_code == 200
|
||||
|
||||
def test_short_password_rejected(self, rbac_env):
|
||||
c = rbac_env['client']
|
||||
tok = _login(c, 'op1', 'oldpass123').json()['access_token']
|
||||
h = {'Authorization': f'Bearer {tok}'}
|
||||
r = c.post('/api/v1/auth/change-password', headers=h,
|
||||
json={'current_password': 'oldpass123', 'new_password': 'abc'})
|
||||
assert r.status_code == 400
|
||||
|
||||
def test_auth_events_logged(self, rbac_env):
|
||||
from backend.models import AuthEvent
|
||||
c = rbac_env['client']
|
||||
_login(c, 'op1', 'oldpass123')
|
||||
_login(c, 'op1', 'WRONG')
|
||||
types = {e.event_type for e in rbac_env['db'].query(AuthEvent).all()}
|
||||
assert 'login' in types and 'login_failed' in types
|
||||
Reference in New Issue
Block a user