如何在具有联合凭据身份验证的 Github Actions 工作流程中运行 Python Azure SDK

问题描述 投票:0回答:1

我有一个Python工具想要与Azure SDK交互(列出和上传blob)。通常我会使用类似的东西:

from azure.identity import DefaultAzureCredential
from azure.storage.blob import BlobServiceClient

creds = DefaultAzureCredential()
client = BlobServiceClient(..., credential=creds)

考虑到我依赖通过 CLI 中当前登录的用户进行身份验证,我想在 GitHub Action 工作流程中运行此脚本。

问题是 GitHub 操作中的身份验证方法必须通过联合凭据。

我有一个配置了联合身份的服务主体,我可以使用

azure/login
操作进行身份验证和登录。问题是,在必须运行 Python 脚本的步骤中,上一步中登录的 CLI 的上下文似乎没有保留,因为我看到以下错误:

azure.core.exceptions.HttpResponseError: This request is not authorized to perform this operation using this permission.

我的工作流程如下(简化):

jobs:
  execute:
    runs-on: ubuntu-latest
    steps:
      # checkout and other stuff
      # ...

      - name: AZ Login
        uses: Azure/login@v1
        with:
          client-id: ${{ vars.AZURE_CLIENT_ID }}
          subscription-id: ${{ vars.AZURE_SUBSCRIPTION_ID }}
          tenant-id: ${{ vars.AZURE_TENANT_ID }}

      # setup python and install packages
      # ...

      - name: Run script
        run: |
          .venv/bin/python script.py

问题是:如何将凭证从

azure/login
步骤传播到 python 运行时环境中?

python azure federated-identity azure-sdk
1个回答
0
投票
This request is not authorized to perform this operation using this permission.

如果服务主体没有读取或执行操作的权限,您将收到上述错误。要解决此错误,请将

storage data contributor
角色添加到 Blob 存储帐户的服务原则中,如下所示: 步骤1:进入存储账户的IAM,点击
Add
,选择
Add role assignment
,如下图:

enter image description here

Step2:搜索

storage data contributor
角色并选择,如下图:

enter image description here

Step3:选择服务原则如下:

enter image description here

成功分配角色后,您将能够使用以下代码读取文件或列出容器中的文件:

import io
from azure.storage.blob import BlobServiceClient
import pandas as pd
from azure.identity import ClientSecretCredential

account_url = "https://<storageAccountName>.blob.core.windows.net"
container_name = "<containerNmae>"
file_path = "<filePath>"
TENANT_ID = "<tenantId>"
CLIENT_ID = "clientId"
CLIENT_SECRET = "clientSecret"
credentials = ClientSecretCredential(TENANT_ID, CLIENT_ID, CLIENT_SECRET)
blob_service_client = BlobServiceClient(account_url=account_url, credential=credentials)
container_client = blob_service_client.get_container_client(container_name)
blobs_list = container_client.list_blobs()
for blob in blobs_list:
    print(blob.name)
blob_client = container_client.get_blob_client(file_path) 
blob_data = blob_client.download_blob()
csv_data = blob_data.content_as_text()
df = pd.read_csv(io.StringIO(csv_data))
print("Sample.CSV data:")
print(df)

enter image description here

© www.soinside.com 2019 - 2024. All rights reserved.