kKjd`UdZddlmZddlZddlZddlZddlmZddlm Z m Z m Z ddl Z ddl mZddlmZddlmZmZdd lmZdd lmZdd lmZdd lmZmZdd lmZddlm Z m!Z!ddl"m#Z#m$Z$m%Z%ddl&m'Z'm(Z(ddl)m*Z*e*e+Z,e-e.e.e/e.zfZ0de1d<e2eZ3d&dZ4d'dZ5d(dZ6GddedZ7GddeZ8edddGdd Z9d)d!Z:Gd"d#e!Z;Gd$d%e!Z>wGG4~~h-446==gFF 1(> ??r,cvjd}|yt|tsytfd|DS)NcritFTc3`K|]%}t|t xs|vxs|tv'ywN) isinstancestrSUPPORTED_JWS_HEADER_FIELDS).0 header_nameheaders r* z4_has_unsupported_critical_headers..:sG K {C (( : f $ : 9 9 : s+.)r0r:listany)r?r7s` r*!_has_unsupported_critical_headersrC3sB ::f D | dD !   r,cbeZdZUdZded<ded<ded<ded<ded<ded<d ed <ded <y ) JWKDatazJSON Web Key data structure.r;r.kidusealgne list[str]x5cx5tN__name__ __module__ __qualname____doc____annotations__r,r*rErEBs-& H H H H F F N Hr,rEF)totalceZdZUdZded<y)JWKSDataz JSON Web Key Set data structure.z list[JWKData]keysNrNrTr,r*rWrWOs * r,rWT)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_algorithmr/)rcrd)r^r_)r generate_private_key private_bytesrEncodingPEM PrivateFormatPKCS8 NoEncryptionr2r_ public_bytes PublicFormatSubjectPublicKeyInfor)clsr^ private_pem public_pems r*generatezRSAKeyPair.generate\s..! "//"++// ..44!.!;!;!=0  &/   " " $ \&//33$11FFVG_ !+.!  r,Ncxddi}|r||d<||ttjttj|zd} |r|| d<|rdj|| d<|r| j|t |j j d} tj|| | dg} | S) 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 rHRS256rF)subissiatexpaud scope) algorithms) inttimejoinupdater+r^get_secret_valuer encode) selfsubjectissueraudiencescopesexpires_in_secondsadditional_claimsrFr?payload signing_keytokens r* create_tokenzRSAKeyPair.create_tokens.! F5Mtyy{#tyy{#&88 5  %GEN "xx/GG   NN, -0    - - /  67KWIN r,)returnr])z fastmcp-userzhttps://fastmcp.example.comNNNN)rr;rr;rstr | list[str] | Nonerlist[str] | Nonerr~rzdict[str, Any] | NonerF str | Nonerr;)rOrPrQrRrS classmethodrsrrTr,r*r]r]Us'O! ! J&3+/#'"&37222) 2 ! 2  2122 2r,r]ct|tr|jdd}|jd}t fd|DS)zLReturn True when key text appears to be PEM-encoded asymmetric key material.r/replace)errors)z-----BEGIN PUBLIC KEY-----z-----BEGIN RSA PUBLIC KEY-----z-----BEGIN EC PUBLIC KEY-----z-----BEGIN CERTIFICATE-----c3&K|]}|v ywr9rT)r=markerkey_texts r*r@z-_looks_like_pem_public_key..s< fv! )r:bytesr2striprB)r( pem_markersrs @r*_looks_like_pem_public_keyrsD#ujjj3yy{HK < < < ES256ES384ES512HS256HS384HS512PS256PS384PS512ruRS384RS512r#r$rzhSymmetric HS* algorithms cannot be used with jwks_uri; configure a shared secret via public_key instead.zCSymmetric HS* algorithms require a shared secret, not a public key.)rrrr)r'r%rrsuper__init__r)rrr_rr _http_clientrrOlogger _jwks_cache_jwks_cache_time _cache_ttl) rr_rrrr)rrrrparsed_required_scopes __class__s r*rzJWTVerifier.__init__sAJ(MN N (NO O  k&=E  (   6ykCD D    % H8D Y .=-HL )d  2  #   $  "' * ,.'(r,cK|jr |jS t|}|jd}|j|d{S7#tt t tjf$r}t d||d}~wwxYww)z'Get the verification key for the token.rFNz%Failed to extract key ID from token: ) r_rr0 _get_jwks_keyr'KeyError IndexErrorjsonJSONDecodeError)rrr?rFrJs r*_get_verification_keyz!JWTVerifier._get_verification_key=s ???? " Q&u-F**U#C++C00 00Hj$2F2FG QDQCHIq P Qs:B 0A A AB A#B 6BB  B c|K|js tdtj}||jz |jkre|r||j vr|j |S|sDt |j dk(r,tt|j jS |jd{}i|_|jdgD]?}|jd}t|}|r||j |<1||j d<A||_|rH||j vr+|jjd|td|d |j |St |j dk(r,tt|j jSt |j dkDr td td 7##tt f$r0}|jjd |td ||d}~wt"j$$r}td ||d}~wt&j($r}td||d}~wt*t,t.tf$r0}|jjd|td||d}~wwxYww)zD$$Z06%1D !d...KK%%G%xu4G%HII'',,t''(A-T%5%5%<%<%> ?@@))*Q.$L%%<==C1F>* B KK  I1 M5aS9: A B5aS9: A## ?21#67Q >9h ; D KK  > B7s;@)max_sizetimeoutoverall_timeoutN)r)rr'rrrloadsr contextlib nullcontextr AsyncClientTimeoutr0raise_for_status)rcontentclientresponses r*rzJWTVerifier._fetch_jwkss}}67 7 >>+  $ G ::g& &$$0&&t'8'89&&u}}T/BCDD!'DMM!::))+}} DDDD; DDDDsmAD*D A1D*4D5D*8DD#D; D*DD*D*DD*D'D D'#D*cdD]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). )r|scp)r:r;splitrA)rclaimsclaims r*_extract_scopeszJWTVerifier._extract_scopessU&EfUmS1!%=..00u t4!%=( & r,c LK |j|d{}t||j}t|}t |r|j j dytj|||jgt|jgdj}|jdxs(|jdxs|jdxsd }|jd }|4|tjkr|j jd |y|jrv|jd }d} t|jt r||jv} n||jk(} | s)|j j#d |||jy|j$r|jdd} t|j$t rLtt rt'fd|j$D} nKt)t |j$v} n.tt r|j$v} n|j$k(} | s)|j j#d||j$y|j+|} |j,rbt/| } t/|j,} | j1| s1|j j#d|t3| t3| yt5|t7|| | t9|nd|S7#t:$r|j j dYyt<t>t@tBf$r/}|j j dt7|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. Nz8Token validation failed: unsupported critical JWT headerF)r}strict_check_header)r}registry client_idazprvunknownryz2Bearer token rejected for client %s: token expiredrwzJBearer token rejected for client %s: issuer mismatch (got %r, expected %r)rzc3&K|]}|v ywr9rT)r=expectedrzs r*r@z0JWTVerifier.load_access_token.. s-L4 N$ %L40N$1L44$N!N$N!2%NN$N!!N$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)rrs r* verify_tokenzJWTVerifier.verify_token@s++E2222s )r_zstr | bytes | Nonerrrrrrr)rrrrzAnyHttpUrl | str | Nonerboolrzhttpx.AsyncClient | None)rr;r str | bytes)rFrrr;)rdict[str, Any])rrrrKrr;rzAccessToken | None) rOrPrQrRrrrrrrr __classcell__rs@r*rrs**.#)-+/ $,0,004e'e e ' e ) ee*e*ee.eN Q=D~'."L\ 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 )rN)rrtokens)rrrrs r*rzStaticTokenVerifier.__init__bs 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.Nrrz$Token missing required scopes. Has: z , Required: rr) rr0rrrrrrr)rr token_datarrrrs r*rz StaticTokenVerifier.verify_tokenrs[[__U+  ^^L1  !j499;&>"-   v;L!$"6"67O"++L9 :<. UdTef -!   sCC r9)rzdict[str, dict[str, Any]]rrr)rOrPrQrRrrrrs@r*rrPs)(-1)*  r,r)r(zstr | bytes | JWKKeyDatar)r;)r3rrr;)r?rrr)r(rrr)=rR __future__rrrr dataclassesrtypingrrrrcryptography.hazmat.primitivesr)cryptography.hazmat.primitives.asymmetricr joserfcr r joserfc.errorsr joserfc.jwsr joserfc.registryrpydanticrrtyping_extensionsrfastmcp.server.authrrfastmcp.server.auth.ssrfrrrfastmcp.utilities.authrrfastmcp.utilities.loggingrrOrdictr;rArrS frozensetr<r+r5rCrErWr]rrrrTr,r*rs0" !'' 89$#0*':OOB0 H S#S /12 I2'(;<=@   iu  y  $51\\2\~ =J3-J3Z ? -? r,