- Agregado módulo de auditoría con AuditLog model - Implementado endpoint /api/v1/audit para consulta de logs - Integrado audit_service para registro automático de acciones - Agregado token_service para gestión de refresh tokens - Frontend: Vista de auditoría en panel interno - Actualizada versión en pyproject.toml y package.json - Sistema de auditoría completamente funcional y testeado
149 lines
4.6 KiB
Python
149 lines
4.6 KiB
Python
"""
|
|
Audit Log Model - ServiceManagerWeb
|
|
|
|
Modelo para bitácora de auditoría y compliance.
|
|
Registra todas las acciones importantes del sistema.
|
|
"""
|
|
|
|
from sqlalchemy import String, Text, DateTime, ForeignKey, Index
|
|
from sqlalchemy.orm import Mapped, mapped_column, relationship
|
|
from sqlalchemy.dialects.postgresql import UUID, INET, JSONB
|
|
from typing import Optional, Dict, Any, TYPE_CHECKING
|
|
import uuid
|
|
from datetime import datetime
|
|
|
|
from app.core.database import Base
|
|
|
|
if TYPE_CHECKING:
|
|
from app.models.tenant import Tenant
|
|
from app.models.user import User
|
|
|
|
|
|
class AuditLog(Base):
|
|
"""
|
|
Bitácora de auditoría para tracking completo de acciones.
|
|
|
|
Registra:
|
|
- Qui├®n hizo la acci├│n (user_id)
|
|
- Qu├® hizo (action)
|
|
- Sobre qu├® recurso (resource_type + resource_id)
|
|
- Cuándo lo hizo (created_at)
|
|
- Desde d├│nde (ip_address, user_agent)
|
|
- Qu├® cambi├│ (old_values, new_values)
|
|
"""
|
|
|
|
__tablename__ = "audit_logs"
|
|
|
|
# Multi-tenancy
|
|
tenant_id: Mapped[uuid.UUID] = mapped_column(
|
|
UUID(as_uuid=True),
|
|
ForeignKey("tenants.id", ondelete="CASCADE"),
|
|
nullable=False,
|
|
index=True
|
|
)
|
|
|
|
# Usuario que ejecut├│ la acci├│n (NULL = acci├│n del sistema)
|
|
user_id: Mapped[Optional[uuid.UUID]] = mapped_column(
|
|
UUID(as_uuid=True),
|
|
ForeignKey("users.id", ondelete="SET NULL"),
|
|
nullable=True,
|
|
index=True
|
|
)
|
|
|
|
# Acci├│n realizada (ej: "user.login", "ticket.create", "ticket.assign")
|
|
action: Mapped[str] = mapped_column(
|
|
String(100),
|
|
nullable=False,
|
|
index=True
|
|
)
|
|
|
|
# Tipo de recurso afectado (user, ticket, comment, category, etc.)
|
|
resource_type: Mapped[str] = mapped_column(
|
|
String(50),
|
|
nullable=False,
|
|
index=True
|
|
)
|
|
|
|
# ID del recurso afectado
|
|
resource_id: Mapped[Optional[uuid.UUID]] = mapped_column(
|
|
UUID(as_uuid=True),
|
|
nullable=True
|
|
)
|
|
|
|
# Contexto de la request
|
|
ip_address: Mapped[Optional[str]] = mapped_column(INET, nullable=True)
|
|
user_agent: Mapped[Optional[str]] = mapped_column(Text, nullable=True)
|
|
|
|
# Correlation ID para rastrear requests relacionadas
|
|
correlation_id: Mapped[Optional[uuid.UUID]] = mapped_column(
|
|
UUID(as_uuid=True),
|
|
nullable=True,
|
|
index=True
|
|
)
|
|
|
|
# Valores antes del cambio (JSON)
|
|
old_values: Mapped[Optional[Dict[str, Any]]] = mapped_column(
|
|
JSONB,
|
|
nullable=True
|
|
)
|
|
|
|
# Valores despu├®s del cambio (JSON)
|
|
new_values: Mapped[Optional[Dict[str, Any]]] = mapped_column(
|
|
JSONB,
|
|
nullable=True
|
|
)
|
|
|
|
# Metadata adicional (cualquier info relevante)
|
|
# Nota: 'metadata' está reservado en SQLAlchemy, usamos 'extra_metadata'
|
|
extra_metadata: Mapped[Optional[Dict[str, Any]]] = mapped_column(
|
|
'metadata', # Nombre real de la columna en BD
|
|
JSONB,
|
|
nullable=True
|
|
)
|
|
|
|
# Timestamp
|
|
created_at: Mapped[datetime] = mapped_column(
|
|
DateTime(timezone=True),
|
|
default=datetime.utcnow,
|
|
nullable=False,
|
|
index=True
|
|
)
|
|
|
|
# Relaciones
|
|
tenant: Mapped["Tenant"] = relationship("Tenant", foreign_keys=[tenant_id])
|
|
user: Mapped[Optional["User"]] = relationship("User", foreign_keys=[user_id])
|
|
|
|
# Índices compuestos para queries comunes
|
|
__table_args__ = (
|
|
Index('idx_audit_logs_tenant_action', 'tenant_id', 'action'),
|
|
Index('idx_audit_logs_resource', 'resource_type', 'resource_id'),
|
|
Index('idx_audit_logs_user_created', 'user_id', 'created_at'),
|
|
)
|
|
|
|
# Configuraci├│n del mapper: excluir updated_at porque audit logs son inmutables
|
|
__mapper_args__ = {
|
|
"exclude_properties": ["updated_at"]
|
|
}
|
|
|
|
def __repr__(self) -> str:
|
|
return f"<AuditLog(action='{self.action}', resource='{self.resource_type}:{self.resource_id}')>"
|
|
|
|
@property
|
|
def action_display(self) -> str:
|
|
"""Formato amigable de la acci├│n."""
|
|
parts = self.action.split('.')
|
|
if len(parts) == 2:
|
|
resource, verb = parts
|
|
verb_map = {
|
|
'create': 'cre├│',
|
|
'update': 'actualiz├│',
|
|
'delete': 'elimin├│',
|
|
'login': 'inici├│ sesi├│n',
|
|
'logout': 'cerr├│ sesi├│n',
|
|
'assign': 'asign├│',
|
|
'close': 'cerr├│',
|
|
'reopen': 'reabri├│'
|
|
}
|
|
return f"{verb_map.get(verb, verb)} {resource}"
|
|
return self.action
|