使用Python创建Azure Key Vault遇权限问题及解决

问题场景 #

我使用一段Python代码创建Azure Key Vault，已为服务主体分配Contribute角色，仅将代码中的Object ID替换为我的服务主体对应ID，但运行时出现权限不足的错误。

原报错代码 #

import os
import json
from azure.common.credentials import ServicePrincipalCredentials
from azure.mgmt.keyvault import KeyVaultManagementClient
from azure.mgmt.resource.resources import ResourceManagementClient
from haikunator import Haikunator

haikunator = Haikunator()

WEST_US = "westus"
GROUP_NAME = "azure-sample-group"
KV_NAME = haikunator.haikunate()
# 访问策略对应的用户或应用程序的对象ID，可在Azure门户中查找
OBJECT_ID = "401e9294-xxxx-xxxx-xxxx-xxxx"

# 管理资源和资源组 - 创建、更新和删除资源组，
# 将解决方案部署到资源组，导出ARM模板。创建、读取、更新
# 和删除资源
#
# 此脚本要求设置以下环境变量：
#
# AZURE_TENANT_ID: Azure Active Directory租户ID或域名
# AZURE_CLIENT_ID: Azure Active Directory应用程序客户端ID
# AZURE_CLIENT_SECRET: Azure Active Directory应用程序密钥
# AZURE_SUBSCRIPTION_ID: Azure订阅ID
#


def run_example():
    """资源组管理示例。"""
    # 使用应用程序（服务主体）令牌提供程序创建资源管理器客户端
    subscription_id = os.environ["AZURE_SUBSCRIPTION_ID"]

    credentials = ServicePrincipalCredentials(
        client_id=os.environ["AZURE_CLIENT_ID"],
        secret=os.environ["AZURE_CLIENT_SECRET"],
        tenant=os.environ["AZURE_TENANT_ID"],
    )
    kv_client = KeyVaultManagementClient(credentials, subscription_id)
    resource_client = ResourceManagementClient(credentials, subscription_id)

    # 你可能需要为这些凭据注册KeyVault作为有效提供程序
    # 如果需要，此操作只需为每个凭据执行一次
    resource_client.providers.register("Microsoft.KeyVault")

    # 创建资源组
    print("\n创建资源组")
    resource_group_params = {"location": WEST_US}
    print_item(
        resource_client.resource_groups.create_or_update(
            GROUP_NAME, resource_group_params
        )
    )

    # 创建密钥保管库
    print("\n创建密钥保管库")
    vault = kv_client.vaults.create_or_update(
        GROUP_NAME,
        KV_NAME,
        {
            "location": WEST_US,
            "properties": {
                "sku": {"name": "standard"},
                "tenant_id": os.environ["AZURE_TENANT_ID"],
                "access_policies": [
                    {
                        "tenant_id": os.environ["AZURE_TENANT_ID"],
                        "object_id": OBJECT_ID,
                        "permissions": {"keys": ["all"], "secrets": ["all"]},
                    }
                ],
            },
        },
    )
    print_item(vault)

    # 列出密钥保管库
    print("\n列出密钥保管库")
    for vault in kv_client.vaults.list():
        print_item(vault)

    # 删除资源组及其中所有资源
    print("\n删除资源组")
    delete_async_operation = resource_client.resource_groups.delete(GROUP_NAME)
    delete_async_operation.wait()
    print("\n已删除: {}".format(GROUP_NAME))


def print_item(group):
    """打印实例信息。"""
    print("\t名称: {}".format(group.name))
    print("\tID: {}".format(group.id))
    print("\t位置: {}".format(group.location))
    print("\t标签: {}".format(group.tags))


if __name__ == "__main__":
    run_example()

报错信息 #

权限不足，无法执行操作。

解决方法 #

更换为新版Azure SDK认证方式，使用DefaultAzureCredential替代旧的ServicePrincipalCredentials，并调整Key Vault创建的API调用（使用异步方法begin_create_or_update），以下是可用代码（需确保已正确配置Azure环境变量）：

import os

from azure.identity import DefaultAzureCredential
from azure.mgmt.keyvault import KeyVaultManagementClient
from azure.mgmt.resource import ResourceManagementClient


def main():
    TENANT_ID = os.environ.get("AZURE_TENANT_ID", None)
    SUBSCRIPTION_ID = os.environ.get("AZURE_SUBSCRIPTION_ID", None)
    GROUP_NAME = "resource_group_name"
    VAULT = "vault_name"
    LOCATION = "azure_location_eg_westus"
    OBJECT_ID = "service_principal_object_id"

    # 创建客户端
    resource_client = ResourceManagementClient(
        credential=DefaultAzureCredential(), subscription_id=SUBSCRIPTION_ID
    )
    keyvault_client = KeyVaultManagementClient(
        credential=DefaultAzureCredential(), subscription_id=SUBSCRIPTION_ID
    )

    # 创建资源组
    resource_client.resource_groups.create_or_update(GROUP_NAME, {"location": LOCATION})

    # 创建密钥保管库
    vault = keyvault_client.vaults.begin_create_or_update(
        GROUP_NAME,
        VAULT,
        {
            "location": LOCATION,
            "properties": {
                "tenant_id": TENANT_ID,
                "sku": {"family": "A", "name": "standard"},
                "access_policies": [
                    {
                        "tenant_id": TENANT_ID,
                        "object_id": OBJECT_ID,
                        "permissions": {
                            "keys": [
                                "encrypt",
                                "decrypt",
                                "wrapKey",
                                "unwrapKey",
                                "sign",
                                "verify",
                                "get",
                                "list",
                                "create",
                                "update",
                                "import",
                                "delete",
                                "backup",
                                "restore",
                                "recover",
                                "purge",
                            ],
                            "secrets": [
                                "get",
                                "list",
                                "set",
                                "delete",
                                "backup",
                                "restore",
                                "recover",
                                "purge",
                            ],
                        },
                    }
                ],
                "enabled_for_deployment": True,
                "enabled_for_disk_encryption": True,
                "enabled_for_template_deployment": True,
            },
        },
    ).result()
    print("创建密钥保管库:\n{}".format(vault))

    # 获取密钥保管库信息
    vault = keyvault_client.vaults.get(GROUP_NAME, VAULT)
    print("获取密钥保管库:\n{}".format(vault))

    # 更新密钥保管库
    vault = keyvault_client.vaults.update(
        GROUP_NAME, VAULT, {"tags": {"category": "Marketing"}}
    )
    print("更新密钥保管库:\n{}".format(vault))

    # 删除密钥保管库
    keyvault_client.vaults.delete(GROUP_NAME, VAULT)
    print("删除密钥保管库.\n")

    # 清除已删除的密钥保管库
    keyvault_client.vaults.begin_purge_deleted(VAULT, LOCATION).result()
    print("清除已删除的密钥保管库.\n")

    # 删除资源组
    resource_client.resource_groups.begin_delete(GROUP_NAME).result()


if __name__ == "__main__":
    main()

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