Class OAuth2Authentication

web/pgadmin/authenticate/oauth2.py:118–774

OAuth Authentication Class

class OAuth2Authentication(BaseAuthentication):
    """OAuth Authentication Class"""

    LOGOUT_VIEW = OAUTH2_LOGOUT

    oauth_obj = OAuth(Flask(__name__))
    oauth2_clients = {}
    oauth2_config = {}
    email_keys = ['mail', 'email']

    _WORKLOAD_IDENTITY_ASSERTION_TYPE = (
        'urn:ietf:params:oauth:client-assertion-type:jwt-bearer'
    )

    def __init__(self):
        # Selected provider name (set during authenticate()).
        # Initializing avoids AttributeError in edge cases/tests.
        self.oauth2_current_client = None

        for oauth2_config in config.OAUTH2_CONFIG:

            provider_name = oauth2_config.get('OAUTH2_NAME', '<unknown>')

            client_auth_method = oauth2_config.get(
                'OAUTH2_CLIENT_AUTH_METHOD', 'client_secret'
            )
            if not isinstance(client_auth_method, str):
                client_auth_method = 'client_secret'
            client_auth_method = client_auth_method.strip().lower()
            is_workload_identity = (client_auth_method == 'workload_identity')

            OAuth2Authentication.oauth2_config[
                oauth2_config['OAUTH2_NAME']] = oauth2_config

            # Build client_kwargs with defaults
            client_kwargs = {
                'scope': oauth2_config.get(
                    'OAUTH2_SCOPE', 'email profile'),
                'verify': oauth2_config.get(
                    'OAUTH2_SSL_CERT_VERIFICATION', True)
            }

            pkce_method = oauth2_config.get('OAUTH2_CHALLENGE_METHOD')
            pkce_response_type = oauth2_config.get('OAUTH2_RESPONSE_TYPE')
            pkce_is_configured = any(
                [
                    pkce_method is not None,
                    pkce_response_type is not None
                ]
            )

            raw_client_secret = oauth2_config.get('OAUTH2_CLIENT_SECRET')
            client_secret_is_empty = (
                raw_client_secret is None or
                (isinstance(raw_client_secret, str) and
                 raw_client_secret.strip() == '')
            )