o i>Q@s`dZddlmZddlZddlmZddlmZmZddl 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)GdddeddZ*GdddeZ+eddddGdddZ,GdddeZ-GdddeZ.Gd d!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)NotSetNotSetTc@sReZdZUdZded<ded<ded<ded<ded<ded<d ed <ded <d S) JWKDatazJSON Web Key data structure.strktykidusealgne list[str]x5cx5tN__name__ __module__ __qualname____doc____annotations__r*r*g/var/www/html/karishye-ai-python/venv/lib/python3.10/site-packages/fastmcp/server/auth/providers/jwt.pyrs  rF)totalc@seZdZUdZded<dS)JWKSDataz JSON Web Key Set data structure.z list[JWKData]keysNr$r*r*r*r+r-(s  r-T)frozenkw_onlyreprc@sJeZdZUdZded<ded<edddZ ddddZd S) RSAKeyPairzRSA key pair for JWT testing.r private_keyr public_keyreturncCs`tjddd}|jtjjtjjtd d}| j tjjtj j d d}|t||dS)zt Generate an RSA key pair for testing. Returns: RSAKeyPair: Generated key pair ii)public_exponentkey_size)encodingformatencryption_algorithmutf-8)r8r9)r3r4)r generate_private_key private_bytesr EncodingPEM PrivateFormatPKCS8 NoEncryptiondecoder4 public_bytes PublicFormatSubjectPublicKeyInfor )clsr3 private_pem public_pemr*r*r+generate5s,  zRSAKeyPair.generate fastmcp-userhttps://fastmcp.example.comNsubjectissueraudiencestr | list[str] | Nonescopeslist[str] | Noneexpires_in_secondsintadditional_claimsdict[str, Any] | Noner str | Nonec Csddi}|r ||d<||tttt|d} |r!|| d<|r*d|| d<|r1| |tdg} | || |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 rRS256r)subissiatexpaud scoper;) rUtimejoinupdaterencoder3get_secret_valuerC) selfrNrOrPrRrTrVrheaderpayloadjwt_lib token_bytesr*r*r+ create_tokenYs&     zRSAKeyPair.create_token)r5r2)rKrLNNrMNN)rNrrOrrPrQrRrSrTrUrVrWrrXr5r)r%r&r'r(r) classmethodrJrkr*r*r*r+r2.s  %r2c@seZdZUdZededdZdZded<dZ ded<dZ d ed <dZ ded <dZ d ed <dZ d ed<dZded<edddeddZdS)JWTVerifierSettingsz$Settings for JWT token verification.FASTMCP_SERVER_AUTH_JWT_ignore) env_prefixenv_fileextraNrXr4jwks_urirQrO algorithmrPrSrequired_scopeszAnyHttpUrl | str | Nonebase_urlbefore)modecCst|SNr)rGvr*r*r+ _parse_scopessz!JWTVerifierSettings._parse_scopes)r%r&r'r(rr model_configr4r)rsrOrtrPrurvr rlr{r*r*r*r+rms"         rmcsfeZdZdZeeeeeeedd#fddZd$ddZd%ddZd&ddZd'dd Z d'd!d"Z Z 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 r4rsrOrPrtrurvr4str | NotSetT | NonersrO str | list[str] | NotSetT | NonerPrtrulist[str] | NotSetT | Nonerv!AnyHttpUrl | str | NotSetT | Nonec stdd|||||||dD}|js|jstd|jr(|jr(td|jp,d}|dvr9td|d tj|j |j d ||_|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. cSsi|] \}}|tur||qSr*)r).0krzr*r*r+ s   z(JWTVerifier.__init__..r~z.Either public_key or jwks_uri must be providedz/Provide either public_key or jwks_uri, not bothrY> ES256ES384ES512HS256HS384HS512PS256PS384PS512rYRS384RS512zUnsupported algorithm: .)rvrurrMN)rmmodel_validateitemsr4rs ValueErrorrtsuper__init__rvrurOrPrjwtrr%logger _jwks_cache_jwks_cache_time _cache_ttl) rfr4rsrOrPrtrurvsettings __class__r*r+rsD     zJWTVerifier.__init__tokenrr5c s|jr|jSz1ddl}ddl}|dd}|ddt|d7}|||}|d}||IdHWSt yL}zt d||d}~ww)z'Get the verification key for the token.rNr=rz%Failed to extract key ID from token: ) r4base64jsonsplitlenloadsurlsafe_b64decodeget _get_jwks_key Exceptionr)rfrrr header_b64rgrr r*r*r+_get_verification_key s z!JWTVerifier._get_verification_keyrrXc s|jstdt}||j|jkr2|r ||jvr |j|S|s2t|jdkr2tt|j Szt 4IdH}| |jIdH}| |}WdIdHn 1IdHs]wYi|_| dgD]}| d}t|}|} |r| |j|<qk| |jd<qk||_|r||jvr|jd|td|d |j|WSt|jdkrtt|j WSt|jdkrtd td t jy} ztd | | d} ~ wty} z|jd | td | | d} ~ ww)z(Fetch key from JWKS with simple caching.zJWKS URI not configuredNr.r_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: )rsrrarrrrnextitervalueshttpx AsyncClientrraise_for_statusrr import_keyget_public_keyrdebug HTTPErrorr) rfr current_timeclientresponse jwks_datakey_datakey_kidjwkr4r r*r*r+r s\  (      zJWTVerifier._get_jwks_keyclaimsdict[str, Any]r!cCsNdD]"}||vr$t||tr||St||tr$||SqgS)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) isinstancerrlist)rfrclaimr*r*r+_extract_scopes^s zJWTVerifier._extract_scopesAccessToken | Nonec sFz||IdH}|j||}|dp!|dp!|dp!d}|d}|r@|tkr@|jd||jd|WdS|jrn|d }d }t |jt rV||jv}n||jk}|sn|jd ||jd|WdS|j r|d d }t |j t rt t rt fd d|j D}nt t |j v}nt t r|j v}n|j k}|s|jd||jd|WdS||} |jrt| } t|j} | | s|jd| | |jd|WdSt|t|| |rt|nd|dWSty|jdYdSty"} z|jdt| WYd} ~ dSd} ~ ww)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_idazprZunknownr]z4Token validation failed: expired token for client %sz#Bearer token rejected for client %sr[Fz6Token validation failed: issuer mismatch for client %sr^c3s|]}|vVqdSryr*)rexpectedr^r*r+ s z0JWTVerifier.load_access_token..z8Token validation failed: audience mismatch for client %sz4Token missing required scopes. Has: %s, Required: %srrrR expires_atrz5Token validation failed: JWT signature/format invalidzToken validation failed: %s)rrrCrrarrinforOrrrPanyrrrusetissubsetrrrUrr) rfrverification_keyrrr]r[ issuer_validaudience_validrR token_scopesrur r*rr+load_access_tokenos                   zJWTVerifier.load_access_tokencs||IdHS)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)rfrr*r*r+ verify_tokens zJWTVerifier.verify_token)r4rrsrrOrrPrrtrrurrvr)rrr5r)rrXr5r)rrr5r!rrr5r) r%r&r'r(rrrrrrr __classcell__r*r*rr+r}s T  > vr}cs0eZdZdZ ddfdd Zdd dZZS)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]]rurScstj|d||_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 )ruN)rrr)rfrrurr*r+rs zStaticTokenVerifier.__init__rrr5rcs|j|}|s dS|d}|dur|tkrdS|dg}|jr@t|}t|j}||s@td|d|dSt||d|||dS)z-Verify token against static token dictionary.NrrRz$Token missing required scopes. Has: z , Required: rr) rrrarurrrrr)rfr token_datarrRrrur*r*r+rs.     z StaticTokenVerifier.verify_tokenry)rrrurSr)r%r&r'r(rrrr*r*rr+rs r)0r( __future__rra dataclassesrtypingrrr authlib.joserrauthlib.jose.errorsrcryptography.hazmat.primitivesr )cryptography.hazmat.primitives.asymmetricr pydanticr r r pydantic_settingsrrtyping_extensionsrfastmcp.server.authrrfastmcp.settingsrfastmcp.utilities.authrfastmcp.utilities.loggingrfastmcp.utilities.typesrrr%rrr-r2rmr}rr*r*r*r+s8          _R