Ki>QdZddlmZddlZddlmZddlmZmZddl Z ddl m Z m Z ddl mZddlmZdd lmZdd lmZmZmZdd lmZmZdd lmZdd lmZmZddlm Z ddl!m"Z"ddl#m$Z$ddl%m&Z&m'Z'e$e(Z)GddedZ*GddeZ+edddGddZ,GddeZ-GddeZ.Gd d!eZ/y)"z*TokenVerifier implementations for FastMCP.) annotationsN) dataclass)Anycast) JsonWebKey JsonWebToken) JoseError) serialization)rsa) AnyHttpUrl SecretStrfield_validator) BaseSettingsSettingsConfigDict) TypedDict) AccessToken TokenVerifier)ENV_FILE parse_scopes) get_logger)NotSetNotSetTcbeZdZUdZded<ded<ded<ded<ded<ded<d ed <ded <y ) JWKDatazJSON Web Key data structure.strktykidusealgne list[str]x5cx5tN__name__ __module__ __qualname____doc____annotations__m/home/jay/workspace/scripts/.codegraph-venv/lib/python3.12/site-packages/fastmcp/server/auth/providers/jwt.pyrrs-& H H H H F F N Hr-rF)totalceZdZUdZded<y)JWKSDataz JSON Web Key Set data structure.z list[JWKData]keysNr&r,r-r.r1r1(s * r-r1T)frozenkw_onlyreprcneZdZUdZded<ded<ed dZ d d dZy) RSAKeyPairzRSA key pair for JWT testing.r private_keyr public_keyctjdd}|jtjj tj jtjjd}|jjtjj tjjjd}|t||S)zt Generate an RSA key pair for testing. Returns: RSAKeyPair: Generated key pair ii)public_exponentkey_size)encodingformatencryption_algorithmutf-8)r=r>)r8r9)r generate_private_key private_bytesr EncodingPEM PrivateFormatPKCS8 NoEncryptiondecoder9 public_bytes PublicFormatSubjectPublicKeyInfor )clsr8 private_pem public_pems r.generatezRSAKeyPair.generate5s..! "//"++// ..44!.!;!;!=0  &/   " " $ \&//33$11FFVG_ !+.!  r-Ncddi}|r||d<||ttjttj|zd} |r|| d<|rdj|| d<|r| j|t dg} | j || |j j} | jdS) a Generate a test JWT token for testing purposes. Args: subject: Subject claim (usually user ID) issuer: Issuer claim audience: Audience claim - can be a string or list of strings (optional) scopes: List of scopes to include expires_in_seconds: Token expiration time in seconds additional_claims: Any additional claims to include kid: Key ID to include in header r RS256r)subissiatexpaud scoper@) inttimejoinupdaterencoder8get_secret_valuerH) selfsubjectissueraudiencescopesexpires_in_secondsadditional_claimsrheaderpayloadjwt_lib token_bytess r. create_tokenzRSAKeyPair.create_tokenYs.! F5Mtyy{#tyy{#&88   %GEN "xx/GG   NN, -y)nn GT-->>@ !!'**r-)returnr7)z fastmcp-userzhttps://fastmcp.example.comNNNN)r`rrarrbstr | list[str] | Nonerclist[str] | NonerdrYrezdict[str, Any] | Noner str | Nonerkr)r'r(r)r*r+ classmethodrOrjr,r-r.r7r7.s'O! ! J&3+/#'"&372+2+2+) 2+ ! 2+  2+12+2+ 2+r-r7ceZdZUdZededZdZded<dZ ded<dZ d ed <dZ ded <dZ d ed <dZ d ed<dZded<eddedZy)JWTVerifierSettingsz$Settings for JWT token verification.FASTMCP_SERVER_AUTH_JWT_ignore) env_prefixenv_fileextraNror9jwks_urirmra algorithmrbrnrequired_scopeszAnyHttpUrl | str | Nonebase_urlbefore)modect|SNr)rLvs r. _parse_scopesz!JWTVerifierSettings._parse_scopessAr-)r'r(r)r*rr model_configr9r+rxraryrbrzr{rrprr,r-r.rrrrs.%-L "J !Hj%)F ") Iz '+H$+(,O%,(,H%,&X67r-rrcxeZdZdZeeeeeeed d fdZd dZd dZd dZd dZ d dZ xZ S) JWTVerifiera JWT token verifier supporting both asymmetric (RSA/ECDSA) and symmetric (HMAC) algorithms. This verifier validates JWT tokens using various signing algorithms: - **Asymmetric algorithms** (RS256/384/512, ES256/384/512, PS256/384/512): Uses public/private key pairs. Ideal for external clients and services where only the authorization server has the private key. - **Symmetric algorithms** (HS256/384/512): Uses a shared secret for both signing and verification. Perfect for internal microservices and trusted environments where the secret can be securely shared. Use this when: - You have JWT tokens issued by an external service (asymmetric) - You need JWKS support for automatic key rotation (asymmetric) - You have internal microservices sharing a secret key (symmetric) - Your tokens contain standard OAuth scopes and claims r9rxrarbryrzr{c tj|||||||djD cic]\}} | tur|| c} }} | js| j s t d| jr| j r t d| jxsd}|dvrt d|dt |%| j| j||_| j|_ | j|_ | j|_| j |_t|jg|_t!t"|_i|_d |_d |_y cc} }w) a Initialize a JWTVerifier configured to validate JWTs using either a static key or a JWKS endpoint. Parameters: public_key (str | NotSetT | None): PEM-encoded public key for asymmetric algorithms or shared secret for symmetric algorithms. jwks_uri (str | NotSetT | None): URI to fetch a JSON Web Key Set; used when verifying tokens with remote JWKS. issuer (str | list[str] | NotSetT | None): Expected issuer claim value or list of allowed issuer values. audience (str | list[str] | NotSetT | None): Expected audience claim value or list of allowed audience values. algorithm (str | NotSetT | None): JWT signing algorithm to accept (default: "RS256"). Supported: HS256/384/512, RS256/384/512, ES256/384/512, PS256/384/512. required_scopes (list[str] | NotSetT | None): Scopes that must be present in validated tokens. base_url (AnyHttpUrl | str | NotSetT | None): Base URL passed to the parent TokenVerifier. Raises: ValueError: If neither or both of `public_key` and `jwks_uri` are provided, or if `algorithm` is unsupported. rz.Either public_key or jwks_uri must be providedz/Provide either public_key or jwks_uri, not bothrQ> ES256ES384ES512HS256HS384HS512PS256PS384PS512rQRS384RS512zUnsupported algorithm: .)r{rzrrlN)rrmodel_validateitemsrr9rx ValueErrorrysuper__init__r{rzrarbrjwtrr'logger _jwks_cache_jwks_cache_time _cache_ttl) r_r9rxrarbryrzr{krsettings __class__s r.rzJWTVerifier.__init__sm4'55#- ($ (!*'6 (%' AqF?1   ""8+<+<MN N   8#4#4NO O&&1'   6ykCD D &&$44  #oo  )) "-- ))  01 * ,.'(o sE' czK|jr |jS ddl}ddl}|jdd}|ddt |dzz zz }|j |j |}|jd}|j|d{S7#t$r}td||d}~wwxYww)z'Get the verification key for the token.rNr=rz%Failed to extract key ID from token: ) r9base64jsonsplitlenloadsurlsafe_b64decodeget _get_jwks_key Exceptionr)r_tokenrr header_b64rfrr"s r._get_verification_keyz!JWTVerifier._get_verification_key s ???? " Q  S)!,J #S_q%8!89 9JZZ 8 8 DEF**U#C++C00 00 QDQCHIq P Qs;B;A8BBBB;B B8$B33B88B;cK|js tdtj}||jz |jkre|r||j vr|j |S|sDt |j dk(r,tt|j jS tj4d{}|j|jd{}|j|j}dddd{i|_jdgD]Y}|jd}tj |}|j#} |r| |j |<K| |j d<[||_|rH||j vr+|j$j'd|td|d |j |St |j dk(r,tt|j jSt |j dkDr td td 77p7C#1d{7swYTxYw#tj($r} td | | d} ~ wt*$r2} |j$j'd | td | | d} ~ wwxYww)z(Fetch key from JWKS with simple caching.zJWKS URI not configuredNr2r_defaultz-JWKS key lookup failed: key ID '%s' not foundzKey ID 'z' not found in JWKSz2Multiple keys in JWKS but no key ID (kid) in tokenzNo keys found in JWKSzFailed to fetch JWKS: zJWKS fetch failed: )rxrrZrrrrnextitervalueshttpx AsyncClientrraise_for_statusrr import_keyget_public_keyrdebug HTTPErrorr) r_r current_timeclientresponse jwks_datakey_datakey_kidjwkr9r"s r.rzJWTVerifier._get_jwks_key s}}67 7yy{  $// /$// Asd...'',,S!1!12a7D!1!1!8!8!:;<<, B((* , ,f!'DMM!::))+$MMO  , , "D %MM&"5 >",,u- ++H5 //1 0:D$$W-4>D$$Z0 >%1D !d...KK%%G%xu4G%HII'',,t''(A-T%5%5%<%<%> ?@@))*Q.$L%%<==K ,: , , , ,N B5aS9: A B KK   3A37 85aS9: A BsB-K0I7I I7 I!+I,$I! I7IC I7%K&AI7)K*/I7I!I7!I4'I* (I4/I77K J K%-KKKcdD]I}||vst||tr||jcSt||tsD||cSgS)z Extract scopes from JWT claims. Supports both 'scope' and 'scp' claims. Checks the `scope` claim first (standard OAuth2 claim), then the `scp` claim (used by some Identity Providers). )rXscp) isinstancerrlist)r_claimsclaims r._extract_scopeszJWTVerifier._extract_scopes^sZ& )EfUmS1!%=..00u t4!%=(  ) r-c K |j|d{}|jj||}|jdxs(|jdxs|jdxsd}|jd}|rP|t jkr9|j j d||j jd|y|jr|jd }d }t|jtr||jv}n||jk(}|s9|j j d ||j jd|y|jr|jd  d }t|jtrLt trt fd |jD}nK tt|jv}n.t tr|j v}n |jk(}|s9|j j d||j jd|y|j|} |jrkt!| } t!|j} | j#| s:|j j d| | |j jd|yt%|t'|| |r t)|nd|S7#t*$r|j j dYyt,$r/} |j j dt'| Yd} ~ yd} ~ wwxYww)a Validate a JWT bearer token and return an AccessToken when the token is valid. Parameters: token (str): The JWT bearer token string to validate. Returns: AccessToken | None: An AccessToken populated from token claims if the token is valid; `None` if the token is expired, has an invalid signature or format, fails issuer/audience/scope validation, or any other validation error occurs. N client_idazprRunknownrUz4Token validation failed: expired token for client %sz#Bearer token rejected for client %srSFz6Token validation failed: issuer mismatch for client %srVc3&K|]}|v ywrr,).0expectedrVs r. z0JWTVerifier.load_access_token..s-08HO-sz8Token validation failed: audience mismatch for client %sz4Token missing required scopes. Has: %s, Required: %srrrc expires_atrz5Token validation failed: JWT signature/format invalidzToken validation failed: %s)rrrHrrZrrinforarrrbanyrrrzsetissubsetrrrYr r)r_rverification_keyrrrUrS issuer_validaudience_validrc token_scopesrzr"rVs @r.load_access_tokenzJWTVerifier.load_access_tokenosj %)%?%?%FF XX__U,<=F ;'::e$::e$ **U#CsTYY[( !!JI   !F R{{jj' % dkk40#&$++#5L$'$++#5L#KK%%P!KK$$%JIV}}jj'"'dmmT2!#t,),-<@MM-* *-T4==0I)I"#t,)-#)=), )=%KK%%R!KK$$%JIV))&1F##"6{ "%d&:&:";&// =KK%%N$' KK$$%JIVi.'*3s8  w GF  KK  U V  KK  ;SV D s~M)L LB=L M)BL )M)*C-L M)BL M) %L M)L $M&-M)/M&7%M!M)!M&&M)c@K|j|d{S7w)a\ Verify a bearer token and return access info if valid. This method implements the TokenVerifier protocol by delegating to our existing load_access_token method. Args: token: The JWT token string to validate Returns: AccessToken object if valid, None if invalid or expired N)r)r_rs r. verify_tokenzJWTVerifier.verify_tokens++E2222s )r9str | NotSetT | Nonerxrra str | list[str] | NotSetT | Nonerbrryrrzzlist[str] | NotSetT | Noner{z!AnyHttpUrl | str | NotSetT | None)rrrkr)rrorkr)rzdict[str, Any]rkr#rrrkzAccessToken | None) r'r(r)r*rrrrrrr __classcell__rs@r.rrs*,2)/395;*06<6<R)R' R 1 R 3 R(R4R4RhQ(<B|"tl 3r-rc6eZdZdZ d dfd ZddZxZS)StaticTokenVerifiera Simple static token verifier for testing and development. This verifier validates tokens against a predefined dictionary of valid token strings and their associated claims. When a token string matches a key in the dictionary, the verifier returns the corresponding claims as if the token was validated by a real authorization server. Use this when: - You're developing or testing locally without a real OAuth server - You need predictable tokens for automated testing - You want to simulate different users/scopes without complex setup - You're prototyping and need simple API key-style authentication WARNING: Never use this in production - tokens are stored in plain text! c4t||||_y)a Initialize the static token verifier. Args: tokens: Dict mapping token strings to token metadata Each token should have: client_id, scopes, expires_at (optional) required_scopes: Required scopes for all tokens )rzN)rrtokens)r_rrzrs r.rzStaticTokenVerifier.__init__s 9 r-cK|jj|}|sy|jd}||tjkry|jdg}|jrMt |}t |j}|j |st jd|d|yt||d|||Sw)z-Verify token against static token dictionary.Nrrcz$Token missing required scopes. Has: z , Required: rr) rrrZrzrrrrr)r_r token_datarrcrrzs r.rz StaticTokenVerifier.verify_tokens[[__U+  ^^L1  !j499;&>"-   v;L!$"6"67O"++L9 :<. UdTef -!   sCC r)rzdict[str, dict[str, Any]]rzrnr)r'r(r)r*rrrrs@r.rrs)(-1)*  r-r)0r* __future__rrZ dataclassesrtypingrrr authlib.joserrauthlib.jose.errorsr cryptography.hazmat.primitivesr )cryptography.hazmat.primitives.asymmetricr pydanticr r rpydantic_settingsrrtyping_extensionsrfastmcp.server.authrrfastmcp.settingsrfastmcp.utilities.authrfastmcp.utilities.loggingrfastmcp.utilities.typesrrr'rrr1r7rrrrr,r-r.rs0" ! 1)89;;>':%/03 H   iu  y  $51\+\+2\+~,.M3-M3` ? -? r-