Allowing Linkedin oauth2 login into a flask site using OpenID Connect

Question

I am trying to allow users to log in into my flask site using Linkedin oauth2. There has been recent changes in how LinkedIn allows this signin process using OpenID Connect on top of oauth2. I am trying to model my redirect callback function similar to this approach described here.

My setup and callback function seem to be partially working. But I cannot successfully retrieve the information of the user after oauth2 is successfully completed by Linkedin, log that information into my user database and then redirect the user to the profile page. I would appreciate any help to debug this.

This is how I am using Flask-dance to define the blueprint in my init.py:

linkedin_bp = make_linkedin_blueprint(
    client_id=linkedin_client_id,
    client_secret=linkedin_client_secret,
    redirect_to='linkedin_authorized',  # This should match the name of your callback route
    scope=["openid", "profile", "email"],  # Add required scopes here
)

app.register_blueprint(linkedin_bp, url_prefix="/login")

And here is the view for the callback function defined in my views.py. I have defined linkedin_redirect_uri, linkedin_client_id and linkedin_client_secret variables that matches the same at my linkedin developer site.

@app.route("/login/linkedin/authorized")
def linkedin_authorized():

    # Step 1: Get 'code' and 'state' from the query parameters
    code = request.args.get("code")
    state = request.args.get("state")

    token_data = {
        "grant_type": "authorization_code",
        "code": code,
        "redirect_uri": linkedin_redirect_uri,
        "client_id": linkedin_client_id,
        "client_secret": linkedin_client_secret,
    }

    # Step 2: Exchange 'code' for access token
    token_url = "https://www.linkedin.com/oauth/v2/accessToken"
    token_response = requests.post(token_url, data=token_data, headers={"Content-Type": "application/x-www-form-urlencoded"})

    # Check if the token request was successful
    if token_response.status_code != 200:
        return jsonify({"error": "Failed to get access token"}), 400

    token_json = token_response.json()

    access_token = token_json["access_token"]

    # Step 3: Fetch user profile using the access token
    userinfo_url = "https://api.linkedin.com/v2/userinfo"
    profile_response = requests.get(userinfo_url, headers={"Authorization": f"Bearer {access_token}"})

    profile_json = profile_response.json()

    # Extract required user details from the profile
    user_profile = {
        "first_name": profile_json.get("given_name"),
        "last_name": profile_json.get("family_name"),
        "email": profile_json.get("email"),
        "linkedin_id": profile_json.get("sub"),
    }

    linkedin_id = user_profile["linkedin_id"]
    email = user_profile["email"]
    username = user_profile["first_name"] if user_profile["first_name"] else user_profile["last_name"]

    if not linkedin_id:
        app.logger.error("LinkedIn ID missing from profile")

    # Step 4: Check if the user exists in the database
    user = User.query.filter_by(linkedin_id=linkedin_id).first()

    # Debugging: Log whether the user exists in the database
    if user is None:
        # Create a new user if they don't exist
        user = User(
            email=email,
            username=username,  # Assuming first name is available
            linkedin_id=linkedin_id,
            active=True,
        )

        try:
            db.session.add(user)
            db.session.commit()
            app.logger.debug("New user created and added to the database.")
        except Exception as e:
            db.session.rollback()
            app.logger.error(f"Database error: {e}")
    else:
        app.logger.debug(f"User with LinkedIn ID {linkedin_id} found.")

    # Step 5: Log the user in
    login_user(user)

    # Step 6: Redirect to the profile page
    return redirect(url_for('profile', username=user.username))