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.
99 lines
3.4 KiB
Python
99 lines
3.4 KiB
Python
"""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 |