Superset集成Keycloak后单会话限制问题技术求助

我们已通过Flask-OIDC2.1.1和Flask-OpenID1.3.0将Keycloak 24.0.2与Apache Superset 3.1.2完成集成。需求是用户发起新会话时自动失效旧会话，同一时间仅保留一个活跃会话。

Keycloak端已配置「User session count limiter」，实现新登录时移除该用户旧会话，但Superset未同步会话失效状态，旧登录会话仍保持活跃，需要解决此并发会话限制问题。

配置详情 #

Superset_config.py #

from flask import session
from flask import Flask
from datetime import timedelta
import os  # 补充原配置缺失的os模块导入

def make_session_permanent():
    '''
    为session cookie启用maxAge属性
    '''
    session.permanent = True

# 设置会话最大有效期为5分钟（原注释标注24小时，以代码实际配置为准）
PERMANENT_SESSION_LIFETIME = timedelta(minutes=5)
def FLASK_APP_MUTATOR(app: Flask) -> None:
    app.before_request_funcs.setdefault(None, []).append(make_session_permanent)

from flask_appbuilder.security.manager import AUTH_OID
AUTH_TYPE = AUTH_OID
curr = os.path.abspath(os.getcwd())
OIDC_CLIENT_SECRETS= curr + '/docker/pythonpath_dev/client_secret.json'
OIDC_CLOCK_SKEW = 560
AUTH_USER_REGISTRATION = True
AUTH_USER_REGISTRATION_ROLE = 'Gamma'
from custom_sso_security_manager import OIDCSecurityManager
CUSTOM_SECURITY_MANAGER = OIDCSecurityManager
AUTH_ROLES_MAPPING = { "superset_users": ["Gamma","Alpha"], "superset_admins": ["Admin"],"Admin": ["Admin"], "Gamma" : ["Gamma"] }
OIDC_INTROSPECTION_AUTH_METHOD = 'client_secret_post'
OIDC_TOKEN_TYPE_HINT = 'access_token'
OIDC_SCOPES = ['openid', 'email', 'profile']

自定义安全管理器：custom_sso_security_manager.py #

from flask import redirect, request, session
from flask_appbuilder.security.manager import AUTH_OID
from superset.security import SupersetSecurityManager
from flask_oidc import OpenIDConnect
from flask_appbuilder.security.views import AuthOIDView
from flask_login import login_user
from urllib.parse import quote
from flask_appbuilder.views import ModelView, SimpleFormView, expose
import time
import logging


class OIDCSecurityManager(SupersetSecurityManager):

    def __init__(self, appbuilder):
        super(OIDCSecurityManager, self).__init__(appbuilder)
        if self.auth_type == AUTH_OID:
            self.oid = OpenIDConnect(self.appbuilder.get_app)
        self.authoidview = AuthOIDCView

class AuthOIDCView(AuthOIDView):

    @expose('/login/', methods=['GET', 'POST'])
    def login(self, flag=True):
        sm = self.appbuilder.sm
        oidc = sm.oid
        superset_roles = ["Admin", "Alpha", "Gamma", "Public", "granter", "sql_lab"]
        default_role = "Public"

        @self.appbuilder.sm.oid.require_login
        def handle_login():
            user = sm.auth_user_oid(oidc.user_getfield('email'))
            if user is None:
                user = sm.find_user(username=oidc.user_getfield('preferred_username'))
                if user is None:
                    logging.debug('添加用户登录：{0}.'.format(oidc.user_getfield('email')))
                    info = oidc.user_getinfo(['preferred_username', 'given_name', 'family_name', 'email', 'roles'])
                    roles = [role for role in superset_roles if role in info.get('roles', [])]
                    roles += [default_role, ] if not roles else []
                    user = sm.add_user(info.get('preferred_username'), info.get('given_name'), info.get('family_name'),
                           info.get('email'), [sm.find_role(role) for role in roles])
                else:
                    logging.debug('更新用户登录：{0}.'.format(oidc.user_getfield('email')))
                    user.email=oidc.user_getfield('email')
                    sm.update_user(user)

            logging.debug('用户登录：{0}.'.format(user.email))
            login_user(user, remember=False)
            return redirect(self.appbuilder.get_url_for_index)

        return handle_login()

    @expose('/logout/', methods=['GET', 'POST'])
    def logout(self):
        oidc = self.appbuilder.sm.oid
        redirect_url = request.url_root.strip('/') + self.appbuilder.get_url_for_login
        if oidc.user_loggedin:
            refreshToken=oidc.get_refresh_token()
            oidc.logout()
            super(AuthOIDCView, self).logout()
            time.sleep(1)
            session.clear()
            url = ("logout?client_id={}&refresh_token={}&post_logout_redirect_uri={}".format(
                    oidc.client_secrets.get('client_id'),
                    refreshToken,
                    redirect_url))
            return redirect(oidc.client_secrets.get('issuer') + '/protocol/openid-connect/'+ url)

        else:
            return redirect(redirect_url)

解决方案 #

1. 启用令牌实时验证 #

强制Superset每次请求都调用Keycloak的令牌 introspection 接口，验证令牌是否已被吊销：
在Superset_config.py中添加：

OIDC_CHECK_TOKEN_EXPIRY = True
OIDC_VALIDATE_TOKEN = True

2. 改用服务端会话存储 #

Flask默认客户端cookie会话无法主动失效远端会话，改用Redis存储服务端会话：

• 安装依赖：pip install flask-session redis
• 在Superset_config.py中配置Redis会话：

  from flask_session import Session
  import redis
  
  def FLASK_APP_MUTATOR(app: Flask) -> None:
      app.before_request_funcs.setdefault(None, []).append(make_session_permanent)
      # 配置Redis会话存储
      app.config['SESSION_TYPE'] = 'redis'
      app.config['SESSION_REDIS'] = redis.from_url('redis://你的Redis地址:6379/0')
      app.config['SESSION_USE_SIGNER'] = True
      app.config['SESSION_PERMANENT'] = False
      Session(app)
  

• 在AuthOIDCView的handle_login函数中，添加清除用户旧会话逻辑（登录前执行）：

  # 导入Redis依赖（已在配置中添加）
  redis_conn = redis.from_url(self.appbuilder.get_app.config['SESSION_REDIS'])
  # 根据用户ID匹配旧会话键（需根据实际会话键格式调整）
  user_session_keys = redis_conn.keys(f"session:*:{user.id}")
  if user_session_keys:
      redis_conn.delete(*user_session_keys)
  

3. 配置Keycloak单点登出联动 #

让Keycloak在吊销会话时主动通知Superset失效会话：

• 在Keycloak客户端配置中，启用「Front Channel Logout」和「Back Channel Logout」，设置Back Channel Logout URL为http://你的Superset地址/logout/
• 在custom_sso_security_manager.py中添加Back Channel登出处理：

  @expose('/backchannel_logout/', methods=['POST'])
  def backchannel_logout(self):
      logout_token = request.form.get('logout_token')
      if self.appbuilder.sm.oid.validate_logout_token(logout_token):
          user_info = self.appbuilder.sm.oid._parse_id_token(logout_token)
          user = self.appbuilder.sm.find_user(email=user_info.get('email'))
          if user:
              redis_conn = redis.from_url(self.appbuilder.get_app.config['SESSION_REDIS'])
              user_session_keys = redis_conn.keys(f"session:*:{user.id}")
              if user_session_keys:
                  redis_conn.delete(*user_session_keys)
          return '', 204
      return '', 400
  

4. 缩短会话有效期并强制刷新 #

在Superset_config.py中保持短会话有效期，减少旧会话残留时间：

PERMANENT_SESSION_LIFETIME = timedelta(minutes=5)
SESSION_REFRESH_EACH_REQUEST = True  # 每次请求刷新会话有效期

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