Use azure-identity to authenticate mgmt SDKs that needs "msrestazure/azure.common.credentials" protocol

cred_wrapper.py

# Wrap credentials from azure-identity to be compatible with SDK that needs msrestazure or azure.common.credentials
# Need msrest >= 0.6.0
# See also https://pypi.org/project/azure-identity/

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://management.azure.com/.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()
        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)

if __name__ == "__main__":
    import os
    credentials = CredentialWrapper()
    subscription_id = os.environ.get("AZURE_SUBSCRIPTION_ID", "<subscription_id>")

    from azure.mgmt.resource import ResourceManagementClient
    client = ResourceManagementClient(credentials, subscription_id)
    for rg in client.resource_groups.list():
        print(rg.name)

Hi @lmazuel,

Thanks for your work!

As recommended by MS Azure Documentation, I tried to use it in a py program to authenticate from an Azure CentOS VM (with managed identity) to access to a Azure Datalake repo, based on Azure Python SDK (especially azure.datalake.store.core.AzureDLFileSystem feature).
Here is my code :

from cred_wrapper import CredentialWrapper
from azure.datalake.store import core, lib, multithread
credential = CredentialWrapper()
adl = core.AzureDLFileSystem(token=credential, store_name = "<NAME OF MY DATALAKE STORE>")
print(adl.walk('<NAME OF A FOLDER AT ROOT OF MY DATALAKE STORE>'))

I have no output and the program retries infinitely at the adl.walk step.
However, when I use a default authentication strategy (with tenant_id, client_id, secret_id) it works. For example this works:

from azure.datalake.store import core, lib, multithread
token = lib.auth(tenant_id = azrTenantId, client_secret = azrSClientId, client_id = azrClientId)
adl = core.AzureDLFileSystem(token, store_name = azrStore)
for i in adl.ls(obj):
   print(i)

Would you have a clue of my mistake?
Would you have examples to use your program in a DataLake context with managed identity authentication?

Thanks for your help,

Adrien

Hey @adriencoa

We made an enhanced version of this gist here:
https://github.com/jongio/azidext

Could you see if that works, or create an issue there instead if it's not ?Thanks!

Hi there,

I suspect this is a similar issue, hence I'd like to post it here...

I am trying to use the ManagedIdentityCredential(client_id="user_assigned_id_client_id") constructor to pass the returned object to a wrapper class that constructs a DnsManagementClient object so I can list all DNS zone CName records in a particular DNS Zone resource.

The DNS Client, is constructed usingDnsManagementClient(credentials=user_assigned_identity_client_id).

Whenever I try to use the dns_client.records.list_by_dns_zone(resource_group, dns_zone) method, I keep getting:

ManagedIdentityCredential has no attribute 'signed_session'

error.

By reading this post, it looks like I'll have to replace the ManagedIdentityCredential with the cred_wrapper.py file... It would be great to see azure.mgmt.dns supporting ManagedIdentityCredential.

Thank you!

Hello, adding the main issue that references this need for clarity: Azure/azure-sdk-for-python#9310

100% of Python SDKs have now support for azure-identity natively, this snippet is not recommended nor necessary anymore as of today

Except batch? Azure/azure-sdk-for-python#28739 (comment)

Yes, that's correct, Batch is in the process of doing it, they are actively working on it. I forgot batch I admit :)

Cool thanks for confirmation, can you link to the PR? I'm using your workaround, but would love to get rid of the msrest dependency in
from msrest.authentication import BasicTokenAuthentication

Hello! i still need to use CredentialWrapper with
azure.mgmt.sqlvirtualmachine.SqlVirtualMachineManagementClient and
azure.mgmt.documentdb.DocumentDB
Are they planned for update?