如何在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
相关产品推荐
相关产品推荐

