如何在Python中使用用户分配托管标识对Kusto进行身份验证？

1. 先确认托管标识的Kusto权限 #

确保你持有的用户分配托管标识已被授予Kusto集群/数据库的对应权限（比如查询用Viewer、写入用Ingestor），可以通过Kusto查询窗口执行命令配置：

# 给数据库分配查看权限
.add database <你的数据库名> viewers ('aadapp=<托管标识clientId>')

# 给集群分配查看权限（如果需要集群级访问）
.add cluster viewers ('aadapp=<托管标识clientId>')

2. 把托管标识关联到Azure函数 #

门户操作 #

进入你的函数应用 -> 身份标识 -> 用户分配的标识 -> 添加，搜索并选中目标托管标识后保存。

AZ CLI操作 #

az functionapp identity assign --resource-group <你的资源组名> --name <函数应用名> --identities <托管标识principalId>

3. 在函数代码中实现身份验证 #

根据你使用的语言，结合Kusto SDK和托管标识凭据完成认证：

Python示例（用azure-kusto-data库） #

from azure.kusto.data import KustoClient, KustoConnectionStringBuilder
from azure.identity import ManagedIdentityCredential

# 指定用户分配托管标识的clientId初始化凭据
credential = ManagedIdentityCredential(client_id="<托管标识clientId>")
# 构建Kusto连接字符串
kcsb = KustoConnectionStringBuilder.with_aad_managed_service_identity_authentication(
    "<Kusto集群URL，格式：https://xxx.kusto.windows.net>"
)
kcsb.set_credential(credential)

# 创建客户端并执行查询
client = KustoClient(kcsb)
response = client.execute_query("<你的数据库名>", "StormEvents | take 10")

C#示例（用Microsoft.Azure.Kusto.Data库） #

using Microsoft.Azure.Kusto.Data;
using Microsoft.Azure.Kusto.Data.Common;

var clientId = "<托管标识clientId>";
var clusterUri = new Uri("<Kusto集群URL>");

// 基于用户分配托管标识创建认证提供器
var authProvider = new ManagedIdentityAuthenticationProvider(clientId);
var kustoClient = KustoClientFactory.CreateClient(clusterUri, authProvider);

// 执行查询
var query = "StormEvents | take 10";
var database = "<你的数据库名>";
var response = await kustoClient.ExecuteQueryAsync(database, query);

4. 验证身份是否生效 #

部署函数后，通过执行查询或查看函数日志确认：

• 若能正常返回Kusto数据，说明认证成功；
• 若出现权限错误，检查Kusto的角色分配是否正确；
• 若出现标识未找到的错误，确认函数应用已正确关联目标托管标识。

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