147 lines
4.7 KiB
Python
147 lines
4.7 KiB
Python
"""
|
|
Authentication utilities for public web frontend.
|
|
|
|
Provides authentication checking and decorators for protected routes.
|
|
Uses API backend for session validation.
|
|
"""
|
|
|
|
from functools import wraps
|
|
from flask import session, redirect, url_for, request, flash
|
|
from .logging import get_logger
|
|
|
|
|
|
logger = get_logger(__file__)
|
|
|
|
# Track last API validation time per session to avoid excessive checks
|
|
_SESSION_VALIDATION_KEY = '_api_validated_at'
|
|
|
|
|
|
def get_current_user():
|
|
"""
|
|
Get the currently authenticated user from session.
|
|
|
|
Returns:
|
|
Dictionary with user data if authenticated, None otherwise.
|
|
"""
|
|
# Check if we have user in Flask session
|
|
if 'user' in session and session.get('user'):
|
|
return session['user']
|
|
|
|
return None
|
|
|
|
|
|
def require_auth_web(f):
|
|
"""
|
|
Decorator to require authentication for web routes.
|
|
|
|
Validates the session with the API backend and redirects to
|
|
login if not authenticated.
|
|
|
|
Args:
|
|
f: Flask route function
|
|
|
|
Returns:
|
|
Wrapped function that checks authentication
|
|
"""
|
|
@wraps(f)
|
|
def decorated_function(*args, **kwargs):
|
|
user = get_current_user()
|
|
if user is None:
|
|
logger.info("Unauthenticated access attempt", path=request.path)
|
|
# Store the intended destination
|
|
session['next'] = request.url
|
|
return redirect(url_for('auth_views.login'))
|
|
return f(*args, **kwargs)
|
|
return decorated_function
|
|
|
|
|
|
def clear_user_session():
|
|
"""
|
|
Clear user session data.
|
|
|
|
Should be called after logout.
|
|
"""
|
|
session.pop('user', None)
|
|
session.pop('next', None)
|
|
session.pop('api_session_cookie', None)
|
|
session.pop(_SESSION_VALIDATION_KEY, None)
|
|
logger.debug("User session cleared")
|
|
|
|
|
|
def require_auth_strict(revalidate_interval: int = 300):
|
|
"""
|
|
Decorator to require authentication with API session validation.
|
|
|
|
This decorator validates the session with the API backend periodically
|
|
to ensure the session is still valid on the server side.
|
|
|
|
Args:
|
|
revalidate_interval: Seconds between API validation checks (default 5 minutes).
|
|
Set to 0 to validate on every request.
|
|
|
|
Returns:
|
|
Decorator function.
|
|
|
|
Usage:
|
|
@app.route('/protected')
|
|
@require_auth_strict() # Validates every 5 minutes
|
|
def protected_route():
|
|
pass
|
|
|
|
@app.route('/sensitive')
|
|
@require_auth_strict(revalidate_interval=0) # Validates every request
|
|
def sensitive_route():
|
|
pass
|
|
"""
|
|
def decorator(f):
|
|
@wraps(f)
|
|
def decorated_function(*args, **kwargs):
|
|
import time
|
|
from .api_client import get_api_client, APIAuthenticationError, APIError
|
|
|
|
user = get_current_user()
|
|
if user is None:
|
|
logger.info("Unauthenticated access attempt", path=request.path)
|
|
session['next'] = request.url
|
|
return redirect(url_for('auth_views.login'))
|
|
|
|
# Check if we need to revalidate with API
|
|
current_time = time.time()
|
|
last_validated = session.get(_SESSION_VALIDATION_KEY, 0)
|
|
|
|
if revalidate_interval == 0 or (current_time - last_validated) > revalidate_interval:
|
|
try:
|
|
# Validate session by hitting a lightweight endpoint
|
|
api_client = get_api_client()
|
|
api_client.get("/api/v1/auth/me")
|
|
|
|
# Update validation timestamp
|
|
session[_SESSION_VALIDATION_KEY] = current_time
|
|
session.modified = True
|
|
logger.debug("API session validated", user_id=user.get('id'))
|
|
|
|
except APIAuthenticationError:
|
|
# Session expired on server side
|
|
logger.warning(
|
|
"API session expired",
|
|
user_id=user.get('id'),
|
|
path=request.path
|
|
)
|
|
clear_user_session()
|
|
flash('Your session has expired. Please log in again.', 'warning')
|
|
session['next'] = request.url
|
|
return redirect(url_for('auth_views.login'))
|
|
|
|
except APIError as e:
|
|
# API error - log but allow through (fail open for availability)
|
|
logger.error(
|
|
"API validation error",
|
|
user_id=user.get('id'),
|
|
error=str(e)
|
|
)
|
|
# Don't block the user, but don't update validation timestamp
|
|
|
|
return f(*args, **kwargs)
|
|
return decorated_function
|
|
return decorator
|