kKj0dZddlmZddlZddlZddlZddlmZmZm Z ddl Z ddl m Z m Z ddlmZmZddlmZddlmZdd lmZeeZed ZGd d eZy) a!OAuth 2.0 Token Introspection (RFC 7662) provider for FastMCP. This module provides token verification for opaque tokens using the OAuth 2.0 Token Introspection protocol defined in RFC 7662. It allows FastMCP servers to validate tokens issued by authorization servers that don't use JWT format. Example: ```python from fastmcp import FastMCP from fastmcp.server.auth.providers.introspection import IntrospectionTokenVerifier # Verify opaque tokens via RFC 7662 introspection verifier = IntrospectionTokenVerifier( introspection_url="https://auth.example.com/oauth/introspect", client_id="your-client-id", client_secret="your-client-secret", required_scopes=["read", "write"] ) mcp = FastMCP("My Protected Server", auth=verifier) ``` ) annotationsN)AnyLiteralget_args) AnyHttpUrl SecretStr) AccessToken TokenVerifier) parse_scopes) get_logger) TokenCache)client_secret_basicclient_secret_postcteZdZdZdddddddd d fdZd dZd dZd d ZxZS)IntrospectionTokenVerifiera OAuth 2.0 Token Introspection verifier (RFC 7662). This verifier validates opaque tokens by calling an OAuth 2.0 token introspection endpoint. Unlike JWT verification which is stateless, token introspection requires a network call to the authorization server for each token validation. The verifier authenticates to the introspection endpoint using either: - HTTP Basic Auth (client_secret_basic, default): credentials in Authorization header - POST body authentication (client_secret_post): credentials in request body Both methods are specified in RFC 6749 (OAuth 2.0) and RFC 7662 (Token Introspection). Use this when: - Your authorization server issues opaque (non-JWT) tokens - You need to validate tokens from Auth0, Okta, Keycloak, or other OAuth servers - Your tokens require real-time revocation checking - Your authorization server supports RFC 7662 introspection Caching is disabled by default to preserve real-time revocation semantics. Set ``cache_ttl_seconds`` to enable caching and reduce load on the introspection endpoint (e.g., ``cache_ttl_seconds=300`` for 5 minutes). Example: ```python verifier = IntrospectionTokenVerifier( introspection_url="https://auth.example.com/oauth/introspect", client_id="my-service", client_secret="secret-key", required_scopes=["api:read"] ) ``` r N)client_auth_methodtimeout_secondsrequired_scopesbase_urlcache_ttl_secondsmax_cache_size http_clientc | t|nd} t| || ||_||_t |t r|jn||_tt} || vr*djd| D} td|d| d||_ ||_| |_t!t"|_t'|| |_y) a{ Initialize the introspection token verifier. Args: introspection_url: URL of the OAuth 2.0 token introspection endpoint client_id: OAuth client ID for authenticating to the introspection endpoint client_secret: OAuth client secret for authenticating to the introspection endpoint client_auth_method: Client authentication method. "client_secret_basic" (default) uses HTTP Basic Auth header, "client_secret_post" sends credentials in POST body timeout_seconds: HTTP request timeout in seconds (default: 10) required_scopes: Required scopes for all tokens (optional) base_url: Base URL for TokenVerifier protocol cache_ttl_seconds: How long to cache introspection results in seconds. Caching is disabled by default (None) to preserve real-time revocation semantics. Set to a positive integer to enable caching (e.g., 300 for 5 minutes). max_cache_size: Maximum number of tokens to cache when caching is enabled. Default: 10000. http_client: Optional httpx.AsyncClient for connection pooling. When provided, the client is reused across calls and the caller is responsible for its lifecycle. When None (default), a fresh client is created per call. N)rrz or c3(K|] }d|d yw)'N).0ms |/Users/ahmed/devFolder/Ultron/claude-voice/.venv/lib/python3.12/site-packages/fastmcp/server/auth/providers/introspection.py z6IntrospectionTokenVerifier.__init__..s!BMqAaS(MszInvalid client_auth_method: z . Must be .) ttl_secondsmax_size)r super__init__introspection_url client_id isinstancerget_secret_value client_secretrClientAuthMethodjoin ValueErrorrr _http_clientr __name__loggerr _cache)selfr'r(r+rrrrrrrparsed_required_scopes valid_methodsoptions __class__s r r&z#IntrospectionTokenVerifier.__init__PsL.=-HL )d  ( "";#5#5g#>?FFwOy!!r8c6|jd}|gSt|tr=|jDcgc]#}|j s|j %c}St|t r|Dcgc]}|st|c}SgScc}wcc}w)z Extract scopes from introspection response. RFC 7662 allows scopes to be returned as either: - A space-separated string in the 'scope' field - An array of strings in the 'scope' field (less common but valid) scope)getr)strsplitstriplist)r3introspection_response scope_valuess r _extract_scopesz*IntrospectionTokenVerifier._extract_scopess-009  I k3 ''2'8'8':H':!aggiAGGI':H H k4 ($/5Kq1CFK5 5 I6sBB7B? Bc K|jj|\}}|r|jjd|S |jt j |jntj|j4d{}|dd}ddd}|jd k(r|j|d <n-|jd k(r|j|d <|j|d <|j|j||d{}|j dk7rT|jjd|j |j"r|j"ddnd dddd{y|j%}|jdds-|jjd dddd{y|jd xs|jdd} |jd} | rE| t'j&kr.|jjd|  dddd{y|j)|} |j*r`t-| } t-|j*} | j/| s/|jjd| |  dddd{yt1|t3| | | t5| nd|}|jj-|||cdddd{S77 77^77h7#1d{7swYyxYw#tj6$r)|jjd|jYytj8$r&}|jjd|Yd}~yd}~wt:$r&}|jjd|Yd}~yd}~wwxYww)az Verify a bearer token using OAuth 2.0 Token Introspection (RFC 7662). This method makes a POST request to the introspection endpoint with the token, authenticated using the configured client authentication method (client_secret_basic or client_secret_post). Results are cached in-memory to reduce load on the introspection endpoint. Cache TTL and size are configurable via constructor parameters. Args: token: The opaque token string to validate Returns: AccessToken object if valid and active, None if invalid, inactive, or expired zToken introspection cache hitN)timeout access_token)tokentoken_type_hintz!application/x-www-form-urlencodedzapplication/json)z Content-TypeAcceptr Authorizationrr(r+)dataheadersz(Token introspection failed: HTTP %d - %sactiveFz)Token introspection returned active=falsesubunknownexpz4Token validation failed: expired token for client %sz4Token missing required scopes. Has: %s, Required: %s)rPr(scopes expires_atclaimsz.Token introspection timed out after %d secondsz&Token introspection request failed: %szToken introspection error: %s)r2rDr1debugr/ contextlib nullcontexthttpx AsyncClientrrrAr(r+postr' status_codetextjsontimerLrsetissubsetr rEintTimeoutException RequestError Exception)r3rP is_cached cached_resultclientrTrUresponseintrospection_datar(r[r\ token_scopesrresultes r verify_tokenz'IntrospectionTokenVerifier.verify_tokens$$(;;??5#9 =  KK  = > g $$0&&t'8'89&&t/C/CDEE#'5%H0 **.CC/3/M/M/OGO,,,0DD(,D%,0,>,>D)!'**#"-"''3.KK%%B ,,/7}} ds+"  KEEEN&.]]_" *--h>KK%%&QR]EEEb/22>'++E9= ),,U3TYY[( ))R% $}EEEB--.@A''#&v;L&)$*>*>&?O*33LA ))R(+ $]EEEb%!)n!+.?s3x-  v.sEEE01EEEEEEEEv%%  KK  @$BVBV !!  KK  F J  KK  =q A sI>O AL=LL=AL(6 L=LL=O A,L(3 L=>L"?L=O A,L(0 L=;L$<L=O AL( L=L&L=O L=L(L=L="L=$L=&L=(L:.L1 /L:6L=9O :L==9O6O 8O N+&O + O7OO OO )r'rEr(rEr+zstr | SecretStrrr,rrkrzlist[str] | NonerzAnyHttpUrl | str | Noner int | Nonerrxrzhttpx.AsyncClient | None)returnrE)rIzdict[str, Any]ryz list[str])rPrEryzAccessToken | None) r0 __module__ __qualname____doc__r&rArLrw __classcell__)r7s@r rr-s P0E!,0,0(,%)04D D  D ' D - D D *D *D &D #D .D L" .~r8r)r| __future__rr;r`rhtypingrrrrbpydanticrrfastmcp.server.authr r fastmcp.utilities.authr fastmcp.utilities.loggingr fastmcp.utilities.token_cacher r0r1r,rrr8r rsV.#  )) *:/04 H FGDDr8