""" Authentication Blueprint Handles login, logout, and session management """ from flask import Blueprint, render_template, request, jsonify, session, redirect, url_for from werkzeug.security import check_password_hash from functools import wraps import json import os auth_bp = Blueprint('auth', __name__) # Path to users file USERS_FILE = os.path.join(os.path.dirname(os.path.dirname(__file__)), 'data', 'users.json') def load_users(): """Load users from JSON file""" if os.path.exists(USERS_FILE): try: with open(USERS_FILE, 'r') as f: return json.load(f) except: return [] return [] def login_required(f): """Decorator to require login for routes""" @wraps(f) def decorated_function(*args, **kwargs): if 'user_id' not in session: return redirect(url_for('auth.login_page')) # Check if user is active users = load_users() if session['user_id'] >= len(users): session.clear() return redirect(url_for('auth.login_page', message='User not found')) user = users[session['user_id']] if not user.get('active', True): session.clear() return redirect(url_for('auth.login_page', message='Account is inactive')) return f(*args, **kwargs) return decorated_function def get_current_user(): """Get the currently logged in user""" if 'user_id' not in session: return None users = load_users() if session['user_id'] >= len(users): return None return users[session['user_id']] def can_user(permission, location=None): """ Check if current user has a specific permission. Args: permission (str): Permission name (e.g., 'manage_users') location (str, optional): Check permission for specific location Returns: bool: True if user has permission """ user = get_current_user() if not user: return False if not user.get('active', True): return False # Check global permissions global_permissions = user.get('permissions', {}) if permission in global_permissions: return global_permissions[permission] is True if location == 'global': return False # Check location-specific permissions check_location = location if location else session.get('currentLocation') if not check_location: return False location_settings = user.get('locationSettings', {}) if check_location in location_settings: loc_permissions = location_settings[check_location].get('permissions', {}) if permission in loc_permissions: return loc_permissions[permission] is True return False # Routes @auth_bp.route('/login') def login_page(): """Serve the login page""" if 'user_id' in session: return redirect(url_for('home')) return render_template('login.html') @auth_bp.route('/select-location') def select_location_page(): """Serve the location selection page""" if 'user_id' not in session: return redirect(url_for('auth.login_page')) return render_template('select_location.html') @auth_bp.route('/api/login', methods=['POST']) def login(): """Authenticate user and create session""" try: data = request.json username = data.get('username') password = data.get('password') if not username or not password: return jsonify({ 'status': 'error', 'message': 'Username and password are required' }), 400 users = load_users() # Find user by username user_index = None user = None for i, u in enumerate(users): if u['username'] == username: user_index = i user = u break if not user: return jsonify({ 'status': 'error', 'message': 'Invalid username or password' }), 401 # Check if user is active if not user.get('active', True): return jsonify({ 'status': 'error', 'message': 'Account is inactive. Please contact an administrator.' }), 403 # Verify password if not check_password_hash(user['password'], password): return jsonify({ 'status': 'error', 'message': 'Invalid username or password' }), 401 # Create session session['user_id'] = user_index session['username'] = user['username'] session['defaultLocation'] = user.get('defaultLocation') # Get accessible locations accessible_locations = [] if 'locationSettings' in user: for loc_code, settings in user['locationSettings'].items(): if settings.get('accessible', False): accessible_locations.append(loc_code) session['accessibleLocations'] = accessible_locations # Check if user needs to select a location requiresLocationSelection = len(accessible_locations) > 1 # If only one location or no accessible locations, auto-select default if not requiresLocationSelection: session['currentLocation'] = user.get('defaultLocation') return jsonify({ 'status': 'success', 'message': 'Login successful', 'requiresLocationSelection': requiresLocationSelection }) except Exception as e: return jsonify({ 'status': 'error', 'message': str(e) }), 500 @auth_bp.route('/api/select-location', methods=['POST']) def select_location(): """Select a location for the current session""" if 'user_id' not in session: return jsonify({ 'status': 'error', 'message': 'Not logged in' }), 401 try: data = request.json location = data.get('location') # Verify user has access to this location accessible = session.get('accessibleLocations', []) if location not in accessible: return jsonify({ 'status': 'error', 'message': 'You do not have access to this location' }), 403 session['currentLocation'] = location return jsonify({ 'status': 'success', 'message': 'Location selected', 'currentLocation': location }) except Exception as e: return jsonify({ 'status': 'error', 'message': str(e) }), 500 @auth_bp.route('/logout') def logout(): """Log out user and clear session""" session.clear() return redirect(url_for('auth.login_page')) @auth_bp.route('/api/session', methods=['GET']) def get_session(): """Get current session information""" if 'user_id' not in session: return jsonify({ 'status': 'error', 'message': 'Not logged in' }), 401 # Get user permissions user = get_current_user() permissions = {} if user: permissions = user.get('permissions', {}) return jsonify({ 'status': 'success', 'username': session.get('username'), 'defaultLocation': session.get('defaultLocation'), 'currentLocation': session.get('currentLocation'), 'accessibleLocations': session.get('accessibleLocations', []), 'permissions': permissions })