149 lines
No EOL
4.2 KiB
Python
Executable file
149 lines
No EOL
4.2 KiB
Python
Executable file
"""
|
|
Authentication middleware for Quart routes
|
|
"""
|
|
|
|
import logging
|
|
from functools import wraps
|
|
from typing import Optional, Dict, Any, Callable
|
|
from quart import request, jsonify, g
|
|
|
|
from .msal_auth import msal_auth
|
|
from ..config_runtime import server_config
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
async def extract_user_from_request() -> Optional[Dict[str, Any]]:
|
|
"""
|
|
Extract user information from request headers
|
|
|
|
Returns:
|
|
User information if authenticated, None otherwise
|
|
"""
|
|
# Check for Authorization header
|
|
auth_header = request.headers.get('Authorization')
|
|
if not auth_header:
|
|
return None
|
|
|
|
# Extract token from "Bearer <token>" format
|
|
try:
|
|
scheme, token = auth_header.split(' ', 1)
|
|
if scheme.lower() != 'bearer':
|
|
return None
|
|
except ValueError:
|
|
return None
|
|
|
|
# Validate token using MSAL
|
|
user_info = await msal_auth.validate_token(token)
|
|
return user_info
|
|
|
|
def auth_required(f: Callable) -> Callable:
|
|
"""
|
|
Decorator to require authentication for a route
|
|
Bypassed in development mode
|
|
|
|
Usage:
|
|
@app.route('/api/protected')
|
|
@auth_required
|
|
async def protected_route():
|
|
user = g.current_user
|
|
return jsonify({'message': f'Hello {user["name"]}'})
|
|
"""
|
|
@wraps(f)
|
|
async def decorated_function(*args, **kwargs):
|
|
# Get user information
|
|
user_info = await extract_user_from_request()
|
|
|
|
if user_info:
|
|
# Store user in request context
|
|
g.current_user = user_info
|
|
return await f(*args, **kwargs)
|
|
else:
|
|
# Return 401 Unauthorized
|
|
return jsonify({
|
|
'error': 'unauthorized',
|
|
'message': 'Valid authentication required'
|
|
}), 401
|
|
|
|
return decorated_function
|
|
|
|
def dev_mode_bypass(f: Callable) -> Callable:
|
|
"""
|
|
Decorator that creates a mock user in development mode
|
|
Use this for routes that need user context but should work in dev mode
|
|
|
|
Usage:
|
|
@app.route('/api/user-specific')
|
|
@dev_mode_bypass
|
|
async def user_route():
|
|
user = g.current_user
|
|
return jsonify({'user_id': user['oid']})
|
|
"""
|
|
@wraps(f)
|
|
async def decorated_function(*args, **kwargs):
|
|
if server_config.DEV_MODE:
|
|
# Create mock user for dev mode
|
|
g.current_user = {
|
|
'oid': 'dev-user-id',
|
|
'preferred_username': 'dev@localhost',
|
|
'name': 'Development User',
|
|
'roles': ['user']
|
|
}
|
|
else:
|
|
# Use normal authentication
|
|
user_info = await extract_user_from_request()
|
|
if user_info:
|
|
g.current_user = user_info
|
|
else:
|
|
return jsonify({
|
|
'error': 'unauthorized',
|
|
'message': 'Valid authentication required'
|
|
}), 401
|
|
|
|
return await f(*args, **kwargs)
|
|
|
|
return decorated_function
|
|
|
|
def optional_auth(f: Callable) -> Callable:
|
|
"""
|
|
Decorator that extracts user info if present but doesn't require it
|
|
Sets g.current_user to None if not authenticated
|
|
|
|
Usage:
|
|
@app.route('/api/maybe-protected')
|
|
@optional_auth
|
|
async def maybe_protected():
|
|
if g.current_user:
|
|
return jsonify({'message': f'Hello {g.current_user["name"]}'})
|
|
else:
|
|
return jsonify({'message': 'Hello anonymous user'})
|
|
"""
|
|
@wraps(f)
|
|
async def decorated_function(*args, **kwargs):
|
|
# Try to get user information
|
|
user_info = await extract_user_from_request()
|
|
g.current_user = user_info
|
|
|
|
return await f(*args, **kwargs)
|
|
|
|
return decorated_function
|
|
|
|
async def get_current_user() -> Optional[Dict[str, Any]]:
|
|
"""
|
|
Get current user from request context
|
|
|
|
Returns:
|
|
Current user information or None
|
|
"""
|
|
return getattr(g, 'current_user', None)
|
|
|
|
def get_user_id() -> str:
|
|
"""
|
|
Get current user ID from request context
|
|
|
|
Returns:
|
|
User ID or 'anonymous' if not authenticated
|
|
"""
|
|
user = getattr(g, 'current_user', None)
|
|
if user:
|
|
return user.get('oid', 'unknown-user')
|
|
return 'anonymous' |