Pre-github
This commit is contained in:
Jason
@@ -0,0 +1,253 @@
|
||||
"""
|
||||
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
|
||||
})
|
||||
Reference in New Issue
Block a user