312 lines
9.2 KiB
Python
312 lines
9.2 KiB
Python
"""
|
|
Audit Service - ServiceManagerWeb
|
|
|
|
Funciones helper para facilitar el registro de auditoría.
|
|
Simplifica el proceso de logging en toda la aplicación.
|
|
"""
|
|
|
|
from typing import Optional, Dict, Any
|
|
from sqlalchemy.ext.asyncio import AsyncSession
|
|
from fastapi import Request
|
|
import uuid
|
|
import structlog
|
|
|
|
from app.models.audit import AuditLog
|
|
from app.models.user import User
|
|
|
|
logger = structlog.get_logger(__name__)
|
|
|
|
|
|
class AuditService:
|
|
"""
|
|
Servicio centralizado para registro de auditoría.
|
|
|
|
Uso básico:
|
|
await AuditService.log(
|
|
db=db,
|
|
tenant_id=tenant.id,
|
|
user_id=current_user.id,
|
|
action="ticket.create",
|
|
resource_type="ticket",
|
|
resource_id=new_ticket.id,
|
|
new_values={"subject": "...", "status": "NEW"}
|
|
)
|
|
"""
|
|
|
|
@staticmethod
|
|
async def log(
|
|
db: AsyncSession,
|
|
tenant_id: uuid.UUID,
|
|
action: str,
|
|
resource_type: str,
|
|
resource_id: Optional[uuid.UUID] = None,
|
|
user_id: Optional[uuid.UUID] = None,
|
|
old_values: Optional[Dict[str, Any]] = None,
|
|
new_values: Optional[Dict[str, Any]] = None,
|
|
metadata: Optional[Dict[str, Any]] = None,
|
|
request: Optional[Request] = None
|
|
) -> AuditLog:
|
|
"""
|
|
Registra una acción en la bitácora de auditoría.
|
|
|
|
Args:
|
|
db: Sesión de base de datos
|
|
tenant_id: ID del tenant
|
|
action: Acción realizada (formato: "recurso.verbo")
|
|
Ejemplos: "user.login", "ticket.create", "ticket.assign"
|
|
resource_type: Tipo de recurso ("user", "ticket", "comment", etc.)
|
|
resource_id: ID del recurso afectado (opcional)
|
|
user_id: ID del usuario que ejecutó la acción (opcional = sistema)
|
|
old_values: Valores antes del cambio (opcional)
|
|
new_values: Valores después del cambio (opcional)
|
|
metadata: Información adicional (opcional)
|
|
request: Request de FastAPI para extraer IP y user agent (opcional)
|
|
|
|
Returns:
|
|
AuditLog creado
|
|
"""
|
|
# Extraer información del request si está disponible
|
|
ip_address = None
|
|
user_agent = None
|
|
correlation_id = None
|
|
|
|
if request:
|
|
# IP del cliente
|
|
if request.client:
|
|
ip_address = request.client.host
|
|
|
|
# User agent
|
|
user_agent = request.headers.get("user-agent")
|
|
|
|
# Correlation ID (si existe en el request state)
|
|
correlation_id = getattr(request.state, "correlation_id", None)
|
|
|
|
# Crear registro de auditoría
|
|
audit_log = AuditLog(
|
|
tenant_id=tenant_id,
|
|
user_id=user_id,
|
|
action=action,
|
|
resource_type=resource_type,
|
|
resource_id=resource_id,
|
|
ip_address=ip_address,
|
|
user_agent=user_agent,
|
|
correlation_id=correlation_id,
|
|
old_values=old_values,
|
|
new_values=new_values,
|
|
extra_metadata=metadata # Mapeo metadata -> extra_metadata
|
|
)
|
|
|
|
db.add(audit_log)
|
|
await db.flush() # No commit, se hará con la transacción principal
|
|
|
|
# Log estructurado para debugging
|
|
logger.info(
|
|
"Audit log created",
|
|
action=action,
|
|
resource_type=resource_type,
|
|
resource_id=str(resource_id) if resource_id else None,
|
|
user_id=str(user_id) if user_id else "system",
|
|
tenant_id=str(tenant_id)
|
|
)
|
|
|
|
return audit_log
|
|
|
|
@staticmethod
|
|
async def log_login(
|
|
db: AsyncSession,
|
|
user: User,
|
|
request: Request,
|
|
success: bool = True
|
|
) -> AuditLog:
|
|
"""
|
|
Registra un intento de login.
|
|
|
|
Args:
|
|
db: Sesión de base de datos
|
|
user: Usuario que intentó loguearse
|
|
request: Request de FastAPI
|
|
success: Si el login fue exitoso
|
|
|
|
Returns:
|
|
AuditLog creado
|
|
"""
|
|
return await AuditService.log(
|
|
db=db,
|
|
tenant_id=user.tenant_id,
|
|
user_id=user.id if success else None,
|
|
action="user.login" if success else "user.login_failed",
|
|
resource_type="user",
|
|
resource_id=user.id,
|
|
metadata={
|
|
"success": success,
|
|
"email": user.email
|
|
},
|
|
request=request
|
|
)
|
|
|
|
@staticmethod
|
|
async def log_logout(
|
|
db: AsyncSession,
|
|
user: User,
|
|
request: Request
|
|
) -> AuditLog:
|
|
"""
|
|
Registra un logout.
|
|
|
|
Args:
|
|
db: Sesión de base de datos
|
|
user: Usuario que cerró sesión
|
|
request: Request de FastAPI
|
|
|
|
Returns:
|
|
AuditLog creado
|
|
"""
|
|
return await AuditService.log(
|
|
db=db,
|
|
tenant_id=user.tenant_id,
|
|
user_id=user.id,
|
|
action="user.logout",
|
|
resource_type="user",
|
|
resource_id=user.id,
|
|
request=request
|
|
)
|
|
|
|
@staticmethod
|
|
async def log_create(
|
|
db: AsyncSession,
|
|
tenant_id: uuid.UUID,
|
|
user_id: uuid.UUID,
|
|
resource_type: str,
|
|
resource_id: uuid.UUID,
|
|
new_values: Dict[str, Any],
|
|
request: Optional[Request] = None
|
|
) -> AuditLog:
|
|
"""
|
|
Registra la creación de un recurso.
|
|
|
|
Args:
|
|
db: Sesión de base de datos
|
|
tenant_id: ID del tenant
|
|
user_id: ID del usuario que creó el recurso
|
|
resource_type: Tipo de recurso ("ticket", "user", etc.)
|
|
resource_id: ID del recurso creado
|
|
new_values: Valores del nuevo recurso
|
|
request: Request de FastAPI (opcional)
|
|
|
|
Returns:
|
|
AuditLog creado
|
|
"""
|
|
return await AuditService.log(
|
|
db=db,
|
|
tenant_id=tenant_id,
|
|
user_id=user_id,
|
|
action=f"{resource_type}.create",
|
|
resource_type=resource_type,
|
|
resource_id=resource_id,
|
|
new_values=new_values,
|
|
request=request
|
|
)
|
|
|
|
@staticmethod
|
|
async def log_update(
|
|
db: AsyncSession,
|
|
tenant_id: uuid.UUID,
|
|
user_id: uuid.UUID,
|
|
resource_type: str,
|
|
resource_id: uuid.UUID,
|
|
old_values: Dict[str, Any],
|
|
new_values: Dict[str, Any],
|
|
request: Optional[Request] = None
|
|
) -> AuditLog:
|
|
"""
|
|
Registra la actualización de un recurso.
|
|
|
|
Args:
|
|
db: Sesión de base de datos
|
|
tenant_id: ID del tenant
|
|
user_id: ID del usuario que actualizó
|
|
resource_type: Tipo de recurso
|
|
resource_id: ID del recurso
|
|
old_values: Valores anteriores
|
|
new_values: Valores nuevos
|
|
request: Request de FastAPI (opcional)
|
|
|
|
Returns:
|
|
AuditLog creado
|
|
"""
|
|
return await AuditService.log(
|
|
db=db,
|
|
tenant_id=tenant_id,
|
|
user_id=user_id,
|
|
action=f"{resource_type}.update",
|
|
resource_type=resource_type,
|
|
resource_id=resource_id,
|
|
old_values=old_values,
|
|
new_values=new_values,
|
|
request=request
|
|
)
|
|
|
|
@staticmethod
|
|
async def log_delete(
|
|
db: AsyncSession,
|
|
tenant_id: uuid.UUID,
|
|
user_id: uuid.UUID,
|
|
resource_type: str,
|
|
resource_id: uuid.UUID,
|
|
old_values: Dict[str, Any],
|
|
request: Optional[Request] = None
|
|
) -> AuditLog:
|
|
"""
|
|
Registra la eliminación de un recurso.
|
|
|
|
Args:
|
|
db: Sesión de base de datos
|
|
tenant_id: ID del tenant
|
|
user_id: ID del usuario que eliminó
|
|
resource_type: Tipo de recurso
|
|
resource_id: ID del recurso eliminado
|
|
old_values: Valores del recurso antes de eliminar
|
|
request: Request de FastAPI (opcional)
|
|
|
|
Returns:
|
|
AuditLog creado
|
|
"""
|
|
return await AuditService.log(
|
|
db=db,
|
|
tenant_id=tenant_id,
|
|
user_id=user_id,
|
|
action=f"{resource_type}.delete",
|
|
resource_type=resource_type,
|
|
resource_id=resource_id,
|
|
old_values=old_values,
|
|
request=request
|
|
)
|
|
|
|
@staticmethod
|
|
def sanitize_values(values: Dict[str, Any]) -> Dict[str, Any]:
|
|
"""
|
|
Sanitiza valores sensibles antes de guardarlos en audit log.
|
|
|
|
Remueve campos como passwords, tokens, etc.
|
|
|
|
Args:
|
|
values: Diccionario de valores
|
|
|
|
Returns:
|
|
Diccionario sanitizado
|
|
"""
|
|
sensitive_fields = {
|
|
'password',
|
|
'password_hash',
|
|
'totp_secret',
|
|
'backup_codes',
|
|
'token',
|
|
'access_token',
|
|
'refresh_token'
|
|
}
|
|
|
|
return {
|
|
key: '***REDACTED***' if key in sensitive_fields else value
|
|
for key, value in values.items()
|
|
}
|