File size: 8,439 Bytes
b54637e a4b5887 b54637e 38536e2 fb83204 38536e2 fb83204 38536e2 fb83204 38536e2 b54637e 38536e2 b54637e 38536e2 b54637e 9cd1061 b54637e 9cd1061 b54637e 9cd1061 b54637e 9cd1061 b54637e 7c78952 b54637e 7c78952 ff5244a b54637e 7c78952 b54637e 565d6ed | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 189 190 | 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
|