diff --git a/.github/workflows/tests.yaml b/.github/workflows/tests.yaml index 78ba215..bbae547 100644 --- a/.github/workflows/tests.yaml +++ b/.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: | diff --git a/README.md b/README.md index 4e493fc..d52701f 100644 --- a/README.md +++ b/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 diff --git a/authorization_django/config.py b/authorization_django/config.py index 52bb836..8de25b2 100644 --- a/authorization_django/config.py +++ b/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) diff --git a/authorization_django/extensions/drf.py b/authorization_django/extensions/drf.py index abebc5e..9f86a0d 100644 --- a/authorization_django/extensions/drf.py +++ b/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 ) diff --git a/authorization_django/jwks.py b/authorization_django/jwks.py index de0ab18..f8a6451 100644 --- a/authorization_django/jwks.py +++ b/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.") diff --git a/authorization_django/middleware.py b/authorization_django/middleware.py index f58d4f1..da67b82 100644 --- a/authorization_django/middleware.py +++ b/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,7 +220,7 @@ 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 @@ -230,13 +228,12 @@ def handle_scope(self, authz_func, request: HttpRequest): 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,7 +243,7 @@ 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 @@ -254,7 +251,7 @@ def __call__(self, request: HttpRequest): # 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": diff --git a/authorization_django/utils.py b/authorization_django/utils.py new file mode 100644 index 0000000..00e305b --- /dev/null +++ b/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 diff --git a/pyproject.toml b/pyproject.toml index 7e532de..b984eb6 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -4,7 +4,7 @@ build-backend = "uv_build" [project] name = "datapunt-authorization-django" -version = "2.4.3" +version = "3.0.0" description = "Datapunt authorization check for Django" readme = "README.md" requires-python = ">=3.10" @@ -28,6 +28,7 @@ dependencies = [ "requests>=2.34.2", "jwcrypto>=1.5.8", "pydantic>=2.13.5", + "django-environ>=0.14.0", ] [project.urls] @@ -67,7 +68,7 @@ addopts = [ "--ds=tests.settings", "--cov=authorization_django", "--cov-report=term", - "--cov-fail-under=84", + "--cov-fail-under=96", ] # ==== Coverage ==== diff --git a/tests/test_authorization_django.py b/tests/test_authorization_django.py index eb40e83..96fe721 100644 --- a/tests/test_authorization_django.py +++ b/tests/test_authorization_django.py @@ -6,17 +6,19 @@ import json import time from base64 import urlsafe_b64encode +from copy import deepcopy from typing import Any import pytest from django import conf from django.http import HttpResponse, JsonResponse from django.test import RequestFactory +from jwcrypto.jwk import JWKSet from jwcrypto.jwt import JWT from authorization_django import authorization_middleware, config from authorization_django.exceptions import AuthorizationError -from authorization_django.jwks import JWKSWrapper +from authorization_django.jwks import JWKSWrapper, _load_jwks, _load_jwks_from_url JWKS1 = { "keys": [ @@ -87,8 +89,23 @@ "6": "ES512", } + +def trusted_jwks_item(*, jwks=None, jwks_url=None, iss="iss", aud=None): + item = {} + if jwks is not None: + item["jwks"] = jwks + if jwks_url is not None: + item["jwks_url"] = jwks_url + + claims = {"iss": iss} + if aud is not None: + claims["aud"] = aud + item["claims"] = claims + return item + + TESTSETTINGS: dict[str, Any] = { - "JWKS": json.dumps(JWKS1), + "TRUSTED_JWKS": [trusted_jwks_item(jwks=JWKS1)], "ALLOWED_SIGNING_ALGORITHMS": [ "HS256", "HS384", @@ -104,14 +121,32 @@ } +def settings_with(**overrides): + settings = deepcopy(TESTSETTINGS) + settings.update(overrides) + return settings + + +def use_settings(**overrides): + reload_settings(settings_with(**overrides)) + + def reload_settings(s): conf.settings.DATAPUNT_AUTHZ = s config.init_settings() -def create_token(tokendata, kid, alg): - jwks = JWKSWrapper() - key = jwks.get_key(kid) +def get_jwk(jwks, kid): + keyset = JWKSet() + keyset.import_keyset(json.dumps(jwks)) + for key in keyset["keys"]: + if key.get("kid") == kid: + return key + return None + + +def create_token(tokendata, kid, alg, signing_jwks=JWKS1): + key = get_jwk(signing_jwks, kid) token = JWT(header={"alg": alg, "kid": kid}, claims=tokendata) token.make_signed_token(key) return token @@ -123,7 +158,9 @@ def create_unsigned_token(tokendata): return f"{header}.{tokendata}" -def create_request(tokendata, kid=None, prefix="Bearer", path="/", method="GET"): +def create_request( + tokendata, kid=None, prefix="Bearer", path="/", method="GET", signing_jwks=JWKS1 +): """Django WSGI Request mock. A Django request object contains a META dict that contains the HTTP headers per the WSGI spec, PEP333 (meaning, uppercase, prefixed with HTTP_ and dashes transformed to underscores). @@ -131,7 +168,7 @@ def create_request(tokendata, kid=None, prefix="Bearer", path="/", method="GET") if not kid: token = create_unsigned_token(tokendata) else: - token = create_token(tokendata, kid, ALG_LOOKUP[kid]).serialize() + token = create_token(tokendata, kid, ALG_LOOKUP[kid], signing_jwks).serialize() return RequestFactory().generic( method, path=path, headers={"authorization": f"{prefix} {token}"} @@ -151,7 +188,7 @@ def custom_handler(request, exception): @pytest.fixture def tokendata_missing_scopes(): now = int(time.time()) - return {"exp": now + 30} + return {"exp": now + 30, "iss": "iss"} @pytest.fixture @@ -160,6 +197,7 @@ def tokendata_expired(): return { "iat": now, "exp": now - 100, # 60 second leeway allowed + "iss": "iss", "scopes": ["scope1"], } @@ -170,6 +208,7 @@ def tokendata_scope1(): return { "iat": now, "exp": now + 30, + "iss": "iss", "scopes": ["scope1"], "sub": "test@tester.nl", } @@ -181,6 +220,7 @@ def tokendata_scope2(): return { "iat": now, "exp": now + 30, + "iss": "iss", "scopes": ["scope2"], "sub": "test@tester.nl", } @@ -192,6 +232,7 @@ def tokendata_two_scopes(): return { "iat": now, "exp": now + 30, + "iss": "iss", "scopes": ["scope1", "scope2"], "sub": "test@tester.nl", } @@ -203,6 +244,7 @@ def tokendata_account_id(): return { "iat": now, "exp": now + 30, + "iss": "iss", "scopes": ["scope1", "scope2"], "sub": "ABcD12_Ghi3K", } @@ -227,22 +269,12 @@ def tokendata_zero_scopes(): return { "iat": now, "exp": now + 30, + "iss": "iss", "scopes": [], "sub": "test@tester.nl", } -@pytest.fixture -def tokendata_azure_ad_two_scopes(): - now = int(time.time()) - return { - "iat": now, - "exp": now + 30, - "groups": ["test\\scope_1", "test\\scope_2"], - "unique_name": "test@tester.nl", - } - - @pytest.fixture def tokendata_entra_id_two_scopes(): now = int(time.time()) @@ -317,13 +349,14 @@ def middleware(): def test_missing_conf(): + reload_settings({}) with pytest.raises(config.AuthzConfigurationError): authorization_middleware(None) def test_bad_jwks(): with pytest.raises(config.AuthzConfigurationError): - reload_settings({"JWKS": "iamnotajwks"}) + use_settings(TRUSTED_JWKS=[trusted_jwks_item(jwks="iamnotajwks")]) authorization_middleware(None) @@ -333,7 +366,7 @@ def test_jwks_from_url(requests_mock, tokendata_two_scopes): """ jwks_url = "https://get.your.jwks.here/protocol/openid-connect/certs" requests_mock.get(jwks_url, text=json.dumps(JWKS1)) - reload_settings({"JWKS": None, "JWKS_URL": jwks_url}) + use_settings(TRUSTED_JWKS=[trusted_jwks_item(jwks_url=jwks_url)]) middleware = authorization_middleware(_ok_view) request = create_request(tokendata_two_scopes, "4") middleware(request) @@ -348,17 +381,7 @@ def test_jwks_from_url_entra_success(requests_mock, tokendata_two_scopes_aud_iss """ jwks_url = "https://login.microsoftonline.com/get.your.jwks.here/discovery/keys" requests_mock.get(jwks_url, text=json.dumps(JWKS1)) - reload_settings( - { - "JWKS": None, - "TRUSTED_JWKS": [ - { - "jwks_url": jwks_url, - "claims": {"aud": "aud", "iss": "iss"}, - } - ], - } - ) + use_settings(TRUSTED_JWKS=[trusted_jwks_item(jwks_url=jwks_url, iss="iss", aud="aud")]) middleware = authorization_middleware(_ok_view) request = create_request(tokendata_two_scopes_aud_iss, "4") middleware(request) @@ -371,9 +394,7 @@ def test_jwks_from_url_entra_error(requests_mock): jwks_url = "https://login.microsoftonline.com/get.your.jwks.here/discovery/keys" requests_mock.get(jwks_url, text=json.dumps(JWKS1)) with pytest.raises(config.AuthzConfigurationError) as excinfo: - reload_settings( - {"JWKS": None, "TRUSTED_JWKS": [{"jwks_url": jwks_url, "claims": {"iss": "iss"}}]} - ) + use_settings(TRUSTED_JWKS=[trusted_jwks_item(jwks_url=jwks_url)]) assert "When using Microsoft Entra ID" in str(excinfo.value) @@ -385,20 +406,11 @@ def test_jwks_from_url_list(requests_mock, tokendata_two_scopes_aud_iss): entra_jwks_url = "https://login.microsoftonline.com/get.your.jwks.here/discovery/keys" requests_mock.get(kc_jwks_url, text=json.dumps(JWKS1)) requests_mock.get(entra_jwks_url, text=json.dumps(JWKS1)) - reload_settings( - { - "JWKS": None, - "TRUSTED_JWKS": [ - { - "jwks_url": kc_jwks_url, - "claims": {"iss": "iss"}, - }, - { - "jwks_url": entra_jwks_url, - "claims": {"aud": "aud", "iss": "iss"}, - }, - ], - } + use_settings( + TRUSTED_JWKS=[ + trusted_jwks_item(jwks_url=kc_jwks_url), + trusted_jwks_item(jwks_url=entra_jwks_url, iss="iss", aud="aud"), + ] ) middleware = authorization_middleware(_ok_view) request = create_request(tokendata_two_scopes_aud_iss, "4") @@ -414,7 +426,6 @@ def test_jwks_from_url_list_uses_key_from_second_keyset(requests_mock, tokendata tokendata_two_scopes["iss"] = "second-issuer" reload_settings( { - "JWKS": None, "TRUSTED_JWKS": [ { "jwks_url": first_jwks_url, @@ -428,96 +439,17 @@ def test_jwks_from_url_list_uses_key_from_second_keyset(requests_mock, tokendata } ) middleware = authorization_middleware(_ok_view) - request = create_request(tokendata_two_scopes, "6") + request = create_request(tokendata_two_scopes, "6", signing_jwks=JWKS2) middleware(request) assert request.is_authorized_for("scope1", "scope2") -def test_deprecated_settings_warn_and_trusted_jwks_takes_precedence(caplog): - trusted_jwks = [ - { - "jwks_url": "https://trusted.example/.well-known/jwks.json", - "claims": {"aud": "trusted-aud", "iss": "trusted-iss"}, - }, - { - "jwks_url": "https://trusted-2.example/.well-known/jwks.json", - "claims": {"iss": "another-trusted-issuer"}, - }, - ] - with caplog.at_level("WARNING", logger="authorization_django.config"): - reload_settings( - { - "JWKS": None, - "JWKS_URL": "https://legacy.example/jwks.json", - "JWKS_URLS": ["https://legacy.example/jwks.json"], - "CHECK_CLAIMS": {"aud": "legacy-aud", "iss": "legacy-iss"}, - "TRUSTED_JWKS": trusted_jwks, - } - ) - settings = config.get_settings() - - assert settings["JWKS_URL"] == trusted_jwks[0]["jwks_url"] - assert settings["JWKS_URLS"] == [item["jwks_url"] for item in trusted_jwks] - assert settings["CHECK_CLAIMS"] == trusted_jwks[0]["claims"] - - assert caplog.messages[0] == ( - "Deprecated settings present: CHECK_CLAIMS, JWKS_URL, JWKS_URLS. " - "Please migrate to TRUSTED_JWKS." - ) - assert ( - caplog.messages.count( - "Accessing deprecated setting JWKS_URL. Please migrate to TRUSTED_JWKS." - ) - >= 1 - ) - assert ( - caplog.messages.count( - "Accessing deprecated setting JWKS_URLS. Please migrate to TRUSTED_JWKS." - ) - >= 1 - ) - assert ( - caplog.messages.count( - "Accessing deprecated setting CHECK_CLAIMS. Please migrate to TRUSTED_JWKS." - ) - >= 1 - ) - - -def test_trusted_jwks_synthesized_from_deprecated_settings(): - jwks_url = "https://login.microsoftonline.com/get.your.jwks.here/discovery/keys" - reload_settings( - { - "JWKS": None, - "JWKS_URL": jwks_url, - "CHECK_CLAIMS": {"aud": "aud", "iss": "iss"}, - } - ) - - assert config.get_settings()["TRUSTED_JWKS"] == [ - { - "jwks_url": jwks_url, - "claims": {"aud": "aud", "iss": "iss"}, - } - ] - - def test_trusted_jwks_used_for_runtime_access(requests_mock, tokendata_two_scopes_aud_iss, caplog): jwks_url = "https://login.microsoftonline.com/get.your.jwks.here/discovery/keys" requests_mock.get(jwks_url, text=json.dumps(JWKS1)) - reload_settings( - { - "JWKS": None, - "TRUSTED_JWKS": [ - { - "jwks_url": jwks_url, - "claims": {"aud": "aud", "iss": "iss"}, - } - ], - } - ) + use_settings(TRUSTED_JWKS=[trusted_jwks_item(jwks_url=jwks_url, iss="iss", aud="aud")]) middleware = authorization_middleware(_ok_view) request = create_request(tokendata_two_scopes_aud_iss, "4") @@ -535,18 +467,17 @@ def test_reload_jwks_from_url(requests_mock, tokendata_two_scopes): # Create a request with a token signed with a key from JWKS2 requests_mock.get(jwks_url, text=json.dumps(JWKS2)) - reload_settings({"JWKS": None, "JWKS_URL": jwks_url}) - request = create_request(tokendata_two_scopes, "6") + use_settings(TRUSTED_JWKS=[trusted_jwks_item(jwks_url=jwks_url)]) + request = create_request(tokendata_two_scopes, "6", signing_jwks=JWKS2) # Instantiate the middleware with JWKS1 requests_mock.get(jwks_url, text=json.dumps(JWKS1)) reload_settings( - { - "JWKS": None, - "JWKS_URL": jwks_url, - "MIN_INTERVAL_KEYSET_UPDATE": 0, # Set update interval to 0 secs for the test - } + settings_with( + TRUSTED_JWKS=[trusted_jwks_item(jwks_url=jwks_url)], + MIN_INTERVAL_KEYSET_UPDATE=0, + ) ) - assert requests_mock.call_count == 1, requests_mock.request_history + assert requests_mock.call_count == 0, requests_mock.request_history middleware = authorization_middleware(lambda r: HttpResponse(status=200)) """ Process a request with the middleware. The middleware should now: @@ -556,7 +487,7 @@ def test_reload_jwks_from_url(requests_mock, tokendata_two_scopes): - respond with an invalid_token response """ response = middleware(request) - assert requests_mock.call_count == 3 + assert requests_mock.call_count == 2 assert response.status_code == 401 assert response.content == b"Unauthorized. Invalid token." """ @@ -568,7 +499,7 @@ def test_reload_jwks_from_url(requests_mock, tokendata_two_scopes): """ requests_mock.get(jwks_url, text=json.dumps(JWKS2)) middleware(request) - assert requests_mock.call_count == 4 + assert requests_mock.call_count == 3 assert request.is_authorized_for("scope1", "scope2") @@ -579,7 +510,9 @@ def test_hmac_keys_valid(middleware, tokendata_two_scopes): assert request.is_authorized_for("scope1", "scope2") -def test_keycloak_token(middleware, tokendata_keycloak_two_scopes): +def test_keycloak_token(tokendata_keycloak_two_scopes): + use_settings(TRUSTED_JWKS=[trusted_jwks_item(jwks=JWKS1, iss="https://iam.amsterdam.nl/")]) + middleware = authorization_middleware(_ok_view) request = create_request(tokendata_keycloak_two_scopes, "1") middleware(request) @@ -588,9 +521,7 @@ def test_keycloak_token(middleware, tokendata_keycloak_two_scopes): def test_keycloak_token_check_claims(tokendata_keycloak_two_scopes): - testsettings = TESTSETTINGS.copy() - testsettings["CHECK_CLAIMS"] = {"iss": "https://iam.amsterdam.nl/"} - reload_settings(testsettings) + use_settings(TRUSTED_JWKS=[trusted_jwks_item(jwks=JWKS1, iss="https://iam.amsterdam.nl/")]) request = create_request(tokendata_keycloak_two_scopes, "1") middleware = authorization_middleware(_ok_view) middleware(request) @@ -599,7 +530,9 @@ def test_keycloak_token_check_claims(tokendata_keycloak_two_scopes): assert request.get_token_scopes == {"SCOPE/1", "SCOPE/2"} -def test_keycloak_resource_access_token(middleware, tokendata_keycloak_resource_access_two_scopes): +def test_keycloak_resource_access_token(tokendata_keycloak_resource_access_two_scopes): + use_settings(TRUSTED_JWKS=[trusted_jwks_item(jwks=JWKS1, iss="https://iam.amsterdam.nl/")]) + middleware = authorization_middleware(_ok_view) request = create_request(tokendata_keycloak_resource_access_two_scopes, "1") middleware(request) @@ -608,9 +541,9 @@ def test_keycloak_resource_access_token(middleware, tokendata_keycloak_resource_ def test_entra_id_token_aud_iss(tokendata_entra_id_two_scopes): - testsettings = TESTSETTINGS.copy() - testsettings["CHECK_CLAIMS"] = {"aud": "aud", "iss": "https://sts.windows.net"} - reload_settings(testsettings) + use_settings( + TRUSTED_JWKS=[trusted_jwks_item(jwks=JWKS1, iss="https://sts.windows.net", aud="aud")] + ) request = create_request(tokendata_entra_id_two_scopes, "1") middleware = authorization_middleware(_ok_view) middleware(request) @@ -621,15 +554,16 @@ def test_entra_id_token_aud_iss(tokendata_entra_id_two_scopes): def test_entra_id_token_no_aud(tokendata_entra_id_two_scopes, requests_mock): - testsettings = TESTSETTINGS.copy() jwks_url = "https://login.microsoftonline.com/get.your.jwks.here/discovery/keys" - testsettings["TRUSTED_JWKS"] = [ - { - "jwks_url": jwks_url, - "claims": {"aud": "aud", "iss": "https://login.microsoftonline.com/"}, - } - ] - reload_settings(testsettings) + use_settings( + TRUSTED_JWKS=[ + trusted_jwks_item( + jwks_url=jwks_url, + iss="https://login.microsoftonline.com/", + aud="aud", + ) + ] + ) requests_mock.get(jwks_url, text=json.dumps(JWKS1)) # Remove aud claim tokendata_entra_id_two_scopes.pop("aud", None) @@ -639,15 +573,6 @@ def test_entra_id_token_no_aud(tokendata_entra_id_two_scopes, requests_mock): assert response.status_code == 401 -@pytest.mark.xfail(reason="AD Token not supported for now") -def test_azure_ad_token(middleware, tokendata_azure_ad_two_scopes): - request = create_request(tokendata_azure_ad_two_scopes, "1") - middleware(request) - - assert request.get_token_subject == "test@tester.nl" - assert request.get_token_scopes == {"SCOPE/1", "SCOPE/2"} - - def test_valid_one_scope_request(middleware, tokendata_two_scopes): request = create_request(tokendata_two_scopes, "4") middleware(request) @@ -721,19 +646,8 @@ def test_unknown_kid(tokendata_two_scopes): """ Verify that a token signed with an unknown key results in an "invalid_token" response """ - # Create a request with a token signed with a key from JWKS2 - reload_settings( - { - "JWKS": json.dumps(JWKS2), - } - ) - request = create_request(tokendata_two_scopes, "6") - # Instantiate the middleware with JWKS1 - reload_settings( - { - "JWKS": json.dumps(JWKS1), - } - ) + request = create_request(tokendata_two_scopes, "6", signing_jwks=JWKS2) + reload_settings(TESTSETTINGS) middleware = authorization_middleware(_ok_view) response = middleware(request) assert response.status_code == 401 @@ -764,9 +678,7 @@ def test_no_authorization_header(middleware): def test_check_missing_iss(tokendata_scope1): """Enforce claim checks""" - testsettings = TESTSETTINGS.copy() - testsettings["CHECK_CLAIMS"] = {"iss": "FOOBAR"} - reload_settings(testsettings) + use_settings(TRUSTED_JWKS=[trusted_jwks_item(jwks=JWKS1, iss="FOOBAR")]) middleware = authorization_middleware(_ok_view) request = create_request(tokendata_scope1, "4") response = middleware(request) @@ -776,9 +688,7 @@ def test_check_missing_iss(tokendata_scope1): @pytest.mark.parametrize(["issuer", "expect_code"], [("NOT_FOOBAR", 401), ("FOOBAR", 200)]) def test_check_issuer(tokendata_issuer, issuer, expect_code): """Enforce claim checks""" - testsettings = TESTSETTINGS.copy() - testsettings["CHECK_CLAIMS"] = {"iss": issuer} - reload_settings(testsettings) + use_settings(TRUSTED_JWKS=[trusted_jwks_item(jwks=JWKS1, iss=issuer)]) middleware = authorization_middleware(_ok_view) request = create_request(tokendata_issuer, "4") response = middleware(request) @@ -789,9 +699,7 @@ def test_check_correct_issuer_expired(tokendata_issuer_expired): """When check_claims is given, this also overrides 'exp' checking. Make sure that still works! """ - testsettings = TESTSETTINGS.copy() - testsettings["CHECK_CLAIMS"] = {"iss": "FOOBAR"} - reload_settings(testsettings) + use_settings(TRUSTED_JWKS=[trusted_jwks_item(jwks=JWKS1, iss="FOOBAR")]) middleware = authorization_middleware(_ok_view) request = create_request(tokendata_issuer_expired, "4") response = middleware(request) @@ -810,9 +718,7 @@ def test_check_iss_aud_present_for_entra(tokendata_issuer_expired): def test_min_scope_sufficient(tokendata_scope1): """scope1 is required, scope1 is in token""" - testsettings = TESTSETTINGS.copy() - testsettings["MIN_SCOPE"] = ("scope1",) - reload_settings(testsettings) + use_settings(MIN_SCOPE=("scope1",)) middleware = authorization_middleware(_ok_view) request = create_request(tokendata_scope1, "4") response = middleware(request) @@ -821,9 +727,7 @@ def test_min_scope_sufficient(tokendata_scope1): def test_min_scope_insufficient(): """scope1 is required, request with no token""" - testsettings = TESTSETTINGS.copy() - testsettings["MIN_SCOPE"] = ("scope1",) - reload_settings(testsettings) + use_settings(MIN_SCOPE=("scope1",)) middleware = authorization_middleware(_ok_view) request = create_request_no_auth_header() response = middleware(request) @@ -833,9 +737,7 @@ def test_min_scope_insufficient(): def test_min_scope_as_string_sufficient(tokendata_scope1): """MIN_SCOPE configured as string instead of tuple""" - testsettings = TESTSETTINGS.copy() - testsettings["MIN_SCOPE"] = "scope1" - reload_settings(testsettings) + use_settings(MIN_SCOPE="scope1") middleware = authorization_middleware(_ok_view) request = create_request(tokendata_scope1, "4") response = middleware(request) @@ -844,9 +746,7 @@ def test_min_scope_as_string_sufficient(tokendata_scope1): def test_min_scope_as_string_insufficient(tokendata_scope1): """MIN_SCOPE configured as string instead of tuple""" - testsettings = TESTSETTINGS.copy() - testsettings["MIN_SCOPE"] = "scope1" - reload_settings(testsettings) + use_settings(MIN_SCOPE="scope1") middleware = authorization_middleware(_ok_view) request = create_request_no_auth_header() response = middleware(request) @@ -855,9 +755,7 @@ def test_min_scope_as_string_insufficient(tokendata_scope1): def test_min_scope_multiple_sufficient(tokendata_two_scopes): """Two scopes required, both of them in token""" - testsettings = TESTSETTINGS.copy() - testsettings["MIN_SCOPE"] = ("scope1", "scope2") - reload_settings(testsettings) + use_settings(MIN_SCOPE=("scope1", "scope2")) middleware = authorization_middleware(_ok_view) request = create_request(tokendata_two_scopes, "4") response = middleware(request) @@ -866,9 +764,7 @@ def test_min_scope_multiple_sufficient(tokendata_two_scopes): def test_min_scope_multiple_insufficient(tokendata_scope1): """Two scopes required, only one of them in token""" - testsettings = TESTSETTINGS.copy() - testsettings["MIN_SCOPE"] = ("scope1", "scope2") - reload_settings(testsettings) + use_settings(MIN_SCOPE=("scope1", "scope2")) middleware = authorization_middleware(_ok_view) request = create_request(tokendata_scope1, "4") response = middleware(request) @@ -877,10 +773,7 @@ def test_min_scope_multiple_insufficient(tokendata_scope1): def test_forced_anonymous_routes(rf): - testsettings = TESTSETTINGS.copy() - testsettings["FORCED_ANONYMOUS_ROUTES"] = ("/status",) - testsettings["MIN_SCOPE"] = ("scope1",) - reload_settings(testsettings) + use_settings(FORCED_ANONYMOUS_ROUTES=("/status",), MIN_SCOPE=("scope1",)) empty_request = rf.get("/status/lala") middleware = authorization_middleware(_ok_view) response = middleware(empty_request) @@ -890,9 +783,7 @@ def test_forced_anonymous_routes(rf): def test_options_works_while_min_scope(rf): - testsettings = TESTSETTINGS.copy() - testsettings["MIN_SCOPE"] = ("scope",) - reload_settings(testsettings) + use_settings(MIN_SCOPE=("scope",)) middleware = authorization_middleware(_ok_view) empty_request = rf.options(path="/") response = middleware(empty_request) @@ -902,12 +793,12 @@ def test_options_works_while_min_scope(rf): def test_protected_resources_all_methods(tokendata_scope1, tokendata_two_scopes): - testsettings = TESTSETTINGS.copy() - testsettings["PROTECTED"] = [ - ("/one_scope_required", ["*"], ["scope1"]), - ("/two_scopes_required", ["*"], ["scope1", "scope2"]), - ] - reload_settings(testsettings) + use_settings( + PROTECTED=[ + ("/one_scope_required", ["*"], ["scope1"]), + ("/two_scopes_required", ["*"], ["scope1", "scope2"]), + ] + ) middleware = authorization_middleware(_ok_view) # a token with scope1 gives access via all methods @@ -936,12 +827,12 @@ def test_protected_resources_all_methods(tokendata_scope1, tokendata_two_scopes) def test_protected_resource_read_write_distinction(tokendata_scope1, tokendata_scope2): - testsettings = TESTSETTINGS.copy() - testsettings["PROTECTED"] = [ - ("/read_write_distinction", ["GET", "HEAD"], ["scope1"]), - ("/read_write_distinction", ["PATCH", "PUT", "POST", "DELETE"], ["scope2"]), - ] - reload_settings(testsettings) + use_settings( + PROTECTED=[ + ("/read_write_distinction", ["GET", "HEAD"], ["scope1"]), + ("/read_write_distinction", ["PATCH", "PUT", "POST", "DELETE"], ["scope2"]), + ] + ) middleware = authorization_middleware(_ok_view) request = create_request(tokendata_scope1, "4", "Bearer", "/read_write_distinction", "GET") @@ -959,10 +850,8 @@ def test_protected_resource_read_write_distinction(tokendata_scope1, tokendata_s def test_unknown_config_param(): - testsettings = TESTSETTINGS.copy() - testsettings["lalaland"] = "oscar" with pytest.raises(config.AuthzConfigurationError): - reload_settings(testsettings) + reload_settings(settings_with(lalaland="oscar")) authorization_middleware(None) @@ -973,20 +862,14 @@ def test_protected_resource_syntax_error(): ("/foo", ["*"]), ] for entry in invalid_entries: - testsettings = TESTSETTINGS.copy() - protected = [] - protected.append(entry) - testsettings["PROTECTED"] = protected with pytest.raises(config.ProtectedRecourceSyntaxError): - reload_settings(testsettings) + use_settings(PROTECTED=[entry]) authorization_middleware(None) def test_empty_scopes_error(): - testsettings = TESTSETTINGS.copy() - testsettings["PROTECTED"] = [("/foo/protected", ["*"], [])] with pytest.raises(config.NoRequiredScopesError): - reload_settings(testsettings) + use_settings(PROTECTED=[("/foo/protected", ["*"], [])]) authorization_middleware(None) @@ -994,11 +877,11 @@ def test_protected_route_overruled_error(): """Configuring a protected route that would be overruled by a route in FORCED_ANONYMOUS_ROUTES should lead to a ProtectedRouteConflict """ - testsettings = TESTSETTINGS.copy() - testsettings["PROTECTED"] = [("/foo/protected", ["*"], ["scope1"])] - testsettings["FORCED_ANONYMOUS_ROUTES"] = ("/foo",) with pytest.raises(config.ProtectedRouteConflictError): - reload_settings(testsettings) + use_settings( + PROTECTED=[("/foo/protected", ["*"], ["scope1"])], + FORCED_ANONYMOUS_ROUTES=("/foo",), + ) authorization_middleware(None) @@ -1018,13 +901,106 @@ def test_invalid_request_request_type(middleware, tokendata_expired): def test_custom_exception(middleware): """test custom handler""" - testsettings = TESTSETTINGS.copy() - testsettings["MIN_SCOPE"] = ("scope1",) - testsettings["EXCEPTION_HANDLER"] = custom_handler - reload_settings(testsettings) + use_settings(MIN_SCOPE=("scope1",), EXCEPTION_HANDLER=custom_handler) request = create_request_no_auth_header() middleware = authorization_middleware(_ok_view) response = middleware(request) assert isinstance(response, JsonResponse) assert response.status_code == 401 assert response.content == b'{"message": "Unauthorized"}' + + +def test_always_ok_sets_request_state(): + use_settings(ALWAYS_OK=True) + request = create_request_no_auth_header() + + response = authorization_middleware(_ok_view)(request) + + assert response.status_code == 200 + assert request.is_authorized_for("scope1") + assert request.get_token_subject == "ALWAYS_OK" + + +def test_authorized_request_logs_x_unique_id(tokendata_scope1, caplog): + use_settings(MIN_SCOPE=("scope1",)) + request = create_request(tokendata_scope1, "4") + request.META["HTTP_X_UNIQUE_ID"] = "request-1" + middleware = authorization_middleware(_ok_view) + + with caplog.at_level("INFO", logger="authorization_django.middleware"): + response = middleware(request) + + assert response.status_code == 200 + assert "X-Unique-ID: request-1" in caplog.text + + +def test_azure_group_claims_are_converted(): + now = int(time.time()) + use_settings() + tokendata = { + "iat": now, + "exp": now + 30, + "iss": "iss", + "groups": ["scope_1 admins", "scope_2 readers"], + "unique_name": "test@tester.nl", + } + request = create_request(tokendata, "1") + + authorization_middleware(_ok_view)(request) + + assert request.get_token_subject == "test@tester.nl" + assert request.get_token_scopes == {"SCOPE/1", "SCOPE/2"} + + +def test_trusted_jwks_requires_a_single_key_source(): + with pytest.raises(config.AuthzConfigurationError): + reload_settings(settings_with(TRUSTED_JWKS=[{"claims": {"iss": "iss"}}])) + + with pytest.raises(config.AuthzConfigurationError): + use_settings( + TRUSTED_JWKS=[trusted_jwks_item(jwks=JWKS1, jwks_url="https://example.com/jwks")] + ) + + +def test_protected_settings_validate_container_and_field_types(): + with pytest.raises(config.AuthzConfigurationError): + use_settings(PROTECTED="/not/a/list") + + with pytest.raises(config.AuthzConfigurationError): + use_settings(PROTECTED=[(1, ["GET"], ["scope1"])]) + + +def test_jwks_wrapper_get_key_returns_none_for_unknown_key(): + reload_settings(TESTSETTINGS) + + assert JWKSWrapper().get_key("missing") is None + + +def test_load_jwks_raises_for_invalid_settings_data(): + with pytest.raises(config.AuthzConfigurationError): + _load_jwks(JWKSet(), "not-a-jwks") + + +def test_load_jwks_from_url_raises_for_request_errors(requests_mock): + jwks_url = "https://example.com/failing-jwks" + requests_mock.get(jwks_url, status_code=500) + + with pytest.raises(config.AuthzConfigurationError): + _load_jwks_from_url(JWKSet(), jwks_url) + + +def test_load_jwks_from_url_raises_for_invalid_payload(requests_mock): + jwks_url = "https://example.com/invalid-jwks" + requests_mock.get(jwks_url, text="not json") + + with pytest.raises(config.AuthzConfigurationError): + _load_jwks_from_url(JWKSet(), jwks_url) + + +def test_get_settings_initializes_on_demand(): + conf.settings.DATAPUNT_AUTHZ = TESTSETTINGS + config._settings = {} + + settings = config.get_settings() + + assert settings.TRUSTED_JWKS[0].claims.iss == "iss" diff --git a/tests/test_extensions.py b/tests/test_extensions.py new file mode 100644 index 0000000..bb9bd76 --- /dev/null +++ b/tests/test_extensions.py @@ -0,0 +1,105 @@ +from types import SimpleNamespace + +import pytest + +from authorization_django import authorization_middleware +from tests.test_authorization_django import ( + TESTSETTINGS, + _ok_view, + create_request, + reload_settings, + use_settings, +) + + +def import_drf_extensions(): + pytest.importorskip("rest_framework") + pytest.importorskip("drf_spectacular") + + from authorization_django.extensions.drf import HasTokenScopes, JWTAuthentication + from authorization_django.extensions.scheme import JWTTokenScheme + + return HasTokenScopes, JWTAuthentication, JWTTokenScheme + + +def test_drf_jwt_authentication_returns_claims(): + _, jwt_authentication_class, _ = import_drf_extensions() + now_claims = { + "iat": 1, + "exp": 2, + "scopes": ["scope1", "scope2"], + "sub": "test@tester.nl", + } + request = SimpleNamespace(get_token_claims=now_claims) + + authenticator = jwt_authentication_class() + + assert authenticator.authenticate(request) == (None, now_claims) + + +def test_drf_jwt_authentication_requires_claims(): + _, jwt_authentication_class, _ = import_drf_extensions() + request = SimpleNamespace(get_token_claims=None) + + authenticator = jwt_authentication_class() + + with pytest.raises(Exception) as excinfo: + authenticator.authenticate(request) + + assert excinfo.type.__name__ == "NotAuthenticated" + assert "WWW-Authenticate header field" in str(excinfo.value) + + +def test_drf_jwt_authentication_header_uses_realm(): + _, jwt_authentication_class, _ = import_drf_extensions() + authenticator = jwt_authentication_class() + + assert authenticator.authenticate_header(SimpleNamespace()) == 'Bearer realm="api"' + + +def test_has_token_scopes_uses_explicit_needed_scopes(): + permission_class, _, _ = import_drf_extensions() + reload_settings(TESTSETTINGS) + request = SimpleNamespace(get_token_scopes=["scope1", "scope2"]) + + permission = permission_class("scope1") + + assert permission.has_permission(request, None) + assert permission.has_object_permission(request, None, object()) + + +def test_has_token_scopes_uses_min_scope_from_settings(): + permission_class, _, _ = import_drf_extensions() + use_settings(MIN_SCOPE=("scope1",)) + tokendata_scope1 = { + "iat": 1, + "exp": 4102444800, + "iss": "iss", + "scopes": ["scope1"], + "sub": "test@tester.nl", + } + request = create_request(tokendata_scope1, "4") + authorization_middleware(_ok_view)(request) + + permission = permission_class() + + assert permission.has_permission(request, None) + + +def test_has_token_scopes_allows_when_always_ok(): + permission_class, _, _ = import_drf_extensions() + use_settings(ALWAYS_OK=True, MIN_SCOPE=("scope1",)) + request = SimpleNamespace(get_token_scopes=[]) + + permission = permission_class() + + assert permission.has_permission(request, None) + + +def test_drf_spectacular_jwt_token_scheme_definition(): + _, _, jwt_token_scheme_class = import_drf_extensions() + scheme = jwt_token_scheme_class("authorization_django.extensions.drf.JWTAuthentication") + + assert scheme.name == "JWTAuthentication" + assert scheme.target_class == "authorization_django.extensions.drf.JWTAuthentication" + assert scheme.get_security_definition(None) == {"type": "http", "scheme": "bearer"} diff --git a/tests/test_utils.py b/tests/test_utils.py new file mode 100644 index 0000000..ede3b30 --- /dev/null +++ b/tests/test_utils.py @@ -0,0 +1,48 @@ +import json + +from authorization_django.utils import get_trusted_jwks + + +def test_get_trusted_jwks_loads_custom_provider_and_inline_jwks(monkeypatch): + jwks = {"keys": [{"kid": "test-key"}]} + monkeypatch.setenv("OAUTH_TEST_URL", "https://issuer.example.test/jwks") + monkeypatch.setenv("OAUTH_TEST_CLAIMS", "iss=https://issuer.example.test,aud=test-api") + monkeypatch.setenv("TEST_PUBLIC_JWKS", json.dumps(jwks)) + + assert get_trusted_jwks(providers=["TEST", "NON_EXISTENT"], pub_jwks="TEST_PUBLIC_JWKS") == [ + { + "jwks_url": "https://issuer.example.test/jwks", + "claims": {"iss": "https://issuer.example.test", "aud": "test-api"}, + }, + {"jwks": jwks, "claims": {"iss": "iss"}}, + ] + + +def test_get_trusted_jwks_loads_only_provider_when_pub_jwks_missing(monkeypatch): + monkeypatch.setenv("OAUTH_TEST_URL", "https://issuer.example.test/jwks") + monkeypatch.setenv("OAUTH_TEST_CLAIMS", "iss=https://issuer.example.test,aud=test-api") + + assert get_trusted_jwks(providers=["TEST"]) == [ + { + "jwks_url": "https://issuer.example.test/jwks", + "claims": {"iss": "https://issuer.example.test", "aud": "test-api"}, + }, + ] + + +def test_get_trusted_jwks_loads_default_providers(monkeypatch): + monkeypatch.setenv("OAUTH_ENTRA_URL", "https://issuer.entra.test/jwks") + monkeypatch.setenv("OAUTH_ENTRA_CLAIMS", "iss=https://issuer.entra.test,aud=test-api") + monkeypatch.setenv("OAUTH_KEYCLOAK_URL", "https://issuer.keycloak.test/jwks") + monkeypatch.setenv("OAUTH_KEYCLOAK_CLAIMS", "iss=https://issuer.keycloak.test,aud=test-api") + + assert get_trusted_jwks(pub_jwks="TEST_PUBLIC_JWKS") == [ + { + "jwks_url": "https://issuer.entra.test/jwks", + "claims": {"iss": "https://issuer.entra.test", "aud": "test-api"}, + }, + { + "jwks_url": "https://issuer.keycloak.test/jwks", + "claims": {"iss": "https://issuer.keycloak.test", "aud": "test-api"}, + }, + ] diff --git a/uv.lock b/uv.lock index dd5f22a..3a8f8ec 100644 --- a/uv.lock +++ b/uv.lock @@ -519,10 +519,11 @@ wheels = [ [[package]] name = "datapunt-authorization-django" -version = "2.4.3" +version = "3.0.0" source = { editable = "." } dependencies = [ { name = "django" }, + { name = "django-environ" }, { name = "jwcrypto" }, { name = "pydantic" }, { name = "requests" }, @@ -548,6 +549,7 @@ tests = [ [package.metadata] requires-dist = [ { name = "django", specifier = ">=5.2" }, + { name = "django-environ", specifier = ">=0.14.0" }, { name = "djangorestframework", marker = "extra == 'extended'", specifier = "==3.17.2" }, { name = "drf-spectacular", marker = "extra == 'extended'", specifier = "==0.30.0" }, { name = "jwcrypto", specifier = ">=1.5.8" }, @@ -579,6 +581,15 @@ wheels = [ { url = "https://files.pythonhosted.org/packages/df/f8/ce120525ca78f12b07daf65786679c5d0b54a75285a8958d3ae55e39da35/django-5.2.17-py3-none-any.whl", hash = "sha256:f04fb3b36ee119e1af4fa1d397d5fd6cf12700f49321e84d4f4c642c5b1973db", size = 8315563, upload-time = "2026-08-04T15:03:59.1Z" }, ] +[[package]] +name = "django-environ" +version = "0.14.0" +source = { registry = "https://pypi.org/simple" } +sdist = { url = "https://files.pythonhosted.org/packages/57/e1/4c4ddcf6e90f023e89edb1d92f9a7fffce61ad4b691df497f2f5c0be6e26/django_environ-0.14.0.tar.gz", hash = "sha256:b6c48d93b9d2ff8a3ea14099e90c35aa4f101c1b4d5f262dfee0d27b06742ed7", size = 60417, upload-time = "2026-06-18T22:49:55.847Z" } +wheels = [ + { url = "https://files.pythonhosted.org/packages/1a/fe/67b423fc30f16d10259e901320fbc121746bf168aaaad9043e6ebb0110c1/django_environ-0.14.0-py3-none-any.whl", hash = "sha256:8dbe8a57f0a540ab8abd6f54f230de5e99e3a2c9d797cb9caecb037bca3d47d8", size = 20934, upload-time = "2026-06-18T22:49:54.355Z" }, +] + [[package]] name = "djangorestframework" version = "3.17.2"