基于OpenID Connect实现Flask站点LinkedIn OAuth2登录调试求助

我在Flask站点中尝试通过基于OpenID Connect的LinkedIn OAuth2实现用户登录功能，参考Next.js的配置方式写了回调函数，用Flask-Dance在init.py中定义了蓝图，views.py里编写了回调视图，但目前仅部分生效：在LinkedIn完成OAuth2授权后，无法成功获取用户信息、将信息存入数据库并跳转至个人资料页，附上相关代码求调试。

原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")

原views.py回调视图代码 #

@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))

问题分析与修复方案 #

核心问题 #

你同时使用了Flask-Dance的蓝图和手动实现的回调逻辑，两者冲突：Flask-Dance已经封装了授权码交换、token存储、CSRF状态验证等流程，手动编写的回调跳过了这些关键环节，导致授权后流程异常。

修复步骤 #

1. 移除手动回调的重复逻辑，改用Flask-Dance内置能力

删除手动处理code和state、手动请求token的代码，直接使用Flask-Dance提供的linkedin对象处理授权和用户信息请求。

2. 修改后的代码示例

调整init.py

from flask_dance.contrib.linkedin import make_linkedin_blueprint, linkedin

linkedin_bp = make_linkedin_blueprint(
    client_id=linkedin_client_id,
    client_secret=linkedin_client_secret,
    scope=["openid", "profile", "email"],
    redirect_to="linkedin_authorized",
)

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

重写views.py回调视图

from flask import redirect, url_for, flash
from flask_login import login_user
from flask_dance.contrib.linkedin import linkedin
from your_app import db, User

@app.route("/login/linkedin/authorized")
def linkedin_authorized():
    # 检查授权是否成功
    if not linkedin.authorized:
        flash("LinkedIn授权失败，请重试")
        return redirect(url_for("login"))

    # 用Flask-Dance的session请求用户信息
    resp = linkedin.get("/v2/userinfo")
    if resp.status_code != 200:
        flash("获取用户信息失败")
        return redirect(url_for("login"))
    
    profile_json = resp.json()
    linkedin_id = profile_json.get("sub")
    email = profile_json.get("email")
    first_name = profile_json.get("given_name")
    last_name = profile_json.get("family_name")
    username = first_name if first_name else last_name

    if not linkedin_id:
        flash("无法获取LinkedIn用户ID")
        return redirect(url_for("login"))

    # 检查用户是否存在，不存在则创建
    user = User.query.filter_by(linkedin_id=linkedin_id).first()
    if not user:
        user = User(
            email=email,
            username=username,
            linkedin_id=linkedin_id,
            active=True
        )
        try:
            db.session.add(user)
            db.session.commit()
        except Exception as e:
            db.session.rollback()
            flash(f"创建用户失败: {str(e)}")
            return redirect(url_for("login"))

    # 登录用户并跳转
    login_user(user)
    return redirect(url_for("profile", username=user.username))

额外调试建议 #

• 开启Flask调试模式查看详细日志：app.debug = True
• 确认LinkedIn开发者后台的回调URL与实际地址完全匹配（包括HTTP/HTTPS、端口、路径）
• 检查LinkedIn应用是否已开启OpenID Connect权限，且email、profile权限通过审核（LinkedIn部分权限需人工审核）
• 确保linkedin_redirect_uri参数值与/login/linkedin/authorized完全一致

内容的提问来源于stack exchange，提问作者Kaushik Mallick