使用自定义角色授权 Airflow 2.0 和 Azure AD - 覆盖 AirflowSecurityManager 中的 oauth_user_info 方法答案

【问题标题】：Authorization for Airflow 2.0 and Azure AD using custom role - Override oauth_user_info method in AirflowSecurityManager使用自定义角色授权 Airflow 2.0 和 Azure AD - 覆盖 AirflowSecurityManager 中的 oauth_user_info 方法
【发布时间】：2021-07-14 03:29:15
【问题描述】：

我在 Azure 上部署 Airflow 2.0 并使用 Azure AD 进行身份验证和授权。创建应用注册和自定义角色。使用 AD 进行身份验证有效。需要读取令牌中的角色声明以进行角色映射.. 下面是我的 Web 配置...为此，我将覆盖 AirflowSecurityManager 中的 oauth_user_info 方法以读取 Azure AD 令牌中的角色声明并将其分配给 role_keys.. 问题是当我这样做并分配'SECURITY_MANAGER_CLASS = AzureCustomSecurity，虽然Web UI POD没有任何错误但当我测试时似乎根本没有调用自定义类方法oauth_user_info......如果我删除自定义类然后我可以查看身份验证工作，我可以在日志中看到令牌....所以问题是在身份验证和授权过程中未调用此自定义类。任何人都可以在这里提供帮助？

我可以在此处的帖子中看到类似的东西适用于 AWS congnito AWS Cognito OAuth configuration for Flask Appbuilder

这是我的网络配置

import os
import json
from airflow.configuration import conf
from flask_appbuilder.security.manager import AUTH_OAUTH
from airflow.www.security import AirflowSecurityManager
from customsecmanager import AzureDlabSecurity
SQLALCHEMY_DATABASE_URI = conf.get("core", "SQL_ALCHEMY_CONN")
basedir = os.path.abspath(os.path.dirname(__file__))
CSRF_ENABLED = True
AUTH_TYPE = AUTH_OAUTH
AUTH_USER_REGISTRATION_ROLE = "Public"
AUTH_USER_REGISTRATION = True

class AzureCustomSecurity(AirflowSecurityManager):
  def oauth_user_info(self, provider, response=None):
    log.debug("inside custom method")
    if provider == "azure":
        log.debug("Azure response received : {0}".format(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))
        return {
            "name": me["name"],
            "email": me["upn"],
            "first_name": me["given_name"],
            "last_name": me["family_name"],
            "id": me["oid"],
            "username": me["oid"],
            "role_keys": me["roles"],       
          }    
    else:
        return {}


OAUTH_PROVIDERS = [{
    'name':'azure',
    'token_key':'access_token',
    'icon':'fa-windows',
    'remote_app': {
        'base_url':'https://graph.microsoft.com/v1.0/',
        'request_token_params' :{'scope': 'openid'},
        'access_token_url':'https://login.microsoftonline.com/<tenantid>/oauth2/token',
        'authorize_url':'https://login.microsoftonline.com/<tenantid>/oauth2/authorize',
        'request_token_url': None,
        'client_id':'XXXXXXXX',
        'client_secret':'******'
        }
}]

# a mapping from the values of `userinfo["role_keys"]` to a list of FAB roles
AUTH_ROLES_MAPPING = {
     "Viewer": ["Viewer"],
     "Admin": ["Admin"],
#}    
AUTH_ROLES_SYNC_AT_LOGIN = True
SECURITY_MANAGER_CLASS = AzureCustomSecurity

【问题讨论】：

你有没有找到整合它的方法？我有点类似的问题，我在 Google IAP 后面为 Airflow 编写了一个身份验证方法，现在我无法将它集成回来。
我从 Astronomer 那里找到了这个实现，它基于来自外部代理的 JWT 令牌，非常有用：github.com/astronomer/astronomer-fab-securitymanager/blob/…

标签： azure airflow

【解决方案1】：

我试图重现您的代码，但我没有覆盖oauth_user_info，而是覆盖了get_oauth_user_info 方法（docs link）并且它起作用了。你可以试试这个吗？

class AzureCustomSecurity(AirflowSecurityManager):
  def get_oauth_user_info(self, provider, resp):
     """ you code here... """

【讨论】：