需要为 Airflow 配置统一身份认证，支持 LDAP 和 OAuth2 两种方式。

版本信息

  • App Version: 2.8.3
  • Chart Version: 1.13.1

OAuth 认证（推荐）

Authentik 配置要点

关键问题：userinfo/ 路径必须带斜杠，否则会报错：

1
ERROR - OAUTH userinfo does not have username or email {}

webserver_config.py

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
from flask_appbuilder.security.manager import AUTH_OAUTH
from airflow.auth.managers.fab.security_manager.override import FabAirflowSecurityManagerOverride

AUTH_TYPE = AUTH_OAUTH
AUTH_USER_REGISTRATION = True
AUTH_USER_REGISTRATION_ROLE = "User"  # 默认角色

OAUTH_PROVIDERS = [
    {
        "name": "OAuth",
        "icon": "fa-circle-o",
        "token_key": "access_token",
        "remote_app": {
            "client_id": "your-client-id",
            "api_base_url": "https://authentik.example.com/application/o/",
            "client_secret": "your-client-secret",
            "client_kwargs": {"scope": "openid email profile"},
            "server_metadata_url": "https://authentik.example.com/application/o/oauth/.well-known/openid-configuration",
        },
    }
]

PERMANENT_SESSION_LIFETIME = 259200  # 3 天


class CustomSecurityManager(FabAirflowSecurityManagerOverride):
    def get_oauth_user_info(self, provider, resp):
        if provider != "OAuth":
            return {}

        # 关键：必须是 userinfo/ 而不是 userinfo
        me = self.appbuilder.sm.oauth_remotes[provider].get("userinfo/")
        data = me.json()

        return {
            "email": data["email"],
            "username": data.get("name", ""),
            "first_name": data.get("given_name", ""),
            "role_keys": data.get("groups", []),  # 支持组映射
        }


SECURITY_MANAGER_CLASS = CustomSecurityManager

参考资料

LDAP 认证（备选）

webserver_config.py

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
from flask_appbuilder.security.manager import AUTH_LDAP

AUTH_TYPE = AUTH_LDAP
AUTH_LDAP_SERVER = "ldap://ldap.example.com"
AUTH_LDAP_USE_TLS = False

# 用户注册配置
AUTH_USER_REGISTRATION = True
AUTH_USER_REGISTRATION_ROLE = "Admin"

# 字段映射
AUTH_LDAP_LASTNAME_FIELD = "cn"
AUTH_LDAP_EMAIL_FIELD = "mail"  # 如果为空会设为 {username}@email.notfound

# 搜索配置
AUTH_LDAP_SEARCH = "ou=foo,ou=People,dc=example,dc=com"
AUTH_LDAP_UID_FIELD = "uid"
AUTH_LDAP_BIND_USER = "cn=admin,dc=example,dc=com"
AUTH_LDAP_BIND_PASSWORD = "adminPassword"

# 角色映射
AUTH_ROLES_MAPPING = {
    "cn=g-admin,ou=Group,dc=example,dc=com": ["Admin"],
}

# 同步配置
AUTH_LDAP_GROUP_FIELD = "memberOf"
AUTH_ROLES_SYNC_AT_LOGIN = True  # 每次登录同步角色
PERMANENT_SESSION_LIFETIME = 86400  # 1 天后重新认证

官方 LDAP 文档

API 认证配置

需要通过 API 访问 Airflow，可以启用 Basic Auth：

values.yaml

1
2
3
config:
  api:
    auth_backends: "airflow.api.auth.backend.basic_auth"

使用方式：

1
curl -u "username:password" https://airflow.example.com/api/v1/dags

API 认证文档

调试技巧

排查 OAuth 问题时可以临时添加调试代码：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
def get_oauth_user_info(self, provider, resp):
    import sys
    import traceback

    remote_app = self.appbuilder.sm.oauth_remotes[provider]
    print(f"remote_app: {remote_app}, vars: {vars(remote_app)}")

    try:
        me = remote_app.get("userinfo/")
        print(f"me: {me}, vars: {vars(me)}")
        user_data = me.json()
        print(f"user_data: {user_data}")
    except Exception as e:
        print(f"Error: {e}")
        traceback.print_exc()
        raise

    return {
        "email": user_data["email"],
        "username": user_data.get("name", ""),
    }

常见问题

重置管理员密码

参考 issue#37009

userinfo 路径问题

Authentik 的 userinfo 端点必须是 /userinfo/（带斜杠），这是最常见的坑。

注意事项

  • OAuth 方式更灵活，支持多种 SSO 平台
  • LDAP 方式适合已有 LDAP 基础设施的场景
  • 生产环境建议使用 角色映射 而非给所有用户 Admin 权限
  • API 认证和 Web 认证是独立的配置