Airflow与Keycloak集成异常：未触发Keycloak认证求助

安全阿岳

2026-7-18

Airflow与Keycloak集成故障排查求助

我通过Helm在Kubernetes上部署了Airflow（Helm chart版本1.9.0，应用版本2.5.3）和Keycloak（Helm chart版本15.1.3，应用版本21.1.1），但遇到集成问题：无论怎么调整配置，Airflow始终使用内置认证，完全不触发Keycloak的认证流程。已经开启DEBUG日志级别，但日志里没有任何错误信息。试过Stack Overflow上的《Airflow authentication with RBAC and Key cloak》相关代码片段，问题依旧。

以下是当前使用的ConfigMap内容：

apiVersion: v1
data:
  airflow.cfg: |-
    [celery]
    flower_url_prefix = /
    worker_concurrency = 16

    [celery_kubernetes_executor]
    kubernetes_queue = kubernetes

    [core]
    colored_console_log = False
    dags_folder = /opt/airflow/dags
    executor = CeleryExecutor
    load_examples = False
    remote_logging = False

    [elasticsearch]
    json_format = True
    log_id_template = {dag_id}_{task_id}_{execution_date}_{try_number}

    [elasticsearch_configs]
    max_retries = 3
    retry_timeout = True
    timeout = 30

    [kerberos]
    ccache = /var/kerberos-ccache/cache
    keytab = /etc/airflow.keytab
    principal = airflow@FOO.COM
    reinit_frequency = 3600

    [kubernetes]
    airflow_configmap = airflow-airflow-config
    airflow_local_settings_configmap = airflow-airflow-config
    multi_namespace_mode = False
    namespace = airflow
    pod_template_file = /opt/airflow/pod_templates/pod_template_file.yaml
    worker_container_repository = apache/airflow
    worker_container_tag = 2.5.3

    [kubernetes_executor]
    multi_namespace_mode = False
    namespace = airflow
    pod_template_file = /opt/airflow/pod_templates/pod_template_file.yaml
    worker_container_repository = apache/airflow
    worker_container_tag = 2.5.3

    [logging]
    colored_console_log = False
    remote_logging = False
    logging_level = DEBUG
    fab_logging_level = DEBUG

    [metrics]
    statsd_host = airflow-statsd
    statsd_on = True
    statsd_port = 9125
    statsd_prefix = airflow

    [scheduler]
    run_duration = 41460
    standalone_dag_processor = False
    statsd_host = airflow-statsd
    statsd_on = True
    statsd_port = 9125
    statsd_prefix = airflow

    [webserver]
    enable_proxy_fix = True
    rbac = True
    authenticate = True
  airflow_local_settings.py: |2-

    import os
    import logging
    import re
    import base64
    import yaml
    from flask import session
    from airflow.www.security import AirflowSecurityManager
    from flask_appbuilder.security.manager import AUTH_OAUTH
    basedir = os.path.abspath(os.path.dirname(__file__))

    MY_PROVIDER = 'keycloak'
    LOGGING_LEVEL = logging.DEBUG

    class CustomSecurityManager(AirflowSecurityManager):
      def oauth_user_info(self, provider, resp):
        if provider == MY_PROVIDER:
          log.debug("{0} response received : {1}".format(provider,resp))
          id_token = resp["id_token"]
          log.debug(str(id_token))
          me = self._azure_jwt_token_parse(id_token)
          log.debug("Parse JWT token : {0}".format(me))

          if not me.get("name"):
            firstName = ""
            lastName  = ""
          else:
            firstName = me.get("name").split(' ')[0]
            lastName  = me.get("name").split(' ')[-1]
          return {
            "username":   me.get("email"),
            "email":      me.get("email"),
            "first_name": firstName,
            "last_name":  lastName,
            "role_keys":  me.get("groups", ['Guest'])
          }
        else:
          return {}

    log = logging.getLogger(__name__)

    AUTH_TYPE = AUTH_OAUTH
    AUTH_USER_REGISTRATION = True
    AUTH_USER_REGISTRATION_ROLE = "Guest"
    AUTH_ROLES_SYNC_AT_LOGIN = True
    CSRF_ENABLED = True

    AUTH_ROLES_MAPPING = {
        "Airflow_Users": ["User"],
        "Airflow_Admins": ["Admin"],
    }

    PROVIDER_NAME = 'keycloak'
    CLIENT_ID = 'airflow'
    CLIENT_SECRET = 'redacted'
    OIDC_ISSUER = 'redacted'
    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': 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': 'openid groups',
            'token_endpoint_auth_method': 'client_secret_post'
          },
          'access_token_method': 'POST',
        }
      }
    ]
    SECURITY_MANAGER_CLASS = CustomSecurityManager


    from airflow.www.utils import UIAlert

    DASHBOARD_UIALERTS = [
      UIAlert(
        'Usage of a dynamic webserver secret key detected. We recommend a static webserver secret key instead.'
        ' See the &lt;a href='
        '&quot;https://airflow.apache.org/docs/helm-chart/stable/production-guide.html#webserver-secret-key&quot;&gt;'
        'Helm Chart Production Guide&lt;/a&gt; for more details.',
        category="warning",
        roles=["Admin"],
        html=True,
      )
    ]

排查方向与修复建议

1. 补全airflow.cfg的关键配置

你的[webserver]段缺少auth_backend配置，必须添加：

auth_backend = airflow.www.security.auth_backend

确保authenticate = True和rbac = True已经生效，这两个是开启RBAC和外部认证的基础。

2. 修复airflow_local_settings.py的代码问题

日志初始化顺序错误：log = logging.getLogger(__name__)放在了CustomSecurityManager类之后，导致类内的log.debug会触发未定义错误，直接导致自定义SecurityManager加载失败，Airflow自动回退到内置认证。调整顺序：
```
log = logging.getLogger(__name__)
log.setLevel(logging.DEBUG)

class CustomSecurityManager(AirflowSecurityManager):
  # ... 原有类代码 ...
```

错误使用Azure专用JWT解析方法：self._azure_jwt_token_parse(id_token)是Azure AD的专用方法，Keycloak的JWT解析需要用通用方式，比如：

# 先导入jose库（需要确保镜像里安装了python-jose[cryptography]）
from jose import jwt

def oauth_user_info(self, provider, resp):
  if provider == MY_PROVIDER:
    id_token = resp["id_token"]
    # 生产环境必须验证签名，需要从Keycloak获取公钥，这里先禁用签名验证用于测试
    me = jwt.decode(id_token, options={"verify_signature": False})
    # ... 后续处理逻辑 ...

更可靠的方式是调用Keycloak的用户信息端点：

def oauth_user_info(self, provider, resp):
  if provider == MY_PROVIDER:
    access_token = resp['access_token']
    headers = {'Authorization': f'Bearer {access_token}'}
    # 调用Keycloak的用户信息端点
    user_info = self.remote_app.get('userinfo', headers=headers).json()
    # 处理user_info数据
    firstName = user_info.get("given_name", "")
    lastName = user_info.get("family_name", "")
    return {
      "username": user_info.get("email"),
      "email": user_info.get("email"),
      "first_name": firstName,
      "last_name": lastName,
      "role_keys": user_info.get("groups", ['Guest'])
    }

修正OAUTH_PROVIDERS配置：

api_base_url应该直接用Keycloak的领域URL（即OIDC_ISSUER），而不是拼接/protocol/openid-connect，因为remote_app会自动拼接端点路径
token_key改为id_token，因为OIDC流程中需要解析ID Token获取用户信息
调整后的配置示例：

OIDC_BASE_URL = OIDC_ISSUER
OIDC_TOKEN_URL = f"{OIDC_ISSUER}/protocol/openid-connect/token"
OIDC_AUTH_URL = f"{OIDC_ISSUER}/protocol/openid-connect/auth"

OAUTH_PROVIDERS = [{
  'name': PROVIDER_NAME,
  'token_key': 'id_token',
  'icon': 'fa-key',
  'remote_app': {
    'api_base_url': 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': 'openid email profile groups',
      'token_endpoint_auth_method': 'client_secret_post'
    },
    'access_token_method': 'POST',
  }
}]

3. 验证配置文件挂载有效性

执行以下命令检查Webserver Pod内的配置是否正确：

# 检查airflow.cfg内容
kubectl exec -n airflow <airflow-webserver-pod-name> -- cat /opt/airflow/airflow.cfg
# 检查airflow_local_settings.py内容
kubectl exec -n airflow <airflow-webserver-pod-name> -- cat /opt/airflow/config/airflow_local_settings.py

Helm部署的Airflow默认将自定义配置文件挂载到/opt/airflow/config/目录，必须确保该文件存在且内容正确。

4. 检查Keycloak客户端配置

确认客户端的Access Type设置为confidential（与client_secret_post认证方式匹配）
确认Valid Redirect URIs包含Airflow的完整URL，比如https://your-airflow-domain/*
确认开启了Authorization Code Flow（推荐）或Implicit Flow
确认groups作用域已在客户端的Scope中启用，且测试用户已分配对应组（如Airflow_Users、Airflow_Admins）

5. 查看Webserver启动日志

运行时DEBUG日志可能不包含启动阶段的错误，执行以下命令查看启动日志：

kubectl logs -n airflow <airflow-webserver-pod-name>

重点查找ImportError、AttributeError或与SecurityManager相关的报错，这些都会导致自定义认证逻辑加载失败，Airflow回退到内置认证。

内容的提问来源于stack exchange，提问作者orestlenczyk

Airflow与Keycloak集成异常：未触发Keycloak认证求助

排查方向与修复建议 #

1. 补全airflow.cfg的关键配置 #

2. 修复airflow_local_settings.py的代码问题 #

3. 验证配置文件挂载有效性 #

4. 检查Keycloak客户端配置 #