Nuevas funcionalidades: - Sistema completo de análisis de seguridad con detección de amenazas - Detección de patrones: fuerza bruta, escalada de privilegios, eliminaciones masivas, cuentas comprometidas - Panel de vulnerabilidades con visualización detallada - Acciones de seguridad: bloqueo de IPs, notificaciones, reset de contraseñas - Análisis configurable (24h, 48h, 7 días) Backend (/audit/security/): - GET /analysis: Análisis completo de seguridad con amenazas detectadas - POST /action: Ejecutar acciones de seguridad (solo ADMIN/SUPPORT_MANAGER) - Schemas nuevos: SecurityAnalysisResponse, SecurityThreatPattern, SecurityActionRequest Frontend (/audit/security): - Panel completo de análisis con nivel de riesgo general - Visualización de amenazas con severidad (critical, high, medium, low) - Estadísticas: amenazas, intentos fallidos, IPs sospechosas, acciones críticas - Opciones de acción por amenaza: bloquear IP, resetear contraseña, notificar admin - Modal de ejecución de acciones de seguridad Mejoras: - Sidebar actualizado con enlace 'Seguridad' - Permisos: Solo ADMIN/SUPPORT_MANAGER/AUDITOR pueden ver análisis - Solo ADMIN/SUPPORT_MANAGER pueden ejecutar acciones - Audit log de todas las acciones de seguridad ejecutadas
160 lines
6.2 KiB
Python
160 lines
6.2 KiB
Python
"""
|
|
Audit Schemas - ServiceManagerWeb
|
|
|
|
Schemas Pydantic para endpoints de auditoría
|
|
"""
|
|
|
|
from pydantic import BaseModel, Field, UUID4
|
|
from typing import Optional, Dict, Any
|
|
from datetime import datetime
|
|
|
|
|
|
class AuditLogBase(BaseModel):
|
|
"""Schema base para audit logs."""
|
|
action: str = Field(..., description="Acci├│n realizada (ej: ticket.create)")
|
|
resource_type: str = Field(..., description="Tipo de recurso (ticket, user, etc.)")
|
|
resource_id: Optional[UUID4] = Field(None, description="ID del recurso afectado")
|
|
extra_metadata: Optional[Dict[str, Any]] = Field(None, description="Metadata adicional", alias="metadata")
|
|
|
|
|
|
class AuditLogResponse(AuditLogBase):
|
|
"""
|
|
Schema de respuesta para audit logs.
|
|
|
|
Incluye toda la informaci├│n del log con datos del usuario.
|
|
"""
|
|
id: UUID4
|
|
tenant_id: UUID4
|
|
user_id: Optional[UUID4]
|
|
|
|
# Informaci├│n del usuario (si existe)
|
|
user_email: Optional[str] = None
|
|
user_name: Optional[str] = None
|
|
user_role: Optional[str] = None
|
|
|
|
# Contexto de la acci├│n
|
|
ip_address: Optional[str]
|
|
user_agent: Optional[str]
|
|
correlation_id: Optional[UUID4]
|
|
|
|
# Cambios realizados
|
|
old_values: Optional[Dict[str, Any]]
|
|
new_values: Optional[Dict[str, Any]]
|
|
|
|
# Timestamp
|
|
created_at: datetime
|
|
|
|
# Display friendly
|
|
action_display: str = Field(description="Acci├│n en formato amigable")
|
|
|
|
class Config:
|
|
from_attributes = True
|
|
|
|
|
|
# ===================================
|
|
# SECURITY ANALYSIS SCHEMAS
|
|
# ===================================
|
|
|
|
class SecurityThreatPattern(BaseModel):
|
|
"""Patrón de amenaza detectado."""
|
|
type: str = Field(description="Tipo de amenaza (brute_force, privilege_escalation, etc.)")
|
|
severity: str = Field(description="Severidad: low, medium, high, critical")
|
|
description: str = Field(description="Descripción de la amenaza")
|
|
occurrences: int = Field(description="Número de ocurrencias")
|
|
affected_ips: list[str] = Field(default=[], description="IPs involucradas")
|
|
affected_users: list[str] = Field(default=[], description="Usuarios afectados")
|
|
first_seen: datetime = Field(description="Primera ocurrencia")
|
|
last_seen: datetime = Field(description="Última ocurrencia")
|
|
recommendations: list[str] = Field(default=[], description="Recomendaciones de acción")
|
|
|
|
|
|
class SecurityAnalysisResponse(BaseModel):
|
|
"""Análisis completo de seguridad."""
|
|
overall_risk_level: str = Field(description="Nivel de riesgo general: safe, low, medium, high, critical")
|
|
total_threats_detected: int = Field(description="Total de amenazas detectadas")
|
|
threats: list[SecurityThreatPattern] = Field(description="Lista de amenazas detectadas")
|
|
analysis_period_hours: int = Field(description="Período de análisis en horas")
|
|
generated_at: datetime = Field(description="Timestamp del análisis")
|
|
|
|
# Estadísticas de seguridad
|
|
failed_login_attempts: int = Field(description="Intentos fallidos de login")
|
|
suspicious_ips_count: int = Field(description="IPs sospechosas detectadas")
|
|
critical_actions_count: int = Field(description="Acciones críticas realizadas")
|
|
|
|
# Opciones de acción
|
|
recommended_actions: list[str] = Field(default=[], description="Acciones recomendadas")
|
|
|
|
|
|
class SecurityActionRequest(BaseModel):
|
|
"""Solicitud de acción de seguridad."""
|
|
action_type: str = Field(description="Tipo de acción: block_ip, notify_admin, reset_password, etc.")
|
|
target: str = Field(description="Objetivo de la acción (IP, email, etc.)")
|
|
reason: str = Field(description="Razón de la acción")
|
|
duration_minutes: Optional[int] = Field(None, description="Duración del bloqueo en minutos")
|
|
|
|
|
|
class SecurityActionResponse(BaseModel):
|
|
"""Respuesta de acción de seguridad."""
|
|
success: bool = Field(description="Si la acción fue exitosa")
|
|
message: str = Field(description="Mensaje descriptivo")
|
|
action_id: Optional[UUID4] = Field(None, description="ID de la acción registrada")
|
|
|
|
|
|
class AuditLogFilters(BaseModel):
|
|
"""
|
|
Filtros para consulta de audit logs.
|
|
|
|
Permite filtrar por m├║ltiples criterios.
|
|
"""
|
|
# Paginaci├│n
|
|
page: int = Field(default=1, ge=1, description="Número de página")
|
|
per_page: int = Field(default=50, ge=1, le=100, description="Elementos por página")
|
|
|
|
# Filtros
|
|
user_id: Optional[UUID4] = Field(None, description="Filtrar por usuario")
|
|
action: Optional[str] = Field(None, description="Filtrar por acción específica")
|
|
resource_type: Optional[str] = Field(None, description="Filtrar por tipo de recurso")
|
|
resource_id: Optional[UUID4] = Field(None, description="Filtrar por ID de recurso")
|
|
|
|
# Rango de fechas
|
|
date_from: Optional[datetime] = Field(None, description="Fecha inicio (ISO 8601)")
|
|
date_to: Optional[datetime] = Field(None, description="Fecha fin (ISO 8601)")
|
|
|
|
# B├║squeda
|
|
search: Optional[str] = Field(None, description="B├║squeda en acciones o recursos")
|
|
|
|
|
|
class AuditLogStats(BaseModel):
|
|
"""
|
|
Estadísticas de auditoría.
|
|
|
|
Resumen de actividad del sistema.
|
|
"""
|
|
total_actions: int = Field(description="Total de acciones registradas")
|
|
actions_today: int = Field(description="Acciones en las ├║ltimas 24 horas")
|
|
actions_this_week: int = Field(description="Acciones en los últimos 7 días")
|
|
critical_actions_today: int = Field(description="Acciones críticas hoy (delete, cambios sensibles)")
|
|
|
|
# Top acciones
|
|
top_actions: Dict[str, int] = Field(description="Acciones más frecuentes")
|
|
|
|
# Top usuarios
|
|
top_users: Dict[str, int] = Field(description="Usuarios más activos")
|
|
|
|
# Actividad por tipo de recurso
|
|
by_resource_type: Dict[str, int] = Field(description="Acciones por tipo de recurso")
|
|
|
|
|
|
class AuditLogListResponse(BaseModel):
|
|
"""
|
|
Respuesta paginada de audit logs.
|
|
"""
|
|
logs: list[AuditLogResponse]
|
|
total: int = Field(description="Total de registros")
|
|
page: int = Field(description="Página actual")
|
|
per_page: int = Field(description="Registros por página")
|
|
total_pages: int = Field(description="Total de páginas")
|
|
|
|
class Config:
|
|
from_attributes = True
|