tessl/pypi-msal

Microsoft Authentication Library for Python enabling OAuth2/OIDC authentication with Microsoft identity platform

—

Pending

Overview

Eval results

Files

Token Caching

Name: tessl/pypi-msal
Author: tessl

MSAL Python provides comprehensive token caching capabilities to minimize authentication requests and improve application performance. The cache system supports both in-memory caching for single-session scenarios and persistent caching for applications that need to maintain tokens across sessions.

Capabilities

TokenCache (In-Memory)

Base token cache class that maintains tokens in memory using a unified schema compatible across all MSAL libraries. Automatically handles token refresh and account management.

class TokenCache:
    def __init__(self):
        """Create an in-memory token cache."""

    def find(
        self,
        credential_type: str,
        target=None,
        query=None,
        *,
        now=None
    ) -> list:
        """
        Find matching cache entries.
        
        .. deprecated:: 
            Use list(search(...)) instead to explicitly get a list.

        Parameters:
        - credential_type: Type of credential (ACCESS_TOKEN, REFRESH_TOKEN, ACCOUNT, ID_TOKEN)
        - target: Scope string for filtering
        - query: Additional query parameters dict
        - now: Current time for expiration checking

        Returns:
        List of matching cache entries
        """

    def search(
        self,
        credential_type: str,
        target=None,
        query=None,
        now=None
    ):
        """
        Generator version of find() for memory efficiency.

        Returns:
        Generator yielding matching cache entries
        """

    def add(self, event: dict, now=None):
        """
        Add token to cache from authentication event.

        Parameters:
        - event: Authentication result dictionary
        - now: Current time override
        """

    def modify(
        self,
        credential_type: str,
        old_entry: dict,
        new_key_value_pairs=None
    ):
        """
        Modify existing cache entry.

        Parameters:
        - credential_type: Type of credential to modify
        - old_entry: Existing cache entry
        - new_key_value_pairs: Dict of new values, or None to remove entry
        """

Credential Types

class TokenCache:
    class CredentialType:
        ACCESS_TOKEN = "AccessToken"
        REFRESH_TOKEN = "RefreshToken"  
        ACCOUNT = "Account"
        ID_TOKEN = "IdToken"
        APP_METADATA = "AppMetadata"

    class AuthorityType:
        ADFS = "ADFS"
        MSSTS = "MSSTS"  # AAD v2 for both AAD & MSA

Token Management Methods

Specific methods for managing different token types:

def remove_rt(self, rt_item: dict):
    """Remove refresh token from cache."""

def update_rt(self, rt_item: dict, new_rt: str):
    """Update refresh token in cache."""

def remove_at(self, at_item: dict):
    """Remove access token from cache."""

def remove_idt(self, idt_item: dict):
    """Remove ID token from cache."""

def remove_account(self, account_item: dict):
    """Remove account from cache."""

Usage example:

import msal

# Create custom cache
cache = msal.TokenCache()

# Use cache with application
app = msal.PublicClientApplication(
    client_id="your-client-id",
    token_cache=cache
)

# After authentication, tokens are automatically cached
result = app.acquire_token_interactive(scopes=["User.Read"])

# Find cached access tokens
access_tokens = cache.find(
    credential_type=cache.CredentialType.ACCESS_TOKEN,
    target="User.Read"
)

print(f"Found {len(access_tokens)} cached access tokens")

# Find accounts
accounts = cache.find(credential_type=cache.CredentialType.ACCOUNT)
for account in accounts:
    print(f"Account: {account.get('username')}")

SerializableTokenCache (Persistent)

Extended token cache that supports serialization to/from persistent storage. Tracks state changes to optimize serialization operations.

class SerializableTokenCache(TokenCache):
    def __init__(self):
        """Create a serializable token cache."""

    def serialize(self) -> str:
        """
        Serialize current cache state to JSON string.

        Returns:
        JSON string representation of cache state
        """

    def deserialize(self, state: str):
        """
        Deserialize cache from JSON string.

        Parameters:
        - state: JSON string from previous serialize() call
        """

    def add(self, event: dict, **kwargs):
        """
        Add token and mark cache as changed.

        Parameters:
        - event: Authentication result dictionary
        """

    def modify(
        self,
        credential_type: str,
        old_entry: dict,
        new_key_value_pairs=None
    ):
        """
        Modify cache entry and mark cache as changed.

        Parameters:
        - credential_type: Type of credential to modify
        - old_entry: Existing cache entry
        - new_key_value_pairs: Dict of new values, or None to remove entry
        """

    @property
    def has_state_changed(self) -> bool:
        """Check if cache state has changed since last serialization."""

    def _mark_as_changed(self):
        """Mark cache as changed (internal use)."""

File-Based Cache Implementation

Example implementation of persistent file-based cache:

import msal
import os
import json
import atexit

class FilePersistenceTokenCache(msal.SerializableTokenCache):
    def __init__(self, file_path):
        super().__init__()
        self.file_path = file_path
        
        # Load existing cache
        if os.path.exists(file_path):
            try:
                with open(file_path, 'r') as f:
                    cache_data = f.read()
                    if cache_data:
                        self.deserialize(cache_data)
            except Exception as e:
                print(f"Warning: Failed to load cache: {e}")
        
        # Save cache on exit
        atexit.register(self._save_cache)
    
    def _save_cache(self):
        if self.has_state_changed:
            try:
                with open(self.file_path, 'w') as f:
                    f.write(self.serialize())
            except Exception as e:
                print(f"Warning: Failed to save cache: {e}")

# Usage
cache = FilePersistenceTokenCache("token_cache.json")

app = msal.PublicClientApplication(
    client_id="your-client-id",
    token_cache=cache
)

# First run - tokens will be cached to file
result = app.acquire_token_interactive(scopes=["User.Read"])

# Subsequent runs - tokens loaded from file
accounts = app.get_accounts()
if accounts:
    result = app.acquire_token_silent(scopes=["User.Read"], account=accounts[0])

Cross-Application Token Sharing

Multiple applications can share the same cache for single sign-on experiences:

import msal

# Shared cache file
shared_cache = FilePersistenceTokenCache("shared_cache.json")

# Application 1
app1 = msal.PublicClientApplication(
    client_id="app1-client-id",
    token_cache=shared_cache
)

# Application 2  
app2 = msal.PublicClientApplication(
    client_id="app2-client-id",
    token_cache=shared_cache
)

# User authenticates in app1
result1 = app1.acquire_token_interactive(scopes=["User.Read"])

# App2 can use cached tokens for same user
accounts = app2.get_accounts()
if accounts:
    result2 = app2.acquire_token_silent(
        scopes=["User.Read"], 
        account=accounts[0]
    )

Cache Querying and Management

Advanced cache operations for monitoring and managing cached tokens:

import msal
from datetime import datetime

app = msal.PublicClientApplication(
    client_id="your-client-id",
    token_cache=cache
)

# Get all accounts
accounts = app.get_accounts()
print(f"Cached accounts: {len(accounts)}")

for account in accounts:
    print(f"  User: {account.get('username')}")
    print(f"  Environment: {account.get('environment')}")
    print(f"  Home Account ID: {account.get('home_account_id')}")

# Find all access tokens
access_tokens = cache.find(
    credential_type=cache.CredentialType.ACCESS_TOKEN
)

print(f"\nCached access tokens: {len(access_tokens)}")
for token in access_tokens:
    scopes = token.get('target', 'Unknown scopes')
    expires_on = token.get('expires_on')
    if expires_on:
        expires_dt = datetime.fromtimestamp(int(expires_on))
        print(f"  Scopes: {scopes}")
        print(f"  Expires: {expires_dt}")
        print(f"  Expired: {expires_dt < datetime.now()}")

# Find refresh tokens  
refresh_tokens = cache.find(
    credential_type=cache.CredentialType.REFRESH_TOKEN
)
print(f"\nCached refresh tokens: {len(refresh_tokens)}")

# Remove specific account
if accounts:
    app.remove_account(accounts[0])
    print("Removed first account from cache")

Cache Security Considerations

Important security practices for token caching:

import msal
import os
import stat

class SecureFileCache(msal.SerializableTokenCache):
    def __init__(self, file_path):
        super().__init__()
        self.file_path = file_path
        
        # Load with secure permissions
        if os.path.exists(file_path):
            # Verify file permissions (owner read/write only)
            file_stat = os.stat(file_path)
            if file_stat.st_mode & 0o077:  # Check if group/other have permissions
                print("Warning: Cache file has insecure permissions")
            
            with open(file_path, 'r') as f:
                self.deserialize(f.read())
        
        atexit.register(self._save_cache)
    
    def _save_cache(self):
        if self.has_state_changed:
            # Create file with secure permissions
            with open(self.file_path, 'w') as f:
                f.write(self.serialize())
            
            # Set restrictive permissions (owner only)
            os.chmod(self.file_path, stat.S_IRUSR | stat.S_IWUSR)

# Additional security measures:
# 1. Store cache files in user-specific directories
# 2. Encrypt cache contents for sensitive environments  
# 3. Set appropriate file system permissions
# 4. Consider in-memory cache for highly sensitive applications
# 5. Implement cache expiration and cleanup policies

Error Handling

Cache-related error handling patterns:

try:
    cache = FilePersistenceTokenCache("cache.json")
    
    app = msal.PublicClientApplication(
        client_id="your-client-id",
        token_cache=cache
    )
    
    # Attempt silent authentication
    accounts = app.get_accounts()
    if accounts:
        result = app.acquire_token_silent(
            scopes=["User.Read"],
            account=accounts[0]
        )
        
        if "access_token" not in result:
            # Silent auth failed, try interactive
            print("Silent authentication failed, trying interactive...")
            result = app.acquire_token_interactive(scopes=["User.Read"])
    else:
        # No cached accounts
        print("No cached accounts found, using interactive authentication...")
        result = app.acquire_token_interactive(scopes=["User.Read"])

except Exception as e:
    print(f"Cache operation failed: {e}")
    # Fall back to default in-memory cache
    app = msal.PublicClientApplication(client_id="your-client-id")

Install with Tessl CLI