o iQ@sdZddlZddlmZmZddlmZmZddlm Z ddl Z ddl Z ddl m Z mZddlmZmZmZmZddlmZmZGd d d eZd ed eegeeffd dZGddde ZGdddeZGddde ZGdddeZdS)a OAuth client credential extensions for MCP. Provides OAuth providers for machine-to-machine authentication flows: - ClientCredentialsOAuthProvider: For client_credentials with client_id + client_secret - PrivateKeyJWTOAuthProvider: For client_credentials with private_key_jwt authentication (typically using a pre-built JWT from workload identity federation) - RFC7523OAuthClientProvider: For jwt-bearer grant (RFC 7523 Section 2.1) N) AwaitableCallable)AnyLiteral)uuid4) BaseModelField)OAuthClientProviderOAuthFlowErrorOAuthTokenError TokenStorage)OAuthClientInformationFullOAuthClientMetadatacsveZdZdZ  ddededededed d edBd dffd d ZdddZd e j fddZ d e j fddZ Z S)ClientCredentialsOAuthProvideraOAuth provider for client_credentials grant with client_id + client_secret. This provider sets client_info directly, bypassing dynamic client registration. Use this when you already have client credentials (client_id and client_secret). Example: ```python provider = ClientCredentialsOAuthProvider( server_url="https://api.example.com", storage=my_token_storage, client_id="my-client-id", client_secret="my-client-secret", ) ``` client_secret_basicN server_urlstorage client_id client_secrettoken_endpoint_auth_method)rclient_secret_postscopesreturncsDtddg||d}t|||dddtd||dg||d|_dS)aInitialize client_credentials OAuth provider. Args: server_url: The MCP server URL. storage: Token storage implementation. client_id: The OAuth client ID. client_secret: The OAuth client secret. token_endpoint_auth_method: Authentication method for token endpoint. Either "client_secret_basic" (default) or "client_secret_post". scopes: Optional space-separated list of scopes to request. Nclient_credentials redirect_uris grant_typesrscoper@)rrrrrr)rsuper__init__r _fixed_client_info)selfrrrrrrclient_metadata __class__a/home/ubuntu/.local/lib/python3.10/site-packages/mcp/client/auth/extensions/client_credentials.pyr )s z'ClientCredentialsOAuthProvider.__init__c,|jjIdH|j_|j|j_d|_dSz6Load stored tokens and set pre-configured client_info.NTcontextr get_tokenscurrent_tokensr! client_info _initializedr"r&r&r' _initializeO  z*ClientCredentialsOAuthProvider._initializec|IdHS)z)Perform client_credentials authorization.N"_exchange_token_client_credentialsr0r&r&r'_perform_authorizationUz5ClientCredentialsOAuthProvider._perform_authorizationcstddi}ddi}|j||\}}|j|jjr!|j|d<|jjjr-|jjj|d<|}tj d|||dS) z:Build token exchange request for client_credentials grant. grant_typer Content-Type!application/x-www-form-urlencodedresourcerPOSTdataheaders) r+prepare_token_authshould_include_resource_paramprotocol_versionget_resource_urlr#r_get_token_endpointhttpxRequestr" token_datar? token_urlr&r&r'r5Y zAClientCredentialsOAuthProvider._exchange_token_client_credentials)rNrN)__name__ __module__ __qualname____doc__strr rr r1rErFr6r5 __classcell__r&r&r$r'rs, &rtokenrcdtdtffdd }|S)aCreate an assertion provider that returns a static JWT token. Use this when you have a pre-built JWT (e.g., from workload identity federation) that doesn't need the audience parameter. Example: ```python provider = PrivateKeyJWTOAuthProvider( server_url="https://api.example.com", storage=my_token_storage, client_id="my-client-id", assertion_provider=static_assertion_provider(my_prebuilt_jwt), ) ``` Args: token: The pre-built JWT assertion string. Returns: An async callback suitable for use as an assertion_provider. audiencercsSNr&)rTrRr&r'providersz+static_assertion_provider..providerrP)rRrWr&rVr'static_assertion_providernsrYc@seZdZUdZeddZeed<eddZeed<eddZ eed<ed d d Z eed <ed dd Z e ed<eddd Z eeefdBed<deegeeffddZdS)SignedJWTParametersaParameters for creating SDK-signed JWT assertions. Use `create_assertion_provider()` to create an assertion provider callback for use with `PrivateKeyJWTOAuthProvider`. Example: ```python jwt_params = SignedJWTParameters( issuer="my-client-id", subject="my-client-id", signing_key=private_key_pem, ) provider = PrivateKeyJWTOAuthProvider( server_url="https://api.example.com", storage=my_token_storage, client_id="my-client-id", assertion_provider=jwt_params.create_assertion_provider(), ) ``` z0Issuer for JWT assertions (typically client_id).) descriptionissuerz.providerrX)r"rWr&r0r'create_assertion_providersz-SignedJWTParameters.create_assertion_provider)rLrMrNrOrr\rP__annotations__r]r^rcrfrqrgdictrrrrxr&r&r&r'rZs   rZc seZdZdZ ddedededeegeefdedBddf fd d Zdd d Z de j fd dZ de eefddfddZde j fddZZS)PrivateKeyJWTOAuthProvideraqOAuth provider for client_credentials grant with private_key_jwt authentication. Uses RFC 7523 Section 2.2 for client authentication via JWT assertion. The JWT assertion's audience MUST be the authorization server's issuer identifier (per RFC 7523bis security updates). The `assertion_provider` callback receives this audience value and must return a JWT with that audience. **Option 1: Pre-built JWT via Workload Identity Federation** In production scenarios, the JWT assertion is typically obtained from a workload identity provider (e.g., GCP, AWS IAM, Azure AD): ```python async def get_workload_identity_token(audience: str) -> str: # Fetch JWT from your identity provider # The JWT's audience must match the provided audience parameter return await fetch_token_from_identity_provider(audience=audience) provider = PrivateKeyJWTOAuthProvider( server_url="https://api.example.com", storage=my_token_storage, client_id="my-client-id", assertion_provider=get_workload_identity_token, ) ``` **Option 2: Static pre-built JWT** If you have a static JWT that doesn't need the audience parameter: ```python provider = PrivateKeyJWTOAuthProvider( server_url="https://api.example.com", storage=my_token_storage, client_id="my-client-id", assertion_provider=static_assertion_provider(my_prebuilt_jwt), ) ``` **Option 3: SDK-signed JWT (for testing/simple setups)** For testing or simple deployments, use `SignedJWTParameters.create_assertion_provider()`: ```python jwt_params = SignedJWTParameters( issuer="my-client-id", subject="my-client-id", signing_key=private_key_pem, ) provider = PrivateKeyJWTOAuthProvider( server_url="https://api.example.com", storage=my_token_storage, client_id="my-client-id", assertion_provider=jwt_params.create_assertion_provider(), ) ``` Nrrrassertion_providerrrcsHtddgd|d}t|||ddd||_td|dgd|d|_dS)aInitialize private_key_jwt OAuth provider. Args: server_url: The MCP server URL. storage: Token storage implementation. client_id: The OAuth client ID. assertion_provider: Async callback that takes the audience (authorization server's issuer identifier) and returns a JWT assertion. Use `SignedJWTParameters.create_assertion_provider()` for SDK-signed JWTs, `static_assertion_provider()` for pre-built JWTs, or provide your own callback for workload identity federation. scopes: Optional space-separated list of scopes to request. Nrprivate_key_jwtrr)rrrrr)rrr _assertion_providerr r!)r"rrrr|rr#r$r&r'r s z#PrivateKeyJWTOAuthProvider.__init__cr(r)r*r0r&r&r'r1%r2z&PrivateKeyJWTOAuthProvider._initializecr3)z>Perform client_credentials authorization with private_key_jwt.Nr4r0r&r&r'r6+r7z1PrivateKeyJWTOAuthProvider._perform_authorizationrHcsD|jjs tdt|jjj}||IdH}||d<d|d<dS)IAdd JWT assertion for client authentication to token endpoint parameters./Missing OAuth metadata for private_key_jwt flowNclient_assertion6urn:ietf:params:oauth:client-assertion-type:jwt-bearerclient_assertion_type)r+oauth_metadatar rPr\r~)r"rHrT assertionr&r&r'_add_client_authentication_jwt/s z9PrivateKeyJWTOAuthProvider._add_client_authentication_jwtcstddi}ddi}|j|dIdH|j|jjr!|j|d<|jjjr-|jjj|d<|}tj d |||d S) zOBuild token exchange request for client_credentials grant with private_key_jwt.r8rr9r:rHNr;rr<r=) rr+rArBrCr#rrDrErFrGr&r&r'r5=rJz=PrivateKeyJWTOAuthProvider._exchange_token_client_credentialsrUrK)rLrMrNrOrPr rrr r1rErFr6rzrrr5rQr&r&r$r'r{s(A 'r{c@seZdZUdZedddZedBed<edddZedBed<edddZ edBed <edd dZ edBed <edd dZ e ee fdBed <edddZedBed<edddZedBed<edddZeed<ddedBdefddZdS) JWTParameterszJWT parameters.NzeJWT assertion for JWT authentication. Will be used instead of generating a new assertion if provided.rarzIssuer for JWT assertions.r\z&Subject identifier for JWT assertions.r]zAudience for JWT assertions.rTz%Additional claims for JWT assertions.rwr_r`jwt_signing_algorithmzPrivate key for JWT signing.jwt_signing_keyrdrejwt_lifetime_secondswith_audience_fallbackrcCs|jdur |j}|S|jstd|jstd|jstd|jr%|jn|}|s-tdtt}|j|j|||j|t t d}| |j pJit j||j|jpUdd}|S)Nz(Missing signing key for JWT bearer grantz#Missing issuer for JWT bearer grantz$Missing subject for JWT bearer grantz%Missing audience for JWT bearer grantrhr_ro)rrr r\r]rTrqrrrrPrrsrwrtrur)r"rrrTrvrwr&r&r' to_assertioncs6  zJWTParameters.to_assertionrU)rLrMrNrOrrrPryr\r]rTrwrzrrrrrqrr&r&r&r'rRs  rcseZdZdZ    ddedededeegedfdBdegee eedBffdBd e d e dBd dffd d Z dddedede eefdBd ejffddZd ejffdd Zde eeffddZd ejfddZZS)RFC7523OAuthClientProvideraOAuth client provider for RFC 7523 jwt-bearer grant. .. deprecated:: Use :class:`ClientCredentialsOAuthProvider` for client_credentials with client_id + client_secret, or :class:`PrivateKeyJWTOAuthProvider` for client_credentials with private_key_jwt authentication instead. This provider supports the jwt-bearer authorization grant (RFC 7523 Section 2.1) where the JWT itself is the authorization grant. Nrrr#rredirect_handlercallback_handlertimeoutjwt_parametersrc s8ddl}|jdtddt||||||||_dS)NrzsRFC7523OAuthClientProvider is deprecated. Use ClientCredentialsOAuthProvider or PrivateKeyJWTOAuthProvider instead.) stacklevel)warningswarnDeprecationWarningrr r) r"rr#rrrrrrr$r&r'r s  z#RFC7523OAuthClientProvider.__init__r auth_code code_verifierrHcs<|pi}|jjjdkr|j|dtj|||dIdHS)z9Build token exchange request for authorization_code flow.r}rN)r+r#rrr"_exchange_token_authorization_code)r"rrrHr$r&r'rs  z=RFC7523OAuthClientProvider._exchange_token_authorization_codecs2d|jjjvr|IdH}|StIdHS)zPerform the authorization flow.+urn:ietf:params:oauth:grant-type:jwt-bearerN)r+r#r_exchange_token_jwt_bearerrr6)r" token_requestr$r&r'r6s z1RFC7523OAuthClientProvider._perform_authorizationcCs\|jstd|jjstdt|jjj}|jj|d}||d<d|d<|j|d<dS) rz/Missing JWT parameters for private_key_jwt flowrrrrrrTN)rr r+rrPr\rrC)r"rHr\rr&r&r'rsz9RFC7523OAuthClientProvider._add_client_authentication_jwtcs|jjs td|jstd|jjstdt|jjj}|jj|d}d|d}|j |jj r:|j |d<|jj j rF|jj j |d<|}tjd ||d d id S) z2Build token exchange request for JWT bearer grant.zMissing client infozMissing JWT parameterszMissing OAuth metadatarr)r8rr;rr<r9r:r=)r+r.r rrr rPr\rrArBrCr#rrDrErF)r"r\rrHrIr&r&r'rs(  z5RFC7523OAuthClientProvider._exchange_token_jwt_bearer)NNrN)rLrMrNrOrPrr rrtuplefloatrr rzrrErFrr6rrrQr&r&r$r'rsH  r)rOrrcollections.abcrrtypingrruuidrrErtpydanticrrmcp.client.authr r r r mcp.shared.authr rrrPrYrZr{rrr&r&r&r's"   V74