azure-sdk-for-python
azure-sdk-for-python copied to clipboard
SharedTokenCacheBase not identifying cache due to irrelevant error
- 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))