77 lines
2.7 KiB
Python
77 lines
2.7 KiB
Python
from typing import Optional, List, Dict, Any
|
|
from flask import request, g, has_request_context
|
|
from flask_login import current_user
|
|
import json
|
|
from app import db
|
|
from app.models.audit_log import AuditLog
|
|
from app.models.user import User
|
|
|
|
class AuditService:
|
|
"""Servicio para la captura y consulta de la bitácora inmutable de auditoría."""
|
|
|
|
@staticmethod
|
|
def log(action: str, module: str, entity_id: Optional[int] = None,
|
|
details: Optional[Any] = None, user: Optional[User] = None,
|
|
ip_address: Optional[str] = None) -> AuditLog:
|
|
"""
|
|
Registra un evento de auditoría de forma segura, infiriendo el usuario e IP
|
|
desde el contexto actual de la petición si no se especifican.
|
|
"""
|
|
user_id = None
|
|
user_email = None
|
|
|
|
# 1. Determinar usuario actuante
|
|
if user:
|
|
user_id = user.id
|
|
user_email = user.email
|
|
elif has_request_context():
|
|
if hasattr(g, 'jwt_user') and g.jwt_user:
|
|
user_id = g.jwt_user.id
|
|
user_email = g.jwt_user.email
|
|
elif current_user and current_user.is_authenticated:
|
|
user_id = current_user.id
|
|
user_email = current_user.email
|
|
|
|
# 2. Determinar IP de origen
|
|
if not ip_address and has_request_context():
|
|
if request.headers.get('X-Forwarded-For'):
|
|
ip_address = request.headers.get('X-Forwarded-For').split(',')[0].strip()
|
|
else:
|
|
ip_address = request.remote_addr
|
|
|
|
# 3. Serializar detalles si es estructura de datos
|
|
details_str = None
|
|
if details is not None:
|
|
if isinstance(details, (dict, list)):
|
|
try:
|
|
details_str = json.dumps(details, ensure_ascii=False)
|
|
except Exception:
|
|
details_str = str(details)
|
|
else:
|
|
details_str = str(details)
|
|
|
|
log_entry = AuditLog(
|
|
user_id=user_id,
|
|
user_email=user_email,
|
|
action=action.upper(),
|
|
module=module.lower(),
|
|
entity_id=entity_id,
|
|
details=details_str,
|
|
ip_address=ip_address
|
|
)
|
|
try:
|
|
db.session.add(log_entry)
|
|
db.session.commit()
|
|
except Exception:
|
|
db.session.rollback()
|
|
|
|
return log_entry
|
|
|
|
@staticmethod
|
|
def get_recent_logs(limit: int = 50, module: Optional[str] = None) -> List[AuditLog]:
|
|
"""Obtiene las entradas más recientes de auditoría."""
|
|
query = AuditLog.query
|
|
if module:
|
|
query = query.filter_by(module=module.lower())
|
|
return query.order_by(AuditLog.created_at.desc()).limit(limit).all()
|