"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/dS)"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)NotSetNotSetTcdeZdZUdZded<ded<ded<ded<ded<ded<d ed <ded <d S) JWKDatazJSON Web Key data structure.strktykidusealgne list[str]x5cx5tN__name__ __module__ __qualname____doc____annotations__/Users/kimhansen/Desktop/03 Workspace/ceo-agents/chl-effectiveness/mcp-servers/whoop/.venv/lib/python3.11/site-packages/fastmcp/server/auth/providers/jwt.pyrrs^&& HHH HHH HHH HHH FFF FFFNNN HHHHHr-rF)totalceZdZUdZded<dS)JWKSDataz JSON Web Key Set data structure.z list[JWKData]keysNr&r,r-r.r1r1(s$**r-r1T)frozenkw_onlyreprcXeZdZUdZded<ded<eddZ dddZd S) RSAKeyPairzRSA key pair for JWT testing.r private_keyr public_keyreturnctjdd}|tjjtjjtj d}| tjjtj j  d}|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.!   "//"+/ .4!.!;!=!=0   &//   " " $ $ \&/3$1FVG__ s!+..!    r- fastmcp-userhttps://fastmcp.example.comNsubjectissueraudiencestr | list[str] | Nonescopeslist[str] | Noneexpires_in_secondsintadditional_claimsdict[str, Any] | Noner str | Nonecddi}|r||d<||ttjttj|zd} |r|| d<|rd|| d<|r| |t dg} | || |j} | dS) 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 scoperA) r[timejoinupdaterencoder8get_secret_valuerI) selfrTrUrVrXrZr\rheaderpayloadjwt_lib token_bytess r. create_tokenzRSAKeyPair.create_tokenYs.!  F5Mty{{##ty{{##&88     &%GEN  0"xx//GG   . NN, - - -y))nn GT->>@@  !!'***r-)r:r7)rQrRNNrSNN)rTrrUrrVrWrXrYrZr[r\r]rr^r:r)r'r(r)r*r+ classmethodrPrrr,r-r.r7r7.s''OOO! ! ! [! J&3+/#'"&372+2+2+2+2+2+2+r-r7ceZdZUdZededZdZded<dZ ded<dZ d ed <dZ ded <dZ d ed <dZ d ed<dZded<eddedZdS)JWTVerifierSettingsz$Settings for JWT token verification.FASTMCP_SERVER_AUTH_JWT_ignore) env_prefixenv_fileextraNr^r9jwks_urirWrU algorithmrVrYrequired_scopeszAnyHttpUrl | str | Nonebase_urlbefore)modec t|SNr)rMvs r. _parse_scopesz!JWTVerifierSettings._parse_scopessAr-)r'r(r)r*rr model_configr9r+r{rUr|rVr}r~rrsrr,r-r.rurus..%%-L "J!!!!H%)F)))) I    '+H++++(,O,,,,(,H,,,,_&X666[76r-ruc^eZdZdZeeeeeeeddfdZddZddZd 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 r9r{rUrVr|r}r~r9str | NotSetT | Noner{rU str | list[str] | NotSetT | NonerVr|r}list[str] | NotSetT | Noner~!AnyHttpUrl | str | NotSetT | Nonec td|||||||dD}|js|jst d|jr|jrt d|jpd}|dvrt d|dt|j |j ||_|j |_ |j |_ |j|_|j|_t|jg|_tt |_i|_d |_d |_d S) 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. c,i|]\}}|tu||Sr,)r).0krs r. z(JWTVerifier.__init__..s3   AqF??1#??r-rz.Either public_key or jwks_uri must be providedz/Provide either public_key or jwks_uri, not bothr`> ES256ES384ES512HS256HS384HS512PS256PS384PS512r`RS384RS512zUnsupported algorithm: .)r~r}rrSN)rumodel_validateitemsr9r{ ValueErrorr|super__init__r~r}rUrVrjwtrr'logger _jwks_cache_jwks_cache_time _cache_ttl) rmr9r{rUrVr|r}r~settings __class__s r.rzJWTVerifier.__init__s4'55  #- ($ (!*'6 (%''      " O8+< OMNN N   P8#4 PNOO O&1'   CyCCCDD D &$4    #o  ) "- )  011 ** ,.'(r-tokenrr:cK|jr|jS ddl}ddl}|dd}|ddt |dzz zz }|||}|d}||d{VS#t$r}td||d}~wwxYw)z'Get the verification key for the token.rNr=rz%Failed to extract key ID from token: ) r9base64jsonsplitlenloadsurlsafe_b64decodeget _get_jwks_key Exceptionr)rmrrr header_b64rnrr"s r._get_verification_keyz!JWTVerifier._get_verification_key s ? #? " Q MMM KKKS))!,J #S__q%8!89 9JZZ 8 8 D DEEF**U##C++C00000000 0 Q Q QHQHHIIq P QsBB(( C 2CC rr^cK|jstdtj}||jz |jkre|r||jvr |j|S|sKt |jdkr3tt|j S tj 4d{V}| |jd{V}| |}dddd{Vn#1d{VswxYwYi|_| dgD]V}| d}tj|}|} |r | |j|<L| |jd<W||_|rD||jvr.|jd|td|d |j|St |jdkr3tt|j St |jdkrtd td #tj$r} td | | d} ~ wt*$r5} |jd | td | | d} ~ wwxYw)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: )r{rrhrrrrnextitervalueshttpx AsyncClientrraise_for_statusrr import_keyget_public_keyrdebug HTTPErrorr) rmr current_timeclientresponse jwks_datakey_datakey_kidjwkr9r"s r.rzJWTVerifier._get_jwks_key s} 8677 7y{{  $/ /$/ A A =sd...',, =S!122a77D!1!8!8!:!:;;<<<, B(** , , , , , , ,f!'DM!:!:::::::))+++$MMOO  , , , , , , , , , , , , , , , , , , , , , , , , , , , "D %MM&"55 > >",,u-- +H55 //11 >0:D$W--4>D$Z00$0D ! >d...K%%G%%H%H%H%HIII',,t'((A--T%5%<%<%>%> ? ?@@@)**Q..$L%%<=== B B B9a99:: A B B B K  7A77 8 8 89a99:: A BsW%I,>A DI, D##I,&D#'CI,+A I,66I,,K;J K0K  Kclaimsdict[str, Any]r#cdD]b}||vr\t||tr||cSt||tr ||cScgS)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). )rgscp) isinstancerrlist)rmrclaims r._extract_scopeszJWTVerifier._extract_scopes^sz& ) )EfUmS11)!%=..00000u t44)!%=((( r-AccessToken | Nonec K ||d{V}|j||}|dp+|dp|dpd}|d}|rO|t jkr8|jd||jd|dS|jr|d }d }t|jtr ||jv}n ||jk}|s8|jd ||jd|dS|j r|d  d }t|j trSt tr!t fd |j D}nG tt|j v}n*t tr |j v}n |j k}|s8|jd||jd|dS||} |jrqt!| } t!|j} | | s9|jd| | |jd|dSt%|t'|| |rt)|nd|S#t*$r|jdYdSt,$r3} |jdt'| Yd} ~ dSd} ~ wwxYw)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_idazpraunknownrdz4Token validation failed: expired token for client %sz#Bearer token rejected for client %srbFz6Token validation failed: issuer mismatch for client %srec3 K|]}|vV dSrr,)rexpectedres r. z0JWTVerifier.load_access_token..s8--08HO------r-z8Token validation failed: audience mismatch for client %sz4Token missing required scopes. Has: %s, Required: %srrrX expires_atrz5Token validation failed: JWT signature/format invalidzToken validation failed: %s)rrrIrrhrrinforUrrrVanyrrr}setissubsetrrr[r r)rmrverification_keyrrrdrb issuer_validaudience_validrX token_scopesr}r"res @r.load_access_tokenzJWTVerifier.load_access_tokenosj %)%?%?%F%FFFFFFF X__U,<==F ;''::e$$::e$$ **U##C sTY[[(( !!JI   !F RRRt{ jj'' % dk4006#&$+#5LL$'$+#5L# K%%P!K$$%JIVVV4} jj''"'dmT22>!#t,,J),----<@M---** *-T4=0I0I)I"#t,,>)-#)=), )=% K%%R!K$$%JIVVV4))&11F# "6{{ "%d&:";";&// == K%%N$' K$$%JIVVV4i..'*43s888      K  U V V V44    K  ;SVV D D D44444 s9CL"BL)C-LB L%1L$M:> M:(M55M:c<K||d{VS)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)rmrs r. verify_tokenzJWTVerifier.verify_tokens.++E222222222r-)r9rr{rrUrrVrr|rr}rr~r)rrr:r)rr^r:r)rrr:r#rrr:r) r'r(r)r*rrrrrrr __classcell__rs@r.rrs*,2)/395;*06<6<RRRRRRRRhQQQQ(<B<B<B<B|"ttttl 3 3 3 3 3 3 3 3r-rc0eZdZdZ d dfd Zdd ZxZS)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! Ntokensdict[str, dict[str, Any]]r}rYcZt|||_dS)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 )r}N)rrr)rmrr}rs r.rzStaticTokenVerifier.__init__s+ 999 r-rrr:rcK|j|}|sdS|d}||tjkrdS|dg}|jrZt |}t |j}||s"t d|d|dSt||d|||S)z-Verify token against static token dictionary.NrrXz$Token missing required scopes. Has: z , Required: rr) rrrhr}rrrrr)rmr token_datarrXrr}s r.rz StaticTokenVerifier.verify_tokens[__U++  4 ^^L11  !j49;;&>&>4"--   v;;L!$"677O"++L99  f<ffUdfft -!     r-r)rrr}rYr)r'r(r)r*rrrrs@r.rrse(-1         r-r)0r* __future__rrh 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'rrr1r7rurrr,r-r.rs00"""""" !!!!!! 11111111))))))888888999999;;;;;;;;;;>>>>>>>>''''''::::::::%%%%%%//////00000033333333 H        iu     y  $5111\+\+\+\+\+\+\+21\+~,.M3M3M3M3M3-M3M3M3` ? ? ? ? ? -? ? ? ? ? r-