""" Audit Helpers - ServiceManagerWeb =================================== Funciones auxiliares reutilizables para los endpoints de auditoría. Este archivo contiene: - audit_log_to_dict: Convierte un modelo AuditLog a diccionario - apply_tenant_filter: Aplica filtro de tenant según permisos - get_count_stat: Cuenta registros con filtros opcionales (CORREGIDO) - get_top_items: Obtiene los items más frecuentes - detect_mass_deletions: Detecta eliminaciones masivas sospechosas - detect_brute_force: Detecta ataques de fuerza bruta - detect_privilege_escalation: Detecta escaladas de privilegios CORRECCIÓN APLICADA en get_count_stat: La columna created_at en PostgreSQL es 'timestamp with time zone' (TIMESTAMPTZ), lo que significa que almacena y devuelve fechas CON información de timezone (+00). El bug era que se comparaba un datetime naive (sin timezone) contra una columna TIMESTAMPTZ. PostgreSQL no puede comparar ambos tipos directamente, por lo que el filtro se ignoraba silenciosamente y los tres contadores devolvían el mismo valor (el total histórico completo sin ningún filtro de fecha). La solución es garantizar que TODAS las fechas que se usen en queries tengan timezone info (aware datetime en UTC) usando _ensure_aware_utc(). """ from sqlalchemy import select, func, and_, or_, desc from sqlalchemy.ext.asyncio import AsyncSession from typing import Optional, Dict, List from datetime import datetime, timezone import uuid from app.models.audit import AuditLog from app.models.user import User, UserRole from app.models.tenant import Tenant # ============================================================================= # CONVERSIÓN DE MODELOS # ============================================================================= def audit_log_to_dict(log: AuditLog) -> dict: """ Convierte un objeto AuditLog de SQLAlchemy a un diccionario plano compatible con los schemas de respuesta de Pydantic. Incluye los datos del usuario relacionado si están cargados (requiere que la query use selectinload(AuditLog.user)). """ 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 puede ser un objeto especial de PostgreSQL, convertir a string "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, # extra_metadata evita conflicto con la palabra reservada 'metadata' "metadata": log.extra_metadata, "created_at": log.created_at, "action_display": log.action_display, # Campos del usuario (se llenan abajo si la relación está cargada) "user_email": None, "user_name": None, "user_role": None, } # Solo agregar datos del usuario si la relación fue cargada en la query if log.user: log_dict["user_email"] = log.user.email log_dict["user_name"] = log.user.full_name # El rol puede ser un Enum de Python o un string, manejar ambos casos log_dict["user_role"] = ( log.user.role.value if hasattr(log.user.role, 'value') else str(log.user.role) ) return log_dict # ============================================================================= # FILTRO DE MULTI-TENANCY # ============================================================================= def apply_tenant_filter( query, current_user: User, current_tenant: Tenant, all_tenants: bool = False, specific_tenant_id: Optional[uuid.UUID] = None ): """ Aplica el filtro de tenant a una query de SQLAlchemy según los permisos del usuario actual. Reglas: - ADMIN y SUPPORT_MANAGER pueden ver todos los tenants si all_tenants=True, o filtrar por un tenant específico. - Cualquier otro rol solo puede ver los datos de su propio tenant. """ can_see_all_tenants = current_user.role in [UserRole.ADMIN, UserRole.SUPPORT_MANAGER] if all_tenants and can_see_all_tenants: # Usuario privilegiado pidiendo ver todos los tenants → sin filtro return query elif specific_tenant_id and can_see_all_tenants: # Usuario privilegiado pidiendo un tenant específico return query.where(AuditLog.tenant_id == specific_tenant_id) else: # Cualquier otro caso → solo ver el propio tenant return query.where(AuditLog.tenant_id == current_tenant.id) # ============================================================================= # UTILIDAD DE FECHAS # ============================================================================= def _ensure_aware_utc(dt: datetime) -> datetime: """ Garantiza que un datetime tenga información de timezone en UTC. PROBLEMA QUE RESUELVE: La columna created_at en PostgreSQL es 'timestamp with time zone' (TIMESTAMPTZ). Cuando se compara con un datetime naive (sin timezone), PostgreSQL no puede hacer la comparación correctamente y el filtro de fecha se ignora silenciosamente, devolviendo todos los registros sin importar la fecha. SOLUCIÓN: Siempre convertir las fechas a aware UTC antes de usarlas en queries. Casos que maneja: - datetime naive (sin tzinfo): agrega UTC como timezone - datetime aware (con tzinfo): convierte a UTC si es otra zona horaria Ejemplos: datetime(2026, 2, 24, 15, 0, 0) → datetime(2026, 2, 24, 15, 0, 0, tzinfo=UTC) datetime(2026, 2, 24, 9, 0, 0, tzinfo=CST) → datetime(2026, 2, 24, 15, 0, 0, tzinfo=UTC) """ if dt.tzinfo is None: # Datetime naive → asumir que ya es UTC y agregarle timezone info return dt.replace(tzinfo=timezone.utc) else: # Datetime aware → convertir a UTC (por si viene en otra zona horaria) return dt.astimezone(timezone.utc) # ============================================================================= # CONTADORES DE ESTADÍSTICAS # ============================================================================= async def get_count_stat( db: AsyncSession, tenant_id: Optional[uuid.UUID] = None, date_from: Optional[datetime] = None, action_filter=None ) -> int: """ Cuenta registros de AuditLog con filtros opcionales. Usado por get_audit_stats() para calcular: - total_actions: Sin date_from → cuenta todos los registros - actions_today: date_from = now - 24h → registros del día - actions_this_week: date_from = now - 7d → registros de la semana CORRECCIÓN: Las fechas se convierten a aware UTC con _ensure_aware_utc() antes de usarlas en la query, para que sean compatibles con la columna TIMESTAMPTZ de PostgreSQL y el filtro se aplique correctamente. Args: db: Sesión de base de datos tenant_id: Si se especifica, filtra por ese tenant date_from: Si se especifica, solo cuenta registros desde esa fecha action_filter: Condición SQLAlchemy adicional opcional Returns: Número entero de registros que cumplen los filtros """ query = select(func.count()).select_from(AuditLog) if tenant_id: query = query.where(AuditLog.tenant_id == tenant_id) if date_from: # CORRECCIÓN: convertir a aware UTC para compatibilidad con TIMESTAMPTZ # Sin esto, el filtro se ignora y los tres contadores son idénticos date_from_aware = _ensure_aware_utc(date_from) query = query.where(AuditLog.created_at >= date_from_aware) if action_filter is not None: query = query.where(action_filter) result = await db.execute(query) return result.scalar() or 0 # ============================================================================= # ITEMS MÁS FRECUENTES # ============================================================================= async def get_top_items( db: AsyncSession, field, tenant_id: Optional[uuid.UUID] = None, limit: int = 5, join_user: bool = False ) -> Dict[str, int]: """ Obtiene los valores más frecuentes de un campo, ordenados por conteo. Ejemplos de uso: - get_top_items(db, AuditLog.action, ...) → {"ticket.create": 45} - get_top_items(db, AuditLog.resource_type, ...) → {"ticket": 60} - get_top_items(db, None, ..., join_user=True) → {"admin@empresa.com": 40} Args: db: Sesión de base de datos field: Campo de AuditLog por el que agrupar tenant_id: Si se especifica, filtra por ese tenant limit: Máximo de resultados a devolver (por defecto 5) join_user: Si True, agrupa por email de usuario Returns: Diccionario {valor: conteo} ordenado de mayor a menor """ if join_user: # Modo usuarios: hacer JOIN con tabla User y agrupar por email query = ( select(User.email, func.count(AuditLog.id).label('count')) .join(User, AuditLog.user_id == User.id) ) else: # Modo campo: agrupar por el campo especificado query = select(field, func.count(AuditLog.id).label('count')) if tenant_id: query = query.where(AuditLog.tenant_id == tenant_id) if join_user: query = query.group_by(User.email) else: query = query.group_by(field) query = query.order_by(desc('count')).limit(limit) result = await db.execute(query) return {row[0]: row[1] for row in result} # ============================================================================= # DETECTORES DE INCIDENTES DE SEGURIDAD # ============================================================================= def detect_mass_deletions(logs: List[AuditLog], now: datetime) -> List[dict]: """ Detecta patrones de eliminación masiva agrupando por usuario y día. Lógica: - Agrupa todos los logs de eliminación por (usuario, día) - Si un usuario eliminó >= 3 recursos en un día, genera un incidente - La severidad escala según la cantidad: - >= 3 eliminaciones → medium - >= 5 eliminaciones → high - >= 10 eliminaciones → critical El estado del incidente es: - "active": si la última eliminación fue hace menos de 24 horas - "resolved": si fue hace más de 24 horas """ # Agrupar eliminaciones por usuario y día deletion_groups = {} for log in 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) deletion_groups[key]['first_seen'] = min(deletion_groups[key]['first_seen'], log.created_at) deletion_groups[key]['last_seen'] = max(deletion_groups[key]['last_seen'], log.created_at) incidents = [] for key, group in deletion_groups.items(): if group['count'] < 3: continue if group['count'] >= 10: severity = "critical" elif group['count'] >= 5: severity = "high" else: severity = "medium" # Convertir ambas fechas a aware UTC para comparación segura now_aware = _ensure_aware_utc(now) last_seen_aware = _ensure_aware_utc(group['last_seen']) hours_since_last = (now_aware - last_seen_aware).total_seconds() / 3600 incident_status = "active" if hours_since_last <= 24 else "resolved" incidents.append({ "id": f"mass_del_{key.replace('_', '-')}", "title": f"Eliminaciones masivas - {group['user']}", "description": ( f"{group['user']} elimino {group['count']} elementos " f"el {group['date']}" ), "severity": severity, "status": incident_status, "incident_type": "mass_deletion", "affected_user": group['user'], "source_ip": ( str(group['logs'][0].ip_address) if group['logs'][0].ip_address else None ), "evidence": [ f"{log.action} - {log.resource_type} - {log.created_at.strftime('%H:%M:%S')}" for log in group['logs'][: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'] }) return incidents def detect_brute_force(logs: List[AuditLog], now: datetime) -> List[dict]: """ Detecta ataques de fuerza bruta agrupando intentos fallidos por IP. Lógica: - Agrupa todos los intentos fallidos de login por dirección IP - Si una IP tiene >= 5 intentos, genera un incidente - La severidad escala según la cantidad: - >= 5 intentos → medium - >= 10 intentos → high - >= 20 intentos → critical El estado del incidente es: - "active": si el último intento fue hace menos de 24 horas - "investigating": si fue hace más de 24 horas """ ip_groups = {} for log in 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) ip_groups[ip]['first_seen'] = min(ip_groups[ip]['first_seen'], log.created_at) ip_groups[ip]['last_seen'] = max(ip_groups[ip]['last_seen'], log.created_at) if log.user and log.user.email: ip_groups[ip]['users'].add(log.user.email) incidents = [] for ip, group in ip_groups.items(): if group['count'] < 5: continue if group['count'] >= 20: severity = "critical" elif group['count'] >= 10: severity = "high" else: severity = "medium" # Convertir ambas fechas a aware UTC para comparación segura now_aware = _ensure_aware_utc(now) last_seen_aware = _ensure_aware_utc(group['last_seen']) seconds_since_last = (now_aware - last_seen_aware).total_seconds() incident_status = "active" if seconds_since_last <= 86400 else "investigating" incidents.append({ "id": f"brute_force_{ip.replace('.', '-')}", "title": f"Posible ataque de fuerza bruta desde {ip}", "description": ( f"Se detectaron {group['count']} intentos fallidos de " f"login desde la IP {ip}" ), "severity": severity, "status": incident_status, "incident_type": "brute_force_attack", "affected_user": ( ', '.join(list(group['users'])[:3]) if group['users'] else None ), "source_ip": ip, "evidence": [ f"Login fallido - " f"{log.user.email if log.user else 'Desconocido'} - " f"{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'] }) return incidents def detect_privilege_escalation(logs: List[AuditLog]) -> List[dict]: """ Detecta escaladas de privilegios comparando el rol anterior y nuevo. Lógica: - Analiza cada log de cambio de rol (user.update con campo 'role') - Si el nuevo rol tiene más privilegios que el anterior, es sospechoso - Cada cambio que represente una escalada genera un incidente Jerarquía de roles (de menor a mayor privilegio): CLIENT_USER(1) < CLIENT_ADMIN(2) < AGENT(3) < SUPPORT_MANAGER(4) < ADMIN(5) """ role_hierarchy = { 'CLIENT_USER': 1, 'CLIENT_ADMIN': 2, 'AGENT': 3, 'SUPPORT_MANAGER': 4, 'ADMIN': 5 } incidents = [] for log in 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') old_level = role_hierarchy.get(old_role, 0) new_level = role_hierarchy.get(new_role, 0) # Solo generar incidente si el nuevo rol tiene MÁS privilegios if new_level <= old_level: continue severity = "high" if new_role in ['ADMIN', 'SUPPORT_MANAGER'] else "medium" incidents.append({ "id": f"priv_esc_{log.id}", "title": f"Escalada de privilegios - {log.user.email}", "description": ( f"Usuario {log.user.email} cambio de rol " f"{old_role} a {new_role}" ), "severity": severity, "status": "investigating", "incident_type": "privilege_escalation", "affected_user": log.user.email, "source_ip": str(log.ip_address) if log.ip_address else None, "evidence": [ f"Cambio de rol: {old_role} → {new_role} - " f"{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 }) return incidents