Unable to integrate keycloak provider with Airflow 2.7.3

[geoffo-dev]

Apache Airflow version

Other Airflow 2 version (please specify below)

If "Other Airflow 2 version" selected, which one?

2.7.3

What happened?

Hello,

I am trying to integrate keycloak with airflow which I have managed successfully in the past however I am currently struggling - possibly due to it being a newer version of airflow where I understand certain authentication mechanisms have changed.

Having followed the guide have the following webserver_config.py file:

from airflow.www.security import AirflowSecurityManager
from flask_appbuilder.security.manager import AUTH_OAUTH
import os

AUTH_TYPE = AUTH_OAUTH
AUTH_ROLES_SYNC_AT_LOGIN = True
AUTH_USER_REGISTRATION = True

AUTH_ROLE_ADMIN = os.environ['KEYCLOAK_ADMIN_ROLE_NAME']
AUTH_ROLE_OP = os.environ['KEYCLOAK_OP_ROLE_NAME']
AUTH_ROLE_USER = os.environ['KEYCLOAK_USER_ROLE_NAME']
AUTH_ROLE_VIEWER = os.environ['KEYCLOAK_VIEWER_ROLE_NAME']
AUTH_ROLE_PUBLIC = os.environ['KEYCLOAK_PUBLIC_ROLE_NAME']

# Will allow user self registration
AUTH_USER_REGISTRATION = True
AUTH_USER_REGISTRATION_ROLE = "Public"
AUTH_ROLES_SYNC_AT_LOGIN = True
AUTH_ROLES_MAPPING = {
    "Admin": [AUTH_ROLE_ADMIN],
    "Op": [AUTH_ROLE_OP],
    "User": [AUTH_ROLE_USER],
    "Viewer": [AUTH_ROLE_VIEWER],
    "Public": [AUTH_ROLE_PUBLIC],
}

PROVIDER_NAME = 'keycloak'
CLIENT_ID = os.environ['KEYCLOAK_CLIENT_ID']

# Get the OIDC SECRET
with open('/secret/oidc-secret/client-secret', 'r') as file:
    CLIENT_SECRET = file.read().rstrip()

OIDC_ISSUER = os.environ['OIDC_ISSUER_URL']
OIDC_BASE_URL = "{oidc_issuer}/protocol/openid-connect".format(oidc_issuer=OIDC_ISSUER)
OIDC_TOKEN_URL = "{oidc_base_url}/token".format(oidc_base_url=OIDC_BASE_URL)
OIDC_AUTH_URL = "{oidc_base_url}/auth".format(oidc_base_url=OIDC_BASE_URL)

OAUTH_PROVIDERS = [{
    'name': PROVIDER_NAME,
    'token_key':'access_token',
    'icon':'fa-globe',
    'remote_app': {
        'client_id': CLIENT_ID,
        'client_secret': CLIENT_SECRET,
        'api_base_url': OIDC_BASE_URL,
        'client_kwargs': {
            'scope': 'email profile'
        },
        'access_token_url': OIDC_TOKEN_URL,
        'authorize_url': OIDC_AUTH_URL,
        'request_token_url': None,              
    }
}]


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

and the following keycloakAuthorizer.py which should be in the same path:

from airflow.www.security import AirflowSecurityManager
import logging
from typing import Any, List, Union
import os

# Create some logging
log = logging.getLogger(__name__)
log.setLevel(os.getenv("AIRFLOW__LOGGING__FAB_LOGGING_LEVEL", "INFO"))

class KeycloakAuthorizer(AirflowSecurityManager):
CLIENT_ID = os.environ['KEYCLOAK_CLIENT_ID']
def get_oauth_user_info(self, provider: str, resp: Any) -> dict[str, Union[str, list[str]]]:
    remote_app = self.appbuilder.sm.oauth_remotes[provider]
    me = remote_app.get("user")
    roles = me["resource_access"][CLIENT_ID]["roles"]
    if len(roles) < 1:
        roles = ["public"]

    userinfo = {
        "username": me.get("preferred_username"),
        "email": me.get("email"),
        "first_name": me.get("given_name"),
        "last_name": me.get("family_name"),
        "role_keys": roles,
    }
    log.debug(f"User info from Keycloak: {userinfo}\Role info from Keycloak: {roles}")
    return userinfo

However when the webserver launches, there is no option to login via keycloak - only standard username and password.

The only error I can see that might be relevant when the webserver loads which relates to the google package (for what I believe is required for keycloak to work):

/home/airflow/.local/lib/python3.11/site-packages/flask_limiter/extension.py:336 UserWarning: Using the in-memory storage for tracking rate limits as no storage was explicitly specified. This is not recommended for production use. See: https://flask-limiter.readthedocs.io#configuring-a-storage-backend for documentation about configuring the storage backend.
[2024-01-15 15:53:42 +0000] [13] [INFO] Starting gunicorn 21.2.0
[2024-01-15T15:53:49.840+0000] {providers_manager.py:271} INFO - Optional provider feature disabled when importing 'airflow.providers.google.leveldb.hooks.leveldb.LevelDBHook' from 'apache-airflow-providers-google' package
/home/airflow/.local/lib/python3.11/site-packages/snowflake/connector/options.py:103 UserWarning: You have an incompatible version of 'pyarrow' installed (11.0.0), please install a version that adheres to: 'pyarrow<10.1.0,>=10.0.1; extra == "pandas"'
[2024-01-15 15:53:53 +0000] [13] [INFO] Listening at: http://0.0.0.0:8080 (13)
[2024-01-15 15:53:53 +0000] [13] [INFO] Using worker: sync
[2024-01-15 15:53:53 +0000] [25] [INFO] Booting worker with pid: 25
[2024-01-15 15:53:53 +0000] [26] [INFO] Booting worker with pid: 26
[2024-01-15 15:53:53 +0000] [27] [INFO] Booting worker with pid: 27
[2024-01-15 15:53:53 +0000] [28] [INFO] Booting worker with pid: 28
10.38.0.0 - - [15/Jan/2024:15:53:59 +0000] "GET /health HTTP/1.1" 200 318 "-" "kube-probe/1.28+"
10.38.0.0 - - [15/Jan/2024:15:53:59 +0000] "GET /health HTTP/1.1" 200 318 "-" "kube-probe/1.28+"

I have tried version 2.8.0 as well, but due to #36702 - this does not work.

I am at a bit of a loss as to how to resolve, I have tried a number of images, but not certain if this is supported anymore or if there is another way to resolve.

What you think should happen instead?

User should be presented with a login option with keycloak

How to reproduce

Have described above

Operating System

Tags 2.7.3-python3.11 and 2.7.3 in docker

Versions of Apache Airflow Providers

apache-airflow-providers-amazon==8.10.0
apache-airflow-providers-celery==3.4.1
apache-airflow-providers-cncf-kubernetes==7.8.0
apache-airflow-providers-common-sql==1.8.0
apache-airflow-providers-daskexecutor==1.1.0
apache-airflow-providers-docker==3.8.0
apache-airflow-providers-elasticsearch==5.1.0
apache-airflow-providers-ftp==3.6.0
apache-airflow-providers-google==10.11.0
apache-airflow-providers-grpc==3.3.0
apache-airflow-providers-hashicorp==3.5.0
apache-airflow-providers-http==4.6.0
apache-airflow-providers-imap==3.4.0
apache-airflow-providers-microsoft-azure==8.1.0
apache-airflow-providers-mysql==5.4.0
apache-airflow-providers-odbc==4.1.0
apache-airflow-providers-openlineage==1.2.0
apache-airflow-providers-postgres==5.7.1
apache-airflow-providers-redis==3.4.0
apache-airflow-providers-sendgrid==3.3.0
apache-airflow-providers-sftp==4.7.0
apache-airflow-providers-slack==8.3.0
apache-airflow-providers-snowflake==5.1.0
apache-airflow-providers-sqlite==3.5.0
apache-airflow-providers-ssh==3.8.1

Deployment

Official Apache Airflow Helm Chart

Deployment details

Deploying on EKS v1.28.3
Helm Version v3.13.1
Helm Chart Version 1.11.0

Anything else?

Would be happy to support updating a PR, but not really sure where to start!

Are you willing to submit PR?

• Yes I am willing to submit a PR!

Code of Conduct

• I agree to follow this project's Code of Conduct

[boring-cyborg[bot]]

Thanks for opening your first issue here! Be sure to follow the issue template! If you are willing to raise PR to address this issue please do so, no need to wait for approval.

[eladkal]

Hi @geoffo-dev can you please clarify what is the bug?
Your report seems more like a support request.

[geoffo-dev]

Hi @eladkal - So sorry it might be support or it might be a bug. I think I put it here as I have followed the configuration steps and the files are visible in the pod - however nothing is happening. The only thing I can see is the info in the log above.

Is this a bug... I dont know..! I havent seen many examples specific to keycloak since Airflow 2.5.0.

[eladkal]

We use Issues for bug reports since for this moment your problem is more of community support I'm converting this thread to GitHub Discussion.

I hope someone from the community might be able to help you.

[potiuk]

Yeah. I think also we would very much welcome someone who uses keycloak, creates and contribute a native KeyCloak provider for Airflow 2.8.0+ following the AIP-56 introduction - see https://cwiki.apache.org/confluence/display/AIRFLOW/AIP-56+Extensible+user+management the FAB integration with KeyCloak was indirect and we did not have the expert knowledge here, and it was very limiting and opinionated.

Seems that implementing a natve KeyClock AuthManager that would not use FAB but go straight the AIP-56 route is a much better long-term approach - we were also discusing it at the first Airflow Town Hall meeting https://lists.apache.org/thread/gxk5hm1cbxvo84smmcqp9m3omko325p5 where the "call to action" for that was made by @vincbeck

I think if somoene who uses and needs Keycloak would like to implement it natively, that would be a FAR better approach than trying to go the "old FAB" route - and if it can be contributed back, with documentation, some tests etc. even if it will be a simple form of it, that would be more than fantastic.

So @geoffo-dev - maybe you would like to pick the challenge? I guess - since the maintainers are asking for help here and community (and especially someone who already uses KeyCloak and have the need to get it integrated) you woudl also get great help from the maintainers (I am happy to help and I am sure @vincbeck as well alongside few other people - as long as you'd take the lead and implement it (and test it in your own deployment - which is the main reason why we think it has a chance to be working, because you could prove it works for you).

What say you @geoffo-dev ?

[vincbeck]

+1000 on this one. I am very happy to help if you take on that challenge. I really dont think it is complicated to implement an auth manager for KeyCloak and I do think, it can help so many users in the community. The documentation for auth managers is not yet built (it will happen when Airflow 2.9 is released) but you can access it through the code here.

[geoffo-dev]

Hey @potiuk and @vincbeck! Thanks for the info!

So I need to read and digest user management bits and auth manager documentation too - probably over the weekend if that is ok? I am obviously very keen to support the OpenSource community and give back in some way and if I have the skills I certainly would like to!

So yes up for the challenge, but might take me a little while to get familiar with the codebase and things..!

[eladkal]

So yes up for the challenge, but might take me a little while to get familiar with the codebase and things..!

This is awesome. We have very extensive guides about contributing
https://github.com/apache/airflow/blob/main/CONTRIBUTORS_QUICK_START.rst
https://github.com/apache/airflow/blob/main/CONTRIBUTING.rst

Feel free to join the Slack workspace we have dedicated channels to assist contributors when need

[vincbeck]

Hey @potiuk and @vincbeck! Thanks for the info!

So I need to read and digest user management bits and auth manager documentation too - probably over the weekend if that is ok? I am obviously very keen to support the OpenSource community and give back in some way and if I have the skills I certainly would like to!

So yes up for the challenge, but might take me a little while to get familiar with the codebase and things..!

No pressure! Being motivated is the most important :) So I am glad you are. This is really awesome! And dont worry, you're not alone so if you dont understand something or need help, feel free to Slack me, I'll be happy to help

[Rafnel]

Hi there, my team uses Keycloak + Airflow. I'm currently migrating us from Airflow 2.6.3 to the current Airflow version, 2.8.4. I have a working (post-migration) webserver_config.py I can share in case that helps anyone. I believe this is still using the "old FAB" route. Not sure if any of this is worth adding to the repo in some way.

"""
Copy this to the {AIRFLOW_HOME}/ directory. Often ~/airflow/.
This file configures Airflow to use Keycloak for authentication/authorization.
"""

from __future__ import annotations

import os
import logging
import jwt
import requests
from base64 import b64decode
from cryptography.hazmat.primitives import serialization
from airflow.auth.managers.fab.security_manager.override import AUTH_OAUTH
from airflow.auth.managers.fab.security_manager.override import (
    FabAirflowSecurityManagerOverride,
)

basedir = os.path.abspath(os.path.dirname(__file__))
log = logging.getLogger(__name__)

# Flask-WTF flag for CSRF
WTF_CSRF_ENABLED = True
WTF_CSRF_TIME_LIMIT = None

# ----------------------------------------------------
# AUTHENTICATION CONFIG
# ----------------------------------------------------
# For details on how to set up each of the following, see
# http://flask-appbuilder.readthedocs.io/en/latest/security.html#authentication-methods

# The authentication type
# AUTH_OID : Is for OpenID
# AUTH_DB : For Airflow DB (Default)
# AUTH_LDAP : Is for LDAP
# AUTH_REMOTE_USER : Is for using REMOTE_USER from web server
# AUTH_OAUTH : Is for OAuth <- Keycloak uses OAuth
AUTH_TYPE = AUTH_OAUTH

# Required for Airflow to recognize users on their first time authenticating with Keycloak
AUTH_USER_REGISTRATION = True

# The default role if the user hasn't been assigned one in Keycloak
AUTH_USER_REGISTRATION_ROLE = "Public"

AUTH_ROLES_SYNC_AT_LOGIN = True
# airflow_xyz is the arbitrarily-chosen name of the role defined in Keycloak.
# Whereas "Admin", "Op", etc. are defined by Airflow: https://airflow.apache.org/docs/apache-airflow/stable/security/access-control.html
AUTH_ROLES_MAPPING = {
    "airflow_admin": ["Admin"],
    "airflow_op": ["Op"],
    "airflow_user": ["User"],
    "airflow_viewer": ["Viewer"],
    "airflow_public": ["Public"],
    # add custom roles here 
}
# Keycloak variables
PROVIDER_NAME = "keycloak"
CLIENT_ID = "YOUR_CLIENT_ID"
CLIENT_SECRET = "YOUR_CLIENT_SECRET"

OIDC_ISSUER = "https://your.keycloak.site/realms/yourrealm"
OIDC_BASE_URL = "{oidc_issuer}/protocol/openid-connect".format(oidc_issuer=OIDC_ISSUER)
OIDC_TOKEN_URL = "{oidc_base_url}/token".format(oidc_base_url=OIDC_BASE_URL)
OIDC_AUTH_URL = "{oidc_base_url}/auth".format(oidc_base_url=OIDC_BASE_URL)

OAUTH_PROVIDERS = [
    {
        "name": PROVIDER_NAME,
        "token_key": "access_token",
        "icon": "fa-circle-o",
        "remote_app": {
            "api_base_url": f"{OIDC_BASE_URL}/",
            "access_token_url": OIDC_TOKEN_URL,
            "authorize_url": OIDC_AUTH_URL,
            "request_token_url": None,
            "client_id": CLIENT_ID,
            "client_secret": CLIENT_SECRET,
            "client_kwargs": {"scope": "email profile"},
        },
    }
]

# Get public key of your Keycloak instance
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):
    # Airflow expects to receive a user info dict containing the logged-in user's
    # username, email, first_name, last_name, and role_keys. Add custom parsing logic to get that from the JWT from Keycloak.
    def oauth_user_info(self, provider, response):
        if provider == PROVIDER_NAME:
            token = response["access_token"]  # Their JWT
            decoded_jwt = jwt.decode(
                token, public_key, algorithms=["HS256", "RS256"], audience=CLIENT_ID
            )
            # Sample of how to find user's roles in decoded JWT
            # {
            #   "resource_access": { "YOUR_CLIENT_ID": { "roles": ["airflow_admin"] }}
            # }
            groups = decoded_jwt["resource_access"][CLIENT_ID]["roles"]
            if not groups:
                # If user hasn't been assigned a role, give them the default minimal access role.
                groups = ["airflow_public"]
            else:
                groups = [str for str in groups if "airflow" in str]
            user_info = {
                "username": decoded_jwt.get("preferred_username"),
                "email": decoded_jwt.get("email"),
                "first_name": decoded_jwt.get("given_name"),
                "last_name": decoded_jwt.get("family_name"),
                "role_keys": groups,
            }
            # These logs show up in stdout when running the webserver
            log.info(f"User is logging in with info: {user_info}")
            return user_info
        else:
            return {}


SECURITY_MANAGER_CLASS = CustomSecurityManager