azure-sdk-for-python icon indicating copy to clipboard operation
azure-sdk-for-python copied to clipboard

SharedTokenCacheBase not identifying cache due to irrelevant error

Open kcphila opened this issue 4 months ago • 9 comments

  • azure_identity:
  • **Package **: 1.15
  • Operating System: Windows
  • Python Version: 3.11

Describe the bug Attempting to use azure.identity.SharedTokenCacheCredential fails because no accounts are found even when valid refresh tokens exist. This is because the refresh tokens do not have family_id

To Reproduce

See script excerpt function at the bottom.

Expected behavior

azure.identity.SharedTokenCacheCredential should be able to identify a single matching refresh token in the cache without the user having to manually construct an authentication records.

Solution

This is due to the check for family_id in shared_token_cache, around line 185.

There is a comment that says "there should always be a family id." In our runtime, the tokens do not have it. This can be solved by removing the family ID check. I'd be happy to submit a PR.

Additional context

Script to demonstrate the issue. We're using azure.identity for msgraph, and so I do include it but it's just to verify authentication and fetch a record.

import sys
import traceback
import asyncio
import azure, azure.identity
import msal
import msgraph

async def run_program(client_id, tenant_id, scopes):

    # Set up Cache Persistence

    azure_cache = azure.identity.TokenCachePersistenceOptions(
            name = 'azure_bug'
        )

    # Authenticate via Interactive Browser,
    # which should create the cache file
    interactive_cred = azure.identity.InteractiveBrowserCredential(
        client_id = client_id,
        tenant_id = tenant_id,
        redirect_uri='http://localhost:8000',
        cache_persistence_options = azure_cache,
    )
    interactive_client = msgraph.GraphServiceClient(
            interactive_cred,
            scopes,
        )
    verify = await interactive_client.me.drives.get()
    assert verify

    # Set up Shared Token Cache Cred with same cache details
    shared_credential = azure.identity.SharedTokenCacheCredential(
            #client_id = client_id,
            tenant_id = tenant_id,
            # Not sure why, but it seems like authority is required
            authority = 'https://login.microsoftonline.com/' + tenant_id,
            cache_persistence_options = azure_cache,
        )

    # --- this is going to fail, but we'll use it
    # to trigger the default cache instantiation in
    # the SharedToken cred
    shared_client = msgraph.GraphServiceClient(
        shared_credential,
        CONFIG['user_scopes'],
    )
    try:
        fails = await shared_client.me.drives.get()
        raise Exception("This fails in our example")
    except:
        print("Regular Authentication fails")

    # Look under the hood - verify it should work
    refresh_tokens = shared_credential._credential._get_cache_items_for_authority(msal.TokenCache.CredentialType.REFRESH_TOKEN, False)

    all_accounts =shared_credential._credential._get_cache_items_for_authority(msal.TokenCache.CredentialType.ACCOUNT, False)

    assert refresh_tokens
    assert all_accounts
    assert len(refresh_tokens) == 1 or len(all_accounts) == 1

    #--------------------------------------------------------
    # This code is copied from and slightly modified
    # azure-sdk-for-python/sdk/identity/azure-identity/azure/identity/_internal/shared_token_cache.py
    # around line 174
    # https://github.com/Azure/azure-sdk-for-python/blob/azure-identity_1.15.0/sdk/identity/azure-identity/azure/identity/_internal/shared_token_cache.py

    found_account = None
    for refresh_token in refresh_tokens:
        home_account_id = refresh_token.get("home_account_id")
        if not home_account_id:
            continue
        for account in all_accounts:
            # When the token has no family, msal.net falls back to matching client_id,
            # which won't work for the shared cache because we don't know the IDs of
            # all contributing apps. It should be unnecessary anyway because the
            # apps should all belong to the family.
            if home_account_id == account.get("home_account_id"):# BAD CODE:  and "family_id" in refresh_token:
                if found_account:
                    raise Exception("too many matching accounts")
                found_account = account
    #--------------------------------------------------------

    assert found_account

    # now fake it working and verify the refresh token
    # would work
    manufactured_auth_record = azure.identity.AuthenticationRecord(
            tenant_id=found_account['realm'],
            client_id=refresh_token['client_id'],
            authority=refresh_token['environment'],
            home_account_id=found_account['home_account_id'],
            username=found_account['username'],
       )

    shared_credential = azure.identity.SharedTokenCacheCredential(
            tenant_id = tenant_id,
            authority = 'https://login.microsoftonline.com/' + tenant_id,
            cache_persistence_options = azure_cache,
            authentication_record = manufactured_auth_record
        )
    shared_client = msgraph.GraphServiceClient(
        shared_credential,
        CONFIG['user_scopes'],
    )
    succeeds = await shared_client.me.drives.get()
    assert succeeds
    print("Account and Refresh Token appear to match and be valid")
    pass


if __name__ == '__main__':
    client_id = # SOMETHING
    tenant_id = # SOMETHING
    scopes = # SOMETHING
    asyncio.run(run_program(client_id, tenant_id, scopes))

kcphila avatar Mar 05 '24 14:03 kcphila