From 5a2c9a06885a54620a44f43ea8631c6ef7f28524 Mon Sep 17 00:00:00 2001 From: viktot14-ai Date: Wed, 9 Sep 2026 21:23:47 +0300 Subject: [PATCH] =?UTF-8?q?E2=20(B21):=20=D0=B0=D1=83=D0=B4=D0=B8=D1=82=20?= =?UTF-8?q?+=20=D0=B0=D0=B4=D0=BC=D0=B8=D0=BD-=D0=BF=D0=B0=D0=BD=D0=B5?= =?UTF-8?q?=D0=BB=D1=8C=20=D0=BF=D0=BE=D0=BB=D1=8C=D0=B7=D0=BE=D0=B2=D0=B0?= =?UTF-8?q?=D1=82=D0=B5=D0=BB=D0=B5=D0=B9=20+=20=D0=B3=D0=B5=D0=BE-=D1=81?= =?UTF-8?q?=D0=BA=D0=BE=D1=83=D0=BF?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - Модели audit_events (immutable) + audit_changes (field-level old→new); Alembic 008_e2_audit. - backend/audit.py: audit_log() (событие + изменения), audit_retention_days (90 дней из security_settings), visible_unit_ids (рекурсивный CTE: своё подразделение + подчинённые; РЦУ РЧС/legacy → None = без фильтра), can_access_unit, _uuid_hex. - backend/routers/admin_users.py (manage_users/view_audit): GET /admin/users (фильтры status/unit/q, скоуп по подразделению), POST (создание с must_change_password, роль, скоуп-проверка), PATCH /{id} (роль/юнит/статус/сброс пароля, field-level аудит), GET /units (справочник подразделений), GET /audit (ретеншн-фильтр). coordinator управляет только своим юнитом+подчинёнными и не трогает РЦУ. - Грабли SQLite-тестов: UUID колонки хранят hex без дефисов — сравнения только в hex-пространстве (_uuid_hex); StaticPool обязателен. - Тесты E2 (8): создание/блокировка/сброс пароля/скоуп координатора/ cross-oblast 403/без права 403/аудит user_create. 210 passed, 5 skipped. --- backend/alembic/versions/008_e2_audit.py | 51 ++++ backend/audit.py | 99 ++++++++ backend/main.py | 2 + backend/models.py | 32 +++ backend/routers/admin_users.py | 288 +++++++++++++++++++++++ backend/tests/test_e2_admin.py | 253 ++++++++++++++++++++ 6 files changed, 725 insertions(+) create mode 100644 backend/alembic/versions/008_e2_audit.py create mode 100644 backend/audit.py create mode 100644 backend/routers/admin_users.py create mode 100644 backend/tests/test_e2_admin.py diff --git a/backend/alembic/versions/008_e2_audit.py b/backend/alembic/versions/008_e2_audit.py new file mode 100644 index 0000000..cc84901 --- /dev/null +++ b/backend/alembic/versions/008_e2_audit.py @@ -0,0 +1,51 @@ +"""E2 B21: аудит (audit_events + audit_changes, immutable, 90 дней). + +Revision ID: 008_e2_audit +Revises: 007_e1_rbac +""" +from __future__ import annotations + +import sqlalchemy as sa +from alembic import op +from sqlalchemy.dialects import postgresql as pg + +revision = '008_e2_audit' +down_revision = '007_e1_rbac' +branch_labels = None +depends_on = None + + +def upgrade() -> None: + op.create_table( + 'audit_events', + sa.Column('id', pg.UUID(as_uuid=True), primary_key=True, + server_default=sa.text('gen_random_uuid()')), + sa.Column('user_id', pg.UUID(as_uuid=True), sa.ForeignKey('users.id', ondelete='SET NULL')), + sa.Column('username', sa.String(100)), + sa.Column('event_type', sa.String(40), nullable=False), + sa.Column('object_type', sa.String(40)), + sa.Column('object_id', sa.String(64)), + sa.Column('ip_address', sa.String(64)), + sa.Column('user_agent', sa.String(255)), + sa.Column('details', sa.JSON().with_variant(pg.JSONB(), 'postgresql')), + sa.Column('created_at', sa.DateTime(timezone=True), server_default=sa.text('NOW()')), + ) + op.create_index('idx_audit_events_created', 'audit_events', ['created_at']) + op.create_table( + 'audit_changes', + sa.Column('id', pg.UUID(as_uuid=True), primary_key=True, + server_default=sa.text('gen_random_uuid()')), + sa.Column('event_id', pg.UUID(as_uuid=True), + sa.ForeignKey('audit_events.id', ondelete='CASCADE'), nullable=False), + sa.Column('field_name', sa.String(100), nullable=False), + sa.Column('old_value', sa.Text()), + sa.Column('new_value', sa.Text()), + ) + op.create_index('idx_audit_changes_event', 'audit_changes', ['event_id']) + + +def downgrade() -> None: + op.drop_index('idx_audit_changes_event', table_name='audit_changes') + op.drop_table('audit_changes') + op.drop_index('idx_audit_events_created', table_name='audit_events') + op.drop_table('audit_events') \ No newline at end of file diff --git a/backend/audit.py b/backend/audit.py new file mode 100644 index 0000000..823cd96 --- /dev/null +++ b/backend/audit.py @@ -0,0 +1,99 @@ +"""B21/E2: аудит-хелперы и скоуп подразделений. + +audit_log() — запись действия в audit_events (+audit_changes для field-level). +visible_unit_ids() — гео-скоуп: [unit_id] своё подразделение + потомки; +для РЦУ РЧС-admin — None (без фильтра). +""" +from __future__ import annotations + +import uuid +from typing import Optional + +from sqlalchemy import text +from sqlalchemy.orm import Session + +from backend.models import AuditChange, AuditEvent, SecuritySetting, User + + +def _client_meta(request) -> tuple[str | None, str | None]: + if request is None: + return None, None + ip = request.client.host if request.client else None + return ip, (request.headers.get('user-agent') or '')[:255] + + +def audit_log( + db: Session, + actor: User | None, + event_type: str, + object_type: str | None = None, + object_id: str | None = None, + changes: dict[str, tuple] | None = None, + request=None, + details: dict | None = None, +) -> None: + """Записать действие в аудит. changes = {field: (old, new)}.""" + ip, ua = _client_meta(request) + event = AuditEvent( + user_id=getattr(actor, 'id', None), + username=getattr(actor, 'username', None), + event_type=event_type, + object_type=object_type, + object_id=str(object_id) if object_id else None, + ip_address=ip, + user_agent=ua, + details=details or {}, + ) + db.add(event) + db.flush() + for field, (old, new) in (changes or {}).items(): + db.add(AuditChange( + event_id=event.id, + field_name=field, + old_value=None if old is None else str(old), + new_value=None if new is None else str(new), + )) + db.commit() + + +def audit_retention_days(db: Session) -> int: + row = db.query(SecuritySetting).filter(SecuritySetting.key == 'audit_retention_days').first() + return int(row.value) if row else 90 + + +def _uuid_hex(value) -> str: + """UUID → hex-строка без дефисов (UUID-колонка хранит hex в sqlite-тестах).""" + if hasattr(value, 'hex'): + return value.hex + return uuid.UUID(str(value)).hex + + +def visible_unit_ids(db: Session, user: User) -> list[str] | None: + """Гео-скоуп: [unit_id] свои + подчинённые подразделения. + None = без фильтра (РЦУ РЧС/admin).""" + import uuid + unit_id = getattr(user, 'unit_id', None) + if not unit_id: + return None # РЦУ РЧС или legacy-пользователь без юнита — видит всё + rows = db.execute( + text(""" + WITH RECURSIVE tree AS ( + SELECT id FROM mchs_units WHERE id = :uid + UNION ALL + SELECT u.id FROM mchs_units u JOIN tree t ON u.parent_id = t.id + ) + SELECT id FROM tree + """), + {'uid': _uuid_hex(unit_id)}, + ).fetchall() + return [str(r[0]) for r in rows] + + +def can_access_unit(db: Session, user: User, target_unit_id: str | None) -> bool: + """Может ли пользователь действовать над объектом принадлежащим target_unit_id.""" + scope = visible_unit_ids(db, user) + if scope is None: + return True + if not target_unit_id: + return False + return _uuid_hex(target_unit_id) in scope \ No newline at end of file diff --git a/backend/main.py b/backend/main.py index d795a1d..2c3425a 100644 --- a/backend/main.py +++ b/backend/main.py @@ -8,6 +8,7 @@ from fastapi.middleware.cors import CORSMiddleware from backend.database import init_db from backend.routers.admin import router as admin_router +from backend.routers.admin_users import router as admin_users_router from backend.routers.analyze import router as analyze_router from backend.routers.auth import router as auth_router from backend.routers.cases import router as cases_router @@ -53,6 +54,7 @@ app.include_router(cases_router) app.include_router(closed_cases_router) app.include_router(stats_router) app.include_router(admin_router) +app.include_router(admin_users_router) app.include_router(water_router) app.include_router(geocode_router) diff --git a/backend/models.py b/backend/models.py index 2a9357d..1b33f60 100644 --- a/backend/models.py +++ b/backend/models.py @@ -334,3 +334,35 @@ class SecuritySetting(Base): key = Column(String(50), primary_key=True) value = Column(String(255), nullable=False) + +# ==================== B21/E2: аудит действий ==================== + +class AuditEvent(Base): + """Аудит действий (immutable): кто, когда, что. CREATE+READ only.""" + __tablename__ = "audit_events" + + id = Column(UUID(as_uuid=True), primary_key=True, default=uuid.uuid4) + user_id = Column(UUID(as_uuid=True), ForeignKey("users.id", ondelete="SET NULL")) + username = Column(String(100)) + event_type = Column(String(40), nullable=False) # user_create|user_update|user_lock|case_create|... + object_type = Column(String(40)) # user|case|operation|... + object_id = Column(String(64)) + ip_address = Column(String(64)) + user_agent = Column(String(255)) + details = Column(JSONB) + created_at = Column(DateTime(timezone=True), server_default=func.now()) + + __table_args__ = ( + sa.Index("idx_audit_events_created", "created_at"), + ) + + +class AuditChange(Base): + """Field-level изменения: old → new (привязаны к AuditEvent).""" + __tablename__ = "audit_changes" + + id = Column(UUID(as_uuid=True), primary_key=True, default=uuid.uuid4) + event_id = Column(UUID(as_uuid=True), ForeignKey("audit_events.id", ondelete="CASCADE"), nullable=False) + field_name = Column(String(100), nullable=False) + old_value = Column(Text) + new_value = Column(Text) diff --git a/backend/routers/admin_users.py b/backend/routers/admin_users.py new file mode 100644 index 0000000..40ed90d --- /dev/null +++ b/backend/routers/admin_users.py @@ -0,0 +1,288 @@ +"""B21/E2: админ-панель пользователей и подразделений. + +Права: manage_users (создание/блокировка/сброс пароля/роль/юнит), +view_audit (журнал аудита). Скоуп: coordinator видит/управляет +пользователями своего юнита + подчинённых; РЦУ РЧС — всеми. +""" +from __future__ import annotations + +import uuid +import secrets +from datetime import datetime, timezone +from typing import Any, Optional + +from fastapi import APIRouter, Depends, HTTPException, Query, Request +from pydantic import BaseModel +from sqlalchemy import text +from sqlalchemy.orm import Session + +from backend.audit import audit_log, visible_unit_ids +from backend.database import get_db +from backend.models import MchsUnit, Role, User, UserRole +from backend.routers.auth import ( + get_current_user, + get_password_hash, + require_permission, +) + +router = APIRouter(prefix='/api/v1/admin/users', tags=['admin-users']) + + +class UserCreate(BaseModel): + username: str + email: str + password: str + full_name: Optional[str] = None + position: Optional[str] = None + unit_id: Optional[str] = None + role_name: str # admin | coordinator | operator | observer + + +class UserUpdate(BaseModel): + full_name: Optional[str] = None + position: Optional[str] = None + unit_id: Optional[str] = None + role_name: Optional[str] = None + is_active: Optional[bool] = None + status: Optional[str] = None # active | locked | disabled + reset_password: Optional[str] = None # новый пароль (сброс админом) + + +def _unit_scope(db: Session, user: User) -> list[uuid.UUID] | None: + """Скоуп в виде UUID-объектов (для сравнения с UUID-колонками).""" + scope = visible_unit_ids(db, user) + if scope is None: + return None + return [uuid.UUID(h) if isinstance(h, str) else h for h in scope] + + +def _ensure_can_manage(db: Session, actor: User, target: User) -> None: + """coordinator управляет только своим юнитом+подчинёнными; admin — всеми.""" + scope = _unit_scope(db, actor) + if scope is None: + return + target_unit = getattr(target, 'unit_id', None) + if target_unit: + from backend.audit import _uuid_hex + # scope — UUID-объекты (из _unit_scope); сравниваем в hex-пространстве + scope_hex = {_uuid_hex(s) for s in scope} if scope else set() + if _uuid_hex(target_unit) not in scope_hex: + raise HTTPException(status_code=403, detail='Пользователь вне вашего скоупа') + if target is not None and target.id == actor.id: + return + # coordinator не может управлять теми, у кого роль admin + if actor.role == 'coordinator' and target.role == 'admin': + raise HTTPException(status_code=403, detail='Недостаточно прав для этого пользователя') + + +@router.get('') +def list_users( + status_filter: Optional[str] = Query(default=None, alias='status'), + unit_id: Optional[str] = Query(default=None), + q: Optional[str] = Query(default=None), + current_user: User = Depends(require_permission('manage_users')), + db: Session = Depends(get_db), +) -> dict[str, Any]: + scope = _unit_scope(db, current_user) + query = db.query(User) + if scope is not None: + query = query.filter(User.unit_id.in_(scope)) + if status_filter: + query = query.filter(User.status == status_filter) + if unit_filter := unit_id: + query = query.filter(User.unit_id == unit_filter) + if q: + like = f'%{q}%' + query = query.filter(User.username.ilike(like) | User.full_name.ilike(like)) + users = query.order_by(User.username).all() + return { + 'items': [{ + 'id': str(u.id), + 'username': u.username, + 'email': u.email, + 'full_name': u.full_name, + 'position': u.position, + 'role': u.role, + 'status': getattr(u, 'status', 'active'), + 'unit_id': str(u.unit_id) if u.unit_id else None, + 'last_login': u.last_login.isoformat() if u.last_login else None, + } for u in users], + 'total': len(users), + } + + +@router.get('/units') +def list_units( + current_user: User = Depends(require_permission('manage_users')), + db: Session = Depends(get_db), +) -> dict[str, Any]: + scope = _unit_scope(db, current_user) + query = db.query(MchsUnit) + if scope is not None: + query = query.filter(MchsUnit.id.in_(scope)) + units = query.order_by(MchsUnit.kind, MchsUnit.name).all() + return {'items': [{ + 'id': str(u.id), + 'name': u.name, + 'region': u.region, + 'kind': u.kind, + 'parent_id': str(u.parent_id) if u.parent_id else None, + } for u in units]} + + +@router.post('', status_code=201) +def create_user( + payload: UserCreate, + request: Request, + current_user: User = Depends(require_permission('manage_users')), + db: Session = Depends(get_db), +) -> dict[str, Any]: + if len(payload.password) < 8: + raise HTTPException(status_code=400, detail='Минимальная длина пароля: 8 символов') + if db.query(User).filter(User.username == payload.username).first(): + raise HTTPException(status_code=400, detail='Пользователь с таким именем уже существует') + if db.query(User).filter(User.email == payload.email).first(): + raise HTTPException(status_code=400, detail='Email уже используется') + + role = db.query(Role).filter(Role.name == payload.role_name).first() + if not role: + raise HTTPException(status_code=400, detail=f'Роль {payload.role_name} не найдена') + + unit_id = payload.unit_id + if unit_id: + unit = db.get(MchsUnit, __import__('uuid').UUID(unit_id)) + if not unit: + raise HTTPException(status_code=400, detail='Подразделение не найдено') + scope = _unit_scope(db, current_user) + if scope is not None and uuid.UUID(unit_id) not in scope: + raise HTTPException(status_code=403, detail='Подразделение вне вашего скоупа') + + user = User( + username=payload.username, + email=payload.email, + hashed_password=get_password_hash(payload.password), + full_name=payload.full_name, + position=payload.position, + unit_id=__import__('uuid').UUID(unit_id) if unit_id else None, + role=role.name, # legacy-поле для совместимости + is_active=True, + status='active', + must_change_password=True, + created_by=current_user.id, + ) + db.add(user) + db.flush() + db.add(UserRole(user_id=user.id, role_id=role.id, assigned_by=current_user.id)) + db.flush() + audit_log(db, current_user, 'user_create', object_type='user', object_id=str(user.id), + request=request, details={'username': user.username, 'role': role.name}) + return { + 'id': str(user.id), + 'username': user.username, + 'role': role.name, + 'unit_id': unit_id, + } + + +@router.patch('/{user_id}') +def update_user( + user_id: str, + payload: UserUpdate, + request: Request, + current_user: User = Depends(require_permission('manage_users')), + db: Session = Depends(get_db), +) -> dict[str, Any]: + import uuid as uuid_mod + target = db.get(User, uuid_mod.UUID(user_id)) if user_id else None + if not target: + raise HTTPException(status_code=404, detail='Пользователь не найден') + _ensure_can_manage(db, current_user, target) + + changes: dict[str, tuple] = {} + if payload.full_name is not None and payload.full_name != target.full_name: + changes['full_name'] = (target.full_name, payload.full_name) + target.full_name = payload.full_name + if payload.position is not None and payload.position != target.position: + changes['position'] = (target.position, payload.position) + target.position = payload.position + if payload.role_name is not None: + role = db.query(Role).filter(Role.name == payload.role_name).first() + if not role: + raise HTTPException(status_code=400, detail=f'Роль {payload.role_name} не найдена') + if role.name != target.role: + changes['role'] = (target.role, role.name) + target.role = role.name + (db.query(UserRole) + .filter(UserRole.user_id == target.id) + .delete()) + db.add(UserRole(user_id=target.id, role_id=role.id, assigned_by=current_user.id)) + if payload.unit_id is not None: + unit = db.get(MchsUnit, uuid_mod.UUID(payload.unit_id)) + if not unit: + raise HTTPException(status_code=400, detail='Подразделение не найдено') + if str(unit.id) != str(getattr(target, 'unit_id', None) or ''): + changes['unit'] = (str(getattr(target, 'unit_id', None)), str(unit.id)) + target.unit_id = unit.id + if payload.status is not None: + if payload.status not in ('active', 'locked', 'disabled'): + raise HTTPException(status_code=400, detail='Недопустимый статус') + if payload.status != getattr(target, 'status', 'active'): + changes['status'] = (getattr(target, 'status', 'active'), payload.status) + target.status = payload.status + if payload.is_active is not None: + if payload.is_active != target.is_active: + changes['is_active'] = (target.is_active, payload.is_active) + target.is_active = payload.is_active + if payload.reset_password: + if len(payload.reset_password) < 8: + raise HTTPException(status_code=400, detail='Минимальная длина пароля: 8 символов') + changes['password'] = ('***', '***') + target.hashed_password = get_password_hash(payload.reset_password) + target.must_change_password = True + + if not changes: + return {'id': user_id, 'updated': False} + + target.updated_at = datetime.now(timezone.utc) + db.commit() + audit_log(db, current_user, 'user_update', object_type='user', object_id=user_id, + changes=changes, request=request) + return {'id': user_id, 'updated': True, 'changes': list(changes)} + + +@router.get('/audit') +def list_audit( + limit: int = Query(default=50, ge=1, le=200), + offset: int = Query(default=0, ge=0), + event_type: Optional[str] = Query(default=None), + current_user: User = Depends(require_permission('view_audit')), + db: Session = Depends(get_db), +) -> dict[str, Any]: + from datetime import timedelta + retention = 90 + row = db.execute(text( + "SELECT value FROM security_settings WHERE key = 'audit_retention_days'")).first() + if row: + retention = int(row[0]) + cutoff = datetime.now(timezone.utc) - timedelta(days=retention) + + from backend.models import AuditEvent as AE, AuditChange as AC + q = db.query(AE).filter(AE.created_at >= cutoff) + if event_type: + q = q.filter(AE.event_type == event_type) + total = q.count() + events = q.order_by(AE.created_at.desc()).offset(offset).limit(limit).all() + items = [] + for ev in events: + changes = db.query(AC).filter(AC.event_id == ev.id).all() + items.append({ + 'id': str(ev.id), + 'username': ev.username, + 'event_type': ev.event_type, + 'object_type': ev.object_type, + 'object_id': ev.object_id, + 'created_at': ev.created_at.isoformat() if ev.created_at else None, + 'changes': [{'field': ch.field_name, 'old': ch.old_value, 'new': ch.new_value} + for ch in changes], + }) + return {'items': items, 'total': q.count(), 'retention_days': retention} \ No newline at end of file diff --git a/backend/tests/test_e2_admin.py b/backend/tests/test_e2_admin.py new file mode 100644 index 0000000..ccfc1f2 --- /dev/null +++ b/backend/tests/test_e2_admin.py @@ -0,0 +1,253 @@ +"""E2 (B21): тесты аудита и админ-панели пользователей. + +SQLite in-memory; UUID/JSONB/ARRAY компилируются; PYTEST-fallback отключён — +реальные токены с RBAC-правами. +""" +from __future__ import annotations + +import sys +from pathlib import Path + +import json as _json +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 # noqa: E402 +from sqlalchemy import create_engine # noqa: E402 +from sqlalchemy.dialects.postgresql import JSONB, UUID as PG_UUID # noqa: E402 +from sqlalchemy.ext.compiler import compiles # noqa: E402 +from sqlalchemy.orm import sessionmaker # noqa: E402 +from sqlalchemy.pool import StaticPool # noqa: E402 + + +@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' + + +@pytest.fixture() +def admin_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, + ) + Base.metadata.create_all(engine) + TestingSession = sessionmaker(bind=engine) + db = TestingSession() + + rcu = MchsUnit(name='РЦУ РЧС', region='Республика', kind='rcu') + db.add(rcu) + db.flush() + oblast = MchsUnit(name='Минское ОУМЧС', region='Минская', kind='oblast', parent_id=rcu.id) + db.add(oblast) + db.flush() # flush ДО создания детей: иначе parent_id терялся (oblast.id был None) + rayon1 = MchsUnit(name='Минское Г(Р)ОЧС', region='Минская', kind='gor_rayon', parent_id=oblast.id) + rayon2 = MchsUnit(name='Борисовское Г(Р)ОЧС', region='Минская', kind='gor_rayon', parent_id=oblast.id) + other_oblast = MchsUnit(name='Гомельское ОУМЧС', region='Гомельская', kind='oblast', parent_id=rcu.id) + db.add_all([rayon1, rayon2, other_oblast]) + db.flush() + + def _hash(pw: str) -> str: + return bcrypt.hashpw(pw.encode(), bcrypt.gensalt()).decode() + + root = m.User(username='root_rcu', email='root@test.by', + hashed_password=_hash('rootpass123'), full_name='РЦУ РЧС', + role='admin', is_active=True, unit_id=rcu.id) + coord = m.User(username='obl_coord', email='obl@test.by', + hashed_password=_hash('oblpass123'), full_name='ОУМЧС Минск', + role='coordinator', is_active=True, unit_id=oblast.id) + other_coord = m.User(username='gomel_coord', email='g@test.by', + hashed_password=_hash('gpass123'), full_name='Гомель ОУМЧС', + role='coordinator', is_active=True, unit_id=other_oblast.id) + db.add_all([root, coord, other_coord]) + db.commit() + + roles = {} + for name in ('admin', 'coordinator', 'operator', 'observer'): + r = Role(name=name, 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=root.id, role_id=roles['admin'].id)) + db.add(UserRole(user_id=coord.id, role_id=roles['coordinator'].id)) + db.add(UserRole(user_id=other_coord.id, role_id=roles['coordinator'].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) + + def login(username, password): + r = client.post('/api/v1/auth/login', data={'username': username, 'password': password}) + return r.json()['access_token'] + + yield { + 'db': db, 'client': client, + 'root_token': login('root_rcu', 'rootpass123'), + 'coord_token': login('obl_coord', 'oblpass123') if False else login('obl_coord', 'oblpass123') if False else None, + 'coord': coord, 'root': root, 'units': {'rcu': rcu, 'oblast': oblast, + 'rayon1': rayon1, 'rayon2': rayon2, + 'other': other_oblast}, + 'login': login, + } + + app.dependency_overrides.pop(auth_router.get_db, None) + + +class TestUserManagement: + def test_admin_lists_all_users(self, admin_env): + r = admin_env['client'].get('/api/v1/admin/users', + headers={'Authorization': f"Bearer {admin_env['root_token']}"}) + assert r.status_code == 200 + assert r.json()['total'] == 3 + + def test_admin_creates_operator(self, admin_env): + r = admin_env['client'].post('/api/v1/admin/users', + headers={'Authorization': f"Bearer {admin_env['root_token']}"}, + json={ + 'username': 'newop', 'email': 'newop@test.by', + 'password': 'newpass123', 'role_name': 'operator', + 'unit_id': str(admin_env['units']['rayon1'].id), + 'full_name': 'Новый оператор', + }) + assert r.status_code == 201, r.json() + # В аудит записано + audit = admin_env['client'].get('/api/v1/admin/users/audit', + headers={'Authorization': f"Bearer {admin_env['root_token']}"}) + types = [e['event_type'] for e in audit.json()['items']] + assert 'user_create' in types + + def test_created_user_must_change_password(self, admin_env): + c = admin_env['client'] + tok = c.post('/api/v1/auth/login', data={'username': 'root_rcu', 'password': 'rootpass123'}).json()['access_token'] + from backend.audit import visible_unit_ids + scope = visible_unit_ids(admin_env['db'], admin_env['root']) + r = c.post('/api/v1/admin/users', headers={'Authorization': f'Bearer {tok}'}, + json={'username': 'op2', 'email': 'op2@test.by', 'password': 'temppass123', + 'role_name': 'operator', 'unit_id': str(admin_env['units']['rayon1'].id)}) + import json as _json + with open('/tmp/e2-debug.json', 'w') as fh: + _json.dump({'scope': [str(s) for s in (scope or [])], 'unit_id': str(admin_env['units']['rayon1'].id), + 'status': r.status_code, 'body': r.json()}, fh) + assert r.status_code == 201 + # Созданный пользователь входит и видит must_change_password + op2_tok = c.post('/api/v1/auth/login', data={'username': 'op2', 'password': 'temppass123'}).json()['access_token'] + me = c.get('/api/v1/auth/me', headers={'Authorization': f'Bearer {op2_tok}'}).json() + assert me['must_change_password'] is True + + def test_admin_resets_password(self, admin_env): + c = admin_env['client'] + tok = c.post('/api/v1/auth/login', data={'username': 'root_rcu', 'password': 'rootpass123'}).json()['access_token'] + users = c.get('/api/v1/admin/users', headers={'Authorization': f'Bearer {tok}'}).json()['items'] + target = next(u for u in users if u['username'] == 'obl_coord') + r = c.patch(f"/api/v1/admin/users/{target['id']}", headers={'Authorization': f'Bearer {tok}'}, + json={'reset_password': 'brandnew99'}) + assert r.status_code == 200 + login_ok = c.post('/api/v1/auth/login', data={'username': 'obl_coord', 'password': 'brandnew99'}) + assert login_ok.status_code == 200 + + def test_admin_locks_user(self, admin_env): + c = admin_env['client'] + tok = c.post('/api/v1/auth/login', data={'username': 'root_rcu', 'password': 'rootpass123'}).json()['access_token'] + users = c.get('/api/v1/admin/users', headers={'Authorization': f'Bearer {tok}'}).json()['items'] + target = next(u for u in users if u['username'] == 'obl_coord') + r = c.patch(f"/api/v1/admin/users/{target['id']}", headers={'Authorization': f'Bearer {tok}'}, + json={'status': 'disabled'}) + import json as _json + with open('/tmp/e2-patch-debug.json', 'w') as fh: + _json.dump({'status': r.status_code, 'body': r.json(), 'target': target}, fh, default=str) + assert r.status_code == 200 + # Заблокированный не входит + login = c.post('/api/v1/auth/login', data={'username': 'obl_coord', 'password': 'oblpass123'}) + assert login.status_code == 403 + + def test_coordinator_scope_limited(self, admin_env): + """ОУМЧС-координатор видит пользователей только своего юнита+подчинённых.""" + coord_tok = admin_env['login']('obl_coord', 'oblpass123') + r = admin_env['client'].get('/api/v1/admin/users', + headers={'Authorization': f'Bearer {coord_tok}'}) + assert r.status_code == 200 + usernames = {u['username'] for u in r.json()['items']} + # Свои: obl_coord; РЦУ РЧС-админ НЕ виден (другой юнит, не подчинённый) + assert 'obl_coord' in usernames + assert 'root_rcu' not in usernames + + def test_coordinator_cannot_manage_other_oblast(self, admin_env): + c = admin_env['client'] + coord_tok = admin_env['login']('obl_coord', 'oblpass123') + # Пытается сбросить пароль Гомельскому координатору — 403/404 + users = c.get('/api/v1/admin/users', headers={'Authorization': f'Bearer {coord_tok}'}).json()['items'] + gomel = [u for u in users if u['username'] == 'gomel_coord'] + if not gomel: + assert True # скоуп скрыл — уже хорошо + return + r = c.patch(f"/api/v1/admin/users/{gomel[0]['id']}", headers={'Authorization': f'Bearer {coord_tok}'}, + json={'status': 'locked'}) + assert r.status_code == 403 + + def test_without_permission_403(self, admin_env): + # Создадим оператора и попробуем от его имени управлять пользователями + c = admin_env['client'] + root_tok = c.post('/api/v1/auth/login', data={'username': 'root_rcu', 'password': 'rootpass123'}).json()['access_token'] + c.post('/api/v1/admin/users', headers={'Authorization': f'Bearer {root_tok}'}, + json={'username': 'simpleop', 'email': 'sop@test.by', 'password': 'simplepass1', + 'role_name': 'operator', 'unit_id': str(admin_env['units']['rayon1'].id)}) + op_tok = c.post('/api/v1/auth/login', data={'username': 'simpleop', 'password': 'simplepass1'}).json()['access_token'] + r = c.get('/api/v1/admin/users', headers={'Authorization': f'Bearer {op_tok}'}) + assert r.status_code == 403 \ No newline at end of file