Device Code Flow

def acquire_user_code(self, resource, client_id, language=None):
    """
    Get device code and user code for device flow authentication.
    
    Parameters:
    - resource (str): URI identifying the target resource
    - client_id (str): OAuth2 client ID of the application
    - language (str, optional): Language code for user messages (e.g., 'en-us')
    
    Returns:
    dict: Device code response containing:
        - userCode (str): Code for user to enter at verification URL
        - deviceCode (str): Device code for polling token endpoint
        - verificationUrl (str): URL where user enters the code
        - expiresIn (int): Seconds until codes expire (typically 900)
        - interval (int): Recommended polling interval in seconds
        - message (str): User-friendly instructions
    """

import adal
import time

context = adal.AuthenticationContext('https://login.microsoftonline.com/tenant-id')

# Get device code
user_code_info = context.acquire_user_code(
    resource='https://management.azure.com/',
    client_id='your-client-id'
)

print(f"Please visit {user_code_info['verificationUrl']}")
print(f"And enter this code: {user_code_info['userCode']}")
print(f"Message: {user_code_info['message']}")

def acquire_token_with_device_code(self, resource, user_code_info, client_id):
    """
    Poll for token using device code after user authentication.
    
    Parameters:
    - resource (str): URI identifying the target resource
    - user_code_info (dict): Device code response from acquire_user_code()
    - client_id (str): OAuth2 client ID of the application
    
    Returns:
    dict: Authentication result with access token, refresh token, and metadata
    
    Raises:
    AdalError: If user denies consent, code expires, or other authentication errors
    """

# After displaying user code, poll for token
try:
    token = context.acquire_token_with_device_code(
        resource='https://management.azure.com/',
        user_code_info=user_code_info,
        client_id='your-client-id'
    )
    print("Authentication successful!")
    print(f"Access token: {token['accessToken']}")
    
except adal.AdalError as e:
    print(f"Device authentication failed: {e}")

def cancel_request_to_get_token_with_device_code(self, user_code_info):
    """
    Cancel ongoing device code authentication request.
    
    Parameters:
    - user_code_info (dict): Device code response from acquire_user_code()
    
    Returns:
    None
    """

# Cancel if user requests abort or timeout occurs
context.cancel_request_to_get_token_with_device_code(user_code_info)
print("Device authentication cancelled")

import adal
import time
import threading

def device_flow_authentication():
    authority_url = 'https://login.microsoftonline.com/your-tenant-id'
    context = adal.AuthenticationContext(authority_url)
    
    resource = 'https://management.azure.com/'
    client_id = 'your-client-id'
    
    try:
        # Step 1: Get device code
        print("Initiating device code flow...")
        user_code_info = context.acquire_user_code(resource, client_id)
        
        # Step 2: Display instructions to user
        print(f"\nTo sign in, use a web browser to open the page:")
        print(f"{user_code_info['verificationUrl']}")
        print(f"\nAnd enter the code: {user_code_info['userCode']}")
        print(f"\nMessage: {user_code_info['message']}")
        print(f"\nWaiting for authentication...")
        
        # Optional: Set up cancellation after timeout
        def cancel_after_timeout():
            time.sleep(user_code_info['expiresIn'])
            try:
                context.cancel_request_to_get_token_with_device_code(user_code_info)
                print("\nDevice code expired, authentication cancelled")
            except:
                pass  # May already be completed
        
        timeout_thread = threading.Thread(target=cancel_after_timeout)
        timeout_thread.daemon = True
        timeout_thread.start()
        
        # Step 3: Poll for token
        token = context.acquire_token_with_device_code(
            resource, user_code_info, client_id
        )
        
        print("\nAuthentication successful!")
        print(f"User: {token.get('userId', 'Unknown')}")
        print(f"Token expires: {token.get('expiresOn', 'Unknown')}")
        
        return token
        
    except adal.AdalError as e:
        print(f"\nDevice authentication failed: {e}")
        return None

# Run the flow
token = device_flow_authentication()

try:
    token = context.acquire_token_with_device_code(
        resource, user_code_info, client_id
    )
except adal.AdalError as e:
    if "authorization_pending" in str(e):
        # Still waiting for user - continue polling
        pass
    elif "authorization_declined" in str(e):
        print("User declined authorization")
    elif "expired_token" in str(e):
        print("Device code expired")
    else:
        print(f"Other error: {e}")