# Archivo: UserAccess.py # Ruta: src\App\Security\UserAccess.py # Lenguaje: Python con Flask # Versión: 2.0 - Mejorada con logging, cache y validaciones robustas import logging import traceback from datetime import datetime, timedelta from functools import lru_cache from flask import request, render_template, redirect, url_for, jsonify, session from Consultas_SQL.Security.UserPasswordSQL import ( get_ModuleID, get_RoleID, get_permiso, get_CompanyID_DivisionID_DepartamentID_Profile ) import os class AccessController: """Controlador de acceso con logging profesional y cache""" # Divisiones que NO deben tener acceso amplio (solo acceso específico) RESTRICTED_DIVISIONS = ['CLIE', 'EXTERNOS', 'PROVEEDORES'] def __init__(self): self.env = os.getenv('FLASK_ENV', 'development') self.logger = self._setup_logger() self.cache_duration = 300 if self.env == 'production' else 60 # 5min prod, 1min dev def _setup_logger(self): """Configura el sistema de logging según el entorno (UTF-8 seguro y compatible con Windows, VSCode y Flask)""" import os import logging import re import sys logger = logging.getLogger('user_access') # Evitar duplicar handlers si ya existen if logger.handlers: return logger # Nivel de logging según entorno level = logging.INFO if self.env == 'production' else logging.DEBUG logger.setLevel(level) # Formato general formatter = logging.Formatter( '%(asctime)s - ACCESS_CONTROL - %(levelname)s - %(message)s' ) # === HANDLER PARA ARCHIVO (UTF-8) === try: log_dir = os.path.join(os.path.dirname(__file__), '..', '..', 'logs') os.makedirs(log_dir, exist_ok=True) log_file = os.path.join(log_dir, 'access_control.log') file_handler = logging.FileHandler(log_file, encoding='utf-8') # ✅ aquí está el fix real file_handler.setFormatter(formatter) logger.addHandler(file_handler) except Exception as e: print(f"No se pudo crear el archivo de log: {e}") # === HANDLER PARA CONSOLA === console_handler = logging.StreamHandler(stream=sys.stdout) console_handler.setFormatter(formatter) # Filtro opcional para limpiar emojis si la consola no soporta UTF-8 class NoEmojiFilter(logging.Filter): def filter(self, record): # Remueve caracteres fuera de rango ASCII extendido record.msg = re.sub(r'[^\x00-\x7F]+', '', str(record.msg)) return True # Si estás en Windows modo desarrollo, aplica el filtro if os.name == 'nt' and self.env == 'development': console_handler.addFilter(NoEmojiFilter()) logger.addHandler(console_handler) return logger def _validate_database_integrity(self, user_id, module_data, profile_data): """Valida la integridad de los datos de la base de datos""" errors = [] warnings = [] # Validar datos del módulo if not module_data: errors.append("Datos del módulo no encontrados") else: if module_data[0] is None: # ModuloID errors.append("ModuloID es NULL en la base de datos") # Validar datos del perfil if not profile_data: errors.append("Perfil de usuario no encontrado") else: if profile_data[0] is None: # CompanyID warnings.append("CompanyID del usuario es NULL") # Validar UserID if not user_id: errors.append("UserID no proporcionado") return errors, warnings def _log_access_attempt(self, user_id, ruta, step, result, data=None, error=None): """Registra intentos de acceso con información detallada""" log_data = { 'user_id': user_id, 'ruta': ruta, 'step': step, 'result': result, 'timestamp': datetime.now().isoformat(), 'env': self.env } if data: log_data['data'] = data if error: log_data['error'] = str(error) # Logging según resultado if result == 'SUCCESS': file_msg = f"✅ ACCESO CONCEDIDO | User: {user_id} | Ruta: {ruta} | Método: {step}" console_msg = f"ACCESO CONCEDIDO | User: {user_id} | Ruta: {ruta} | Método: {step}" self.logger.info(file_msg) # Con emoji para archivo if self.env == 'development' and data: debug_msg = f" Datos utilizados: {data}" self.logger.debug(debug_msg) elif result == 'DENIED': file_msg = f"🚫 ACCESO DENEGADO | User: {user_id} | Ruta: {ruta} | Paso fallido: {step}" console_msg = f"ACCESO DENEGADO | User: {user_id} | Ruta: {ruta} | Paso fallido: {step}" self.logger.warning(file_msg) # Con emoji para archivo if data: warning_msg = f" Datos del fallo: {data}" self.logger.warning(warning_msg) elif result == 'ERROR': file_msg = f"❌ ERROR EN VALIDACIÓN | User: {user_id} | Ruta: {ruta} | Paso: {step}" console_msg = f"ERROR EN VALIDACIÓN | User: {user_id} | Ruta: {ruta} | Paso: {step}" self.logger.error(file_msg) # Con emoji para archivo if error: error_msg = f" Error: {error}" self.logger.error(error_msg) @lru_cache(maxsize=100) def _get_cached_permissions(self, role_id, module_id): """Cache de permisos con TTL""" try: return get_permiso(role_id, module_id) except Exception as e: self.logger.error(f"Error obteniendo permisos cached: {e}") return None def _validate_superuser(self, user_id, ruta): """Validación 1: Superusuario""" self.logger.info(f">>> METODO 1: VALIDACION DE SUPERUSUARIO") try: role_id = get_RoleID(user_id) self.logger.info(f" Datos obtenidos: RoleID = '{role_id}'") if role_id == 'SU': self.logger.info(f" RESULTADO: ACCESO CONCEDIDO - Es SUPERUSUARIO") return True, role_id else: self.logger.info(f" RESULTADO: No es superusuario, continuando...") return False, role_id except Exception as e: self.logger.error(f" ✗ ERROR obteniendo RoleID: {e}") return False, None def _validate_index_access(self, user_id, ruta): """Validación 2: Acceso especial a index.html""" self.logger.info(f">>> METODO 2: VALIDACION ESPECIAL INDEX.HTML") self.logger.info(f" Verificando si ruta = '/index.html'") if ruta == '/index.html': self.logger.info(f" RESULTADO: ACCESO CONCEDIDO - Ruta es index.html (acceso libre)") return True else: self.logger.info(f" RESULTADO: Ruta '{ruta}' no es index.html, continuando...") return False def _validate_department_access(self, user_id, ruta, module_data, profile_data, Restricted_Access): """Validación 3: Acceso por departamento/división/compañía""" self.logger.info(f">>> METODO 3: VALIDACION POR DEPARTAMENTO") if Restricted_Access == True: self.logger.info(f" Restricted_Access = True - SALTANDO validacion departamental") self.logger.info(f" RESULTADO: Validacion departamental deshabilitada, continuando...") return False # Extraer datos del módulo ModuloID = module_data[0] if module_data[0] is not None else None M_CompanyID = module_data[1] if module_data[1] is not None else None M_DivisionID = module_data[2] if module_data[2] is not None else None M_DepartamentID = module_data[3] if module_data[3] is not None else None # Extraer datos del perfil P_CompanyID = profile_data[0] if profile_data[0] is not None else None P_DivisionID = profile_data[1] if profile_data[1] is not None else None P_DepartamentID = profile_data[2] if profile_data[2] is not None else None self.logger.info(f" DATOS DEL MODULO:") self.logger.info(f" Company: '{M_CompanyID}' | Division: '{M_DivisionID}' | Department: '{M_DepartamentID}'") self.logger.info(f" DATOS DE TU PERFIL:") self.logger.info(f" Company: '{P_CompanyID}' | Division: '{P_DivisionID}' | Department: '{P_DepartamentID}'") # Verificar si el usuario está en una división restringida if P_DivisionID in self.RESTRICTED_DIVISIONS: self.logger.info(f" DIVISION RESTRINGIDA detectada: '{P_DivisionID}'") self.logger.info(f" Divisiones restringidas: {self.RESTRICTED_DIVISIONS}") self.logger.info(f" RESULTADO: Solo se permite acceso especifico (Company+Division+Department)") # Para usuarios restringidos, solo permitir acceso específico completo if (M_CompanyID == P_CompanyID and M_DivisionID == P_DivisionID and M_DepartamentID == P_DepartamentID and M_DivisionID and M_DivisionID != '' and M_DepartamentID and M_DepartamentID != ''): self.logger.info(f" RESULTADO: ACCESO CONCEDIDO - Acceso especifico completo para division restringida") return True else: self.logger.info(f" RESULTADO: ACCESO DENEGADO - Division restringida requiere acceso especifico completo") self.logger.info(f" Company coincide: {M_CompanyID == P_CompanyID}") self.logger.info(f" Division coincide: {M_DivisionID == P_DivisionID}") self.logger.info(f" Department coincide: {M_DepartamentID == P_DepartamentID}") self.logger.info(f" Modulo tiene Division especifica: {M_DivisionID and M_DivisionID != ''}") self.logger.info(f" Modulo tiene Department especifico: {M_DepartamentID and M_DepartamentID != ''}") else: # Lógica normal para usuarios no restringidos self.logger.info(f" Division normal: '{P_DivisionID}' - Aplicando logica jerarquica normal") # Determinar nivel de validación según los campos del módulo if not M_DivisionID or M_DivisionID == '': # Solo validar Company self.logger.info(f" Modulo sin division especifica (vacio/None) - Validando solo Company") if M_CompanyID == P_CompanyID: self.logger.info(f" RESULTADO: ACCESO CONCEDIDO - Company coincide") return True else: self.logger.info(f" RESULTADO: Company no coincide") self.logger.info(f" Company coincide: {M_CompanyID == P_CompanyID}") elif not M_DepartamentID or M_DepartamentID == '': # Validar Company + Division self.logger.info(f" Modulo sin departamento especifico (vacio/None) - Validando Company + Division") if M_CompanyID == P_CompanyID and M_DivisionID == P_DivisionID: self.logger.info(f" RESULTADO: ACCESO CONCEDIDO - Company y Division coinciden") return True else: self.logger.info(f" RESULTADO: Company o Division no coinciden") self.logger.info(f" Company coincide: {M_CompanyID == P_CompanyID}") self.logger.info(f" Division coincide: {M_DivisionID == P_DivisionID}") else: # Validar Company + Division + Department self.logger.info(f" Modulo con departamento especifico - Validando Company + Division + Department") if (M_CompanyID == P_CompanyID and M_DivisionID == P_DivisionID and M_DepartamentID == P_DepartamentID): self.logger.info(f" RESULTADO: ACCESO CONCEDIDO - Company, Division y Department coinciden") return True else: self.logger.info(f" RESULTADO: No todos los valores coinciden") self.logger.info(f" Company coincide: {M_CompanyID == P_CompanyID}") self.logger.info(f" Division coincide: {M_DivisionID == P_DivisionID}") self.logger.info(f" Department coincide: {M_DepartamentID == P_DepartamentID}") self.logger.info(f" Continuando al siguiente metodo...") return False def _validate_permission_access(self, user_id, ruta, role_id, module_id): """Validación 4: Permisos específicos""" self.logger.info(f">>> METODO 4: VALIDACION POR PERMISOS ESPECIFICOS") if not role_id: self.logger.warning(f"RoleID no definido para UserID={user_id}, no se pueden validar permisos específicos.") return False try: permission = self._get_cached_permissions(role_id, module_id) self.logger.info(f" DATOS DE PERMISOS:") self.logger.info(f" Tu RoleID: '{role_id}' | ModuloID: '{module_id}'") self.logger.info(f" Valor de permiso obtenido: '{permission}'") if permission is not None: permission_level = int(permission) self.logger.info(f" Permiso convertido a entero: {permission_level}") if permission_level > 0: self.logger.info(f" RESULTADO: ACCESO CONCEDIDO - Permiso > 0") return True else: self.logger.info(f" RESULTADO: ACCESO DENEGADO - Permiso = 0 (sin acceso)") else: self.logger.info(f" RESULTADO: ACCESO DENEGADO - No se encontro permiso (NULL)") return False except Exception as e: self.logger.error(f" ✗ ERROR obteniendo permisos: {e}") return False def _validate_access_cascade(self, UserID, ruta, module_data, profile_data, Restricted_Access, ModuloID): """ Ejecuta los 4 métodos de validación en cascada. Retorna: (resultado: bool, metodo: str, role_id: str | None) """ # === MÉTODO 1: SUPERUSUARIO === is_superuser, role_id = self._validate_superuser(UserID, ruta) if is_superuser: return True, "METODO 1 - SUPERUSUARIO", role_id # === MÉTODO 2: INDEX.HTML === if self._validate_index_access(UserID, ruta): return True, "METODO 2 - INDEX.HTML", role_id # === MÉTODO 3: DEPARTAMENTO === if self._validate_department_access(UserID, ruta, module_data, profile_data, Restricted_Access): return True, "METODO 3 - DEPARTAMENTO", role_id # === MÉTODO 4: PERMISOS ESPECIFICOS === if role_id and self._validate_permission_access(UserID, ruta, role_id, ModuloID): return True, "METODO 4 - PERMISOS ESPECIFICOS", role_id # === SIN ACCESO === return False, "NINGUNO", role_id def _render_success_template(self, ruta, user_id, start_time=None): """Renderiza el template exitosamente con manejo especial para index.html""" execution_time = (datetime.now() - start_time).total_seconds() if start_time else 0 self.logger.info( f"RENDERIZANDO TEMPLATE | User: {user_id} | Ruta: {ruta} | " f"Tiempo ejecucion: {execution_time:.3f}s" ) if ruta == '/index.html': user_email = session.get('email', 'Usuario') return render_template(ruta, user_email=user_email) return render_template(ruta) def check_user_access(UserID, ruta, Restricted_Access): """ Valida el acceso del usuario a la ruta solicitada. Aplica validaciones en cascada y maneja errores de integridad. """ from Consultas_SQL.Security.UserPasswordSQL import ( get_ModuleID, get_CompanyID_DivisionID_DepartamentID_Profile ) from flask import render_template, session, redirect, url_for from datetime import datetime import traceback start_time = datetime.now() access_controller = AccessController() try: # =============================================================== # 0️⃣ VALIDACIONES DE ENTRADA (sesión y parámetros) # =============================================================== if not UserID: access_controller.logger.warning("Intento de acceso sin UserID en sesión - Redirigiendo al login") session.clear() # Limpia la sesión corrupta return redirect(url_for('login')) if not ruta: error_msg = "Ruta no proporcionada" access_controller._log_access_attempt(UserID, ruta, "INPUT_VALIDATION", "ERROR", error=error_msg) return render_template("Security/AccessDened.html", message=error_msg) access_controller.logger.info("=== INICIANDO VALIDACION DE ACCESO ===") access_controller.logger.info(f"User: {UserID} | Ruta: {ruta} | Restricted: {Restricted_Access}") access_controller.logger.info("=== VALIDANDO 4 METODOS EN CASCADA ===") # =============================================================== # 1️⃣ OBTENER DATOS DE LA BASE DE DATOS # =============================================================== # Validar que el módulo exista try: module_data = get_ModuleID(ruta) except LookupError: execution_time = (datetime.now() - start_time).total_seconds() error_msg = f"No se encontró el módulo en la BD: {ruta}" access_controller._log_access_attempt(UserID, ruta, "DB_INTEGRITY", "ERROR", error=error_msg) access_controller.logger.error( f"ERROR EN VALIDACIÓN | User: {UserID} | Ruta: {ruta} | " f"Tiempo ejecución: {execution_time:.3f}s | {error_msg}" ) return render_template("Security/AccessDened.html", message="El módulo no está registrado en la base de datos.") # Obtener datos del perfil del usuario profile_data = get_CompanyID_DivisionID_DepartamentID_Profile(UserID) # Validar integridad entre módulo y perfil errors, warnings = access_controller._validate_database_integrity(UserID, module_data, profile_data) # Registrar advertencias (no bloquean acceso) for warning in warnings: access_controller.logger.warning(f"WARNING: {warning}") # Si hay errores críticos de integridad if errors: execution_time = (datetime.now() - start_time).total_seconds() error_msg = f"Errores de integridad en BD: {'; '.join(errors)}" access_controller._log_access_attempt(UserID, ruta, "DB_INTEGRITY", "ERROR", error=error_msg) access_controller.logger.error( f"ERROR EN VALIDACIÓN | User: {UserID} | Ruta: {ruta} | " f"Tiempo ejecución: {execution_time:.3f}s | {error_msg}" ) return render_template("Security/AccessDened.html", message="Error al validar datos de usuario.") # Extraer ModuloID para siguientes pasos ModuloID = module_data[0] if module_data and module_data[0] is not None else None # =============================================================== # 2️⃣ VALIDAR ACCESO EN CASCADA # =============================================================== resultado, metodo, role_id = access_controller._validate_access_cascade( UserID, ruta, module_data, profile_data, Restricted_Access, ModuloID ) # =============================================================== # 3️⃣ REGISTRAR RESULTADO FINAL # =============================================================== execution_time = (datetime.now() - start_time).total_seconds() if resultado: access_controller.logger.info( f"ACCESO FINAL CONCEDIDO POR: {metodo} | " f"User: {UserID} | RoleID={role_id or 'N/A'} | ModuloID={ModuloID or 'N/A'} | " f"Ruta: {ruta} | Tiempo: {execution_time:.3f}s" ) return access_controller._render_success_template(ruta, UserID, start_time) else: # ---- Contexto de comparación detallado ---- if profile_data: user_cmp = ( f"CompanyID={profile_data[0]} | DivisionID={profile_data[1]} | " f"DepartamentID={profile_data[2]} | RoleID={role_id or 'N/A'}" ) else: user_cmp = "Perfil de usuario no disponible" if module_data: module_cmp = ( f"ModuloID={module_data[0]} | CompanyID={module_data[1]} | " f"DivisionID={module_data[2]} | DepartamentID={module_data[3]}" ) else: module_cmp = "Datos del módulo no disponibles" access_controller.logger.warning( f"ACCESO FINAL DENEGADO - TODOS LOS 4 MÉTODOS FALLARON | " f"User: {UserID} | RoleID={role_id or 'N/A'} | Ruta: {ruta} | Tiempo: {execution_time:.3f}s" ) access_controller.logger.warning(f" Comparación → Usuario: {user_cmp} | Módulo: {module_cmp}") access_controller._log_access_attempt( UserID, ruta, "ACCESS_DENIED", "ERROR", error=( f"Diferencias detectadas entre perfil y módulo " f"(Usuario: {user_cmp} vs Módulo: {module_cmp})" ) ) return render_template( "Security/AccessDened.html", message="No tienes permisos para acceder a este módulo." ) except Exception as e: # =============================================================== # 4️⃣ MANEJO DE ERRORES CRÍTICOS # =============================================================== execution_time = (datetime.now() - start_time).total_seconds() access_controller.logger.error( f"ERROR CRÍTICO | User: {UserID} | Ruta: {ruta} | " f"Tiempo ejecución: {execution_time:.3f}s | Error: {str(e)}" ) if access_controller.env == 'development': access_controller.logger.error(f"Traceback completo:\n{traceback.format_exc()}") access_controller._log_access_attempt(UserID, ruta, "GENERAL", "ERROR", error=str(e)) return render_template("Security/AccessDened.html", message=f"Error al verificar acceso: {str(e)}")