Coverage for /pythoncovmergedfiles/medio/medio/usr/local/lib/python3.8/site-packages/jwt/jwks_client.py: 26%
73 statements
« prev ^ index » next coverage.py v7.3.2, created at 2023-12-09 07:17 +0000
« prev ^ index » next coverage.py v7.3.2, created at 2023-12-09 07:17 +0000
1import json
2import urllib.request
3from functools import lru_cache
4from ssl import SSLContext
5from typing import Any, Dict, List, Optional
6from urllib.error import URLError
8from .api_jwk import PyJWK, PyJWKSet
9from .api_jwt import decode_complete as decode_token
10from .exceptions import PyJWKClientConnectionError, PyJWKClientError
11from .jwk_set_cache import JWKSetCache
14class PyJWKClient:
15 def __init__(
16 self,
17 uri: str,
18 cache_keys: bool = False,
19 max_cached_keys: int = 16,
20 cache_jwk_set: bool = True,
21 lifespan: int = 300,
22 headers: Optional[Dict[str, Any]] = None,
23 timeout: int = 30,
24 ssl_context: Optional[SSLContext] = None,
25 ):
26 if headers is None:
27 headers = {}
28 self.uri = uri
29 self.jwk_set_cache: Optional[JWKSetCache] = None
30 self.headers = headers
31 self.timeout = timeout
32 self.ssl_context = ssl_context
34 if cache_jwk_set:
35 # Init jwt set cache with default or given lifespan.
36 # Default lifespan is 300 seconds (5 minutes).
37 if lifespan <= 0:
38 raise PyJWKClientError(
39 f'Lifespan must be greater than 0, the input is "{lifespan}"'
40 )
41 self.jwk_set_cache = JWKSetCache(lifespan)
42 else:
43 self.jwk_set_cache = None
45 if cache_keys:
46 # Cache signing keys
47 # Ignore mypy (https://github.com/python/mypy/issues/2427)
48 self.get_signing_key = lru_cache(maxsize=max_cached_keys)(self.get_signing_key) # type: ignore
50 def fetch_data(self) -> Any:
51 jwk_set: Any = None
52 try:
53 r = urllib.request.Request(url=self.uri, headers=self.headers)
54 with urllib.request.urlopen(
55 r, timeout=self.timeout, context=self.ssl_context
56 ) as response:
57 jwk_set = json.load(response)
58 except (URLError, TimeoutError) as e:
59 raise PyJWKClientConnectionError(
60 f'Fail to fetch data from the url, err: "{e}"'
61 )
62 else:
63 return jwk_set
64 finally:
65 if self.jwk_set_cache is not None:
66 self.jwk_set_cache.put(jwk_set)
68 def get_jwk_set(self, refresh: bool = False) -> PyJWKSet:
69 data = None
70 if self.jwk_set_cache is not None and not refresh:
71 data = self.jwk_set_cache.get()
73 if data is None:
74 data = self.fetch_data()
76 if not isinstance(data, dict):
77 raise PyJWKClientError("The JWKS endpoint did not return a JSON object")
79 return PyJWKSet.from_dict(data)
81 def get_signing_keys(self, refresh: bool = False) -> List[PyJWK]:
82 jwk_set = self.get_jwk_set(refresh)
83 signing_keys = [
84 jwk_set_key
85 for jwk_set_key in jwk_set.keys
86 if jwk_set_key.public_key_use in ["sig", None] and jwk_set_key.key_id
87 ]
89 if not signing_keys:
90 raise PyJWKClientError("The JWKS endpoint did not contain any signing keys")
92 return signing_keys
94 def get_signing_key(self, kid: str) -> PyJWK:
95 signing_keys = self.get_signing_keys()
96 signing_key = self.match_kid(signing_keys, kid)
98 if not signing_key:
99 # If no matching signing key from the jwk set, refresh the jwk set and try again.
100 signing_keys = self.get_signing_keys(refresh=True)
101 signing_key = self.match_kid(signing_keys, kid)
103 if not signing_key:
104 raise PyJWKClientError(
105 f'Unable to find a signing key that matches: "{kid}"'
106 )
108 return signing_key
110 def get_signing_key_from_jwt(self, token: str) -> PyJWK:
111 unverified = decode_token(token, options={"verify_signature": False})
112 header = unverified["header"]
113 return self.get_signing_key(header.get("kid"))
115 @staticmethod
116 def match_kid(signing_keys: List[PyJWK], kid: str) -> Optional[PyJWK]:
117 signing_key = None
119 for key in signing_keys:
120 if key.key_id == kid:
121 signing_key = key
122 break
124 return signing_key