from flask import render_template, request, session, redirect, url_for, flash, current_app, jsonify from .permissions import APP_PERMISSIONS, ROLE_HIERARCHY, ACTIONS, get_all_permissions, get_default_permissions_for_role from .db_pool import get_db_connection from .logging_config import get_logger import mariadb import os import json from contextlib import contextmanager logger = get_logger('settings') # Global permission cache to avoid repeated database queries _permission_cache = {} @contextmanager def db_connection_context(): """ Context manager for database connections from the pool. Ensures connections are properly closed and committed/rolled back. """ logger.debug("Acquiring database connection from pool (settings)") conn = get_db_connection() try: logger.debug("Database connection acquired successfully") yield conn except Exception as e: logger.error(f"Error in settings database operation: {e}", exc_info=True) conn.rollback() raise e finally: if conn: logger.debug("Closing database connection (settings)") conn.close() def check_permission(permission_key, user_role=None): """ Check if the current user (or specified role) has a specific permission. Args: permission_key (str): The permission key like 'settings.user_management.create' user_role (str, optional): Role to check. If None, uses current session role. Returns: bool: True if user has the permission, False otherwise """ logger.debug(f"Checking permission '{permission_key}' for role '{user_role or session.get('role')}'") if user_role is None: user_role = session.get('role') if not user_role: logger.warning(f"Cannot check permission - no role provided") return False # Superadmin always has all permissions if user_role == 'superadmin': logger.debug(f"Superadmin bypass - permission '{permission_key}' granted") return True # Check cache first cache_key = f"{user_role}:{permission_key}" if cache_key in _permission_cache: logger.debug(f"Permission '{permission_key}' found in cache: {_permission_cache[cache_key]}") return _permission_cache[cache_key] try: logger.debug(f"Checking permission '{permission_key}' for role '{user_role}' in database") with db_connection_context() as conn: cursor = conn.cursor() cursor.execute(""" SELECT granted FROM role_permissions WHERE role = %s AND permission_key = %s """, (user_role, permission_key)) result = cursor.fetchone() # Cache the result has_permission = bool(result and result[0]) _permission_cache[cache_key] = has_permission logger.info(f"Permission '{permission_key}' for role '{user_role}': {has_permission}") return has_permission except Exception as e: logger.error(f"Error checking permission {permission_key} for role {user_role}: {e}", exc_info=True) return False def clear_permission_cache(): """Clear the permission cache (call after permission updates)""" global _permission_cache _permission_cache = {} def require_permission(permission_key): """ Decorator to require a specific permission for a route. Usage: @require_permission('settings.user_management.create') def create_user(): ... """ def decorator(f): from functools import wraps @wraps(f) def decorated_function(*args, **kwargs): if not check_permission(permission_key): flash(f'Access denied: You do not have permission to {permission_key}') return redirect(url_for('main.dashboard')) return f(*args, **kwargs) return decorated_function return decorator def get_user_permissions(user_role): """Get all permissions for a specific role""" if user_role == 'superadmin': return [p['key'] for p in get_all_permissions()] try: conn = get_external_db_connection() cursor = conn.cursor() cursor.execute(""" SELECT permission_key FROM role_permissions WHERE role = %s AND granted = TRUE """, (user_role,)) result = cursor.fetchall() conn.close() return [row[0] for row in result] except Exception as e: print(f"Error getting permissions for role {user_role}: {e}") return [] # Settings module logic # Helper to check if current user is superadmin def is_superadmin(): return session.get('role') == 'superadmin' # Route handler for role permissions management def role_permissions_handler(): if not is_superadmin(): flash('Access denied: Superadmin only.') return redirect(url_for('main.dashboard')) try: # Get roles and their current permissions conn = get_external_db_connection() cursor = conn.cursor() # Get roles from role_hierarchy table cursor.execute("SELECT role_name, display_name, description, level FROM role_hierarchy ORDER BY level DESC") role_data = cursor.fetchall() roles = {} for role_name, display_name, description, level in role_data: roles[role_name] = { 'display_name': display_name, 'description': description, 'level': level } # Get current role permissions cursor.execute(""" SELECT role, permission_key FROM role_permissions WHERE granted = TRUE """) permission_data = cursor.fetchall() role_permissions = {} for role, permission_key in permission_data: if role not in role_permissions: role_permissions[role] = [] role_permissions[role].append(permission_key) conn.close() # Convert to JSON for JavaScript permissions_json = json.dumps(get_all_permissions()) role_permissions_json = json.dumps(role_permissions) return render_template('role_permissions.html', roles=roles, pages=APP_PERMISSIONS, action_names=ACTIONS, permissions_json=permissions_json, role_permissions_json=role_permissions_json) except Exception as e: flash(f'Error loading role permissions: {e}') return redirect(url_for('main.settings')) def settings_handler(): if 'role' not in session or session['role'] not in ['superadmin', 'admin']: flash('Access denied: Admin or Superadmin required.') return redirect(url_for('main.dashboard')) # Get users from external MariaDB database users = [] try: conn = get_external_db_connection() cursor = conn.cursor() # Create users table if it doesn't exist cursor.execute(''' CREATE TABLE IF NOT EXISTS users ( id INT AUTO_INCREMENT PRIMARY KEY, username VARCHAR(50) UNIQUE NOT NULL, password VARCHAR(255) NOT NULL, role VARCHAR(50) NOT NULL, email VARCHAR(255) ) ''') # Get all users from external database cursor.execute("SELECT id, username, password, role, modules FROM users") users_data = cursor.fetchall() # Convert to list of dictionaries for template compatibility users = [] for user_data in users_data: users.append({ 'id': user_data[0], 'username': user_data[1], 'password': user_data[2], 'role': user_data[3], 'modules': user_data[4] if len(user_data) > 4 else None }) conn.close() except Exception as e: print(f"Error fetching users from external database: {e}") flash(f'Error loading users: {e}') # Load external database settings from the instance folder external_settings = {} settings_file = os.path.join(current_app.instance_path, 'external_server.conf') if os.path.exists(settings_file): with open(settings_file, 'r') as f: for line in f: line = line.strip() # Skip empty lines and comments if not line or line.startswith('#'): continue if '=' in line: key, value = line.split('=', 1) external_settings[key] = value return render_template('settings.html', users=users, external_settings=external_settings, current_user={'role': session.get('role', '')}) # Helper function to get external database connection def get_external_db_connection(): """ DEPRECATED: Use get_db_connection() from db_pool.py instead. This function is kept for backward compatibility. Returns a connection from the managed connection pool. """ return get_db_connection() # User management handlers def save_role_permissions_handler(): """Save role permissions via AJAX""" if not is_superadmin(): return jsonify({'success': False, 'error': 'Access denied: Superadmin only.'}) try: data = request.get_json() role = data.get('role') permissions = data.get('permissions', []) if not role: return jsonify({'success': False, 'error': 'Role is required'}) conn = get_external_db_connection() cursor = conn.cursor() # Clear existing permissions for this role cursor.execute("DELETE FROM role_permissions WHERE role = %s", (role,)) # Add new permissions current_user = session.get('username', 'system') for permission_key in permissions: cursor.execute(""" INSERT INTO role_permissions (role, permission_key, granted, granted_by) VALUES (%s, %s, TRUE, %s) """, (role, permission_key, current_user)) # Log the change cursor.execute(""" INSERT INTO permission_audit_log (role, permission_key, action, changed_by, reason) VALUES (%s, %s, 'bulk_update', %s, %s) """, (role, f"Updated {len(permissions)} permissions", current_user, f"Bulk update via UI")) conn.commit() conn.close() # Clear permission cache since permissions changed clear_permission_cache() return jsonify({'success': True, 'message': f'Saved {len(permissions)} permissions for {role}'}) except Exception as e: return jsonify({'success': False, 'error': str(e)}) def reset_role_permissions_handler(): """Reset role permissions to defaults""" if not is_superadmin(): return jsonify({'success': False, 'error': 'Access denied: Superadmin only.'}) try: data = request.get_json() role = data.get('role') if not role: return jsonify({'success': False, 'error': 'Role is required'}) # Get default permissions for the role default_permissions = get_default_permissions_for_role(role) conn = get_external_db_connection() cursor = conn.cursor() # Clear existing permissions for this role cursor.execute("DELETE FROM role_permissions WHERE role = %s", (role,)) # Add default permissions current_user = session.get('username', 'system') for permission_key in default_permissions: cursor.execute(""" INSERT INTO role_permissions (role, permission_key, granted, granted_by) VALUES (%s, %s, TRUE, %s) """, (role, permission_key, current_user)) # Log the change cursor.execute(""" INSERT INTO permission_audit_log (role, permission_key, action, changed_by, reason) VALUES (%s, %s, 'reset_defaults', %s, %s) """, (role, f"Reset {len(default_permissions)} permissions", current_user, "Reset to default permissions")) conn.commit() conn.close() # Clear permission cache since permissions changed clear_permission_cache() return jsonify({ 'success': True, 'permissions': default_permissions, 'message': f'Reset {len(default_permissions)} permissions for {role} to defaults' }) except Exception as e: return jsonify({'success': False, 'error': str(e)}) def save_all_role_permissions_handler(): """Save all role permissions at once""" if not is_superadmin(): return jsonify({'success': False, 'error': 'Access denied: Superadmin only.'}) try: data = request.get_json() permissions_data = data.get('permissions', {}) if not permissions_data: return jsonify({'success': False, 'error': 'No permissions data provided'}) conn = get_external_db_connection() cursor = conn.cursor() current_user = session.get('username', 'system') total_updated = 0 # Process each role's permissions for role, role_permissions in permissions_data.items(): # Clear existing permissions for this role cursor.execute("DELETE FROM role_permissions WHERE role = %s", (role,)) # Convert nested permissions to flat permission keys permission_keys = [] for page_key, page_perms in role_permissions.items(): for section_key, actions in page_perms.items(): for action in actions: permission_key = f"{page_key}.{section_key}.{action}" permission_keys.append(permission_key) # Insert new permissions for permission_key in permission_keys: cursor.execute(""" INSERT INTO role_permissions (role, permission_key, granted, granted_by) VALUES (%s, %s, TRUE, %s) """, (role, permission_key, current_user)) total_updated += 1 # Log the change cursor.execute(""" INSERT INTO permission_audit_log (role, permission_key, action, changed_by, reason) VALUES (%s, %s, 'bulk_update', %s, %s) """, (role, f"Updated {len(permission_keys)} permissions", current_user, "Bulk permission update")) conn.commit() conn.close() # Clear permission cache since permissions changed clear_permission_cache() return jsonify({ 'success': True, 'message': f'Successfully updated {total_updated} permissions across {len(permissions_data)} roles' }) except Exception as e: return jsonify({'success': False, 'error': str(e)}) def reset_all_role_permissions_handler(): """Reset all role permissions to defaults""" if not is_superadmin(): return jsonify({'success': False, 'error': 'Access denied: Superadmin only.'}) try: # Get all roles conn = get_external_db_connection() cursor = conn.cursor() cursor.execute("SELECT role_name FROM role_hierarchy") roles = [row[0] for row in cursor.fetchall()] current_user = session.get('username', 'system') total_reset = 0 # Reset each role to defaults for role in roles: # Clear existing permissions cursor.execute("DELETE FROM role_permissions WHERE role = %s", (role,)) # Get default permissions for the role default_permissions = get_default_permissions_for_role(role) # Add default permissions for permission_key in default_permissions: cursor.execute(""" INSERT INTO role_permissions (role, permission_key, granted, granted_by) VALUES (%s, %s, TRUE, %s) """, (role, permission_key, current_user)) total_reset += 1 # Log the change cursor.execute(""" INSERT INTO permission_audit_log (role, permission_key, action, changed_by, reason) VALUES (%s, %s, 'reset_all_defaults', %s, %s) """, (role, f"Reset {len(default_permissions)} permissions", current_user, "Reset all to default permissions")) conn.commit() conn.close() # Clear permission cache since permissions changed clear_permission_cache() return jsonify({ 'success': True, 'message': f'Successfully reset {total_reset} permissions across {len(roles)} roles to defaults' }) except Exception as e: return jsonify({'success': False, 'error': str(e)})