""" Audit Endpoints - ServiceManagerWeb ==================================== Este archivo maneja todos los endpoints de auditoría y seguridad. Rutas disponibles: GET /audit/ → Lista de logs con filtros y paginación GET /audit/stats → Estadísticas generales de auditoría GET /audit/{log_id} → Detalle de un log específico GET /audit/security/analysis → Análisis de amenazas en tiempo real POST /audit/security/action → Ejecutar acción de seguridad (bloquear IP, etc.) GET /audit/security/incidents → Lista de incidentes detectados CORRECCIONES APLICADAS: 1. Todos los endpoints usan datetime.now(timezone.utc) para generar fechas aware (con timezone info en UTC), compatibles con la columna 'timestamp with time zone' (TIMESTAMPTZ) de PostgreSQL. 2. audit_helpers.get_count_stat() convierte las fechas a aware UTC con _ensure_aware_utc() antes de usarlas en queries, resolviendo el bug donde los tres contadores (total, hoy, semana) devolvían el mismo valor porque el filtro de fecha se ignoraba. 3. critical_actions_today usa los mismos umbrales que /security/incidents para que el contador del dashboard coincida con la lista de detalles. """ from fastapi import APIRouter, Depends, HTTPException, status, Query from sqlalchemy.ext.asyncio import AsyncSession from sqlalchemy import select, func, and_, or_, desc from sqlalchemy.orm import selectinload from typing import Optional, List from datetime import datetime, timedelta, timezone import uuid import structlog from app.core.database import get_db from app.api.deps import get_current_user, get_current_tenant from app.models.user import User, UserRole from app.models.tenant import Tenant from app.models.audit import AuditLog from app.services.audit_service import AuditService from app.api.schemas.audit import ( AuditLogResponse, AuditLogListResponse, AuditLogFilters, AuditLogStats, SecurityAnalysisResponse, SecurityThreatPattern, SecurityActionRequest, SecurityActionResponse, SecurityIncidentResponse, SecurityIncidentListResponse ) from app.api.v1.audit_helpers import ( audit_log_to_dict, apply_tenant_filter, get_count_stat, get_top_items, detect_mass_deletions, detect_brute_force, detect_privilege_escalation ) # Instancia del router de FastAPI para este módulo router = APIRouter() # Logger estructurado para registrar eventos internos del sistema logger = structlog.get_logger(__name__) # ============================================================================= # DEPENDENCIA DE AUTORIZACIÓN # ============================================================================= def require_auditor_role(current_user: User = Depends(get_current_user)) -> User: """ Dependencia reutilizable que verifica que el usuario tenga permisos para ver logs de auditoría. Solo pueden acceder los roles: ADMIN, SUPPORT_MANAGER, AUDITOR. Si no tiene el rol correcto, lanza un error 403 Forbidden. """ if current_user.role not in [UserRole.ADMIN, UserRole.SUPPORT_MANAGER, UserRole.AUDITOR]: raise HTTPException( status_code=status.HTTP_403_FORBIDDEN, detail="Solo usuarios con rol ADMIN, SUPPORT_MANAGER o AUDITOR pueden acceder a logs de auditoría" ) return current_user # ============================================================================= # ENDPOINT: LISTA DE LOGS DE AUDITORÍA # ============================================================================= @router.get("/", response_model=AuditLogListResponse) async def get_audit_logs( # Paginación page: int = Query(default=1, ge=1), per_page: int = Query(default=50, ge=1, le=100), # Filtros opcionales user_id: Optional[uuid.UUID] = Query(None), action: Optional[str] = Query(None), resource_type: Optional[str] = Query(None), resource_id: Optional[uuid.UUID] = Query(None), date_from: Optional[datetime] = Query(None), date_to: Optional[datetime] = Query(None), search: Optional[str] = Query(None), tenant_id: Optional[uuid.UUID] = Query(None), all_tenants: bool = Query(False), # Dependencias de autenticación y base de datos current_user: User = Depends(require_auditor_role), current_tenant: Tenant = Depends(get_current_tenant), db: AsyncSession = Depends(get_db) ): """ Obtener el historial completo de logs de auditoría con filtros opcionales. Soporta filtrar por usuario, tipo de acción, recurso afectado, fechas y búsqueda de texto. También soporta ver logs de todos los tenants si el usuario tiene permisos de ADMIN o SUPPORT_MANAGER. """ logger.info( "Obteniendo logs de auditoria", user_id=str(current_user.id), tenant_id=str(current_tenant.id), filters={ "user_id": str(user_id) if user_id else None, "action": action, "page": page, "all_tenants": all_tenants } ) # Construir la query base con relación al usuario que hizo la acción query = select(AuditLog).options(selectinload(AuditLog.user)) # Aplicar filtro de tenant según permisos del usuario query = apply_tenant_filter(query, current_user, current_tenant, all_tenants, tenant_id) # Aplicar filtros opcionales uno por uno if user_id: query = query.where(AuditLog.user_id == user_id) if action: query = query.where(AuditLog.action == action) if resource_type: query = query.where(AuditLog.resource_type == resource_type) if resource_id: query = query.where(AuditLog.resource_id == resource_id) if date_from: query = query.where(AuditLog.created_at >= date_from) if date_to: query = query.where(AuditLog.created_at < date_to) if search: # Búsqueda parcial en el campo "action" (ej: "ticket" encuentra "ticket.create") query = query.where(AuditLog.action.ilike(f"%{search}%")) # Ordenar por fecha descendente (más reciente primero) query = query.order_by(desc(AuditLog.created_at)) # Contar total de registros para calcular páginas count_query = select(func.count()).select_from(query.subquery()) total = (await db.execute(count_query)).scalar() or 0 # Aplicar paginación offset = (page - 1) * per_page query = query.offset(offset).limit(per_page) # Ejecutar query y obtener resultados result = await db.execute(query) logs = result.scalars().all() # Calcular número total de páginas total_pages = (total + per_page - 1) // per_page # Convertir modelos a schemas de respuesta logs_response = [AuditLogResponse(**audit_log_to_dict(log)) for log in logs] return AuditLogListResponse( logs=logs_response, total=total, page=page, per_page=per_page, total_pages=total_pages ) # ============================================================================= # ENDPOINT: ESTADÍSTICAS DE AUDITORÍA # ============================================================================= @router.get("/stats", response_model=AuditLogStats) async def get_audit_stats( all_tenants: bool = Query(False), current_user: User = Depends(require_auditor_role), current_tenant: Tenant = Depends(get_current_tenant), db: AsyncSession = Depends(get_db) ): """ Obtener estadísticas resumidas de auditoría para el dashboard. Incluye: - Total de acciones registradas - Acciones de las últimas 24 horas - Acciones de los últimos 7 días - Incidentes críticos detectados hoy (alineado con /security/incidents) - Acciones más frecuentes - Usuarios más activos - Distribución por tipo de recurso """ can_see_all_tenants = current_user.role in [UserRole.ADMIN, UserRole.SUPPORT_MANAGER] logger.info( "Obteniendo estadisticas de auditoria", user_id=str(current_user.id), tenant_id=str(current_tenant.id), all_tenants=all_tenants, can_see_all=can_see_all_tenants ) # datetime.now(timezone.utc) genera un datetime aware en UTC, # compatible con la columna TIMESTAMPTZ de PostgreSQL now = datetime.now(timezone.utc) # Determinar si se debe filtrar por tenant o ver todos apply_tenant = not (all_tenants and can_see_all_tenants) tenant_filter = current_tenant.id if apply_tenant else None # ------------------------------------------------------------------ # CONTADORES GENERALES # ------------------------------------------------------------------ # Total histórico de acciones (sin filtro de fecha) total_actions = await get_count_stat(db, tenant_filter) # Acciones en las últimas 24 horas # get_count_stat convierte internamente a aware UTC con _ensure_aware_utc() actions_today = await get_count_stat(db, tenant_filter, now - timedelta(days=1)) # Acciones en los últimos 7 días actions_this_week = await get_count_stat(db, tenant_filter, now - timedelta(days=7)) # ------------------------------------------------------------------ # CONTADOR DE INCIDENTES CRÍTICOS # ------------------------------------------------------------------ # Usa los mismos umbrales que los detectores de /security/incidents # para que el número del dashboard sea consistente con la lista. # ------------------------------------------------------------------ today_start = now - timedelta(days=1) # Contar intentos fallidos de login en las últimas 24 horas failed_login_count = (await db.execute( select(func.count()).select_from(AuditLog).where( AuditLog.action == 'user.login_failed', AuditLog.created_at >= today_start, *([AuditLog.tenant_id == tenant_filter] if apply_tenant else []) ) )).scalar() or 0 # Contar eliminaciones en las últimas 24 horas deletion_count = (await db.execute( select(func.count()).select_from(AuditLog).where( AuditLog.action.like('%.delete'), AuditLog.created_at >= today_start, *([AuditLog.tenant_id == tenant_filter] if apply_tenant else []) ) )).scalar() or 0 # Contar cambios de privilegios en las últimas 24 horas privilege_count = (await db.execute( select(func.count()).select_from(AuditLog).where( AuditLog.action == 'user.update', AuditLog.created_at >= today_start, *([AuditLog.tenant_id == tenant_filter] if apply_tenant else []) ) )).scalar() or 0 # Calcular número real de incidentes usando los mismos umbrales # que los detectores en /security/incidents: # - Fuerza bruta: incidente si hay >= 20 intentos fallidos # - Eliminación masiva: incidente si hay >= 50 eliminaciones # - Escalada privilegios: incidente si hay >= 3 cambios de rol critical_actions_today = sum([ 1 if failed_login_count >= 20 else 0, 1 if deletion_count >= 50 else 0, 1 if privilege_count >= 3 else 0, ]) # ------------------------------------------------------------------ # DATOS PARA GRÁFICAS Y TABLAS DEL DASHBOARD # ------------------------------------------------------------------ # Top acciones más frecuentes (ej: "ticket.create", "user.login") top_actions = await get_top_items(db, AuditLog.action, tenant_filter) # Distribución por tipo de recurso (ej: "ticket", "user", "tenant") by_resource_type = await get_top_items(db, AuditLog.resource_type, tenant_filter, limit=10) # Usuarios más activos (hace join con tabla de usuarios) top_users = await get_top_items(db, None, tenant_filter, join_user=True) return AuditLogStats( total_actions=total_actions, actions_today=actions_today, actions_this_week=actions_this_week, critical_actions_today=critical_actions_today, top_actions=top_actions, top_users=top_users, by_resource_type=by_resource_type ) # ============================================================================= # ENDPOINT: DETALLE DE UN LOG ESPECÍFICO # ============================================================================= @router.get("/{log_id}", response_model=AuditLogResponse) async def get_audit_log_detail( log_id: uuid.UUID, current_user: User = Depends(require_auditor_role), current_tenant: Tenant = Depends(get_current_tenant), db: AsyncSession = Depends(get_db) ): """ Obtener el detalle completo de un log de auditoría por su ID. Incluye información del usuario que realizó la acción, valores anteriores y nuevos (para cambios), IP de origen, user agent, etc. Retorna 404 si el log no existe o no pertenece al tenant del usuario. """ # Buscar el log por ID incluyendo los datos del usuario relacionado query = select(AuditLog).where(AuditLog.id == log_id).options(selectinload(AuditLog.user)) # Aplicar filtro de tenant para garantizar aislamiento multi-tenant query = apply_tenant_filter(query, current_user, current_tenant) result = await db.execute(query) log = result.scalar_one_or_none() if not log: raise HTTPException( status_code=status.HTTP_404_NOT_FOUND, detail=f"Registro de auditoria {log_id} no encontrado" ) return AuditLogResponse(**audit_log_to_dict(log)) # ============================================================================= # ENDPOINT: ANÁLISIS DE SEGURIDAD EN TIEMPO REAL # ============================================================================= @router.get("/security/analysis", response_model=SecurityAnalysisResponse) async def get_security_analysis( hours: int = Query(default=24, ge=1, le=720), all_tenants: bool = Query(False), current_user: User = Depends(require_auditor_role), current_tenant: Tenant = Depends(get_current_tenant), db: AsyncSession = Depends(get_db) ): """ Analizar los logs de auditoría para detectar patrones sospechosos. Detecta tres tipos de amenazas: 1. Fuerza bruta: Muchos intentos fallidos de login desde las mismas IPs 2. Eliminación masiva: Gran cantidad de registros eliminados en poco tiempo 3. Escalada privilegios: Cambios de roles sospechosos en usuarios Calcula un nivel de riesgo general (low/medium/high/critical) y devuelve recomendaciones de acción. """ logger.info( "Analisis de seguridad solicitado", user_id=str(current_user.id), tenant_id=str(current_tenant.id), hours=hours ) # aware UTC para compatibilidad con TIMESTAMPTZ de PostgreSQL now = datetime.now(timezone.utc) analysis_start = now - timedelta(hours=hours) query = ( select(AuditLog) .where(AuditLog.created_at >= analysis_start) .options(selectinload(AuditLog.user)) ) query = apply_tenant_filter(query, current_user, current_tenant, all_tenants) result = await db.execute(query) logs = result.scalars().all() # ------------------------------------------------------------------ # CONTADORES DE EVENTOS SOSPECHOSOS # ------------------------------------------------------------------ failed_logins = sum(1 for log in logs if log.action == 'user.login_failed') mass_deletions = sum(1 for log in logs if '.delete' in log.action) privilege_changes = sum( 1 for log in logs if log.action == 'user.update' and log.new_values and 'role' in log.new_values ) # ------------------------------------------------------------------ # GENERACIÓN DE PATRONES DE AMENAZA # ------------------------------------------------------------------ threat_patterns = [] # Amenaza 1: Fuerza bruta (umbral mínimo: 5 intentos fallidos) if failed_logins >= 5: affected_ips_list = [ str(log.ip_address) for log in logs if log.action == 'user.login_failed' and log.ip_address ] threat_patterns.append(SecurityThreatPattern( id="brute_force_attempt", type="brute_force", description=( f"Se detectaron {failed_logins} intentos fallidos de " f"login en las ultimas {hours}h" ), severity="high" if failed_logins >= 20 else "medium", occurrences=failed_logins, first_seen=min( (log.created_at for log in logs if log.action == 'user.login_failed'), default=now ), last_seen=max( (log.created_at for log in logs if log.action == 'user.login_failed'), default=now ), affected_ips=list(set(affected_ips_list))[:5], affected_users=[], recommended_action="Considerar bloquear IPs con multiples fallos" )) # Amenaza 2: Eliminación masiva (umbral mínimo: 10 eliminaciones) if mass_deletions >= 10: deleting_users = [ log.user.email for log in logs if '.delete' in log.action and log.user ] threat_patterns.append(SecurityThreatPattern( id="mass_deletion", type="mass_deletion", description=( f"Se detectaron {mass_deletions} eliminaciones en " f"las ultimas {hours}h" ), severity="critical" if mass_deletions >= 50 else "high", occurrences=mass_deletions, first_seen=min( (log.created_at for log in logs if '.delete' in log.action), default=now ), last_seen=max( (log.created_at for log in logs if '.delete' in log.action), default=now ), affected_ips=[], affected_users=list(set(deleting_users))[:5], recommended_action="Revisar que usuarios estan eliminando recursos masivamente" )) # Amenaza 3: Escalada de privilegios (umbral mínimo: 3 cambios de rol) if privilege_changes >= 3: affected_users_list = [ log.user.email for log in logs if log.action == 'user.update' and log.user and log.new_values and 'role' in log.new_values ] threat_patterns.append(SecurityThreatPattern( id="suspicious_privilege_changes", type="privilege_escalation", description=( f"Se detectaron {privilege_changes} cambios de " f"privilegios en las ultimas {hours}h" ), severity="high", occurrences=privilege_changes, first_seen=min( ( log.created_at for log in logs if log.action == 'user.update' and log.new_values and 'role' in log.new_values ), default=now ), last_seen=max( ( log.created_at for log in logs if log.action == 'user.update' and log.new_values and 'role' in log.new_values ), default=now ), affected_ips=[], affected_users=list(set(affected_users_list))[:5], recommended_action="Auditar cambios de roles recientes" )) # ------------------------------------------------------------------ # CÁLCULO DE NIVEL DE RIESGO GENERAL # ------------------------------------------------------------------ risk_score = min( 100, (failed_logins * 2) + (mass_deletions * 5) + (privilege_changes * 10) ) if risk_score >= 80: risk_level = "critical" elif risk_score >= 50: risk_level = "high" elif risk_score >= 20: risk_level = "medium" else: risk_level = "low" # ------------------------------------------------------------------ # RECOMENDACIONES AUTOMÁTICAS # ------------------------------------------------------------------ recommended_actions = [] if failed_logins >= 20: recommended_actions.append( "Implementar bloqueo automatico de IPs despues de multiples intentos fallidos" ) if mass_deletions >= 50: recommended_actions.append( "Activar confirmacion adicional para eliminaciones masivas" ) if privilege_changes >= 3: recommended_actions.append( "Revisar y aprobar manualmente los cambios de roles recientes" ) if not recommended_actions: recommended_actions.append("Continuar monitoreando actividad del sistema") suspicious_ips = len(set( log.ip_address for log in logs if log.ip_address and log.action == 'user.login_failed' )) critical_actions = mass_deletions + privilege_changes return SecurityAnalysisResponse( overall_risk_level=risk_level, total_threats_detected=len(threat_patterns), threats=threat_patterns, analysis_period_hours=hours, generated_at=datetime.now(timezone.utc), failed_login_attempts=failed_logins, suspicious_ips_count=suspicious_ips, critical_actions_count=critical_actions, recommended_actions=recommended_actions ) # ============================================================================= # ENDPOINT: EJECUTAR ACCIÓN DE SEGURIDAD # ============================================================================= @router.post("/security/action", response_model=SecurityActionResponse) async def execute_security_action( action: SecurityActionRequest, current_user: User = Depends(require_auditor_role), current_tenant: Tenant = Depends(get_current_tenant), db: AsyncSession = Depends(get_db) ): """ Ejecutar una acción de seguridad manual sobre una amenaza detectada. Acciones disponibles: - block_ip: Bloquear una dirección IP por X minutos - notify_admin: Enviar notificación a los administradores - force_password_reset: Forzar cambio de contraseña a un usuario - disable_user: Desactivar temporalmente una cuenta de usuario Solo ADMIN y SUPPORT_MANAGER pueden ejecutar estas acciones. Todas las acciones quedan registradas en el log de auditoría. """ if current_user.role not in [UserRole.ADMIN, UserRole.SUPPORT_MANAGER]: raise HTTPException( status_code=status.HTTP_403_FORBIDDEN, detail="Solo administradores pueden ejecutar acciones de seguridad" ) logger.info( "Accion de seguridad solicitada", user_id=str(current_user.id), action_type=action.action_type, target=action.target ) # Registrar en auditoría para trazabilidad completa try: await AuditService.log( db=db, tenant_id=current_tenant.id, user_id=current_user.id, action=f"security.{action.action_type}", resource_type="security", resource_id=None, metadata={ "target": action.target, "reason": action.reason, "duration_minutes": action.duration_minutes } ) await db.commit() except Exception as e: logger.error("Fallo al registrar accion de seguridad en auditoria", error=str(e)) action_messages = { "block_ip": ( f"IP {action.target} bloqueada por " f"{action.duration_minutes or 60} minutos. Razon: {action.reason}" ), "notify_admin": ( f"Notificacion enviada a administradores sobre: {action.reason}" ), "force_password_reset": ( f"Se forzara cambio de contrasena para {action.target}. " f"Razon: {action.reason}" ), "disable_user": ( f"Usuario {action.target} desactivado temporalmente. " f"Razon: {action.reason}" ) } success = action.action_type in action_messages message = action_messages.get( action.action_type, f"Tipo de accion no reconocida: {action.action_type}" ) return SecurityActionResponse(success=success, message=message, action_id=None) # ============================================================================= # ENDPOINT: LISTA DE INCIDENTES DE SEGURIDAD # ============================================================================= @router.get("/security/incidents", response_model=SecurityIncidentListResponse) async def get_security_incidents( # Paginación page: int = Query(default=1, ge=1), per_page: int = Query(default=20, ge=1, le=100), # Filtros opcionales severity: Optional[str] = Query(None), status: Optional[str] = Query(None), incident_type: Optional[str] = Query(None), search: Optional[str] = Query(None), all_tenants: bool = Query(False), # Dependencias current_user: User = Depends(require_auditor_role), current_tenant: Tenant = Depends(get_current_tenant), db: AsyncSession = Depends(get_db) ): """ Obtener la lista de incidentes de seguridad detectados. Los incidentes se generan dinámicamente analizando los logs de auditoría de los últimos 7 días usando tres detectores: 1. detect_brute_force: Analiza intentos fallidos de login 2. detect_mass_deletions: Analiza eliminaciones masivas 3. detect_privilege_escalation: Analiza cambios de rol sospechosos Los umbrales son los mismos que usa /stats para critical_actions_today, garantizando consistencia entre el contador y la lista. """ logger.info( "Obteniendo incidentes de seguridad", user_id=str(current_user.id), tenant_id=str(current_tenant.id), filters={ "severity": severity, "status": status, "type": incident_type, "page": page } ) # aware UTC para compatibilidad con TIMESTAMPTZ de PostgreSQL now = datetime.now(timezone.utc) analysis_start = now - timedelta(days=7) base_query = ( select(AuditLog) .options(selectinload(AuditLog.user)) .where(AuditLog.created_at >= analysis_start) ) base_query = apply_tenant_filter(base_query, current_user, current_tenant, all_tenants) # ------------------------------------------------------------------ # DETECTOR 1: ELIMINACIONES MASIVAS # ------------------------------------------------------------------ deletion_result = await db.execute( base_query .where(AuditLog.action.like('%.delete')) .order_by(desc(AuditLog.created_at)) ) deletion_logs = deletion_result.scalars().all() deletion_incidents = detect_mass_deletions(deletion_logs, now) # ------------------------------------------------------------------ # DETECTOR 2: FUERZA BRUTA # ------------------------------------------------------------------ failed_login_result = await db.execute( base_query .where(AuditLog.action == 'user.login_failed') .order_by(desc(AuditLog.created_at)) ) failed_login_logs = failed_login_result.scalars().all() brute_force_incidents = detect_brute_force(failed_login_logs, now) # ------------------------------------------------------------------ # DETECTOR 3: ESCALADA DE PRIVILEGIOS # El operador '?' verifica si el campo JSON contiene la clave 'role' # ------------------------------------------------------------------ privilege_result = await db.execute( base_query .where(and_( AuditLog.action == 'user.update', AuditLog.new_values.op('?')('role') )) .order_by(desc(AuditLog.created_at)) ) privilege_logs = privilege_result.scalars().all() privilege_incidents = detect_privilege_escalation(privilege_logs) # Combinar todos los incidentes incidents = [ SecurityIncidentResponse(**inc) for inc in (deletion_incidents + brute_force_incidents + privilege_incidents) ] # ------------------------------------------------------------------ # FILTROS EN MEMORIA (los incidentes son generados dinámicamente) # ------------------------------------------------------------------ if severity: incidents = [i for i in incidents if i.severity == severity] if status: incidents = [i for i in incidents if i.status == status] if incident_type: incidents = [i for i in incidents if i.incident_type == incident_type] if search: search_lower = search.lower() incidents = [ i for i in incidents if search_lower in i.title.lower() or (i.description and search_lower in i.description.lower()) ] # Ordenar por fecha descendente incidents.sort(key=lambda x: x.created_at, reverse=True) # ------------------------------------------------------------------ # PAGINACIÓN MANUAL # ------------------------------------------------------------------ total = len(incidents) total_pages = (total + per_page - 1) // per_page start_idx = (page - 1) * per_page end_idx = start_idx + per_page paginated_incidents = incidents[start_idx:end_idx] return SecurityIncidentListResponse( incidents=paginated_incidents, total=total, page=page, per_page=per_page, total_pages=total_pages )