Sadmin
5a2c9a0688
- Модели 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.
288 lines
12 KiB
Python
288 lines
12 KiB
Python
"""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} |