diff --git a/authentik/providers/oauth2/tests/test_provider_info.py b/authentik/providers/oauth2/tests/test_provider_info.py new file mode 100644 index 000000000000..0596922f4b5e --- /dev/null +++ b/authentik/providers/oauth2/tests/test_provider_info.py @@ -0,0 +1,100 @@ +"""Provider info (OpenID discovery) tests""" + +from unittest.mock import patch + +from django.urls import reverse + +from authentik.blueprints.tests import apply_blueprint +from authentik.core.models import Application +from authentik.core.tests.utils import create_test_flow +from authentik.lib.generators import generate_id +from authentik.providers.oauth2.models import ( + OAuth2Provider, + RedirectURI, + RedirectURIMatchingMode, + ScopeMapping, +) +from authentik.providers.oauth2.tests.utils import OAuthTestCase +from authentik.providers.oauth2.views.provider import ProviderInfoView + + +class TestProviderInfo(OAuthTestCase): + """Test provider info view""" + + @apply_blueprint("system/providers-oauth2.yaml") + def setUp(self) -> None: + super().setUp() + self.provider = OAuth2Provider.objects.create( + name=generate_id(), + client_id=generate_id(), + authorization_flow=create_test_flow(), + redirect_uris=[RedirectURI(RedirectURIMatchingMode.STRICT, "http://local.invalid")], + signing_key=self.keypair, + ) + self.provider.property_mappings.set(ScopeMapping.objects.all()) + self.app = Application.objects.create( + name=generate_id(), slug=generate_id(), provider=self.provider + ) + + def get_info(self, app: Application | None = None) -> dict: + """Fetch the discovery document""" + response = self.client.get( + reverse( + "authentik_providers_oauth2:provider-info", + kwargs={"application_slug": (app or self.app).slug}, + ) + ) + self.assertEqual(response.status_code, 200) + return response.json() + + def test_info(self): + """Test discovery document""" + body = self.get_info() + self.assertTrue(body["issuer"].endswith(f"/application/o/{self.app.slug}/")) + self.assertTrue(body["token_endpoint"].endswith("/token/")) + self.assertIn("openid", body["scopes_supported"]) + self.assertIn("email", body["scopes_supported"]) + for claim in ["sub", "iss", "aud", "exp", "iat", "acr", "amr", "nonce"]: + self.assertIn(claim, body["claims_supported"]) + # Claims from the default scope mappings + self.assertIn("email", body["claims_supported"]) + self.assertIn("preferred_username", body["claims_supported"]) + + def test_claims_cached(self): + """Test claims are only evaluated once""" + claims = self.get_info()["claims_supported"] + # Anything hitting get_claims again means the cache didn't take + with patch.object(ProviderInfoView, "get_claims", return_value=claims) as get_claims: + self.assertEqual(self.get_info()["claims_supported"], claims) + get_claims.assert_not_called() + + def test_claims_cached_per_provider(self): + """Test cached claims aren't shared between providers""" + self.get_info() + other = OAuth2Provider.objects.create( + name=generate_id(), + client_id=generate_id(), + authorization_flow=create_test_flow(), + redirect_uris=[RedirectURI(RedirectURIMatchingMode.STRICT, "http://local.invalid")], + signing_key=self.keypair, + ) + other.property_mappings.set( + ScopeMapping.objects.filter(managed="goauthentik.io/providers/oauth2/scope-openid") + ) + other_app = Application.objects.create( + name=generate_id(), slug=generate_id(), provider=other + ) + self.assertNotIn("email", self.get_info(other_app)["claims_supported"]) + self.assertIn("email", self.get_info()["claims_supported"]) + + def test_claims_mapping_error(self): + """Test failing scope mapping is skipped""" + self.provider.property_mappings.add( + ScopeMapping.objects.create( + name=generate_id(), + scope_name=generate_id(), + expression="raise Exception('test')", + ) + ) + body = self.get_info() + self.assertIn("email", body["claims_supported"]) diff --git a/authentik/providers/oauth2/views/provider.py b/authentik/providers/oauth2/views/provider.py index eb105de61951..baf05e09fea7 100644 --- a/authentik/providers/oauth2/views/provider.py +++ b/authentik/providers/oauth2/views/provider.py @@ -3,6 +3,7 @@ from typing import Any from django.apps import apps +from django.core.cache import cache from django.http import HttpRequest, HttpResponse, JsonResponse from django.shortcuts import get_object_or_404, reverse from django.views import View @@ -117,7 +118,7 @@ def get_info(self, provider: OAuth2Provider) -> dict[str, Any]: "scopes_supported": scopes, # https://openid.net/specs/openid-connect-core-1_0.html#RequestObject "request_parameter_supported": False, - "claims_supported": self.get_claims(provider), + "claims_supported": self.get_claims_cached(provider), "claims_parameter_supported": False, "code_challenge_methods_supported": [PKCE_METHOD_PLAIN, PKCE_METHOD_S256], "dpop_signing_alg_values_supported": sorted(DPOP_SUPPORTED_ALGS), @@ -138,6 +139,16 @@ def get_info(self, provider: OAuth2Provider) -> dict[str, Any]: pass return config + def get_claims_cached(self, provider: OAuth2Provider) -> list[str]: + """Same as self.get_claims but cached to avoid re-evaluating property-mappings""" + key = f"authentik/providers/oauth2/provider_info/claims/{provider.pk}" + ttl_seconds = 60 * 60 + claims = cache.get(key) + if not claims: + claims = self.get_claims(provider) + cache.set(key, claims, ttl_seconds) + return claims + def get_claims(self, provider: OAuth2Provider) -> list[str]: """Get a list of supported claims based on configured scope mappings""" default_claims = [