坚持使用python中的azure函数应用程序，使用托管标识

Question

我正在尝试编写一个函数应用程序，它将从日志分析工作区获取数据，并使用python3推送到事件中心。函数应用程序使用托管标识。我正在使用azure sdk for python。我当前的代码如下所示：

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时，我得到访问令牌错误，并且根据建议，我正在使用在internet上找到的包装器。当我使用它时，我得到异常：ErrorResponseException:（InvalidTokenError）提供的身份验证对此资源无效。 所以我很困惑如何使用LogAnalyticsSDK客户端。我在本地和门户网站上进行测试。我的最终目标是使用系统管理的身份和IAM角色来访问LA workspace的功能应用程序。我已将工作区上的监视读取器角色授予SMI。仍然面临这个问题

Answer 1

如果要在Azure函数中使用AzureMSI调用Azure日志分析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}"

1. 代码

我的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
    )