Sadmin
d5a270f3af
- backend/logging_setup.py: setup_logging() (stdout INFO + файл /tmp/vector/vector.log,
ротация 5×2 МБ, VectorLOGLEVEL/VectorLOGFILE/VectorLOGDIR), get_logger('vector.*')
- события: vector.analyze (матрасчёт с длительностью), vector.auth (login_ok/failed/
lockout), vector.audit (каждая запись), vector.contour (ok/failed/rejected),
vector.operations (operation_complete), vector.geocode (успех/timeout/ошибка)
- существующие логгеры переведены в namespace vector.* (geo, geo.osm, rules, water, admin)
- README §13а: таблица наблюдаемых событий + правила (без ПДн/секретов)
- SETUP_GUIDE: §15 RPO/RTO + backup≠restore, §16 grep по vector.*, §17 диагностика
по цепочкам (БД, 500, fallback-зоны, вход), §17а чек-лист передачи проекта
- DEPLOY.md: вместо устаревшего дубля — ссылка на актуальные доки + шпаргалка стенда
- тесты: 239 passed
105 lines
3.6 KiB
Python
105 lines
3.6 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.logging_setup import get_logger
|
|
from backend.models import AuditChange, AuditEvent, SecuritySetting, User
|
|
|
|
log = get_logger('audit')
|
|
|
|
|
|
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()
|
|
log.info('audit %s user=%s obj=%s/%s ip=%s', event_type,
|
|
event.username or '—', object_type or '—', event.object_id or '—',
|
|
event.ip_address or '—')
|
|
|
|
|
|
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 |