Merge pull request #19 from Jawshua/static-jwks

Add support for static JWKS URIs

CHANGELOG.md

@@ -7,6 +7,18 @@ this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.htm
 
 ## [Unreleased]
 
+## [0.7.0] - 2024-01-17
+
+### Added
+
+- Support for static configurations,
+  [PR #19](https://github.com/ioxiocom/pyjwt-key-fetcher/pull/19).
+
+### Fixed
+
+- Security updates to libraries (aiohttp, cryptography).
+- Minor adjustments to error descriptions.
+
 ## [0.6.0] - 2023-09-19
 
 ### Changed
@@ -86,7 +98,8 @@ this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.htm
 
 - Everything for the initial release
 
-[unreleased]: https://github.com/ioxiocom/pyjwt-key-fetcher/compare/0.6.0...HEAD
+[unreleased]: https://github.com/ioxiocom/pyjwt-key-fetcher/compare/0.7.0...HEAD
+[0.7.0]: https://github.com/ioxiocom/pyjwt-key-fetcher/compare/0.6.0...0.7.0
 [0.6.0]: https://github.com/ioxiocom/pyjwt-key-fetcher/compare/0.5.0...0.6.0
 [0.5.0]: https://github.com/ioxiocom/pyjwt-key-fetcher/compare/0.4.0...0.5.0
 [0.4.0]: https://github.com/ioxiocom/pyjwt-key-fetcher/compare/0.3.0...0.4.0

README.md

@@ -103,6 +103,22 @@ You can override the config path when creating the `AsyncKeyFetcher` like this:
 AsyncKeyFetcher(config_path="/.well-known/dataspace/party-configuration.json")
 ```
 
+#### Using static configuration
+
+If you use an issuer that does not provide a configuration (they are for example missing
+the `/.well-known/openid-configuration`), you can create a static configuration to use
+for that issuer instead and in it specify the `jwks_uri` like this:
+
+```python
+AsyncKeyFetcher(
+    static_issuer_config={
+        "https://example.com": {
+            "jwks_uri": "https://example.com/.well-known/jwks.json",
+        },
+    },
+)
+```
+
 #### Using your own HTTP Client
 
 The library ships with a `DefaultHTTPClient` that uses `aiohttp` for fetching the JSON

pyjwt_key_fetcher/__init__.py

@@ -2,3 +2,4 @@
 from pyjwt_key_fetcher.fetcher import AsyncKeyFetcher
 from pyjwt_key_fetcher.http_client import DefaultHTTPClient
 from pyjwt_key_fetcher.key import Key
+from pyjwt_key_fetcher.provider import OpenIDConfigurationTypeDef

pyjwt_key_fetcher/fetcher.py

@@ -1,12 +1,12 @@
-from typing import Any, Dict, Iterable, MutableMapping, Optional
+from typing import Any, Dict, Iterable, Mapping, MutableMapping, Optional
 
 import jwt
 from cachetools import TTLCache
 
 from pyjwt_key_fetcher.errors import JWTFormatError, JWTInvalidIssuerError
 from pyjwt_key_fetcher.http_client import DefaultHTTPClient, HTTPClient
 from pyjwt_key_fetcher.key import Key
-from pyjwt_key_fetcher.provider import Provider
+from pyjwt_key_fetcher.provider import OpenIDConfigurationTypeDef, Provider
 
 
 class AsyncKeyFetcher:
@@ -17,6 +17,7 @@ def __init__(
         cache_ttl: int = 3600,
         cache_maxsize: int = 32,
         config_path: str = "/.well-known/openid-configuration",
+        static_issuer_config: Optional[Dict[str, OpenIDConfigurationTypeDef]] = None,
     ) -> None:
         """
         Initialize a new AsyncKeyFetcher.
@@ -28,8 +29,11 @@ def __init__(
         :param cache_maxsize: The maximum size of the cache.
         :param config_path: The path from which the configuration is fetched from the
         issuer.
+        :param static_issuer_config: A dict of static configurations for specific token
+        issuers. Fetching from config_path will be skipped for these issuers.
         """
         self.config_path = config_path
+        self.static_issuer_config = static_issuer_config or {}
 
         if not http_client:
             http_client = DefaultHTTPClient()
@@ -100,12 +104,17 @@ def _get_provider(self, iss: str) -> Provider:
         try:
             provider = self._cache[iss]
         except KeyError:
-            provider = Provider(iss, self._http_client, self.config_path)
+            provider = Provider(
+                iss,
+                self._http_client,
+                self.config_path,
+                self.static_issuer_config.get(iss, None),
+            )
             self._cache[iss] = provider
 
         return provider
 
-    async def get_configuration(self, iss: str) -> Dict[str, Any]:
+    async def get_configuration(self, iss: str) -> Mapping[str, Any]:
         """
         Get the configuration from the issuer.
 

pyjwt_key_fetcher/provider.py

@@ -1,4 +1,4 @@
-from typing import Any, Callable, Dict, Optional
+from typing import Any, Callable, Dict, Mapping, Optional, TypedDict
 from uuid import uuid4
 
 import aiocache  # type: ignore
@@ -37,16 +37,25 @@ def key_builder(
     return key
 
 
+class OpenIDConfigurationTypeDef(TypedDict):
+    """
+    Type definition for the OpenID configuration values relevant to JWT validation.
+    """
+
+    jwks_uri: str
+
+
 class Provider:
     def __init__(
         self,
         iss: str,
         http_client: HTTPClient,
         config_path: str = "/.well-known/openid-configuration",
+        static_config: Optional[OpenIDConfigurationTypeDef] = None,
     ) -> None:
         self.iss = iss
         self.http_client = http_client
-        self._configuration: Optional[Dict[str, Any]] = None
+        self._configuration: Optional[Mapping[str, Any]] = static_config
         self._jwk_map: Dict[str, Dict[str, Any]] = {}
         self.keys: Dict[str, Key] = {}
         self.config_path = config_path
@@ -61,7 +70,7 @@ async def _config_uri(self) -> str:
         """
         return f"{self.iss.rstrip('/')}{self.config_path}"
 
-    async def get_configuration(self) -> Dict[str, Any]:
+    async def get_configuration(self) -> Mapping[str, Any]:
         """
         Get the configuration as a dictionary.
 
@@ -99,14 +108,14 @@ async def _fetch_jwk_map(self) -> Dict[str, Dict[str, Any]]:
         :return: A mapping of {kid: {<data_for_the_kid>}, ...}
         :raise JWTHTTPFetchError: If there's a problem fetching the data.
         :raise JWTProviderConfigError: If the config doesn't contain "jwks_uri".
-        :raise JWTProviderJWKSError: If the jwks_uri is missing the "jwks".
+        :raise JWTProviderJWKSError: If the jwks_uri is missing the "keys".
         """
         jwks_uri = await self._get_jwks_uri()
         data = await self.http_client.get_json(jwks_uri)
         try:
             jwks_list = data["keys"]
         except KeyError as e:
-            raise JWTProviderJWKSError(f"Missing 'jwks' in {jwks_uri}") from e
+            raise JWTProviderJWKSError(f"Missing 'keys' in {jwks_uri}") from e
 
         jwk_map = {jwk["kid"]: jwk for jwk in jwks_list}
 
@@ -120,7 +129,7 @@ async def get_jwk_data(self, kid: str) -> Dict[str, Any]:
         :return: The raw JWK data as a dictionary.
         :raise JWTHTTPFetchError: If there's a problem fetching the data.
         :raise JWTProviderConfigError: If the config doesn't contain "jwks_uri".
-        :raise JWTProviderJWKSError: If the jwks_uri is missing the "jwks".
+        :raise JWTProviderJWKSError: If the jwks_uri is missing the "keys".
         :raise JWTKeyNotFoundError: If no matching kid was found.
         """
         if kid not in self._jwk_map:
@@ -138,7 +147,7 @@ async def get_key(self, kid: str) -> Key:
         :return: The Key.
         :raise JWTHTTPFetchError: If there's a problem fetching the data.
         :raise JWTProviderConfigError: If the config doesn't contain "jwks_uri".
-        :raise JWTProviderJWKSError: If the jwks_uri is missing the "jwks".
+        :raise JWTProviderJWKSError: If the jwks_uri is missing the "keys".
         :raise JWTKeyNotFoundError: If no matching kid was found.
         """
         if kid not in self.keys:

pyjwt_key_fetcher/tests/test_fetcher.py

@@ -115,3 +115,16 @@ async def test_issuer_validation(create_provider_fetcher_and_client, create_prov
 
     assert client.get_configuration.call_count == 1
     assert client.get_jwks.call_count == 1
+
+
+@pytest.mark.asyncio
+async def test_static_issuer_config():
+    issuer = "https://valid.example.com"
+
+    fetcher = pyjwt_key_fetcher.AsyncKeyFetcher(
+        static_issuer_config={issuer: {"jwks_uri": f"{issuer}/.well-known/jwks.json"}}
+    )
+    provider = fetcher._get_provider(issuer)
+
+    provider_config = await provider.get_configuration()
+    assert provider_config == {"jwks_uri": f"{issuer}/.well-known/jwks.json"}

pyproject.toml

@@ -1,19 +1,17 @@
 [tool.poetry]
 name = "pyjwt-key-fetcher"
-version = "0.6.0"
+version = "0.7.0"
 description = "Async library to fetch JWKs for JWT tokens"
 authors = ["IOXIO Ltd"]
 license = "BSD-3-Clause"
 readme = "README.md"
 repository = "https://github.com/ioxiocom/pyjwt-key-fetcher"
-packages = [
-    {include="pyjwt_key_fetcher", from="."}
-]
+packages = [{ include = "pyjwt_key_fetcher", from = "." }]
 
 [tool.poetry.dependencies]
 python = "^3.8"
-PyJWT = {version = "^2.8.0", extras = ["crypto"]}
-aiohttp = {version = "^3.8.6", extras = ["speedups"]}
+PyJWT = { version = "^2.8.0", extras = ["crypto"] }
+aiohttp = { version = "^3.9.1", extras = ["speedups"] }
 cachetools = "^5.3.2"
 aiocache = "^0.12.2"