You need to enable JavaScript to run this app.
优惠活动
大模型
产品
解决方案
定价
更多

如何在Flask AppBuilder应用中检查OAuth令牌过期时间?

在Flask AppBuilder(含Superset)中实现OAuth令牌过期预检与自动刷新

要解决令牌过期后仍允许请求执行直到后端报错的问题,核心是在每次请求处理前检查令牌有效性,自动尝试刷新,失败则强制登出。以下是具体实现方案:

核心思路

利用Flask的before_request钩子(FAB完全兼容),在请求到达业务逻辑前完成以下操作:

  • 跳过登录、登出等无需认证的路由
  • 从session中读取OAuth令牌,检查过期时间
  • 若令牌过期,调用Authlib的刷新令牌接口更新令牌
  • 刷新失败则清除session,跳转至登录页

具体实现(以Superset为例)

方案1:在配置文件中注册全局钩子

在superset_config.py中添加以下代码:

from flask import request, redirect, session, url_for
from datetime import datetime
from superset.extensions import oauth

def check_oauth_token_validity():
    # 跳过无需认证的路由,避免循环跳转
    excluded_endpoints = {'security.login', 'security.logout', 'static'}
    if request.endpoint in excluded_endpoints:
        return

    # 读取session中的OAuth令牌数据
    oauth_session = session.get('oauth')
    if not oauth_session:
        return redirect(url_for('security.login'))

    token = oauth_session.get('token')
    if not token:
        session.clear()
        return redirect(url_for('security.login'))

    # 检查令牌是否过期(exp为Unix时间戳)
    expires_at = token.get('exp')
    if expires_at and datetime.utcnow().timestamp() >= expires_at:
        # 尝试刷新令牌(替换为你的OAuth客户端名称,比如'keycloak')
        client = oauth.create_client('keycloak')
        try:
            new_token = client.refresh_token(token['refresh_token'])
            # 更新session中的令牌
            session['oauth']['token'] = new_token
            session.permanent = True
        except Exception:
            # 刷新失败(如refresh_token过期),清除会话并跳转登录
            session.pop('oauth', None)
            session.pop('_user_id', None)
            return redirect(url_for('security.login'))

# 注册全局请求钩子
from superset import app
app.before_request(check_oauth_token_validity)

方案2:扩展AuthOAuthView实现自定义检查

如果需要更紧密地集成FAB的认证流程,可以自定义OAuth视图类:

from flask_appbuilder.security.views import AuthOAuthView
from flask import session, redirect, url_for
from datetime import datetime

class CustomAuthOAuthView(AuthOAuthView):
    def before_request(self, *args, **kwargs):
        super().before_request(*args, **kwargs)
        
        # 仅对已登录用户检查令牌
        if not self.appbuilder.sm.is_authenticated():
            return

        oauth_session = session.get('oauth')
        if not oauth_session:
            return redirect(url_for('security.login'))

        token = oauth_session.get('token')
        expires_at = token.get('exp')
        if expires_at and datetime.utcnow().timestamp() >= expires_at:
            # 调用FAB已配置的OAuth客户端刷新令牌
            client = self.appbuilder.sm.oauth_remotes['keycloak']
            try:
                new_token = client.refresh_token(token['refresh_token'])
                session['oauth']['token'] = new_token
            except Exception:
                # 刷新失败,清除会话
                session.clear()
                return redirect(url_for('security.login'))

然后在superset_config.py中指定自定义视图:

AUTH_OAUTH_VIEW = 'path.to.your.CustomAuthOAuthView'

关键注意事项

  • Keycloak配置:确保客户端已开启Refresh Token支持,并且设置了合理的令牌过期时长(在Keycloak客户端的Settings→Advanced Settings中调整)
  • 异常处理:刷新令牌时需捕获网络错误、refresh_token过期等异常,避免请求崩溃
  • 路由排除:必须跳过登录、登出路由,否则会陷入"检查令牌→跳转登录→再检查令牌"的循环

内容的提问来源于stack exchange,提问作者Pat Buxton

相关产品推荐
方舟 Agent Plan

超全模态模型 × Harness 升级,最新支持 Deepseek-V4.1-Flash、GLM-5.3 系列、Doubao-Seedream-5.0-pro、Kimi-K3 (部分), 限时 9.9 元起

最近更新时间:2026.06.22 05:32:37