python - 使用托管身份在 python 中被 azure 函数应用程序卡住

Question

我正在尝试编写函数应用程序代码，该应用程序将从日志分析工作区获取数据并使用 python3 推送到事件中心。函数应用程序使用托管身份。我正在使用适用于Python的azure sdk。我当前的代码如下所示:

def getAzureEventData():
    """
    if "MSI_ENDPOINT" in os.environ:
        print("GeTTING MSI Authentication")
        creds = MSIAuthentication()
    else:
        creds, *_ = get_azure_cli_credentials()         
    """

   ## want to find out which one is correct tested each one.
    creds = DefaultAzureCredential()
    creds=CredentialWrapper()
    creds = MSIAuthentication()
    #creds, _ = get_azure_cli_credentials(resource="https://api.loganalytics.io")

    log_client = LogAnalyticsDataClient(creds)
    
    laQuery = 'ActivityLog | where TimeGenerated > ago(1d)'
    result = log_client.query(cisalog_workspace_id, QueryBody(query=laQuery))

根据我见过的例子，

creds, _ = get_azure_cli_credentials(resource="https://api.loganalytics.io")

使用了

，但是当我在没有任何 DefaultCredential() 的情况下使用该函数时，我收到 404 错误，表示未启用系统管理身份。当我使用 DefualtCrednetial 时，我收到 access_token 错误，并且根据建议我使用在互联网上找到的包装器。当我使用它时，我得到异常:ErrorResponseException:(InvalidTokenError)提供的身份验证对此资源无效。所以我很困惑如何使用 Loganalytics SDK 客户端。我正在本地和门户中进行测试。我的最终目标是一个功能应用程序，使用系统管理的身份和 IAM 角色来访问 LA 工作区。我已将工作区上的“监视读者”角色授予 SMI。仍然面临问题。

Answer 1

如果要使用 AzureMSI 调用 Azure 函数中的 Azure Log Analytics Rest API，则需要将 Azure RABC 角色 Log Analytics Reader 分配给 MSI。更多详情请引用here .

例如

1. Enable Azure Function MSI

2. 分配角色

New-AzRoleAssignment -ObjectId "<the objectId of Azure function MSI>" -RoleDefinitionName "Log Analytics Reader" -Scope "/subscriptions/{subId}"

代码

我的cred_wrapper.py

from msrest.authentication import BasicTokenAuthentication
from azure.core.pipeline.policies import BearerTokenCredentialPolicy
from azure.core.pipeline import PipelineRequest, PipelineContext
from azure.core.pipeline.transport import HttpRequest

from azure.identity import DefaultAzureCredential


class CredentialWrapper(BasicTokenAuthentication):
    def __init__(self, credential=None, resource_id="https://westus2.api.loganalytics.io/.default", **kwargs):
        """Wrap any azure-identity credential to work with SDK that needs azure.common.credentials/msrestazure.
        Default resource is ARM (syntax of endpoint v2)
        :param credential: Any azure-identity credential (DefaultAzureCredential by default)
        :param str resource_id: The scope to use to get the token (default ARM)
        """
        super(CredentialWrapper, self).__init__(None)
        if credential is None:
            #credential = DefaultAzureCredential()
            credential = DefaultAzureCredential()
        self._policy = BearerTokenCredentialPolicy(
            credential, resource_id, **kwargs)

    def _make_request(self):
        return PipelineRequest(
            HttpRequest(
                "CredentialWrapper",
                "https://fakeurl"
            ),
            PipelineContext(None)
        )

    def set_token(self):
        """Ask the azure-core BearerTokenCredentialPolicy policy to get a token.
        Using the policy gives us for free the caching system of azure-core.
        We could make this code simpler by using private method, but by definition
        I can't assure they will be there forever, so mocking a fake call to the policy
        to extract the token, using 100% public API."""
        request = self._make_request()
        self._policy.on_request(request)
        # Read Authorization, and get the second part after Bearer
        token = request.http_request.headers["Authorization"].split(" ", 1)[1]
        self.token = {"access_token": token}

    def signed_session(self, session=None):
        self.set_token()
        return super(CredentialWrapper, self).signed_session(session)

我的函数代码

import logging
from azure.loganalytics import LogAnalyticsDataClient
from .cred_wrapper import CredentialWrapper
import azure.functions as func
from azure.loganalytics.models import QueryBody
import json


def main(req: func.HttpRequest) -> func.HttpResponse:
    logging.info('Python HTTP trigger function processed a request.')

    creds = CredentialWrapper()
    client = LogAnalyticsDataClient(creds)
    result = client.query(workspace_id='',
                          body=QueryBody(query='Heartbeat | take 10'))
    return func.HttpResponse(
        json.dumps(result.tables[0].rows),
        status_code=200
    )