1 i!ddlmZddlZddlZddlmZddlmZddl m Z ddl m Z m Z ddlmZmZdd lmZdd lmZmZdd lmZGd d ZdS)) annotationsN) lru_cache) SSLContext)Any) HTTPErrorURLError)PyJWKPyJWKSet)decode_complete)PyJWKClientConnectionErrorPyJWKClientError) JWKSetCachecjeZdZ d'd(dZd)dZd*d+dZd*d,dZd-d Zd.d#Ze d/d&Z dS)0 PyJWKClientFT,Nuristr cache_keysboolmax_cached_keysint cache_jwk_setlifespanfloatheadersdict[str, Any] | Nonetimeout ssl_contextSSLContext | Nonec |i}||_d|_||_||_||_|r.|dkrt d|dt ||_nd|_|r't||j} | |_dSdS)uA client for retrieving signing keys from a JWKS endpoint. ``PyJWKClient`` uses a two-tier caching system to avoid unnecessary network requests: **Tier 1 — JWK Set cache** (enabled by default): Caches the entire JSON Web Key Set response from the endpoint. Controlled by: - ``cache_jwk_set``: Set to ``True`` (the default) to enable this cache. When enabled, the JWK Set is fetched from the network only when the cache is empty or expired. - ``lifespan``: Time in seconds before the cached JWK Set expires. Defaults to ``300`` (5 minutes). Must be greater than 0. **Tier 2 — Signing key cache** (disabled by default): Caches individual signing keys (looked up by ``kid``) using an LRU cache with **no time-based expiration**. Keys are evicted only when the cache reaches its maximum size. Controlled by: - ``cache_keys``: Set to ``True`` to enable this cache. Defaults to ``False``. - ``max_cached_keys``: Maximum number of signing keys to keep in the LRU cache. Defaults to ``16``. :param uri: The URL of the JWKS endpoint. :type uri: str :param cache_keys: Enable the per-key LRU cache (Tier 2). :type cache_keys: bool :param max_cached_keys: Max entries in the signing key LRU cache. :type max_cached_keys: int :param cache_jwk_set: Enable the JWK Set response cache (Tier 1). :type cache_jwk_set: bool :param lifespan: TTL in seconds for the JWK Set cache. :type lifespan: float :param headers: Optional HTTP headers to include in requests. :type headers: dict or None :param timeout: HTTP request timeout in seconds. :type timeout: float :param ssl_context: Optional SSL context for the request. :type ssl_context: ssl.SSLContext or None Nrz/Lifespan must be greater than 0, the input is "")maxsize) r jwk_set_cacherr r!rrrget_signing_key) selfrrrrrrr r!r's C:\Users\Dell Inspiron 16\Desktop\tws\AgrotaPowerBi\back-agrota-powerbi\mcp-client-agrota\venv\Lib\site-packages\jwt/jwks_client.py__init__zPyJWKClient.__init__sj ?G15  &  &1}}&QhQQQ"-X!6!6D  !%D   3@i@@@AUVVO#2D  3 3returnrcPd} tj|j|j}tj||j|j5}tj |}dddn #1swxYwY||j |j |SS#ttf$rB}t|tr|t#d|d|d}~wwxYw#|j |j |wwxYw)aeFetch the JWK Set from the JWKS endpoint. Makes an HTTP request to the configured ``uri`` and returns the parsed JSON response. If the JWK Set cache is enabled, the response is stored in the cache. :returns: The parsed JWK Set as a dictionary. :raises PyJWKClientConnectionError: If the HTTP request fails. N)urlr)r contextz'Fail to fetch data from the url, err: "r$)urllibrequestRequestrrurlopenr r!jsonloadr&putr TimeoutError isinstancercloser )r(jwk_setrresponsees r) fetch_datazPyJWKClient.fetch_data_sz 0&&48T\&JJA''4<1A( .)H-- . . . . . . . . . . . . . . .!-"&&w////.,'   !Y''  ,>!>>>  !-"&&w////.sNAB+A<0 B+<BB+BB+D+C><=C99C>>D$D%refreshr cd}|j|s|j}||}t|tst dt j|S)aNReturn the JWK Set, using the cache when available. :param refresh: Force a fresh fetch from the endpoint, bypassing the cache. :type refresh: bool :returns: The JWK Set. :rtype: PyJWKSet :raises PyJWKClientError: If the endpoint does not return a JSON object. Nz.The JWKS endpoint did not return a JSON object)r&getr>r8dictrr from_dict)r(r?datas r) get_jwk_setzPyJWKClient.get_jwk_set|sr   )' )%))++D <??$$D$%% U"#STT T!$'''r+ list[PyJWK]ct||}d|jD}|std|S)aReturn all signing keys from the JWK Set. Filters the JWK Set to keys whose ``use`` is ``"sig"`` (or unspecified) and that have a ``kid``. :param refresh: Force a fresh fetch from the endpoint, bypassing the cache. :type refresh: bool :returns: A list of signing keys. :rtype: list[PyJWK] :raises PyJWKClientError: If no signing keys are found. c2g|]}|jdv |j|S))sigN)public_key_usekey_id).0 jwk_set_keys r) z0PyJWKClient.get_signing_keys..s8   )]::{?Q: :::r+z2The JWKS endpoint did not contain any signing keys)rEkeysr)r(r?r: signing_keyss r)get_signing_keyszPyJWKClient.get_signing_keyssW""7++  &|    Y"#WXX Xr+kidr c|}|||}|sA|d}|||}|std|d|S)aReturn the signing key matching the given ``kid``. If no match is found in the current JWK Set, the set is refreshed from the endpoint and the lookup is retried once. :param kid: The key ID to look up. :type kid: str :returns: The matching signing key. :rtype: PyJWK :raises PyJWKClientError: If no matching key is found after refreshing. T)r?z,Unable to find a signing key that matches: "r$)rQ match_kidr)r(rRrP signing_keys r)r'zPyJWKClient.get_signing_keys,,.. nn\377  000>>L..s;;K &I3IIIr+token str | bytesct|ddi}|d}||dS)aGReturn the signing key for a JWT by reading its ``kid`` header. Extracts the ``kid`` from the token's unverified header and delegates to :meth:`get_signing_key`. :param token: The encoded JWT. :type token: str or bytes :returns: The matching signing key. :rtype: PyJWK verify_signatureF)optionsheaderrR) decode_tokenr'rA)r(rV unverifiedr[s r)get_signing_key_from_jwtz$PyJWKClient.get_signing_key_from_jwtsF"%2De1LMMM H%##FJJu$5$5666r+rP PyJWK | Nonec2d}|D]}|j|kr|}n|S)a7Find a key in *signing_keys* that matches *kid*. :param signing_keys: The list of keys to search. :type signing_keys: list[PyJWK] :param kid: The key ID to match. :type kid: str :returns: The matching key, or ``None`` if not found. :rtype: PyJWK or None N)rK)rPrRrUkeys r)rTzPyJWKClient.match_kids<   CzS  ! !r+)FrTrNrN)rrrrrrrrrrrrr rr!r")r,r)F)r?rr,r )r?rr,rF)rRrr,r )rVrWr,r )rPrFrRrr,r_) __name__ __module__ __qualname__r*r>rErQr'r^ staticmethodrTr+r)rrs!!")-)-L3L3L3L3L3\0000:(((((.28 7 7 7 7\r+r) __future__rr4urllib.requestr0 functoolsrsslrtypingr urllib.errorrrapi_jwkr r api_jwtr r\ exceptionsr rr&rrrfr+r)rps """""" ,,,,,,,,$$$$$$$$444444DDDDDDDD&&&&&&YYYYYYYYYYr+