如何使用 MFA（应用程序和用户）配置合作伙伴中心 API？

Question

我正在尝试按照此链接中的步骤操作： https://github.com/MicrosoftDocs/partner-rest/blob/docs/partner-rest/develop/api-authentication.md

但是创建应用并用户访问后，并没有显示如何调用API。 我正在尝试使用 Python/Spark 进行连接，但无法进一步进行，因为没有文档说明如何逐步进行连接。

下面的代码是仅应用程序身份验证的示例，它可以工作，但不允许获取一些需要应用程序+用户身份验证的数据。

import requests, json
from pyspark.sql import SparkSession
from pyspark.sql.types import StructField, StringType, IntegerType, StructType

access_var = {
    "auth_url": "https://login.microsoftonline.com/tenantid/oauth2/token",
    "resource": "https://graph.windows.net",
    "client_id": {mssparkutils.credentials.getSecret('https://vaultname.vault.azure.net/', 'KV-Client-ID')},
    "client_st": {mssparkutils.credentials.getSecret('https://vaultname.vault.azure.net/', 'KV-Client-Secret')},
    "grant_type": "client_credentials"
}

_schema = StructType([
    StructField("Tenant ID", StringType(), True),
    StructField("Domain", StringType(), True),
    StructField("Company", StringType(), True),
    StructField("URI", StringType(), True)
])

def get_access():
    _body = {
            "resource": access_var["resource"],
            "client_id": access_var["client_id"],
            "client_secret": access_var["client_st"],
            "grant_type": access_var["grant_type"]
    }

    _headers = {"Content-Type": "application/x-www-form-urlencoded"}
    response = requests.post(access_var["auth_url"], data=_body, headers=_headers)
    response.raise_for_status()
    return response.json()

response = get_access()
access_t = response["access_token"]

def catch(api_url):
    "https://api.partnercenter.microsoft.com/v1/customers"

def fetch():
    api_url = "https://api.partnercenter.microsoft.com/v1/customers"
    api_auth = {
        "Authorization": f"Bearer {access_t}",
        "Accept": "application/json"
    }
    api_response = requests.get(api_url, headers=api_auth)
    return api_response.json()

def create_df(spark, result_list):
    return spark.createDataFrame(result_list, schema=_schema)

if __name__ == "__main__":
    spark = SparkSession.builder.appName("API_CALL").getOrCreate()
    all_results = []

    data = fetch()
    result_list = data

    for item in result_list["items"]:
        tenant_id = item["companyProfile"].get("tenantId", "N/A").strip()
        domain = item["companyProfile"].get("domain", "N/A").strip()
        company_name = item["companyProfile"].get("companyName").strip()
        uri = item["links"]["self"]["uri"]

        all_results.append((tenant_id, domain, company_name, uri))
    
    df = create_df(spark, all_results)
    df.show()

Answer 1

启用用户的 MFA 时，您需要使用授权代码或交互流等委托流来为合作伙伴中心 API 生成令牌。

注册Azure AD应用程序并授予

API permissions

的委托类型，如下所示：

使用交互流时，您必须包含移动和桌面应用程序平台的重定向URI：

确保在您的应用程序注册中启用公共客户端流，如下所示：

现在，使用以下 Python 代码使用要求用户登录的交互式流程为合作伙伴中心 API 生成访问令牌：

import msal

tenant_id = "tenantId"
client_id = "appId"
authority = f'https://login.microsoftonline.com/{tenant_id}'

scope = ["https://api.partnercenter.microsoft.com/user_impersonation"]

app = msal.PublicClientApplication(
    client_id,
    authority=authority,
)

token_response = app.acquire_token_interactive(scopes=scope)
access_token = token_response['access_token']
print("Access token:", access_token)

回复：

您可以在 jwt.ms 中解码上述令牌并检查

aud

和

scp

声明是否具有 valid 值：

使用上述令牌，您可以使用以下更新的Python代码向合作伙伴中心API发出请求：

import msal

tenant_id = "tenantId"
client_id = "appId"
authority = f'https://login.microsoftonline.com/{tenant_id}'

scope = "https://api.partnercenter.microsoft.com/user_impersonation"

app = msal.PublicClientApplication(
    client_id,
    authority=authority,
)

token_response = app.acquire_token_interactive(scopes=scope)
access_token = token_response['access_token']

# Make a request to the Partner Center API
api_url = "https://api.partnercenter.microsoft.com/v1/customers"
headers = {
    'Authorization': 'Bearer ' + access_token,
    'Content-Type': 'application/json',
}

response = requests.get(api_url, headers=headers)

if response.status_code == 200:
    print("API request successful")
    print(response.json())
else:
    print(f"API request failed with status code {response.status_code}")
    print(response.text)

参考： 合作伙伴中心认证-合作伙伴应用开发者|微软