Airflow 3.0.2 + Helm + Keycloak SSO: User role changes from "Admin" to "Viewer" after login

We are using Apache Airflow 3.0.2 with the official Helm chart version 1.17.0, deployed on Kubernetes via Terraform. We're integrating SSO using Keycloak.

Problem

After successful SSO login, users with the "Admin" role in Keycloak are being mapped to "Viewer" in Airflow. We expect users with the Admin role in Keycloak to retain their Admin role in Airflow as well.

Configuration

We have the following webserver_config.py to configure Keycloak as the OAuth provider:

import os
from flask_appbuilder.security.manager import AUTH_OAUTH
import logging

log = logging.getLogger(__name__)

AUTH_TYPE = AUTH_OAUTH
BASE_URL = os.environ.get("BASE_URL", "https://airflow-example.com")

OAUTH_PROVIDERS = [
    {
        "name": "keycloak",
        "icon": "fa-key",
        "token_key": "access_token",
        "remote_app": {
            "api_base_url": os.getenv("api_base_url"),
            "client_kwargs": {"scope": "openid email profile"},
            "request_token_url": None,
            "access_token_url": os.getenv("access_token_url"),
            "authorize_url": os.getenv("authorize_url"),
            "client_id": os.getenv("client_id"),
            "client_secret": os.getenv("client_secret"),
            "server_metadata_url": os.getenv("server_metadata_url"),
        },
        "role_keys": "realm_access.roles",
    },
]

AUTH_ROLES_MAPPING = {
    "Admin": ["Admin"],
    "Viewer": ["Viewer"],
    "User": ["User"],
    "Public": ["Public"],
    "Op": ["Op"],
}

AUTH_ROLE_ADMIN = "Admin"
AUTH_ROLE_PUBLIC = "Public"
AUTH_USER_REGISTRATION = True
AUTH_USER_REGISTRATION_ROLE = "Viewer"
AUTH_ROLES_SYNC_AT_LOGIN = True
OIDC_ISSUER = os.getenv("OIDC_ISSUER")

Despite this configuration, the user gets assigned the "Viewer" role after login.

Attempted Solution

We implemented a custom SecurityManager to override the role assignment behavior, as follows:

import os
from flask_appbuilder.security.manager import AUTH_OAUTH
from base64 import b64decode

import jwt
import requests
from cryptography.hazmat.primitives import serialization

from airflow.providers.fab.auth_manager.security_manager.override import FabAirflowSecurityManagerOverride
import logging
log = logging.getLogger(__name__)

basedir = os.path.abspath(os.path.dirname(__file__))
# Flask-WTF flag for CSRF
WTF_CSRF_ENABLED = True
# ----------------------------------------------------
# AUTHENTICATION CONFIG
# ----------------------------------------------------
AUTH_TYPE = AUTH_OAUTH
BASE_URL = os.environ.get("BASE_URL", "https://aiflow-example.com")

OAUTH_PROVIDERS = [
    {
        "name": "keycloak",
        "icon": "fa-key",
        "token_key": "access_token",
        "remote_app": {
            "api_base_url": os.getenv("api_base_url"),
            "client_kwargs": {"scope": "openid email profile"},
            "request_token_url": None,
            "access_token_url": os.getenv("access_token_url"),
            "authorize_url": os.getenv("authorize_url"),
            "client_id": os.getenv("client_id"),
            "client_secret": os.getenv("client_secret"),
            "server_metadata_url": os.getenv("server_metadata_url"),
        },
        "role_keys": "realm_access.roles",
    },
]

AUTH_ROLES_MAPPING = {
    "Viewer": ["Viewer"],
    "Admin": ["Admin"],
    "User": ["User"],
    "Public": ["Public"],
    "Op": ["Op"],
}

# Uncomment to setup Full admin role name
AUTH_ROLE_ADMIN = "Admin"
# Uncomment to setup Public role name, no authentication needed
AUTH_ROLE_PUBLIC = "Public"
# Will allow user self registration
AUTH_USER_REGISTRATION = True
# The default user self registration role
AUTH_USER_REGISTRATION_ROLE = "Viewer"
AUTH_ROLES_SYNC_AT_LOGIN = True
OIDC_ISSUER = os.getenv("OIDC_ISSUER")

# Fetch public key
req = requests.get(OIDC_ISSUER)
key_der_base64 = req.json()["public_key"]
key_der = b64decode(key_der_base64.encode())
public_key = serialization.load_der_public_key(key_der)


class CustomSecurityManager(FabAirflowSecurityManagerOverride):
    def get_oauth_user_info(self, provider, response):
        if provider == "keycloak":
            token = response["access_token"]
            me = jwt.decode(token, public_key, algorithms=["HS256", "RS256"], audience=os.getenv("client_id"))

            # Extract roles from resource access
            realm_access = me.get("realm_access", {})
            groups = realm_access.get("roles", [])

            log.info("groups: {0}".format(groups))

            if not groups:
                groups = ["Viewer"]

            userinfo = {
                "username": me.get("preferred_username"),
                "email": me.get("email"),
                "first_name": me.get("given_name"),
                "last_name": me.get("family_name"),
                "role_keys": groups,
            }

            log.info("user info: {0}".format(userinfo))

            return userinfo
        else:
            return {}


# Make sure to replace this with your own implementation of AirflowSecurityManager class
SECURITY_MANAGER_CLASS = CustomSecurityManager

Result

Now, users with the Admin role do get mapped correctly and can log in as admins.

Issue

However, after this change, we are seeing frequent 500 Internal Server Error responses on various Admin pages (e.g., under Security or Manage sections in the UI).

Question

• Are we missing something in the way role_keys are passed or how AUTH_ROLES_MAPPING interacts with CustomSecurityManager?
• Is this approach incompatible with Airflow 3.0.2 / Flask-AppBuilder?
• Are there additional role-to-permission bindings we need to declare manually in Airflow 3.x?

Any guidance or examples of working SSO setup with role preservation in Airflow 3.0+ would be really appreciated.