Can we write a custom code to validate the oauth token which we already implemented with CustomSsoSecurityManager every request

[thanushcst]

We need to check each request every time on the auth token still valid if that is already invalid redirect into login page.

`class CustomSsoSecurityManager(SupersetSecurityManager):

authoauthview = CustomSsoAuthOAuthView

def oauth_user_info(self, provider, response=None):
if provider == 'wso2':
wso2_base_url = os.getenv("WSO2_BASE_URL")
me = self.appbuilder.sm.oauth_remotes[provider].get(f'{wso2_base_url}/oauth2/userinfo').json()
logging.debug(" user_data: %s", me)
prefix = 'Superset'
return {
'username' : me['username'],
'id': me['sub'],
'email': me['email'],
'roles' : me['roles']
} `

[rad-pat]

Hi, did you figure this out? We have the same requirement.

[thanushcst]

I have customised the code to resole that. One changes on superset_config/views/base.py file another one is added custom_sso_security_manager.py

base.py changes
@superset_app.before_request def before_request(): request_url = request.url access_token = session.get('access_token') if not '/static/assets/' in request_url and not access_token is None: if not appbuilder.sm.validate_oauth_token(): session.clear() return redirect('/logout/')

`custom_sso_security_manager.py code

from superset.security import SupersetSecurityManager
from flask_appbuilder.security.views import AuthOAuthView
from flask_appbuilder.baseviews import expose
from flask import redirect, session
import os
import logging
import requests

class CustomSsoAuthOAuthView(AuthOAuthView):

@expose("/logout/")
def logout(self, provider="{provider_name}", register=None):
    provider_obj = self.appbuilder.sm.oauth_remotes[provider]
    url = ("{}/oidc/logout?client_id={}&redirect_uri={}".format(
                os.getenv("AUTH_BASE_URL"),
                provider_obj.client_id,
                provider_obj.server_metadata.get("logout_redirect_uri")
                ) )

    super().logout()
    return redirect(url)

class CustomSsoSecurityManager(SupersetSecurityManager):
# override the logout function
authoauthview = CustomSsoAuthOAuthView

def oauth_user_info(self, provider, response=None):
   if provider == "{auth_provider_name}":
        auth_base_url = os.getenv("AUTH_BASE_URL")
        me = self.appbuilder.sm.oauth_remotes[provider].get(f'{auth_base_url}/oauth2/userinfo').json()
        session["access_token"] = response["access_token"]
        prefix = "Superset"

        return {
                "username" : me["username"],
                "id": me['sub'],
                "email": me["email"],
                "roles" : me["roles"]
        }

def validate_oauth_token(self, response=None):
    access_token = session.get("access_token")
    #refresh_token = session.get("refresh_token")
    # OAuth token validation endpoint provided by your OAuth provider
    token_validation_url = ("{}/oauth2/introspect".format(
                os.getenv("AUTH_BASE_URL")) )
    
    data = {
        "token": {access_token}
    }

    headers = {
        "Content-Type": "application/x-www-form-urlencoded"
    }
    try:
        # Send a GET request to the token validation endpoint
        response = requests.post(token_validation_url, headers=headers, data=data)
        if response.status_code == 200:
            token_info = response.json()
            # Token is valid, return token information
            if token_info['active']:
                return True
            else: 
                session.clear()
                return False
        else:
            session.clear()
            return False

    except requests.RequestException as e:
        logging.info("Token validation request failed %s :", e)
        session.clear()
        return False`