We have integrated Keycloak 24.0.2 with Apache Superset 3.1.2 using Flask-OIDC==2.1.1 and Flask-OpenID==1.3.0 plugins. However, we have the requirement to expire an old session if a new session is initiated by a user, i.e. to have one session active at a time. In Keyloak this can be done using “User session count limiter”, i.e. it removes/expired old session of that user on every new login.

However, the same information is not received at Superset, so the login session on Superset remains active. So, we need help on this to limit concurrent session Superset.

Details of configuration are as follows

Superset_config.py

from flask import session
from flask import Flask
from datetime import timedelta


def make_session_permanent():
    '''
    Enable maxAge for the cookie 'session'
    '''
    session.permanent = True

# Set up max age of session to 24 hours
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 security Manager: 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('add login {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('update login {0}.'.format(oidc.user_getfield('email')))
                    user.email=oidc.user_getfield('email')
                    sm.update_user(user)

            logging.debug('user login {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)