feat: Mejoras en auditoría - incidentes de seguridad y esquema de colores
- Backend: * Agregado endpoint /v1/audit/security/incidents con paginación y filtros * Nuevos schemas SecurityIncidentResponse y SecurityIncidentListResponse * Fix timezone: datetime.utcnow() → datetime.now(timezone.utc) en 4 ubicaciones * Detección automática de incidentes: mass deletion, brute force, privilege escalation - Frontend (Internal): * Nueva sección de Incidentes de Seguridad con modal de detalles * Filtros por severidad, estado y tipo de incidente * Conversión completa a esquema grayscale (gray-100 a gray-900) * Eliminados todos los emojis de páginas audit y security * Implementada paginación para incidentes - Fixes: * Resuelto error 500: TypeError con datetimes timezone-aware/naive * Resuelto error 404: endpoint de incidentes faltante
This commit is contained in:
@@ -10,7 +10,7 @@ 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
|
||||
from datetime import datetime, timedelta, timezone
|
||||
import uuid
|
||||
import structlog
|
||||
|
||||
@@ -28,7 +28,9 @@ from app.api.schemas.audit import (
|
||||
SecurityAnalysisResponse,
|
||||
SecurityThreatPattern,
|
||||
SecurityActionRequest,
|
||||
SecurityActionResponse
|
||||
SecurityActionResponse,
|
||||
SecurityIncidentResponse,
|
||||
SecurityIncidentListResponse
|
||||
)
|
||||
|
||||
router = APIRouter()
|
||||
@@ -227,7 +229,7 @@ async def get_audit_stats(
|
||||
can_see_all=can_see_all_tenants
|
||||
)
|
||||
|
||||
now = datetime.utcnow()
|
||||
now = datetime.now(timezone.utc)
|
||||
|
||||
# Determinar si aplicar filtro de tenant
|
||||
apply_tenant_filter = not (all_tenants and can_see_all_tenants)
|
||||
@@ -429,7 +431,7 @@ async def get_security_analysis(
|
||||
hours=hours
|
||||
)
|
||||
|
||||
now = datetime.utcnow()
|
||||
now = datetime.now(timezone.utc)
|
||||
analysis_start = now - timedelta(hours=hours)
|
||||
|
||||
threats = []
|
||||
@@ -713,3 +715,537 @@ async def execute_security_action(
|
||||
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
|
||||
)
|
||||
|
||||
Reference in New Issue
Block a user