Spaces:

dn-scribe
/

attstore

Runtime error

feat: support tenant deletion

38536e2 10 months ago

8.44 kB

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

	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