o ưi@sdZddlZddlmZddlmZmZmZmZeGdddZ eGdddeZ Gd d d Z Gd d d Z Gd ddZ dS)z Credential providers for proxy authentication. This module provides a provider-agnostic interface for obtaining OAuth2/JWT tokens. It follows the same TokenCredential protocol used by Azure SDK. N) dataclass)AnyOptionalProtocolruntime_checkablec@s"eZdZUdZeed<eed<dS) AccessTokena Represents an OAuth2 access token with expiration. This matches the structure used by azure.core.credentials.AccessToken. Attributes: token: The access token string (typically a JWT). expires_on: Unix timestamp when the token expires. token expires_onN)__name__ __module__ __qualname____doc__str__annotations__intrrU/home/app/Keep/.python/lib/python3.10/site-packages/litellm/proxy_auth/credentials.pyr s  rc@s"eZdZdZdedefddZdS)TokenCredentialz Protocol for credential providers. This matches the azure.core.credentials.TokenCredential interface, allowing any Azure SDK credential to be used directly. Any class implementing get_token(scope) -> AccessToken can be used. scopereturncCsdS)z Get an access token for the specified scope. Args: scope: The OAuth2 scope to request (e.g., "api://my-app/.default") Returns: AccessToken with the token string and expiration timestamp. Nr)selfrrrr get_token(s zTokenCredential.get_tokenN)r r r r rrrrrrrrs rc@s6eZdZdZd deefddZdedefdd Z dS) AzureADCredentiala Wrapper for Azure Identity credentials. This wraps any azure-identity credential (DefaultAzureCredential, ClientSecretCredential, ManagedIdentityCredential, etc.) and converts the token to our AccessToken format. If no credential is provided, it will use DefaultAzureCredential which tries multiple authentication methods automatically. Example: # Use default credential chain (env vars, managed identity, CLI, etc.) cred = AzureADCredential() # Or provide a specific credential from azure.identity import ClientSecretCredential azure_cred = ClientSecretCredential(tenant_id, client_id, client_secret) cred = AzureADCredential(credential=azure_cred) N credentialcCs||_|du|_dS)z Initialize with an optional Azure credential. Args: credential: An azure-identity credential object. If None, DefaultAzureCredential will be used on first token request. N) _credential _initialized)rrrrr__init__JszAzureADCredential.__init__rrcCsX|jszddlm}||_d|_Wn tytdw|j|}t|j|jdS)a Get an access token from Azure AD. Args: scope: The OAuth2 scope (e.g., "api://my-app/.default") Returns: AccessToken with the JWT and expiration. Raises: ImportError: If azure-identity is not installed. r)DefaultAzureCredentialTz]azure-identity is required for AzureADCredential. Install it with: pip install azure-identityrr ) rZazure.identityrr ImportErrorrrrr )rrrresultrrrrUs     zAzureADCredential.get_token)N) r r r r rrrrrrrrrrr5s rc@s8eZdZdZdededefddZdedefd d Zd S) GenericOAuth2Credentialay Generic OAuth2 client credentials flow. This works with any OAuth2 provider (Okta, Auth0, Keycloak, etc.) that supports the client_credentials grant type. Example: cred = GenericOAuth2Credential( client_id="my-client-id", client_secret="my-client-secret", token_url="https://my-idp.com/oauth2/token" ) client_id client_secret token_urlcCs||_||_||_d|_dS)z Initialize OAuth2 client credentials. Args: client_id: OAuth2 client ID client_secret: OAuth2 client secret token_url: Token endpoint URL (e.g., "https://idp.com/oauth2/token") N)r"r#r$ _cached_token)rr"r#r$rrrrs  z GenericOAuth2Credential.__init__rrcCs|jr|jjtdkr|jSddl}|j|jd|j|j|dd}|| }t |dt t| dd d |_|jS) a Get an access token using OAuth2 client credentials flow. Tokens are cached and reused until they expire (with 60s buffer). Args: scope: The OAuth2 scope to request Returns: AccessToken with the token and expiration. <rNZclient_credentials)Z grant_typer"r#r)dataZ access_tokenZ expires_inir) r%r timehttpxpostr$r"r#raise_for_statusjsonrrget)rrr)responser'rrrrs$  z!GenericOAuth2Credential.get_tokenN)r r r r rrrrrrrrr!rsr!c@s>eZdZdZdedefddZdefddZde fd d Z d S) ProxyAuthHandlera{ Manages OAuth2/JWT token lifecycle for proxy authentication. This handler: - Obtains tokens from the configured credential provider - Caches tokens to avoid unnecessary requests - Automatically refreshes tokens before they expire (60s buffer) - Generates Authorization headers for HTTP requests Set this as litellm.proxy_auth to automatically inject auth headers into all requests to your LiteLLM Proxy. Example: import litellm from litellm.proxy_auth import AzureADCredential, ProxyAuthHandler litellm.proxy_auth = ProxyAuthHandler( credential=AzureADCredential(), scope="api://my-litellm-proxy/.default" ) litellm.api_base = "https://my-proxy.example.com" # Auth headers are now automatically injected response = litellm.completion(model="gpt-4", messages=[...]) rrcCs||_||_d|_dS)a Initialize the proxy auth handler. Args: credential: A TokenCredential implementation (AzureADCredential, GenericOAuth2Credential, or any custom implementation) scope: The OAuth2 scope to request tokens for N)rrr%)rrrrrrrs  zProxyAuthHandler.__init__rcCs0|jr |jjtdkr|j|j|_|jS)z Get a valid access token, refreshing if necessary. Returns: AccessToken that is valid for at least 60 more seconds. r&)r%r r(rrr)rrrrrszProxyAuthHandler.get_tokencCs|}dd|jiS)z Get HTTP headers for authentication. Returns: Dict with Authorization header containing Bearer token. AuthorizationzBearer )rr)rrrrrget_auth_headerssz!ProxyAuthHandler.get_auth_headersN) r r r r rrrrrdictr1rrrrr/s   r/)r r( dataclassesrtypingrrrrrrrr!r/rrrrs =B