mirror of
https://github.com/idrainformatica/PecFlow.git
synced 2026-06-16 12:45:42 +02:00
123 lines
4.5 KiB
Python
123 lines
4.5 KiB
Python
"""
|
||
Router Audit Log – consultazione ed esportazione degli eventi di sistema.
|
||
|
||
Endpoint:
|
||
GET /api/v1/audit-log – lista paginata con filtri (solo admin/super_admin)
|
||
GET /api/v1/audit-log/export – esportazione CSV o PDF (solo admin/super_admin)
|
||
|
||
Permessi:
|
||
- admin: vede solo gli eventi del proprio tenant
|
||
- super_admin: vede tutti i tenant (filtrabile per tenant_id)
|
||
"""
|
||
|
||
import uuid
|
||
from datetime import datetime
|
||
from typing import Optional
|
||
|
||
from fastapi import APIRouter, Query
|
||
from fastapi.responses import Response, StreamingResponse
|
||
|
||
from app.dependencies import AdminUser, DB
|
||
from app.schemas.audit_log import AuditLogListResponse
|
||
from app.services.audit_service import AuditService
|
||
|
||
router = APIRouter(prefix="/audit-log", tags=["Audit Log"])
|
||
|
||
|
||
def _effective_tenant_id(current_user, tenant_id: Optional[uuid.UUID]) -> Optional[uuid.UUID]:
|
||
"""Determina il tenant_id effettivo in base al ruolo dell'utente."""
|
||
if current_user.is_super_admin:
|
||
return tenant_id # None = tutti i tenant
|
||
return current_user.tenant_id # sempre vincolato al proprio tenant
|
||
|
||
|
||
@router.get("", response_model=AuditLogListResponse)
|
||
async def list_audit_log(
|
||
current_user: AdminUser,
|
||
db: DB,
|
||
page: int = Query(1, ge=1, description="Numero di pagina"),
|
||
page_size: int = Query(25, ge=1, le=100, description="Elementi per pagina"),
|
||
action: Optional[str] = Query(None, description="Filtra per azione (es. auth.login, user.*)"),
|
||
user_id: Optional[uuid.UUID] = Query(None, description="Filtra per utente"),
|
||
outcome: Optional[str] = Query(None, pattern="^(success|failure)$", description="Esito: success o failure"),
|
||
date_from: Optional[datetime] = Query(None, description="Data inizio (ISO 8601)"),
|
||
date_to: Optional[datetime] = Query(None, description="Data fine (ISO 8601)"),
|
||
resource_type: Optional[str] = Query(None, description="Tipo risorsa (user, mailbox, message, ...)"),
|
||
tenant_id: Optional[uuid.UUID] = Query(None, description="Filtra per tenant (solo super_admin)"),
|
||
) -> AuditLogListResponse:
|
||
"""Restituisce la lista paginata degli eventi di audit."""
|
||
svc = AuditService(db)
|
||
return await svc.list(
|
||
tenant_id=_effective_tenant_id(current_user, tenant_id),
|
||
page=page,
|
||
page_size=page_size,
|
||
action=action,
|
||
user_id=user_id,
|
||
outcome=outcome,
|
||
date_from=date_from,
|
||
date_to=date_to,
|
||
resource_type=resource_type,
|
||
)
|
||
|
||
|
||
@router.get("/export")
|
||
async def export_audit_log(
|
||
current_user: AdminUser,
|
||
db: DB,
|
||
format: str = Query("csv", pattern="^(csv|pdf)$", description="Formato: csv o pdf"),
|
||
action: Optional[str] = Query(None),
|
||
user_id: Optional[uuid.UUID] = Query(None),
|
||
outcome: Optional[str] = Query(None, pattern="^(success|failure)$"),
|
||
date_from: Optional[datetime] = Query(None),
|
||
date_to: Optional[datetime] = Query(None),
|
||
resource_type: Optional[str] = Query(None),
|
||
tenant_id: Optional[uuid.UUID] = Query(None),
|
||
) -> Response:
|
||
"""
|
||
Esporta i log di audit in formato CSV o PDF.
|
||
Applica gli stessi filtri dell'endpoint lista.
|
||
"""
|
||
svc = AuditService(db)
|
||
effective_tid = _effective_tenant_id(current_user, tenant_id)
|
||
|
||
# Nome file con periodo
|
||
suffix = ""
|
||
if date_from:
|
||
suffix += f"_dal_{date_from.strftime('%Y%m%d')}"
|
||
if date_to:
|
||
suffix += f"_al_{date_to.strftime('%Y%m%d')}"
|
||
|
||
if format == "csv":
|
||
csv_content = await svc.export_csv(
|
||
tenant_id=effective_tid,
|
||
action=action,
|
||
user_id=user_id,
|
||
outcome=outcome,
|
||
date_from=date_from,
|
||
date_to=date_to,
|
||
resource_type=resource_type,
|
||
)
|
||
filename = f"audit_log{suffix}.csv"
|
||
return Response(
|
||
content=csv_content.encode("utf-8-sig"), # BOM per compatibilita' Excel
|
||
media_type="text/csv; charset=utf-8",
|
||
headers={"Content-Disposition": f'attachment; filename="{filename}"'},
|
||
)
|
||
|
||
else: # pdf
|
||
pdf_bytes = await svc.export_pdf_bytes(
|
||
tenant_id=effective_tid,
|
||
action=action,
|
||
user_id=user_id,
|
||
outcome=outcome,
|
||
date_from=date_from,
|
||
date_to=date_to,
|
||
resource_type=resource_type,
|
||
)
|
||
filename = f"audit_log{suffix}.pdf"
|
||
return Response(
|
||
content=pdf_bytes,
|
||
media_type="application/pdf",
|
||
headers={"Content-Disposition": f'attachment; filename="{filename}"'},
|
||
)
|