authorization_django v3.0.0

.github/workflows/tests.yaml

@@ -39,7 +39,7 @@ jobs:
 
       - name: Install Packages
         run: |
-          uv sync --python ${{ matrix.python }} --extra tests --no-default-groups
+          uv sync --python ${{ matrix.python }} --extra extended --no-default-groups
 
       - name: Run Tests
         run: |

README.md

@@ -206,6 +206,11 @@ We use GitHub pull requests. If your PR should produce a new release of authoriz
 
 Changelog
 ---------
+* v3.0.0
+  * Deprecated JWKS_URL(S)/CHECK_CLAIMS.
+  * Increase test coverage.
+  * Drop SettingsProxy as it is no longer needed.
+  * Add utility method for getting TRUSTED_JWKS from env variables.
 * v2.4.3
   * Bugfix: ensure all jwks entries are tried when keys are missing.
 * v2.4.2

authorization_django/config.py

@@ -4,10 +4,18 @@
 """
 
 import logging
-from collections.abc import Callable, Iterator, Mapping
+from collections.abc import Callable
+from time import time
 
 from django.conf import settings as django_settings
-from pydantic import BaseModel, Field, ValidationError, field_validator, model_validator
+from pydantic import (
+    BaseModel,
+    Field,
+    ValidationError,
+    computed_field,
+    field_validator,
+    model_validator,
+)
 
 logger = logging.getLogger(__name__)
 
@@ -18,6 +26,11 @@ class Claims(BaseModel):
     iss: str
     aud: str | list[str] | None = None
 
+    @computed_field
+    @property
+    def exp(self) -> int:
+        return int(time())
+
 
 class TrustedJwksItem(BaseModel):
     jwks_url: str | None = None
@@ -89,11 +102,7 @@ def from_raw(cls, resource):
 
 
 class Settings(BaseModel):
-    JWKS: str | None = ""
-    JWKS_URL: str | None = ""
-    JWKS_URLS: list[str] = Field(default_factory=list)
-    CHECK_CLAIMS: dict = Field(default_factory=dict)
-    TRUSTED_JWKS: list[TrustedJwksItem] = Field(default_factory=list)
+    TRUSTED_JWKS: list[TrustedJwksItem] = Field(default_factory=list, min_length=1)
     ALLOWED_SIGNING_ALGORITHMS: list[str] = Field(
         default_factory=lambda: [
             "ES256",
@@ -111,28 +120,6 @@ class Settings(BaseModel):
     MIN_INTERVAL_KEYSET_UPDATE: int = 30
     EXCEPTION_HANDLER: Callable | None = None
 
-    @property
-    def effective_jwks_url(self):
-        try:
-            return next(item.jwks_url for item in self.TRUSTED_JWKS if item.jwks_url)
-        except StopIteration:
-            return self.JWKS_URL
-
-    @property
-    def effective_jwks_urls(self):
-        if self.TRUSTED_JWKS:
-            trusted_urls = [item.jwks_url for item in self.TRUSTED_JWKS if item.jwks_url]
-            if trusted_urls:
-                return trusted_urls
-        return self.JWKS_URLS
-
-    @property
-    def effective_check_claims(self):
-        for item in self.TRUSTED_JWKS:
-            if item.claims:
-                return item.claims.model_dump(exclude_none=True)
-        return self.CHECK_CLAIMS
-
     @field_validator("MIN_SCOPE", mode="before")
     def validate_min_scope(cls, v):
         if type(v) is not tuple:
@@ -155,20 +142,6 @@ def validate_protected(cls, value):
 
     @model_validator(mode="after")
     def validate_model(self):
-        if not self.JWKS and not self.effective_jwks_url and not self.effective_jwks_urls:
-            raise AuthzConfigurationError(
-                f"{SETTINGS_KEY}['JWKS'], {SETTINGS_KEY}['JWKS_URL'] or {SETTINGS_KEY}['JWKS_URLS']  must be set, or all"
-            )
-
-        is_entra = (
-            self.effective_jwks_url and self.effective_jwks_url.startswith(MICROSOFT)
-        ) or any(url.startswith(MICROSOFT) for url in self.effective_jwks_urls)
-        if is_entra and {"iss", "aud"}.isdisjoint(self.effective_check_claims):
-            raise AuthzConfigurationError(
-                "When using Microsoft Entra ID, make sure to set an 'iss' and 'aud' claim"
-                f" in the {SETTINGS_KEY}['TRUSTED_JWKS'] settings for entra."
-            )
-
         for resource in self.PROTECTED:
             for anonymous_route in self.FORCED_ANONYMOUS_ROUTES:
                 if resource.route.startswith(anonymous_route):
@@ -206,104 +179,6 @@ class NoRequiredScopesError(AuthzConfigurationError):
     """
 
 
-class SettingsProxy(Mapping):
-    """Read-only settings wrapper with deprecated-setting compatibility."""
-
-    _deprecated_keys = {"JWKS_URL", "JWKS_URLS", "CHECK_CLAIMS"}
-
-    def __init__(self, settings: Settings):
-        self._values = settings.model_dump()
-        self._values["TRUSTED_JWKS"] = [
-            item.model_dump(exclude_none=True) for item in settings.TRUSTED_JWKS
-        ]
-        self._values["PROTECTED"] = [
-            (resource.route, resource.methods, resource.scopes) for resource in settings.PROTECTED
-        ]
-        # Warn if any deprecated keys are present in the initial values
-        deprecated_keys = self._deprecated_keys & self._values.keys()
-        if deprecated_keys:
-            logger.warning(
-                "Deprecated settings present: %s. Please migrate to TRUSTED_JWKS.",
-                ", ".join(sorted(deprecated_keys)),
-            )
-
-    def __getitem__(self, key):
-        if key in self._deprecated_keys:
-            logger.warning("Accessing deprecated setting %s. Please migrate to TRUSTED_JWKS.", key)
-            trusted_value = self._trusted_jwks_value(key)
-            if trusted_value is not None:
-                return trusted_value
-        if key == "TRUSTED_JWKS":
-            if self._values["TRUSTED_JWKS"]:
-                return self._values["TRUSTED_JWKS"]
-            logger.warning(
-                "TRUSTED_JWKS is not set, constructing from JWKS, JWKS_URLS, JWKS_URL, and CHECK_CLAIMS."
-            )
-            logger.warning("This will be deprecated in v3.0.0")
-            return self._compose_trusted_jwks()
-        return self._values[key]
-
-    def _compose_trusted_jwks(self):
-        trusted_jwks = []
-        check_claims = self._values.get("CHECK_CLAIMS")
-        check_claims_no_aud = {k: v for k, v in (check_claims or {}).items() if k != "aud"}
-        if self._values["JWKS"]:
-            trusted_jwks.append(
-                {
-                    "jwks": self._values["JWKS"],
-                    "claims": check_claims_no_aud,
-                }
-            )
-        if self._values["JWKS_URLS"]:
-            trusted_jwks.extend(
-                {
-                    "jwks_url": url,
-                    "claims": check_claims if url.startswith(MICROSOFT) else check_claims_no_aud,
-                }
-                for url in self._values["JWKS_URLS"]
-            )
-        if self._values["JWKS_URL"]:
-            trusted_jwks.append(
-                {
-                    "jwks_url": self._values["JWKS_URL"],
-                    "claims": check_claims
-                    if self._values["JWKS_URL"].startswith(MICROSOFT)
-                    else check_claims_no_aud,
-                }
-            )
-        return trusted_jwks
-
-    def __iter__(self) -> Iterator[str]:
-        return iter(self._values)
-
-    def __len__(self) -> int:
-        return len(self._values)
-
-    def get(self, key, default=None):
-        try:
-            return self[key]
-        except KeyError:
-            return default
-
-    def _trusted_jwks_value(self, key):
-        trusted_jwks = self._values.get("TRUSTED_JWKS") or []
-        if not trusted_jwks:
-            return None
-
-        if key == "JWKS_URL":
-            return trusted_jwks[0].get("jwks_url")
-
-        if key == "JWKS_URLS":
-            return [item["jwks_url"] for item in trusted_jwks if item.get("jwks_url")] or None
-
-        if key == "CHECK_CLAIMS":
-            try:
-                return next(item["claims"] for item in trusted_jwks if item.get("claims"))
-            except StopIteration:
-                return None
-        return None
-
-
 def init_settings():
     global _settings
     _settings = load_settings()
@@ -325,8 +200,6 @@ def load_settings():
     user_settings = dict(getattr(django_settings, SETTINGS_KEY, {}))
 
     try:
-        settings = Settings.model_validate(user_settings, extra="forbid")
+        return Settings.model_validate(user_settings, extra="forbid")
     except ValidationError as e:
         raise AuthzConfigurationError(f"Invalid {SETTINGS_KEY} configuration: {e}") from e
-
-    return SettingsProxy(settings)

authorization_django/extensions/drf.py

@@ -46,14 +46,14 @@ class HasTokenScopes(BasePermission):
     message = "Required scopes not given in token."
 
     def __init__(self, *needed_scopes):
-        self.needed_scopes = frozenset(needed_scopes or get_settings()["MIN_SCOPE"])
+        self.needed_scopes = frozenset(needed_scopes or get_settings().MIN_SCOPE)
 
     def has_permission(self, request, view):
         """Check whether the user has all required scopes"""
         # This essentially does what request.is_authorized_for() does, without the logging.
         # In this scenario it's not clear whether this is the only permission check,
         # so falsely logging that access is granted is a bit premature.
-        return get_settings()["ALWAYS_OK"] or set(request.get_token_scopes).issuperset(
+        return get_settings().ALWAYS_OK or set(request.get_token_scopes).issuperset(
             self.needed_scopes
         )
 

authorization_django/jwks.py

@@ -30,10 +30,10 @@ def init_keyset(self):
         self._keyset: dict[str, JWKSet] = defaultdict(JWKSet)
         self._keyset_last_update = time.time()
 
-        for trusted_jwks_item in self._settings["TRUSTED_JWKS"]:
-            if url := trusted_jwks_item.get("jwks_url"):
+        for trusted_jwks_item in self._settings.TRUSTED_JWKS:
+            if url := trusted_jwks_item.jwks_url:
                 _load_jwks_from_url(self._keyset[url], url)
-            elif jwks := trusted_jwks_item.get("jwks"):
+            elif jwks := trusted_jwks_item.jwks:
                 _load_jwks(self._keyset["JWKS"], jwks)
 
         if not any(len(keyset["keys"]) > 0 for keyset in self._keyset.values()):
@@ -57,16 +57,13 @@ def check_update_keyset(self):
         the url, we set a minimal interval between two checks.
         """
         current_time = time.time()
-        if current_time - self._keyset_last_update >= self._settings["MIN_INTERVAL_KEYSET_UPDATE"]:
+        if current_time - self._keyset_last_update >= self._settings.MIN_INTERVAL_KEYSET_UPDATE:
             self.init_keyset()
 
 
 def _load_jwks(keyset: JWKSet, jwks):
     try:
-        if type(jwks) is str:
-            keyset.import_keyset(jwks)
-        else:
-            keyset.import_keyset(json.dumps(jwks))
+        keyset.import_keyset(json.dumps(jwks))
     except JWException as e:
         raise AuthzConfigurationError("Failed to import keyset from settings") from e
     logger.info("Loaded JWKS from JWKS setting.")

authorization_django/middleware.py

@@ -5,7 +5,6 @@
 
 import json
 import logging
-from time import time
 
 from django.http import HttpRequest, HttpResponse, JsonResponse
 from jwcrypto.common import JWException
@@ -84,7 +83,7 @@ def authorize_forced_anonymous(_):
         raise RuntimeError("Should not call is_authorized_for in anonymous routes")
 
     def handle_exception(self, request, exception):
-        if exception_handler := self.settings["EXCEPTION_HANDLER"]:
+        if exception_handler := self.settings.EXCEPTION_HANDLER:
             return exception_handler(
                 request, exception
             )  # other application takes care of exception handling
@@ -149,14 +148,13 @@ def _get_account_id(self, claims, sub) -> str:
     def _decode_token(self, raw_jwt):
         keyset = self.jwks.keyset
         error = None
-        for trusted_jwks_item in self.settings["TRUSTED_JWKS"]:
-            check_claims = trusted_jwks_item.get("claims", {})
-            check_claims["exp"] = int(time())
+        for trusted_jwks_item in self.settings.TRUSTED_JWKS:
+            check_claims = trusted_jwks_item.claims.model_dump(exclude_none=True)
             try:
                 return JWT(
                     jwt=raw_jwt,
-                    key=keyset[trusted_jwks_item.get("jwks_url", "JWKS")],
-                    algs=self.settings["ALLOWED_SIGNING_ALGORITHMS"],
+                    key=keyset[trusted_jwks_item.jwks_url or "JWKS"],
+                    algs=self.settings.ALLOWED_SIGNING_ALGORITHMS,
                     check_claims=check_claims,
                 )
             except JWTExpired as e:
@@ -222,21 +220,20 @@ def convert_scope(self, scope):
         return scope.upper().replace("_", "/")
 
     def handle_scope(self, authz_func, request: HttpRequest):
-        min_scope = self.settings["MIN_SCOPE"]
+        min_scope = self.settings.MIN_SCOPE
 
         if not authz_func:
             raise InsufficientScopeError
 
         if len(min_scope) > 0 and not authz_func(*min_scope):
             raise InsufficientScopeError
 
-        PROTECTED = self.settings["PROTECTED"]
+        PROTECTED = self.settings.PROTECTED
         for resource in PROTECTED:
-            (route, protected_methods, required_scopes) = resource
             if (
-                request.path.startswith(route)
-                and _method_is_protected(request.method, protected_methods)
-                and not authz_func(*required_scopes)
+                request.path.startswith(resource.route)
+                and _method_is_protected(request.method, resource.methods)
+                and not authz_func(*resource.scopes)
             ):
                 raise InsufficientScopeError
 
@@ -246,15 +243,15 @@ def __call__(self, request: HttpRequest):
         """
 
         # Config is set to ALWAYS OK, authorisation check disabled
-        if self.settings["ALWAYS_OK"]:
+        if self.settings.ALWAYS_OK:
             logger.warning("API authz DISABLED")
             request.is_authorized_for = self.always_ok
             request.get_token_subject = "ALWAYS_OK"  # noqa: S105
             return self.get_response(request)
 
         # Path is in forced anonymous routes or method is Options
         forced_anonymous = any(
-            request.path.startswith(route) for route in self.settings["FORCED_ANONYMOUS_ROUTES"]
+            request.path.startswith(route) for route in self.settings.FORCED_ANONYMOUS_ROUTES
         )
 
         if forced_anonymous or request.method == "OPTIONS":

authorization_django/utils.py

@@ -0,0 +1,29 @@
+import json
+import os
+
+import environ
+
+env = environ.Env()
+
+
+def get_trusted_jwks(
+    providers: list[str] | None = None,
+    pub_jwks: str = "PUB_JWKS",
+) -> list[dict]:
+    if providers is None:
+        providers = ["ENTRA", "KEYCLOAK"]
+    trusted_jwks = []
+    for claim in providers:
+        url = f"OAUTH_{claim}_URL"
+        claims = f"OAUTH_{claim}_CLAIMS"
+        if jwks_url := os.getenv(url):
+            trusted_jwks.append(
+                {
+                    "jwks_url": jwks_url,
+                    "claims": env.dict(claims, default={}),
+                }
+            )
+    if jwks := os.getenv(pub_jwks):
+        # Only used for testing, issuer is set to a dummy value.
+        trusted_jwks.append({"jwks": json.loads(jwks), "claims": {"iss": "iss"}})
+    return trusted_jwks