icamarillo
bd21207aae
- Nuevo módulo de reportes: backend/app/api/v1/endpoints/reports.py - Schemas de reportes: backend/app/api/schemas/reports.py - Frontend: frontend-internal/src/routes/reports/ - Mejoras al módulo de auditoría (audit.py, audit_helpers.py) - Modelo de auditoría actualizado - Sidebar actualizado con enlace a reportes
779 lines
29 KiB
Python
779 lines
29 KiB
Python
"""
|
|
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
|
|
) |