触发python azure函数从密钥库获取机密

Question

我尝试运行由blob触发的pythonazure函数，该函数将获取保存在密钥库中的个人访问令牌，并运行DevOps管道。我在本地测试了代码，它工作正常，当我将代码包含在init.py文件中时，它不会触发管道。甚至无法调试代码，因为没有提供太多信息

下面是部署前在init.py文件中编写的代码，我在requirement.txt文件中提供了所需的库

    import logging
    from azure.devops.connection import Connection
    from msrest.authentication import BasicAuthentication
    import azure.functions as func

    from azure.identity import ManagedIdentityCredential
    from azure.keyvault.secrets import SecretClient

    credentials = ManagedIdentityCredential()

    secret_client = SecretClient(vault_url="https://myKeyvault.vault.azure.net", credential=credentials)
    Personal_Access_Token = secret_client.get_secret("devops-token")
    print(Personal_Access_Token.value)

    Organization_URL = 'https://dev.azure.com/org/'
    Project_Name = 'ProjectName'

    def create_pipeline_client():
        credentials = BasicAuthentication('',Personal_Access_Token.value)
        connection = Connection(base_url=Organization_URL,creds=credentials)
        pipeline_client = connection.clients_v6_0.get_pipelines_client()
        return pipeline_client
        
    def build_pipeline(pipeline_id,run_params,pipeline_version=None):
        pipeline_client = create_pipeline_client()
        print("Running Pipeline with ID : "+ str(pipeline_id))
        try:
            pipeline_client.run_pipeline(run_parameters=run_params,project=Project_Name,pipeline_id=pipeline_id,pipeline_version=pipeline_version)
            print("Pipeline Run sucessfully activated")
        except Exception as ex:
            print("Pipeline Failed with Exception : " + str(ex))


    def get_pipeline(pipeline_id,pipeline_version=None):
        pipeline_client = create_pipeline_client()
        pipeline = pipeline_client.get_pipeline(project=Project_Name,pipeline_id=pipeline_id,pipeline_version=pipeline_version)
        print(pipeline)
        
    def list_pipelines():
        pipeline_client = create_pipeline_client()
        pipeline_list = pipeline_client.list_pipelines(Project_Name)
        for item in pipeline_list:
            print(item)

    def main(myblob: func.InputStream):
        logging.info(f"Python blob trigger function processed blob \n"
                     f"Name: {myblob.name}\n"
                     f"Blob Size: {myblob.length} bytes")

        run_params = {'branch/tag':'master'}
        build_pipeline(1,run_params,None)

请引导我

Answer 1

在azure功能的identity下创建系统分配/用户分配的访问对象ID（SP）

创建访问策略，并在密钥库中对上述SP进行必要的访问