uPjdZddlZddlmZddlmZmZmZmZeGddZ eGddeZ Gd d Z Gd d Z Gd dZ y)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)AnyFinalProtocolruntime_checkablec&eZdZUdZeed<eed<y) 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__intc/home/oem/CyberBrain26-unified/.venv/lib/python3.12/site-packages/litellm/proxy_auth/credentials.pyr r s JOrr c eZdZdZdedefdZy)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. scopereturncy)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)selfrs r get_tokenzTokenCredential.get_token(s rN)r r rrrr rrrrrrs  s  {  rrc4eZdZdZddedzfdZdedefdZy) 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 credentialc$||_|du|_y)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)rrs r__init__zAzureADCredential.__init__Js!+&d2rrrc|js ddlm}||_d|_|jj |}t |j|jS#t$r t dwxYw)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-identityr r ) r"azure.identityr%r! ImportErrorrr r r )rrr%results rrzAzureADCredential.get_tokenUsx   A#9#; $(! ((2259&:K:KLL  !s s A##A8)N) r r rrrr#rr rrrrrr5s-( 33: 3MsM{Mrrc4eZdZdZdededefdZdedefdZy ) 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_urlc<||_||_||_d|_y)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.s rr#z GenericOAuth2Credential.__init__s"#*"15rrrc|jr:|jjtjdzkDr |jSddl}|j |j d|j |j|d}|j|j}t|dttj|jdd z |_|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. <rNclient_credentials) grant_typer,r-r)data access_token expires_inir&) r0r timehttpxpostr.r,r-raise_for_statusjsonr rget)rrr9responser5s rrz!GenericOAuth2Credential.get_tokens   $"4"4"?"?$))+PRBR"R%% %** NN2!^^!%!3!3 %  !!#mmo(~&499;'$((<*FF !!!rN)r r rrrr#r rrrrr+r+qs4  6# 6c 6c 6""s""{""rr+c8eZdZdZdedefdZdefdZde fdZ y) 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=[...]) rrc.||_||_d|_y)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)rrr0)rrrs rr#zProxyAuthHandler.__init__s% 15rrc|jr.|jjtjdzkr*|jj |j |_|jS)z Get a valid access token, refreshing if necessary. Returns: AccessToken that is valid for at least 60 more seconds. r2)r0r r8rrr)rs rrzProxyAuthHandler.get_tokensS!!T%7%7%B%BdiikTVFV%V!%!:!:4::!FD !!!rcD|j}dd|jiS)z Get HTTP headers for authentication. Returns: Dict with Authorization header containing Bearer token. AuthorizationzBearer )rr )rr s rget_auth_headersz!ProxyAuthHandler.get_auth_headerss&~~'75;;-!899rN) r r rrrrr#r rdictrErrrr@r@s24 6? 63 6 "; ":$:rr@)rr8 dataclassesrtypingrrrrr rrr+r@rrrrIsn !::      h  .9M9Mx?"?"D<:<:r