如何通过REST API获取Microsoft Purview角色分配及用户/服务主体名称

我尝试用以下REST API从Microsoft Purview获取角色分配：

api_endpoint = f"https://{pv_account_name}.purview.azure.com/policystore/metadataroles"

这个API只能拿到角色ID和名称，无法获取用户、服务主体的名称。我写了Python脚本尝试解决，但脚本返回空集合，这是我第一次做这类操作，不知道怎么推进。

需要解决的问题：如何获取数据策展人、集合管理员等角色的分配信息，以及对应的用户/服务主体名称？

我尝试的Python脚本 #

import requests
import json

# Replace these with your actual values
pv_account_name = "purview"
api_version = "2021-07-01"

# Azure AD credentials
client_id = "b" 
client_secret = "L"
tenant_id = "e"
resource = "https://purview.azure.net"

# Construct the token request URL
token_url = f"https://login.microsoftonline.com/{tenant_id}/oauth2/token"

# Set up the token request parameters
token_params = {
    "client_id": client_id,
    "client_secret": client_secret,
    "grant_type": "client_credentials",
    "resource": resource
}

# Make the token request to get an access token
token_response = requests.post(token_url, data=token_params)

# Check the token response status
if token_response.status_code == 200:
    token_data = token_response.json()
    access_token = token_data.get("access_token")

    # Construct the API endpoint for role assignments
    api_endpoint = f"https://{pv_account_name}.purview.azure.com/policystore/metadataroles"


    # Set up headers with the access token
    headers = {
        "Authorization": f"Bearer {access_token}",
        "Content-Type": "application/json",
    }

    # Set up parameters for the request
    params = {
        "api-version": api_version,
        # Add any additional parameters as needed
    }

    # Make the GET request to retrieve role assignments
    response = requests.get(api_endpoint, headers=headers, params=params)

    # Check the response status
    if response.status_code == 200:
        role_assignments = response.json()

        # Initialize a list to store role assignments with user or service principal information
        role_assignments_with_names = []

        # Function to retrieve user or service principal information
        def get_user_info(unique_identifier):
            # Make a request to the Azure AD Graph API to retrieve user or service principal info
            user_info_endpoint = f"https://graph.microsoft.com/v1.0/users/{unique_identifier}"
            user_info_response = requests.get(user_info_endpoint, headers=headers)

            if user_info_response.status_code == 200:
                user_info = user_info_response.json()
                return user_info

        for assignment in role_assignments.get("values", []):
            unique_identifier = assignment.get("principalId")

            if unique_identifier:
                user_info = get_user_info(unique_identifier)

                if user_info:
                    # Combine role assignment and user info
                    role_assignment_with_name = {
                        "role_assignment": assignment,
                        "user_info": user_info,
                    }
                    role_assignments_with_names.append(role_assignment_with_name)

        # Now, role_assignments_with_names contains role assignments with user or service principal info
        print(role_assignments_with_names)

    else:
        print(f"Error: {response.status_code}")
        print(response.text)

else:
    print(f"Error: {token_response.status_code}")
    print(token_response.text)

问题分析与解决方案 #

你的脚本返回空集合主要有3个核心问题：

1. API端点错误：policystore/metadataroles仅返回角色定义（比如数据策展人、集合管理员这些角色本身的信息），不是角色分配记录，需调用/policystore/roleassignments接口获取分配数据。
2. 令牌资源不匹配：调用Purview的令牌是针对https://purview.azure.net的，无法直接用于Microsoft Graph API，需单独获取Graph API的令牌。
3. 主体查询逻辑缺失：Graph API的/users接口只能查用户，服务主体需要调用/servicePrincipals接口查询。

修正后的脚本 #

import requests
import json

# 替换为你的实际信息
pv_account_name = "purview"
api_version = "2021-07-01"
client_id = "你的客户端ID" 
client_secret = "你的客户端密钥"
tenant_id = "你的租户ID"

# 获取Purview API的令牌
def get_purview_token():
    token_url = f"https://login.microsoftonline.com/{tenant_id}/oauth2/token"
    token_params = {
        "client_id": client_id,
        "client_secret": client_secret,
        "grant_type": "client_credentials",
        "resource": "https://purview.azure.net"
    }
    response = requests.post(token_url, data=token_params)
    if response.status_code == 200:
        return response.json().get("access_token")
    else:
        print(f"获取Purview令牌失败: {response.status_code}")
        print(response.text)
        return None

# 获取Graph API的令牌
def get_graph_token():
    token_url = f"https://login.microsoftonline.com/{tenant_id}/oauth2/token"
    token_params = {
        "client_id": client_id,
        "client_secret": client_secret,
        "grant_type": "client_credentials",
        "resource": "https://graph.microsoft.com"
    }
    response = requests.post(token_url, data=token_params)
    if response.status_code == 200:
        return response.json().get("access_token")
    else:
        print(f"获取Graph令牌失败: {response.status_code}")
        print(response.text)
        return None

# 获取主体信息（用户或服务主体）
def get_principal_info(principal_id, graph_token):
    headers = {"Authorization": f"Bearer {graph_token}"}
    # 先尝试查询用户
    user_response = requests.get(f"https://graph.microsoft.com/v1.0/users/{principal_id}", headers=headers)
    if user_response.status_code == 200:
        return {"type": "用户", "info": user_response.json()}
    # 用户不存在则查询服务主体
    sp_response = requests.get(f"https://graph.microsoft.com/v1.0/servicePrincipals/{principal_id}", headers=headers)
    if sp_response.status_code == 200:
        return {"type": "服务主体", "info": sp_response.json()}
    return None

# 主逻辑
purview_token = get_purview_token()
graph_token = get_graph_token()

if purview_token and graph_token:
    # 调用角色分配接口
    api_endpoint = f"https://{pv_account_name}.purview.azure.com/policystore/roleassignments"
    headers = {
        "Authorization": f"Bearer {purview_token}",
        "Content-Type": "application/json",
    }
    params = {"api-version": api_version}
    response = requests.get(api_endpoint, headers=headers, params=params)

    if response.status_code == 200:
        role_assignments = response.json().get("values", [])
        result = []

        for assignment in role_assignments:
            principal_id = assignment.get("principalId")
            role_id = assignment.get("roleId")
            scope = assignment.get("scope")

            # 获取角色名称
            role_info_url = f"https://{pv_account_name}.purview.azure.com/policystore/metadataroles/{role_id}"
            role_response = requests.get(role_info_url, headers=headers, params=params)
            role_name = role_response.json().get("name") if role_response.status_code == 200 else "未知角色"

            # 获取主体信息
            principal_info = get_principal_info(principal_id, graph_token)

            if principal_info:
                result.append({
                    "角色名称": role_name,
                    "主体类型": principal_info["type"],
                    "主体名称": principal_info["info"].get("displayName"),
                    "主体ID": principal_id,
                    "分配范围": scope
                })

        print(json.dumps(result, ensure_ascii=False, indent=2))
    else:
        print(f"获取角色分配失败: {response.status_code}")
        print(response.text)
else:
    print("令牌获取失败，终止执行")

必要权限配置 #

1. Purview权限：你的应用需要在Purview中拥有「集合管理员」或「策略管理员」权限，才能读取角色分配数据。
2. Graph API权限：在Azure AD中给应用添加应用权限：Directory.Read.All（或User.Read.All+Application.Read.All），并完成管理员同意。

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