Confidential Client Applications

class ConfidentialClientApplication(ClientApplication):
    def __init__(
        self,
        client_id: str,
        client_credential,
        authority=None,
        validate_authority=True,
        token_cache=None,
        http_client=None,
        verify=True,
        proxies=None,
        timeout=None,
        client_claims=None,
        app_name=None,
        app_version=None,
        client_capabilities=None,
        azure_region=None,
        exclude_scopes=None,
        http_cache=None,
        instance_discovery=None,
        enable_pii_log=None,
        oidc_authority=None,
        **kwargs
    ):
        """
        Create a confidential client application.

        Parameters:
        - client_id: Your app's client ID from Azure portal
        - client_credential: Client secret string, certificate dict, or assertion callable
        - authority: Authority URL (default: https://login.microsoftonline.com/common)
        - azure_region: Azure region for regional STS endpoints
        - token_cache: Custom token cache instance
        - http_client: Custom HTTP client
        - proxies: HTTP proxy configuration
        - timeout: HTTP timeout in seconds
        """

app = msal.ConfidentialClientApplication(
    client_id="your-client-id",
    client_credential="your-client-secret",
    authority="https://login.microsoftonline.com/your-tenant-id"
)

client_credential = {
    "private_key": "-----BEGIN PRIVATE KEY-----\n...\n-----END PRIVATE KEY-----",
    "thumbprint": "A1B2C3D4E5F6...",
    "passphrase": "optional-passphrase"  # If private key is encrypted
}

app = msal.ConfidentialClientApplication(
    client_id="your-client-id",
    client_credential=client_credential,
    authority="https://login.microsoftonline.com/your-tenant-id"
)

client_credential = {
    "private_key": "-----BEGIN PRIVATE KEY-----\n...\n-----END PRIVATE KEY-----",
    "thumbprint": "A1B2C3D4E5F6...",
    "public_certificate": "-----BEGIN CERTIFICATE-----\n...\n-----END CERTIFICATE-----",
    "passphrase": "optional-passphrase"
}

app = msal.ConfidentialClientApplication(
    client_id="your-client-id", 
    client_credential=client_credential,
    authority="https://login.microsoftonline.com/your-tenant-id"
)

def acquire_token_for_client(
    self,
    scopes: list,
    claims_challenge=None,
    **kwargs
):
    """
    Acquire token for the client application.

    Parameters:
    - scopes: List of scopes (typically ["{resource}/.default"])
    - claims_challenge: Additional claims from resource provider

    Returns:
    Dictionary with 'access_token' on success, 'error' on failure
    """

def remove_tokens_for_client(self):
    """
    Remove all tokens previously acquired via acquire_token_for_client().
    """

import msal

app = msal.ConfidentialClientApplication(
    client_id="your-client-id",
    client_credential="your-client-secret",
    authority="https://login.microsoftonline.com/your-tenant-id"
)

# Acquire token for Microsoft Graph
result = app.acquire_token_for_client(
    scopes=["https://graph.microsoft.com/.default"]
)

if "access_token" in result:
    print("Client credentials authentication successful!")
    access_token = result["access_token"]
    
    # Use the token to call Microsoft Graph API
    import requests
    headers = {"Authorization": f"Bearer {access_token}"}
    response = requests.get("https://graph.microsoft.com/v1.0/users", headers=headers)
    
    if response.status_code == 200:
        users = response.json()
        print(f"Found {len(users.get('value', []))} users")
else:
    print(f"Authentication failed: {result.get('error_description')}")

# Clear client tokens when needed
app.remove_tokens_for_client()

def acquire_token_on_behalf_of(
    self,
    user_assertion: str,
    scopes: list,
    claims_challenge=None,
    **kwargs
):
    """
    Acquire token on behalf of user.

    Parameters:
    - user_assertion: The user's access token received by the service
    - scopes: List of scopes for downstream API
    - claims_challenge: Additional claims from resource provider

    Returns:
    Dictionary with 'access_token' on success, 'error' on failure
    """

import msal
from flask import Flask, request

app_flask = Flask(__name__)

# Configure MSAL
msal_app = msal.ConfidentialClientApplication(
    client_id="your-service-client-id",
    client_credential="your-service-client-secret",
    authority="https://login.microsoftonline.com/your-tenant-id"
)

@app_flask.route('/api/data')
def get_data():
    # Extract user's access token from Authorization header
    auth_header = request.headers.get('Authorization', '')
    if not auth_header.startswith('Bearer '):
        return {"error": "Missing or invalid authorization header"}, 401
    
    user_token = auth_header[7:]  # Remove 'Bearer ' prefix
    
    # Use OBO flow to get token for downstream API
    result = msal_app.acquire_token_on_behalf_of(
        user_assertion=user_token,
        scopes=["https://api.downstream.com/.default"]
    )
    
    if "access_token" in result:
        # Call downstream API with new token
        import requests
        headers = {"Authorization": f"Bearer {result['access_token']}"}
        response = requests.get("https://api.downstream.com/data", headers=headers)
        return response.json()
    else:
        return {"error": result.get("error_description")}, 400

def initiate_auth_code_flow(
    self,
    scopes: list,
    redirect_uri=None,
    state=None,
    prompt=None,
    login_hint=None,
    domain_hint=None,
    claims_challenge=None,
    max_age=None,
    **kwargs
):
    """
    Initiate authorization code flow.

    Returns:
    Dictionary containing auth_uri and state information
    """

def acquire_token_by_auth_code_flow(
    self,
    auth_code_flow: dict,
    auth_response: dict,
    scopes=None,
    **kwargs
):
    """
    Complete authorization code flow.

    Parameters:
    - auth_code_flow: Flow state from initiate_auth_code_flow()
    - auth_response: Authorization response from redirect URI
    - scopes: Optional scopes override

    Returns:
    Dictionary with 'access_token' on success, 'error' on failure
    """

import msal
from flask import Flask, request, redirect, session, url_for

app_flask = Flask(__name__)
app_flask.secret_key = 'your-secret-key'

msal_app = msal.ConfidentialClientApplication(
    client_id="your-webapp-client-id",
    client_credential="your-webapp-client-secret",
    authority="https://login.microsoftonline.com/your-tenant-id"
)

@app_flask.route('/login')
def login():
    # Initiate auth code flow
    auth_flow = msal_app.initiate_auth_code_flow(
        scopes=["User.Read"],
        redirect_uri=url_for('auth_response', _external=True)
    )
    
    # Store flow state in session
    session['auth_flow'] = auth_flow
    
    # Redirect user to authorization URL
    return redirect(auth_flow['auth_uri'])

@app_flask.route('/auth-response')
def auth_response():
    # Get stored flow state
    auth_flow = session.get('auth_flow', {})
    
    # Complete the flow with authorization response
    result = msal_app.acquire_token_by_auth_code_flow(
        auth_code_flow=auth_flow,
        auth_response=request.args
    )
    
    if "access_token" in result:
        # Store tokens in session
        session['tokens'] = result
        return "Login successful!"
    else:
        return f"Login failed: {result.get('error_description')}"

app = msal.ConfidentialClientApplication(
    client_id="your-client-id",
    client_credential="your-client-secret",
    authority="https://login.microsoftonline.com/your-tenant-id",
    azure_region="eastus"  # Specify Azure region
)

# Or use auto-detection in Azure environment
app = msal.ConfidentialClientApplication(
    client_id="your-client-id", 
    client_credential="your-client-secret",
    authority="https://login.microsoftonline.com/your-tenant-id",
    azure_region=msal.ClientApplication.ATTEMPT_REGION_DISCOVERY
)

result = app.acquire_token_for_client(scopes=["https://graph.microsoft.com/.default"])

if "access_token" in result:
    # Success
    access_token = result["access_token"]
    expires_in = result["expires_in"]
elif result.get("error") == "invalid_client":
    # Invalid client credentials
    print("Invalid client ID or secret")
elif result.get("error") == "invalid_scope":
    # Invalid or unauthorized scope
    print(f"Invalid scope: {result.get('error_description')}")
elif result.get("error") == "unauthorized_client":
    # Client not authorized for requested grant type
    print("Client not authorized for client credentials flow")
else:
    # Other error
    print(f"Authentication failed: {result.get('error_description')}")

# Handle OBO-specific errors
obo_result = app.acquire_token_on_behalf_of(
    user_assertion=user_token,
    scopes=["https://api.downstream.com/.default"]
)

if obo_result.get("error") == "invalid_grant":
    # User assertion is invalid or expired
    print("User token is invalid or expired")
elif obo_result.get("error") == "consent_required":
    # Additional consent needed
    print("Additional consent required for downstream API")