""" Audit Endpoints - ServiceManagerWeb Endpoints para consulta de logs de auditor├¡a. Solo accesible por roles: ADMIN, SUPPORT_MANAGER, AUDITOR """ 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 ) router = APIRouter() logger = structlog.get_logger(__name__) def require_auditor_role(current_user: User = Depends(get_current_user)) -> User: """ Dependency que verifica que el usuario tenga rol de auditor. Solo ADMIN, SUPPORT_MANAGER y AUDITOR pueden ver logs de auditor├¡a. """ allowed_roles = [UserRole.ADMIN, UserRole.SUPPORT_MANAGER, UserRole.AUDITOR] if current_user.role not in allowed_roles: 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 @router.get("/", response_model=AuditLogListResponse) async def get_audit_logs( # Paginaci├│n page: int = Query(default=1, ge=1, description="N├║mero de p├ígina"), per_page: int = Query(default=50, ge=1, le=100, description="Registros por p├ígina"), # Filtros user_id: Optional[uuid.UUID] = Query(None, description="Filtrar por usuario"), action: Optional[str] = Query(None, description="Filtrar por acci├│n"), resource_type: Optional[str] = Query(None, description="Filtrar por tipo de recurso"), resource_id: Optional[uuid.UUID] = Query(None, description="Filtrar por ID de recurso"), date_from: Optional[datetime] = Query(None, description="Fecha desde"), date_to: Optional[datetime] = Query(None, description="Fecha hasta"), search: Optional[str] = Query(None, description="B├║squeda en acci├│n o email"), # Multi-tenant filters (solo ADMIN/SUPPORT_MANAGER) tenant_id: Optional[uuid.UUID] = Query(None, description="Ver logs de un tenant específico"), all_tenants: bool = Query(False, description="Ver logs de todos los tenants"), # Dependencies current_user: User = Depends(require_auditor_role), current_tenant: Tenant = Depends(get_current_tenant), db: AsyncSession = Depends(get_db) ): """ Obtener logs de auditor├¡a con filtros y paginaci├│n. **Permisos**: ADMIN, SUPPORT_MANAGER, AUDITOR **Filtros disponibles**: - `user_id`: Acciones de un usuario espec├¡fico - `action`: Tipo de acci├│n (ej: "ticket.create") - `resource_type`: Tipo de recurso (ej: "ticket") - `resource_id`: ID de recurso espec├¡fico - `date_from`, `date_to`: Rango de fechas - `search`: B├║squeda en acciones **Retorna**: Lista paginada de audit logs """ logger.info( "Fetching audit logs", 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, "resource_type": resource_type, "page": page, "tenant_filter": str(tenant_id) if tenant_id else None, "all_tenants": all_tenants } ) # Determinar el filtro de tenant # Solo ADMIN y SUPPORT_MANAGER pueden ver otros tenants o todos los tenants can_see_all_tenants = current_user.role in [UserRole.ADMIN, UserRole.SUPPORT_MANAGER] # Query base con filtro de tenant dinámico query = select(AuditLog).options(selectinload(AuditLog.user)) if all_tenants and can_see_all_tenants: # Ver todos los tenants (no agregar filtro de tenant) pass elif tenant_id and can_see_all_tenants: # Ver un tenant específico query = query.where(AuditLog.tenant_id == tenant_id) else: # Ver solo el tenant actual (comportamiento default) query = query.where(AuditLog.tenant_id == current_tenant.id) # Aplicar filtros 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: # El frontend ya envía el timestamp correcto query = query.where(AuditLog.created_at < date_to) if search: # B├║squeda en action search_filter = AuditLog.action.ilike(f"%{search}%") query = query.where(search_filter) # Ordenar por fecha descendente (m├ís recientes primero) query = query.order_by(desc(AuditLog.created_at)) # Contar total antes de paginar count_query = select(func.count()).select_from(query.subquery()) total_result = await db.execute(count_query) total = total_result.scalar() or 0 # Aplicar paginaci├│n offset = (page - 1) * per_page query = query.offset(offset).limit(per_page) # Ejecutar query result = await db.execute(query) logs = result.scalars().all() # Calcular total de p├íginas total_pages = (total + per_page - 1) // per_page # Convertir a response schema (agregar info del usuario) logs_response = [] for log in logs: log_dict = { "id": log.id, "tenant_id": log.tenant_id, "user_id": log.user_id, "action": log.action, "resource_type": log.resource_type, "resource_id": log.resource_id, "ip_address": str(log.ip_address) if log.ip_address else None, "user_agent": log.user_agent, "correlation_id": log.correlation_id, "old_values": log.old_values, "new_values": log.new_values, "metadata": log.extra_metadata, "created_at": log.created_at, "action_display": log.action_display, "user_email": None, "user_name": None } # Agregar info del usuario si existe if log.user: log_dict["user_email"] = log.user.email log_dict["user_name"] = log.user.full_name log_dict["user_role"] = log.user.role.value if hasattr(log.user.role, 'value') else str(log.user.role) logs_response.append(AuditLogResponse(**log_dict)) return AuditLogListResponse( logs=logs_response, total=total, page=page, per_page=per_page, total_pages=total_pages ) @router.get("/stats", response_model=AuditLogStats) async def get_audit_stats( all_tenants: bool = Query(False, description="Ver stats de todos los tenants"), current_user: User = Depends(require_auditor_role), current_tenant: Tenant = Depends(get_current_tenant), db: AsyncSession = Depends(get_db) ): """ Obtener estadísticas de auditoría del tenant (o todos los tenants si es ADMIN). **Permisos**: ADMIN, SUPPORT_MANAGER, AUDITOR **Retorna**: Estadísticas de actividad """ can_see_all_tenants = current_user.role in [UserRole.ADMIN, UserRole.SUPPORT_MANAGER] logger.info( "Fetching audit stats", user_id=str(current_user.id), tenant_id=str(current_tenant.id), all_tenants=all_tenants, can_see_all=can_see_all_tenants ) now = datetime.now(timezone.utc) # Determinar si aplicar filtro de tenant apply_tenant_filter = not (all_tenants and can_see_all_tenants) # Total de acciones total_query = select(func.count()).select_from(AuditLog) if apply_tenant_filter: total_query = total_query.where(AuditLog.tenant_id == current_tenant.id) total_result = await db.execute(total_query) total_actions = total_result.scalar() or 0 # Acciones hoy (├║ltimas 24 horas) today_start = now - timedelta(days=1) today_query = select(func.count()).select_from(AuditLog).where( AuditLog.created_at >= today_start ) if apply_tenant_filter: today_query = today_query.where(AuditLog.tenant_id == current_tenant.id) today_result = await db.execute(today_query) actions_today = today_result.scalar() or 0 # Acciones esta semana (├║ltimos 7 d├¡as) week_start = now - timedelta(days=7) week_query = select(func.count()).select_from(AuditLog).where( AuditLog.created_at >= week_start ) if apply_tenant_filter: week_query = week_query.where(AuditLog.tenant_id == current_tenant.id) week_result = await db.execute(week_query) actions_this_week = week_result.scalar() or 0 # Top 5 acciones m├ís frecuentes top_actions_query = select( AuditLog.action, func.count(AuditLog.id).label('count') ) if apply_tenant_filter: top_actions_query = top_actions_query.where(AuditLog.tenant_id == current_tenant.id) top_actions_query = top_actions_query.group_by( AuditLog.action ).order_by( desc('count') ).limit(5) top_actions_result = await db.execute(top_actions_query) top_actions = {row.action: row.count for row in top_actions_result} # Acciones por tipo de recurso by_resource_query = select( AuditLog.resource_type, func.count(AuditLog.id).label('count') ) if apply_tenant_filter: by_resource_query = by_resource_query.where(AuditLog.tenant_id == current_tenant.id) by_resource_query = by_resource_query.group_by( AuditLog.resource_type ).order_by( desc('count') ) by_resource_result = await db.execute(by_resource_query) by_resource_type = {row.resource_type: row.count for row in by_resource_result} # Top usuarios (con join a users para obtener nombres) top_users_query = select( User.email, func.count(AuditLog.id).label('count') ).join( User, AuditLog.user_id == User.id ) if apply_tenant_filter: top_users_query = top_users_query.where(AuditLog.tenant_id == current_tenant.id) top_users_query = top_users_query.group_by( User.email ).order_by( desc('count') ).limit(5) top_users_result = await db.execute(top_users_query) top_users = {row.email: row.count for row in top_users_result} # Acciones cr├¡ticas hoy (delete, update sensibles, etc.) critical_conditions = [ AuditLog.created_at >= today_start, or_( AuditLog.action.like('%.delete'), AuditLog.action.like('user.update'), AuditLog.action.like('%.assign'), AuditLog.action.in_(['user.login_failed', 'user.logout']) ) ] if apply_tenant_filter: critical_conditions.append(AuditLog.tenant_id == current_tenant.id) critical_actions_query = select(func.count()).select_from(AuditLog).where( and_(*critical_conditions) ) critical_result = await db.execute(critical_actions_query) critical_actions_today = critical_result.scalar() or 0 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 ) @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 detalle de un audit log espec├¡fico. **Permisos**: ADMIN, SUPPORT_MANAGER, AUDITOR **Retorna**: Detalle completo del audit log """ # Buscar el log query = select(AuditLog).where( and_( AuditLog.id == log_id, AuditLog.tenant_id == current_tenant.id ) ) 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"Audit log {log_id} no encontrado" ) # Convertir a response log_dict = { "id": log.id, "tenant_id": log.tenant_id, "user_id": log.user_id, "action": log.action, "resource_type": log.resource_type, "resource_id": log.resource_id, "ip_address": str(log.ip_address) if log.ip_address else None, "user_agent": log.user_agent, "correlation_id": log.correlation_id, "old_values": log.old_values, "new_values": log.new_values, "metadata": log.extra_metadata, "created_at": log.created_at, "action_display": log.action_display, "user_email": None, "user_name": None } if log.user: log_dict["user_email"] = log.user.email log_dict["user_name"] = log.user.full_name return AuditLogResponse(**log_dict) # =================================== # SECURITY ANALYSIS ENDPOINTS # =================================== @router.get("/security/analysis", response_model=SecurityAnalysisResponse) async def get_security_analysis( hours: int = Query(default=24, ge=1, le=168, description="Período de análisis en horas"), current_user: User = Depends(require_auditor_role), current_tenant: Tenant = Depends(get_current_tenant), db: AsyncSession = Depends(get_db) ): """ Análisis de seguridad y detección de amenazas. **Permisos**: ADMIN, SUPPORT_MANAGER, AUDITOR **Detecta**: - Intentos de fuerza bruta (login_failed) - Escalada de privilegios - Eliminaciones masivas - Accesos desde IPs sospechosas - Patrones anómalos de actividad **Retorna**: Análisis completo con amenazas y recomendaciones """ logger.info( "Security analysis requested", user_id=str(current_user.id), tenant_id=str(current_tenant.id), hours=hours ) now = datetime.now(timezone.utc) analysis_start = now - timedelta(hours=hours) threats = [] failed_login_attempts = 0 suspicious_ips = set() critical_actions_count = 0 # 1. DETECCIÓN DE FUERZA BRUTA brute_force_query = select( AuditLog.ip_address, func.count(AuditLog.id).label('attempts'), func.min(AuditLog.created_at).label('first_seen'), func.max(AuditLog.created_at).label('last_seen') ).where( and_( AuditLog.tenant_id == current_tenant.id, AuditLog.action == 'user.login_failed', AuditLog.created_at >= analysis_start ) ).group_by(AuditLog.ip_address).having(func.count(AuditLog.id) >= 5) brute_force_result = await db.execute(brute_force_query) brute_force_ips = brute_force_result.all() for ip_data in brute_force_ips: if ip_data.ip_address: suspicious_ips.add(str(ip_data.ip_address)) failed_login_attempts += ip_data.attempts severity = "high" if ip_data.attempts > 20 else "medium" if ip_data.attempts > 10 else "low" threats.append(SecurityThreatPattern( type="brute_force_attack", severity=severity, description=f"Ataque de fuerza bruta detectado desde {ip_data.ip_address}", occurrences=ip_data.attempts, affected_ips=[str(ip_data.ip_address)], affected_users=[], first_seen=ip_data.first_seen, last_seen=ip_data.last_seen, recommendations=[ f"Bloquear IP {ip_data.ip_address} temporalmente", "Revisar logs de firewall", "Considerar implementar CAPTCHA", "Notificar al equipo de seguridad" ] )) # 2. ESCALADA DE PRIVILEGIOS privilege_query = select( User.email, func.count(AuditLog.id).label('changes'), func.min(AuditLog.created_at).label('first_seen'), func.max(AuditLog.created_at).label('last_seen') ).join( User, AuditLog.user_id == User.id ).where( and_( AuditLog.tenant_id == current_tenant.id, AuditLog.action == 'user.update', AuditLog.created_at >= analysis_start, AuditLog.new_values.op('?')('role') ) ).group_by(User.email).having(func.count(AuditLog.id) >= 3) privilege_result = await db.execute(privilege_query) privilege_changes = privilege_result.all() for priv_data in privilege_changes: threats.append(SecurityThreatPattern( type="privilege_escalation", severity="critical", description=f"Posible escalada de privilegios - {priv_data.email} ha modificado roles {priv_data.changes} veces", occurrences=priv_data.changes, affected_ips=[], affected_users=[priv_data.email], first_seen=priv_data.first_seen, last_seen=priv_data.last_seen, recommendations=[ f"Revisar permisos del usuario {priv_data.email}", "Auditar todos los cambios de roles realizados", "Verificar si los cambios fueron autorizados", "Considerar revertir cambios no autorizados" ] )) # 3. ELIMINACIONES MASIVAS deletion_query = select( User.email, func.count(AuditLog.id).label('deletions'), func.min(AuditLog.created_at).label('first_seen'), func.max(AuditLog.created_at).label('last_seen') ).join( User, AuditLog.user_id == User.id ).where( and_( AuditLog.tenant_id == current_tenant.id, AuditLog.action.like('%.delete'), AuditLog.created_at >= analysis_start ) ).group_by(User.email).having(func.count(AuditLog.id) >= 10) deletion_result = await db.execute(deletion_query) mass_deletions = deletion_result.all() for del_data in mass_deletions: critical_actions_count += del_data.deletions threats.append(SecurityThreatPattern( type="mass_deletion", severity="high", description=f"Eliminaciones masivas detectadas - {del_data.email} ha eliminado {del_data.deletions} recursos", occurrences=del_data.deletions, affected_ips=[], affected_users=[del_data.email], first_seen=del_data.first_seen, last_seen=del_data.last_seen, recommendations=[ f"Verificar urgentemente las eliminaciones de {del_data.email}", "Comprobar si hay backups disponibles", "Contactar al usuario para verificar la acción", "Revisar sistema de permisos" ] )) # 4. ACCESOS DESDE MÚLTIPLES IPS (Cuenta comprometida) multi_ip_query = select( User.email, func.count(func.distinct(AuditLog.ip_address)).label('ip_count'), func.min(AuditLog.created_at).label('first_seen'), func.max(AuditLog.created_at).label('last_seen') ).join( User, AuditLog.user_id == User.id ).where( and_( AuditLog.tenant_id == current_tenant.id, AuditLog.action.in_(['user.login', 'user.logout']), AuditLog.created_at >= analysis_start ) ).group_by(User.email).having(func.count(func.distinct(AuditLog.ip_address)) >= 5) multi_ip_result = await db.execute(multi_ip_query) multi_ip_users = multi_ip_result.all() for ip_data in multi_ip_users: threats.append(SecurityThreatPattern( type="account_compromise", severity="medium", description=f"Posible cuenta comprometida - {ip_data.email} accedió desde {ip_data.ip_count} IPs diferentes", occurrences=ip_data.ip_count, affected_ips=[], affected_users=[ip_data.email], first_seen=ip_data.first_seen, last_seen=ip_data.last_seen, recommendations=[ f"Contactar a {ip_data.email} para verificar actividad", "Forzar cambio de contraseña", "Revisar ubicaciones de acceso", "Considerar habilitar 2FA obligatorio" ] )) # Calcular nivel de riesgo general critical_count = sum(1 for t in threats if t.severity == "critical") high_count = sum(1 for t in threats if t.severity == "high") medium_count = sum(1 for t in threats if t.severity == "medium") if critical_count > 0: overall_risk = "critical" elif high_count >= 3: overall_risk = "high" elif high_count > 0 or medium_count >= 3: overall_risk = "medium" elif medium_count > 0 or len(threats) > 0: overall_risk = "low" else: overall_risk = "safe" # Recomendaciones generales recommended_actions = [] if failed_login_attempts > 20: recommended_actions.append("Implementar límite de intentos de login por IP") if len(suspicious_ips) > 0: recommended_actions.append(f"Bloquear {len(suspicious_ips)} IPs sospechosas identificadas") if critical_actions_count > 50: recommended_actions.append("Revisar políticas de permisos - demasiadas acciones críticas") if len(threats) == 0: recommended_actions.append("Sistema seguro - continuar monitoreando") return SecurityAnalysisResponse( overall_risk_level=overall_risk, total_threats_detected=len(threats), threats=threats, analysis_period_hours=hours, generated_at=now, failed_login_attempts=failed_login_attempts, suspicious_ips_count=len(suspicious_ips), critical_actions_count=critical_actions_count, recommended_actions=recommended_actions ) @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 acción de seguridad. **Permisos**: ADMIN, SUPPORT_MANAGER (solo ellos pueden ejecutar acciones) **Acciones disponibles**: - `block_ip`: Bloquear IP temporalmente - `notify_admin`: Notificar administradores - `force_password_reset`: Forzar cambio de contraseña - `disable_user`: Desactivar usuario temporalmente **Retorna**: Resultado de la acción """ # Verificar que solo ADMIN y SUPPORT_MANAGER puedan ejecutar acciones 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( "Security action requested", user_id=str(current_user.id), action_type=action.action_type, target=action.target ) # Registrar la acción en auditoría 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("Failed to log security action", error=str(e)) # Por ahora, simular la ejecución (en producción conectar con firewall, email, etc.) message = "" success = True if action.action_type == "block_ip": message = f"IP {action.target} bloqueada por {action.duration_minutes or 60} minutos. Razón: {action.reason}" # TODO: Integrar con firewall/WAF elif action.action_type == "notify_admin": message = f"Notificación enviada a administradores sobre: {action.reason}" # TODO: Enviar email/Slack notification elif action.action_type == "force_password_reset": message = f"Se forzará cambio de contraseña para {action.target}. Razón: {action.reason}" # TODO: Marcar usuario para reset password elif action.action_type == "disable_user": message = f"Usuario {action.target} desactivado temporalmente. Razón: {action.reason}" # TODO: Desactivar usuario en BD else: success = False message = f"Tipo de acción no reconocida: {action.action_type}" return SecurityActionResponse( success=success, message=message, action_id=None # TODO: Retornar ID del audit log creado ) @router.get("/security/incidents", response_model=SecurityIncidentListResponse) async def get_security_incidents( # Paginación page: int = Query(default=1, ge=1, description="Número de página"), per_page: int = Query(default=20, ge=1, le=100, description="Incidentes por página"), # Filtros severity: Optional[str] = Query(None, description="Filtrar por severidad"), status: Optional[str] = Query(None, description="Filtrar por estado"), incident_type: Optional[str] = Query(None, description="Filtrar por tipo"), search: Optional[str] = Query(None, description="Búsqueda en título o descripción"), # Multi-tenant (solo ADMIN/SUPPORT_MANAGER) all_tenants: bool = Query(False, description="Ver incidentes de todos los tenants"), # Dependencies current_user: User = Depends(require_auditor_role), current_tenant: Tenant = Depends(get_current_tenant), db: AsyncSession = Depends(get_db) ): """ Obtener incidentes de seguridad. Los incidentes se generan dinámicamente analizando logs de auditoría para detectar patrones sospechosos y acciones críticas. **Permisos**: ADMIN, SUPPORT_MANAGER, AUDITOR **Retorna**: Lista paginada de incidentes de seguridad """ logger.info( "Fetching security incidents", user_id=str(current_user.id), tenant_id=str(current_tenant.id), filters={ "severity": severity, "status": status, "type": incident_type, "page": page, "per_page": per_page } ) # Generar incidentes a partir de logs de auditoría incidents = [] now = datetime.now(timezone.utc) # Determinar rango de tiempo para análisis (últimos 7 días para mejor performance) analysis_start = now - timedelta(days=7) # Construir query base base_query = select(AuditLog).options( selectinload(AuditLog.user) ).where( AuditLog.created_at >= analysis_start ) # Aplicar filtro de tenant if all_tenants and current_user.role in [UserRole.ADMIN, UserRole.SUPPORT_MANAGER]: # Ver incidentes de todos los tenants pass else: base_query = base_query.where(AuditLog.tenant_id == current_tenant.id) # 1. DETECTAR ELIMINACIONES MASIVAS deletion_query = base_query.where( AuditLog.action.like('%.delete') ).order_by(desc(AuditLog.created_at)) deletion_result = await db.execute(deletion_query) deletion_logs = deletion_result.scalars().all() # Agrupar eliminaciones por usuario y fecha deletion_groups = {} for log in deletion_logs: if not log.user: continue key = f"{log.user.email}_{log.created_at.date()}" if key not in deletion_groups: deletion_groups[key] = { 'user': log.user.email, 'date': log.created_at.date(), 'count': 0, 'logs': [], 'first_seen': log.created_at, 'last_seen': log.created_at } deletion_groups[key]['count'] += 1 deletion_groups[key]['logs'].append(log) if log.created_at < deletion_groups[key]['first_seen']: deletion_groups[key]['first_seen'] = log.created_at if log.created_at > deletion_groups[key]['last_seen']: deletion_groups[key]['last_seen'] = log.created_at # Crear incidentes para eliminaciones masivas (>=3 eliminaciones) for key, group in deletion_groups.items(): if group['count'] >= 3: # Umbral para considerar "masivo" severity = "critical" if group['count'] >= 10 else "high" if group['count'] >= 5 else "medium" incidents.append(SecurityIncidentResponse( id=f"mass_del_{key.replace('_', '-')}", title=f"Eliminaciones masivas - {group['user']}", description=f"{group['user']} eliminó {group['count']} elementos el {group['date']}", severity=severity, status="active" if (now - group['last_seen']).days <= 1 else "resolved", incident_type="mass_deletion", affected_user=group['user'], source_ip=group['logs'][0].ip_address, evidence=[ f"{log.action} - {log.resource_type} - {log.created_at.strftime('%H:%M:%S')}" for log in group['logs'][:5] # Solo mostrar los primeros 5 ], metadata={ "total_deletions": group['count'], "resource_types": list(set(log.resource_type for log in group['logs'])), "time_span_minutes": int((group['last_seen'] - group['first_seen']).total_seconds() / 60) }, created_at=group['first_seen'], updated_at=group['last_seen'] )) # 2. DETECTAR INTENTOS DE LOGIN FALLIDOS failed_login_query = base_query.where( AuditLog.action == 'user.login_failed' ).order_by(desc(AuditLog.created_at)) failed_login_result = await db.execute(failed_login_query) failed_login_logs = failed_login_result.scalars().all() # Agrupar por IP ip_groups = {} for log in failed_login_logs: if not log.ip_address: continue ip = str(log.ip_address) if ip not in ip_groups: ip_groups[ip] = { 'count': 0, 'logs': [], 'first_seen': log.created_at, 'last_seen': log.created_at, 'users': set() } ip_groups[ip]['count'] += 1 ip_groups[ip]['logs'].append(log) if log.created_at < ip_groups[ip]['first_seen']: ip_groups[ip]['first_seen'] = log.created_at if log.created_at > ip_groups[ip]['last_seen']: ip_groups[ip]['last_seen'] = log.created_at if log.user and log.user.email: ip_groups[ip]['users'].add(log.user.email) # Crear incidentes para IPs con muchos fallos (>=5) for ip, group in ip_groups.items(): if group['count'] >= 5: severity = "critical" if group['count'] >= 20 else "high" if group['count'] >= 10 else "medium" incidents.append(SecurityIncidentResponse( id=f"brute_force_{ip.replace('.', '-')}", title=f"Posible ataque de fuerza bruta desde {ip}", description=f"Se detectaron {group['count']} intentos fallidos de login desde la IP {ip}", severity=severity, status="active" if (now - group['last_seen']).total_seconds() <= 86400 else "investigating", # 24 horas incident_type="brute_force_attack", affected_user=', '.join(list(group['users'])[:3]) if group['users'] else None, source_ip=ip, evidence=[ f"Login fallido - {log.user.email if log.user else 'Unknown'} - {log.created_at.strftime('%H:%M:%S')}" for log in group['logs'][:5] ], metadata={ "total_attempts": group['count'], "targeted_users": list(group['users']), "time_span_hours": int((group['last_seen'] - group['first_seen']).total_seconds() / 3600) }, created_at=group['first_seen'], updated_at=group['last_seen'] )) # 3. DETECTAR CAMBIOS DE ROLES/PRIVILEGIOS privilege_query = base_query.where( and_( AuditLog.action == 'user.update', AuditLog.new_values.op('?')('role') ) ).order_by(desc(AuditLog.created_at)) privilege_result = await db.execute(privilege_query) privilege_logs = privilege_result.scalars().all() for log in privilege_logs: if not log.user or not log.new_values or 'role' not in log.new_values: continue old_role = log.old_values.get('role') if log.old_values else 'Unknown' new_role = log.new_values.get('role') # Solo crear incidente si es escalada de privilegios role_hierarchy = {'CLIENT_USER': 1, 'CLIENT_ADMIN': 2, 'AGENT': 3, 'SUPPORT_MANAGER': 4, 'ADMIN': 5} old_level = role_hierarchy.get(old_role, 0) new_level = role_hierarchy.get(new_role, 0) if new_level > old_level: incidents.append(SecurityIncidentResponse( id=f"priv_esc_{log.id}", title=f"Escalada de privilegios - {log.user.email}", description=f"Usuario {log.user.email} cambió de rol {old_role} a {new_role}", severity="high" if new_role in ['ADMIN', 'SUPPORT_MANAGER'] else "medium", status="investigating", incident_type="privilege_escalation", affected_user=log.user.email, source_ip=log.ip_address, evidence=[ f"Cambio de rol: {old_role} → {new_role} - {log.created_at.strftime('%Y-%m-%d %H:%M')}" ], metadata={ "old_role": old_role, "new_role": new_role, "correlation_id": str(log.correlation_id) if log.correlation_id else None }, created_at=log.created_at, updated_at=log.created_at )) # Aplicar filtros de búsqueda filtered_incidents = incidents if severity: filtered_incidents = [i for i in filtered_incidents if i.severity == severity] if status: filtered_incidents = [i for i in filtered_incidents if i.status == status] if incident_type: filtered_incidents = [i for i in filtered_incidents if i.incident_type == incident_type] if search: search_lower = search.lower() filtered_incidents = [ i for i in filtered_incidents if search_lower in i.title.lower() or (i.description and search_lower in i.description.lower()) ] # Ordenar por fecha de creación (más recientes primero) filtered_incidents.sort(key=lambda x: x.created_at, reverse=True) # Aplicar paginación total = len(filtered_incidents) total_pages = (total + per_page - 1) // per_page start_idx = (page - 1) * per_page end_idx = start_idx + per_page paginated_incidents = filtered_incidents[start_idx:end_idx] return SecurityIncidentListResponse( incidents=paginated_incidents, total=total, page=page, per_page=per_page, total_pages=total_pages ) @router.get("/security/incidents", response_model=SecurityIncidentListResponse) async def get_security_incidents( # Paginación page: int = Query(default=1, ge=1, description="Número de página"), per_page: int = Query(default=20, ge=1, le=100, description="Incidentes por página"), # Filtros severity: Optional[str] = Query(None, description="Filtrar por severidad"), status: Optional[str] = Query(None, description="Filtrar por estado"), incident_type: Optional[str] = Query(None, description="Filtrar por tipo"), search: Optional[str] = Query(None, description="Búsqueda en título o descripción"), # Multi-tenant (solo ADMIN/SUPPORT_MANAGER) all_tenants: bool = Query(False, description="Ver incidentes de todos los tenants"), # Dependencies current_user: User = Depends(require_auditor_role), current_tenant: Tenant = Depends(get_current_tenant), db: AsyncSession = Depends(get_db) ): """ Obtener incidentes de seguridad. Los incidentes se generan dinámicamente analizando logs de auditoría para detectar patrones sospechosos y acciones críticas. **Permisos**: ADMIN, SUPPORT_MANAGER, AUDITOR **Retorna**: Lista paginada de incidentes de seguridad """ logger.info( "Fetching security incidents", user_id=str(current_user.id), tenant_id=str(current_tenant.id), filters={ "severity": severity, "status": status, "type": incident_type, "page": page, "per_page": per_page } ) # Generar incidentes a partir de logs de auditoría incidents = [] now = datetime.now(timezone.utc) # Determinar rango de tiempo para análisis (últimos 30 días) analysis_start = now - timedelta(days=30) # Construir query base base_query = select(AuditLog).options( selectinload(AuditLog.user) ).where( AuditLog.created_at >= analysis_start ) # Aplicar filtro de tenant if all_tenants and current_user.role in [UserRole.ADMIN, UserRole.SUPPORT_MANAGER]: # Ver incidentes de todos los tenants pass else: base_query = base_query.where(AuditLog.tenant_id == current_tenant.id) # 1. DETECTAR ELIMINACIONES MASIVAS deletion_query = base_query.where( AuditLog.action.like('%.delete') ).order_by(desc(AuditLog.created_at)) deletion_result = await db.execute(deletion_query) deletion_logs = deletion_result.scalars().all() # Agrupar eliminaciones por usuario y fecha deletion_groups = {} for log in deletion_logs: if not log.user: continue key = f"{log.user.email}_{log.created_at.date()}" if key not in deletion_groups: deletion_groups[key] = { 'user': log.user.email, 'date': log.created_at.date(), 'count': 0, 'logs': [], 'first_seen': log.created_at, 'last_seen': log.created_at } deletion_groups[key]['count'] += 1 deletion_groups[key]['logs'].append(log) if log.created_at < deletion_groups[key]['first_seen']: deletion_groups[key]['first_seen'] = log.created_at if log.created_at > deletion_groups[key]['last_seen']: deletion_groups[key]['last_seen'] = log.created_at # Crear incidentes para eliminaciones masivas (>=5 eliminaciones) for key, group in deletion_groups.items(): if group['count'] >= 5: # Umbral para considerar "masivo" severity = "critical" if group['count'] >= 20 else "high" if group['count'] >= 10 else "medium" incidents.append(SecurityIncidentResponse( id=f"mass_del_{key.replace('_', '-')}", title=f"Eliminaciones masivas detectadas - {group['user']}", description=f"{group['user']} eliminó {group['count']} elementos el {group['date']}", severity=severity, status="active" if (now - group['last_seen']).days <= 1 else "resolved", incident_type="mass_deletion", affected_user=group['user'], source_ip=group['logs'][0].ip_address, evidence=[ f"{log.action} - {log.resource_type} {log.resource_id or 'N/A'} - {log.created_at.isoformat()}" for log in group['logs'][:5] # Solo mostrar los primeros 5 ], metadata={ "total_deletions": group['count'], "resource_types": list(set(log.resource_type for log in group['logs'])), "time_span_minutes": int((group['last_seen'] - group['first_seen']).total_seconds() / 60) }, created_at=group['first_seen'], updated_at=group['last_seen'] )) # 2. DETECTAR INTENTOS DE LOGIN FALLIDOS failed_login_query = base_query.where( AuditLog.action == 'user.login_failed' ).order_by(desc(AuditLog.created_at)) failed_login_result = await db.execute(failed_login_query) failed_login_logs = failed_login_result.scalars().all() # Agrupar por IP ip_groups = {} for log in failed_login_logs: if not log.ip_address: continue ip = str(log.ip_address) if ip not in ip_groups: ip_groups[ip] = { 'count': 0, 'logs': [], 'first_seen': log.created_at, 'last_seen': log.created_at, 'users': set() } ip_groups[ip]['count'] += 1 ip_groups[ip]['logs'].append(log) if log.created_at < ip_groups[ip]['first_seen']: ip_groups[ip]['first_seen'] = log.created_at if log.created_at > ip_groups[ip]['last_seen']: ip_groups[ip]['last_seen'] = log.created_at if log.user and log.user.email: ip_groups[ip]['users'].add(log.user.email) # Crear incidentes para IPs con muchos fallos (>=10) for ip, group in ip_groups.items(): if group['count'] >= 10: severity = "critical" if group['count'] >= 50 else "high" if group['count'] >= 25 else "medium" incidents.append(SecurityIncidentResponse( id=f"brute_force_{ip.replace('.', '-')}", title=f"Posible ataque de fuerza bruta desde {ip}", description=f"Se detectaron {group['count']} intentos fallidos de login desde la IP {ip}", severity=severity, status="active" if (now - group['last_seen']).hours <= 24 else "investigating", incident_type="brute_force_attack", affected_user=', '.join(list(group['users'])[:3]) if group['users'] else None, source_ip=ip, evidence=[ f"Login fallido - {log.user.email if log.user else 'Unknown'} - {log.created_at.isoformat()}" for log in group['logs'][:10] ], metadata={ "total_attempts": group['count'], "targeted_users": list(group['users']), "time_span_hours": int((group['last_seen'] - group['first_seen']).total_seconds() / 3600) }, created_at=group['first_seen'], updated_at=group['last_seen'] )) # 3. DETECTAR CAMBIOS DE ROLES/PRIVILEGIOS privilege_query = base_query.where( and_( AuditLog.action == 'user.update', AuditLog.new_values.op('?')('role') ) ).order_by(desc(AuditLog.created_at)) privilege_result = await db.execute(privilege_query) privilege_logs = privilege_result.scalars().all() for log in privilege_logs: if not log.user or not log.new_values or 'role' not in log.new_values: continue old_role = log.old_values.get('role') if log.old_values else 'Unknown' new_role = log.new_values.get('role') # Solo crear incidente si es escalada de privilegios role_hierarchy = {'CLIENT_USER': 1, 'CLIENT_ADMIN': 2, 'AGENT': 3, 'SUPPORT_MANAGER': 4, 'ADMIN': 5} old_level = role_hierarchy.get(old_role, 0) new_level = role_hierarchy.get(new_role, 0) if new_level > old_level: incidents.append(SecurityIncidentResponse( id=f"priv_esc_{log.id}", title=f"Escalada de privilegios detectada - {log.user.email}", description=f"Usuario {log.user.email} cambió de rol {old_role} a {new_role}", severity="high" if new_role in ['ADMIN', 'SUPPORT_MANAGER'] else "medium", status="investigating", incident_type="privilege_escalation", affected_user=log.user.email, source_ip=log.ip_address, evidence=[ f"Cambio de rol: {old_role} → {new_role} - {log.created_at.isoformat()}" ], metadata={ "old_role": old_role, "new_role": new_role, "changed_by": log.correlation_id # En el futuro, trackear quién hizo el cambio }, created_at=log.created_at, updated_at=log.created_at )) # Aplicar filtros de búsqueda filtered_incidents = incidents if severity: filtered_incidents = [i for i in filtered_incidents if i.severity == severity] if status: filtered_incidents = [i for i in filtered_incidents if i.status == status] if incident_type: filtered_incidents = [i for i in filtered_incidents if i.incident_type == incident_type] if search: search_lower = search.lower() filtered_incidents = [ i for i in filtered_incidents if search_lower in i.title.lower() or (i.description and search_lower in i.description.lower()) ] # Ordenar por fecha de creación (más recientes primero) filtered_incidents.sort(key=lambda x: x.created_at, reverse=True) # Aplicar paginación total = len(filtered_incidents) total_pages = (total + per_page - 1) // per_page start_idx = (page - 1) * per_page end_idx = start_idx + per_page paginated_incidents = filtered_incidents[start_idx:end_idx] return SecurityIncidentListResponse( incidents=paginated_incidents, total=total, page=page, per_page=per_page, total_pages=total_pages )