from functools import wraps from flask import request, jsonify, current_app, session from flask_jwt_extended import verify_jwt_in_request, get_jwt_identity, get_current_user from models import User from logging_utils import log_event def jwt_required_with_logging(f): """JWT required decorator with logging""" @wraps(f) def decorated_function(*args, **kwargs): try: verify_jwt_in_request() user_id = get_jwt_identity() if not user_id: # If session exists, force re-login for browser users accept = request.headers.get('Accept', '') if 'text/html' in accept and 'user_id' in session and 'access_token' in session: from flask import redirect, url_for, flash flash('Session invalid or expired. Please log in again.', 'warning') return redirect(url_for('web.login')) raise Exception('No user_id in JWT') # JWT token is valid, proceed with function return f(*args, **kwargs) except Exception as jwt_ex: # Check if this is a werkzeug HTTP exception (like 404, 403, etc.) # These should be re-raised, not converted to 401 from werkzeug.exceptions import HTTPException if isinstance(jwt_ex, HTTPException): raise jwt_ex # Log the specific JWT error type for debugging import traceback error_details = f"{type(jwt_ex).__name__}: {str(jwt_ex)}" log_event( severity=7, # DEBUG message=f"JWT verification details for {request.method} {request.path}: {error_details}", event_type="jwt_verification_debug", source_ip=request.remote_addr ) # Fallback: allow session-based access for browser (HTML) requests accept = request.headers.get('Accept', '') if 'text/html' in accept and 'user_id' in session and 'access_token' in session: user_id = session['user_id'] request.jwt_user_id = user_id log_event( severity=6, # INFO message=f"Session-based API access: {request.method} {request.path}", event_type="session_api_access", user_id=user_id, source_ip=request.remote_addr ) import flask_jwt_extended original_get_identity = flask_jwt_extended.get_jwt_identity def session_get_identity(): return str(user_id) flask_jwt_extended.get_jwt_identity = session_get_identity try: result = f(*args, **kwargs) flask_jwt_extended.get_jwt_identity = original_get_identity return result except Exception as inner_e: flask_jwt_extended.get_jwt_identity = original_get_identity # Re-raise HTTP exceptions if isinstance(inner_e, HTTPException): raise inner_e raise inner_e # Otherwise, fail as before explanation = 'Authentication failed: JWT expired, invalid, or missing user identity. Please log in again.' log_event( severity=3, # ERROR message=f"Authentication failed for {request.method} {request.path}: {str(jwt_ex)}", event_type="auth_failed", source_ip=request.remote_addr ) if 'text/html' in accept: from flask import redirect, url_for, flash flash(explanation, 'warning') return redirect(url_for('web.login')) else: return jsonify({'error': 'unauthorized', 'explanation': explanation}), 401 return decorated_function def optional_jwt_with_logging(f): """Optional JWT decorator with logging""" @wraps(f) def decorated_function(*args, **kwargs): try: verify_jwt_in_request(optional=True) user_id = get_jwt_identity() # Log the API access for authenticated users if user_id: # All good, user is authenticated return f(*args, **kwargs) else: # No token provided, log unauthenticated access log_event( severity=6, # INFO message=f"Unauthenticated access: {request.method} {request.path}", event_type="unauthenticated_access", source_ip=request.remote_addr ) # Check if we should return 401 (for API endpoints) or continue (for optional endpoints) if request.path.startswith('/api/') or request.path.startswith('/v1/'): return jsonify({'error': 'Authentication required'}), 401 else: return f(*args, **kwargs) except Exception as e: # Token validation failed log_event( severity=4, # WARNING message=f"Authentication error: {request.method} {request.path}: {str(e)}", event_type="auth_token_error", source_ip=request.remote_addr ) # Return 401 for API endpoints if request.path.startswith('/api/') or request.path.startswith('/v1/'): return jsonify({'error': 'Authentication failed'}), 401 # For other endpoints, continue without auth (will be handled by the view function) return f(*args, **kwargs) return decorated_function def get_current_user_safe(): """Get current user safely without raising exceptions""" # First check session-based authentication (prioritize for browser access) try: # Check if session exists (for browser access) if 'user_id' in session: user_id = session.get('user_id') if user_id: return User.query.get(user_id) except Exception as e: # Log the session authentication failure log_event( severity=5, # NOTICE level - slightly more concerning message=f"Session authentication failed in get_current_user_safe: {str(e)}", event_type="auth_session_failure", source_ip=request.remote_addr if request else None ) # If session check failed or no session, try to get user from JWT try: # Verify JWT without requiring it (won't raise exception if missing) verify_jwt_in_request(optional=True) user_id = get_jwt_identity() if user_id: return User.query.get(user_id) except Exception as e: # Log the JWT authentication failure only at debug level since we're falling back log_event( severity=7, # DEBUG level - not an error, just a fallback message=f"JWT authentication failed in get_current_user_safe: {str(e)}", event_type="auth_jwt_fallback", source_ip=request.remote_addr if request else None ) return None def refresh_jwt_for_user(user_id=None, response=None, redirect_url=None): """ Refresh the JWT for the given user_id (or current session user). Updates session['access_token'] and sets cookie if response is provided. If redirect_url is given, returns a redirect response with the new token set. """ from flask_jwt_extended import create_access_token from flask import session, make_response, redirect if user_id is None: user_id = session.get('user_id') if not user_id: return None new_access_token = create_access_token(identity=str(user_id)) session['access_token'] = new_access_token if response is not None: response.set_cookie('access_token', new_access_token, httponly=False, secure=False, samesite='Lax') return response if redirect_url is not None: response = make_response(redirect(redirect_url)) response.set_cookie('access_token', new_access_token, httponly=False, secure=False, samesite='Lax') return response return new_access_token