I’m facing an issue with Flask-Login + OAuth flow where authentication works in some environments but fails in another.

✅ Works in:

• Local

• PCF (Cloud Foundry)

❌ Fails in:

• KOB (Kubernetes + Nginx setup)

Expected Behavior

After OAuth redirect to /callback, the session should persist and load_user() should be triggered.

Working logs (Local / PCF)

[BEFORE_REQUEST_DEBUG] Method: GET, Path: /callback
[LOAD_USER_DEBUG] load_user called with user_id: user_email@whh.com
[BEFORE_REQUEST_DEBUG] Authenticated user email: user_email@whh.com
[BEFORE_REQUEST_DEBUG] User is_dev status (cached): True

Actual Behavior (KOB)

• load_user() is never called

• Session data is missing in /callback

Logs:

[BEFORE_REQUEST_DEBUG] Method: GET, Path: /callback
[CALLBACK_DEBUG] Inside /callback route
[CALLBACK_DEBUG] Session info at /callback: None
[CALLBACK_DEBUG] Request args: {
  'code': '3142c197771a43529f2113dfe060f97c',
  'state': '123'
}

Key Observation

Session is lost between / and /callback, which causes:

• session['user_info'] to be None

• current_user.is_authenticated to be False

• load_user() never gets triggered

Relevant Code

user_loader

@app.before_request
def check_is_dev():
    print(f"[BEFORE_REQUEST_DEBUG] Method: {request.method}, Path: {request.path}")

    if current_user.is_authenticated:
        user_email = session.get('user_info', {}).get('email')
        print(f"[BEFORE_REQUEST_DEBUG] Authenticated user email: {user_email}")

        if user_email:
            cached_is_dev = session.get('is_dev')

            if cached_is_dev is not None:
                g.is_dev = cached_is_dev
                print(f"[BEFORE_REQUEST_DEBUG] User is_dev status (cached): {g.is_dev}")
                print("[BEFORE_REQUEST_DEBUG] User authenticated successfully")
            else:
                try:
                    g.is_dev = user_manager.isDev(user_email)
                    session['is_dev'] = g.is_dev
                    print(f"[BEFORE_REQUEST_DEBUG] User is_dev status (db): {g.is_dev}")
                    print("[BEFORE_REQUEST_DEBUG] User authenticated successfully")
                except Exception as e:
                    print(f"[BEFORE_REQUEST_DEBUG] Error checking is_dev status: {str(e)}")
                    g.is_dev = False
                    session['is_dev'] = False
        else:
            print("[BEFORE_REQUEST_DEBUG] No user email found in session")
            g.is_dev = False
    else:
        g.is_dev = False


@login_manager.user_loader
def load_user(user_id):
    print(f"[LOAD_USER_DEBUG] load_user called with user_id: {user_id}")

    user_info = session.get('user_info')

    if not user_info or user_info.get('sub') != user_id:
        return None

    return User(
        id=user_info['sub'],
        name=user_info['name'],
        email=user_info['email'],
        role=user_info.get('role')
    )


@app.route('/')
def index():
    print("Inside Index")

    if current_user.is_authenticated:
        print("User already authenticated, redirecting to review")
        return redirect(url_for('review'))

    auth_in_progress = session.get('auth_in_progress', False)
    oauth_timestamp = session.get('oauth_timestamp', 0)
    current_time = time.time()

    if auth_in_progress and (current_time - oauth_timestamp) < 60:
        print("OAuth flow already in progress, showing waiting page")

        return """
        
            
                
            
            
                
Authentication in progress...
                
Please wait, this page will refresh automatically.
                

                    If stuck, click here.
                
            
        
        """

    uri, code_verifier = user_manager.generate_auth_url()

    session['code_verifier'] = code_verifier
    session['auth_in_progress'] = True
    session['oauth_timestamp'] = current_time

    print("Starting OAuth flow, redirecting...")
    return redirect(uri)

@app.route('/callback')
def callback():
    print("[CALLBACK_DEBUG] Inside /callback route")
    print(f"[CALLBACK_DEBUG] Session info at /callback: {session.get('user_info')}")
    print(f"[CALLBACK_DEBUG] Request args: {dict(request.args)}")
    print(f"[CALLBACK_DEBUG] Request method: {request.method}")

    code = request.args.get('code')
    state = request.args.get('state')
    error = request.args.get('error')
    error_description = request.args.get('error_description')

    print(f"[CALLBACK_DEBUG] OAuth callback received code: {code if code else 'None'}...")
    print(f"[CALLBACK_DEBUG] OAuth callback received state: {state}")
    print(f"[CALLBACK_DEBUG] OAuth callback received error: {error}")
    print(f"[CALLBACK_DEBUG] OAuth callback received error_description: {error_description}")

    if error:
        session.pop('auth_in_progress', None)
        print(f"[CALLBACK_DEBUG] OAuth error returned: {error} - {error_description}")
        return f"OAuth Error: {error} - {error_description}", 400

    if not code:
        session.pop('auth_in_progress', None)
        print("[CALLBACK_DEBUG] No code received in callback.")
        return "No code received in callback.", 400

    try:
        print("[CALLBACK_DEBUG] Starting user authentication...")

        code_verifier = session.pop('code_verifier', None)
        if not code_verifier:
            print("[CALLBACK_DEBUG] code_verifier missing from session. Likely a duplicate callback request.")
            return redirect(url_for('index'))

        decoded_response = user_manager.authenticateUser(code, code_verifier)
        print(f"[CALLBACK_DEBUG] Authentication result type: {type(decoded_response)}")
        print(f"[CALLBACK_DEBUG] Authentication result: {decoded_response}")

        if isinstance(decoded_response, tuple):
            session.pop('auth_in_progress', None)
            error_message = decoded_response[0] if len(decoded_response) > 0 else str(decoded_response)
            print(f"[CALLBACK_DEBUG] Authentication returned tuple: {decoded_response}")
            return f"Authentication failed: {error_message}", 401

        if isinstance(decoded_response, str) and decoded_response.startswith("Error"):
            session.pop('auth_in_progress', None)
            print(f"[CALLBACK_DEBUG] Authentication returned error string: {decoded_response}")
            return decoded_response, 401

        if not decoded_response:
            session.pop('auth_in_progress', None)
            print("[CALLBACK_DEBUG] Authentication returned empty/None response")
            return "Authentication returned empty response", 401

        print("[CALLBACK_DEBUG] Authentication successful, converting to User object...")

        user = user_manager.jsonOauthUserToUser(decoded_response)
        print(f"[CALLBACK_DEBUG] User object created: {user}")
        print(f"[CALLBACK_DEBUG] User details - ID: {user.id}, Name: {user.name}, Email: {user.email}, Role: {user.role}")

        # Login user
        login_user(user)

        role = user_manager.check_user_role(user.email)
        print(f"[CALLBACK_DEBUG] User role from database: {role}")

        # Update session
        session.permanent = True
        session['_user_id'] = user.id
        session['user_info'] = {
            'sub': user.id,
            'name': user.name,
            'email': user.email,
            'role': role
        }

        print(f"[CALLBACK_DEBUG] Session updated with user info: {session.get('user_info', {})}")

        # Check if user exists in DB
        exists = user_manager.user_exists(user.email)
        print(f"[CALLBACK_DEBUG] User exists in database: {exists}")

        now = datetime.now()
        formatted_time = now.strftime('%Y-%m-%d %H:%M:%S')
        user_last_login = datetime.strptime(formatted_time, '%Y-%m-%d %H:%M:%S')

        if exists:
            print(f"[CALLBACK_DEBUG] Updating last_login for {user.email}")
            try:
                user_manager.update_last_login(user.email, user_last_login)
                print("[CALLBACK_DEBUG] Last login updated successfully")
            except Exception as e:
                print(f"[CALLBACK_DEBUG] Error updating last login: {str(e)}")
        else:
            print(f"[CALLBACK_DEBUG] Adding new user: {user.email}")
            try:
                user_manager.add_user_if_not_exists(user.email, user.name, user_last_login)
                user_manager.add_default_user_role(user.email)
                print("[CALLBACK_DEBUG] New user added successfully")
            except Exception as e:
                print(f"[CALLBACK_DEBUG] Error adding new user: {str(e)}")

        # Clear auth flags
        session.pop('auth_in_progress', None)
        session.pop('oauth_timestamp', None)

        print("[CALLBACK_DEBUG] Redirecting to /review")
        return redirect(url_for('review'))

    except Exception as e:
        print(f"[CALLBACK_DEBUG] Callback error: {str(e)}")
        print(f"[CALLBACK_DEBUG] Error type: {type(e)}")

        import traceback
        print(f"[CALLBACK_DEBUG] Full traceback: {traceback.format_exc()}")

        session.clear()
        return f"Internal Server Error during callback: {str(e)}", 500