248 lines
8.2 KiB
Python
248 lines
8.2 KiB
Python
"""
|
|
Audit Log Service
|
|
"""
|
|
from datetime import datetime, timedelta, date, time
|
|
from decimal import Decimal
|
|
import uuid
|
|
import pytz
|
|
from typing import Optional, List, Dict, Any
|
|
from sqlalchemy.orm import Session
|
|
from ..models import AuditLog
|
|
from .core import AuditMapper, ReferenceGenerator
|
|
from core.security import verify_token # keep if needed or simpler just remove if unused
|
|
# We don't need security import here anymore as context is passed explicitly or handled by events
|
|
|
|
|
|
def _make_json_safe(obj: Any) -> Any:
|
|
"""Recursively convert non-JSON-serializable types to serializable equivalents."""
|
|
if obj is None:
|
|
return None
|
|
if isinstance(obj, dict):
|
|
return {k: _make_json_safe(v) for k, v in obj.items()}
|
|
if isinstance(obj, (list, tuple)):
|
|
return [_make_json_safe(v) for v in obj]
|
|
if isinstance(obj, datetime):
|
|
return obj.isoformat()
|
|
if isinstance(obj, date):
|
|
return obj.isoformat()
|
|
if isinstance(obj, time):
|
|
return obj.isoformat()
|
|
if isinstance(obj, Decimal):
|
|
return float(obj)
|
|
if isinstance(obj, uuid.UUID):
|
|
return str(obj)
|
|
if isinstance(obj, bytes):
|
|
return obj.decode("utf-8", errors="replace")
|
|
|
|
if hasattr(obj, "__dict__"):
|
|
d = dict(obj.__dict__)
|
|
d.pop("_sa_instance_state", None)
|
|
return {k: _make_json_safe(v) for k, v in d.items()}
|
|
|
|
if not isinstance(obj, (int, float, str, bool)):
|
|
return str(obj)
|
|
|
|
return obj
|
|
|
|
|
|
class AuditService:
|
|
|
|
@staticmethod
|
|
def create_audit_log(
|
|
db: Session,
|
|
reference: str,
|
|
procedure: str,
|
|
movement: str,
|
|
username: str,
|
|
system: str = "SCAF",
|
|
# Extra context
|
|
table_name: Optional[str] = None,
|
|
record_id: Optional[str] = None,
|
|
operation_type: Optional[str] = None,
|
|
old_values: Optional[Dict] = None,
|
|
new_values: Optional[Dict] = None,
|
|
changed_fields: Optional[List[str]] = None,
|
|
# HTTP Context
|
|
ip_address: Optional[str] = None,
|
|
user_agent: Optional[str] = None,
|
|
endpoint: Optional[str] = None,
|
|
request_method: Optional[str] = None,
|
|
session_id: Optional[str] = None,
|
|
company_id: Optional[int] = None,
|
|
tenant_id: Optional[int] = None,
|
|
) -> AuditLog:
|
|
"""
|
|
Low-level creation of an Audit Log entry
|
|
"""
|
|
# Timezone handling: Use UTC for consistency across regions.
|
|
# Frontend will convert to user's local time.
|
|
now = datetime.now(pytz.UTC)
|
|
|
|
log = AuditLog(
|
|
reference=reference,
|
|
procedure=procedure,
|
|
movement=movement,
|
|
username=username,
|
|
date=now.date(),
|
|
time=now.time(),
|
|
timestamp=now,
|
|
system=system,
|
|
table_name=table_name,
|
|
record_id=record_id,
|
|
operation_type=operation_type,
|
|
old_values=_make_json_safe(old_values),
|
|
new_values=_make_json_safe(new_values),
|
|
changed_fields=changed_fields,
|
|
ip_address=ip_address,
|
|
user_agent=user_agent,
|
|
endpoint=endpoint,
|
|
request_method=request_method,
|
|
session_id=session_id,
|
|
company_id=company_id,
|
|
tenant_id=tenant_id
|
|
)
|
|
|
|
db.add(log)
|
|
db.commit()
|
|
db.refresh(log)
|
|
return log
|
|
|
|
@staticmethod
|
|
def log_crud_operation(
|
|
db: Session,
|
|
table_name: str,
|
|
operation_type: str,
|
|
record_data: Dict[str, Any],
|
|
username: str,
|
|
record_id: Optional[str] = None,
|
|
old_values: Optional[Dict] = None,
|
|
new_values: Optional[Dict] = None,
|
|
# Context
|
|
ip_address: Optional[str] = None,
|
|
user_agent: Optional[str] = None,
|
|
company_id: Optional[int] = None,
|
|
tenant_id: Optional[int] = None
|
|
):
|
|
"""
|
|
High-level wrapper to log CRUD operations automatically mapping to Legacy format
|
|
"""
|
|
|
|
# 1. Map to Legacy Base Format
|
|
legacy_data = AuditMapper.map_to_legacy_format(
|
|
table_name=table_name,
|
|
record_id=record_id,
|
|
operation_type=operation_type,
|
|
username=username
|
|
)
|
|
|
|
# 2. Refine Reference based on specific table logic
|
|
reference = legacy_data["reference"]
|
|
|
|
if table_name == "invoice_header":
|
|
generated_ref = ReferenceGenerator.generate_invoice_reference(record_data)
|
|
# Use generated ref only if meaningful, else keep default
|
|
if generated_ref != "FUSE0-000-00":
|
|
reference = generated_ref
|
|
|
|
elif table_name == "pedimentos":
|
|
reference = ReferenceGenerator.generate_pedimento_reference(record_data)
|
|
|
|
elif table_name == "clients_and_providers":
|
|
reference = record_data.get("rfc") or reference
|
|
|
|
elif table_name == "parts":
|
|
reference = record_data.get("part_number") or reference
|
|
|
|
elif table_name == "companies":
|
|
reference = record_data.get("rfc") or reference
|
|
|
|
elif table_name == "classes":
|
|
reference = record_data.get("class_code") or reference
|
|
|
|
|
|
# 3. Detect Changed Fields (for Update)
|
|
changed_fields = None
|
|
if operation_type == "UPDATE" and old_values and new_values:
|
|
changed_fields = [
|
|
k for k in new_values.keys()
|
|
if old_values.get(k) != new_values.get(k)
|
|
]
|
|
|
|
# 4. Create Log
|
|
return AuditService.create_audit_log(
|
|
db=db,
|
|
reference=reference,
|
|
procedure=legacy_data["procedure"],
|
|
movement=legacy_data["movement"],
|
|
username=username,
|
|
system=legacy_data["system"],
|
|
table_name=table_name,
|
|
record_id=record_id,
|
|
operation_type=operation_type,
|
|
old_values=old_values,
|
|
new_values=new_values,
|
|
changed_fields=changed_fields,
|
|
ip_address=ip_address,
|
|
user_agent=user_agent,
|
|
company_id=company_id,
|
|
tenant_id=tenant_id
|
|
)
|
|
|
|
@staticmethod
|
|
def log_login(db: Session, username: str, ip_address: str = None, user_agent: str = None):
|
|
|
|
# Prevent duplicate login logs (debounce 5 seconds)
|
|
# This handles cases where frontend might submit twice or redirects trigger re-auth
|
|
try:
|
|
# Timezone handling: Use UTC for consistency
|
|
now = datetime.now(pytz.UTC)
|
|
|
|
five_seconds_ago = now - timedelta(seconds=5)
|
|
|
|
# Check for recent login from same user
|
|
existing = db.query(AuditLog).filter(
|
|
AuditLog.username == username,
|
|
AuditLog.operation_type == "LOGIN",
|
|
# Compare against timestamp (timezone aware)
|
|
AuditLog.timestamp >= five_seconds_ago
|
|
).first()
|
|
|
|
if existing:
|
|
return existing
|
|
|
|
except Exception as e:
|
|
import traceback
|
|
traceback.print_exc()
|
|
|
|
return AuditService.create_audit_log(
|
|
db=db,
|
|
reference="LOGIN",
|
|
procedure="SYSTEM SCAF",
|
|
movement="SYSTEM LOGIN",
|
|
username=username,
|
|
operation_type="LOGIN",
|
|
ip_address=ip_address,
|
|
user_agent=user_agent
|
|
)
|
|
|
|
@staticmethod
|
|
def log_logout(db: Session, username: str, ip_address: str = None, user_agent: str = None):
|
|
"""
|
|
Registra un evento de cierre de sesión
|
|
"""
|
|
try:
|
|
# Reutilizamos create_audit_log para mantener consistencia
|
|
AuditService.create_audit_log(
|
|
db=db,
|
|
reference="LOGOUT",
|
|
procedure="SYSTEM AUTH",
|
|
movement="SYSTEM LOGOUT",
|
|
username=username,
|
|
operation_type="LOGOUT",
|
|
ip_address=ip_address,
|
|
user_agent=user_agent
|
|
)
|
|
except Exception as e:
|
|
# No re-lanzamos la excepción para no interrumpir el flujo de logout
|
|
pass
|