feat: Implement comprehensive audit logging for authentication events, including login, logout, user agent, and IP address, and add a script for audit verification.

This commit is contained in:
Galindo97
2026-02-11 10:56:43 -06:00
parent 4b7a3a8f4a
commit a594676de7
10 changed files with 300 additions and 60 deletions

View File

@@ -40,8 +40,8 @@ class AuditService:
"""
Low-level creation of an Audit Log entry
"""
# Timezone handling set to Mexico City as requested implicitly by legacy format example
tz = pytz.timezone('America/Mexico_City')
# Timezone handling set to Hermosillo (Sonora) to match user preference (-1h vs CDMX)
tz = pytz.timezone('America/Hermosillo')
now = datetime.now(tz)
log = AuditLog(
@@ -120,6 +120,9 @@ class AuditService:
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)
@@ -150,7 +153,30 @@ class AuditService:
)
@staticmethod
def log_login(db: Session, username: str, ip_address: str = None):
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 set to Hermosillo (Sonora) to match user preference (-1h vs CDMX)
tz = pytz.timezone('America/Hermosillo')
now = datetime.now(tz)
five_seconds_ago = now - datetime.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:
print(f"[AUDIT DEBUG] Duplicate login skipped for {username} within 5s")
return existing
except Exception as e:
print(f"[AUDIT WARNING] Failed to check duplicate login: {e}")
return AuditService.create_audit_log(
db=db,
reference="LOGIN",
@@ -158,5 +184,27 @@ class AuditService:
movement="SYSTEM LOGIN",
username=username,
operation_type="LOGIN",
ip_address=ip_address
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
print(f"Error logging logout: {e}")