providers/oauth2: allow cross provider token introspection for federated providers (cherry-pick #21513 to version-2025.12) (#21747)

Cherry-pick #21513 to version-2025.12 (with conflicts)

This cherry-pick has conflicts that need manual resolution.

Original PR: #21513
Original commit: c84c8d8

Signed-off-by: Jens Langhammer <jens@goauthentik.io>
Co-authored-by: Jens L. <jens@goauthentik.io>

Author: authentik-automation[bot]

authentik/providers/oauth2/tests/test_authorize.py

@@ -372,7 +372,7 @@ def test_full_implicit(self):
                     "nonce": generate_id(),
                 },
             )
-            token: AccessToken = AccessToken.objects.filter(user=user).first()
+            token = AccessToken.objects.filter(user=user).first()
             expires = timedelta_from_string(provider.access_token_validity).total_seconds()
             self.assertEqual(
                 response.url,
@@ -444,7 +444,7 @@ def test_full_implicit_enc(self):
                 },
             )
             self.assertEqual(response.status_code, 302)
-            token: AccessToken = AccessToken.objects.filter(user=user).first()
+            token = AccessToken.objects.filter(user=user).first()
             expires = timedelta_from_string(provider.access_token_validity).total_seconds()
             jwt = self.validate_jwe(token, provider)
             self.assertEqual(jwt["amr"], ["pwd"])
@@ -543,7 +543,7 @@ def test_full_form_post_id_token(self):
         response = self.client.get(
             reverse("authentik_api:flow-executor", kwargs={"flow_slug": flow.slug}),
         )
-        token: AccessToken = AccessToken.objects.filter(user=user).first()
+        token = AccessToken.objects.filter(user=user).first()
         self.assertIsNotNone(token)
         self.assertJSONEqual(
             response.content.decode(),

authentik/providers/oauth2/tests/test_backchannel_logout.py

@@ -4,22 +4,19 @@
 
 import jwt
 from django.test import RequestFactory
-from django.utils import timezone
 from dramatiq.results.errors import ResultFailure
 from requests import Response
 from requests.exceptions import HTTPError, Timeout
 
-from authentik.core.models import Application, AuthenticatedSession, Session
+from authentik.core.models import Application
 from authentik.core.tests.utils import create_test_admin_user, create_test_flow
 from authentik.lib.generators import generate_id
 from authentik.providers.oauth2.id_token import hash_session_key
 from authentik.providers.oauth2.models import (
-    AccessToken,
     OAuth2LogoutMethod,
     OAuth2Provider,
     RedirectURI,
     RedirectURIMatchingMode,
-    RefreshToken,
 )
 from authentik.providers.oauth2.tasks import send_backchannel_logout_request
 from authentik.providers.oauth2.tests.utils import OAuthTestCase
@@ -45,52 +42,6 @@ def setUp(self) -> None:
         self.app.provider = self.provider
         self.app.save()
 
-    def _create_session(self, session_key=None):
-        """Create a session with the given key or a generated one"""
-        session_key = session_key or f"session-{generate_id()}"
-        session = Session.objects.create(
-            session_key=session_key,
-            expires=timezone.now() + timezone.timedelta(hours=1),
-            last_ip="255.255.255.255",
-        )
-        auth_session = AuthenticatedSession.objects.create(
-            session=session,
-            user=self.user,
-        )
-        return auth_session
-
-    def _create_token(
-        self, provider, user, session=None, token_type="access", token_id=None
-    ):  # nosec
-        """Create a token of the specified type"""
-        token_id = token_id or f"{token_type}-token-{generate_id()}"
-        kwargs = {
-            "provider": provider,
-            "user": user,
-            "session": session,
-            "token": token_id,
-            "_id_token": "{}",
-            "auth_time": timezone.now(),
-        }
-
-        if token_type == "access":  # nosec
-            return AccessToken.objects.create(**kwargs)
-        else:  # refresh
-            return RefreshToken.objects.create(**kwargs)
-
-    def _create_provider(self, name=None):
-        """Create an OAuth2 provider"""
-        name = name or f"provider-{generate_id()}"
-        provider = OAuth2Provider.objects.create(
-            name=name,
-            authorization_flow=create_test_flow(),
-            redirect_uris=[
-                RedirectURI(RedirectURIMatchingMode.STRICT, f"http://{name}/callback"),
-            ],
-            signing_key=self.keypair,
-        )
-        return provider
-
     def _create_logout_token(
         self,
         provider: OAuth2Provider | None = None,

authentik/providers/oauth2/tests/test_introspect.py

@@ -14,6 +14,7 @@
 from authentik.providers.oauth2.id_token import IDToken
 from authentik.providers.oauth2.models import (
     AccessToken,
+    ClientTypes,
     OAuth2Provider,
     RedirectURI,
     RedirectURIMatchingMode,
@@ -43,7 +44,7 @@ def setUp(self) -> None:
 
     def test_introspect_refresh(self):
         """Test introspect"""
-        token: RefreshToken = RefreshToken.objects.create(
+        token = RefreshToken.objects.create(
             provider=self.provider,
             user=self.user,
             token=generate_id(),
@@ -75,7 +76,7 @@ def test_introspect_refresh(self):
 
     def test_introspect_access(self):
         """Test introspect"""
-        token: AccessToken = AccessToken.objects.create(
+        token = AccessToken.objects.create(
             provider=self.provider,
             user=self.user,
             token=generate_id(),
@@ -130,7 +131,7 @@ def test_introspect_invalid_provider(self):
         )
         auth = b64encode(f"{provider.client_id}:{provider.client_secret}".encode()).decode()
 
-        token: AccessToken = AccessToken.objects.create(
+        token = AccessToken.objects.create(
             provider=self.provider,
             user=self.user,
             token=generate_id(),
@@ -169,3 +170,76 @@ def test_introspect_invalid_auth(self):
                 "active": False,
             },
         )
+
+    def test_introspect_provider_public(self):
+        """Test introspect"""
+        self.provider.client_type = ClientTypes.PUBLIC
+        self.provider.save()
+        token = AccessToken.objects.create(
+            provider=self.provider,
+            user=self.user,
+            token=generate_id(),
+            auth_time=timezone.now(),
+            _scope="openid user profile",
+            _id_token=json.dumps(
+                asdict(
+                    IDToken("foo", "bar"),
+                )
+            ),
+        )
+        res = self.client.post(
+            reverse("authentik_providers_oauth2:token-introspection"),
+            HTTP_AUTHORIZATION=f"Basic {self.auth}",
+            data={"token": token.token},
+        )
+        self.assertEqual(res.status_code, 200)
+        self.assertJSONEqual(
+            res.content.decode(),
+            {
+                "active": False,
+            },
+        )
+
+    def test_introspect_provider_fed(self):
+        """Test introspect with federation. self.provider is a confidential
+        client and other_provider is a public client."""
+        other_provider = OAuth2Provider.objects.create(
+            name=generate_id(),
+            authorization_flow=create_test_flow(),
+            redirect_uris=[RedirectURI(RedirectURIMatchingMode.STRICT, "")],
+            signing_key=create_test_cert(),
+            client_type=ClientTypes.PUBLIC,
+        )
+        Application.objects.create(name=generate_id(), slug=generate_id(), provider=other_provider)
+
+        other_provider.jwt_federation_providers.add(self.provider)
+
+        token = AccessToken.objects.create(
+            provider=other_provider,
+            user=self.user,
+            token=generate_id(),
+            auth_time=timezone.now(),
+            _scope="openid user profile",
+            _id_token=json.dumps(
+                asdict(
+                    IDToken("foo", "bar"),
+                )
+            ),
+        )
+        res = self.client.post(
+            reverse("authentik_providers_oauth2:token-introspection"),
+            HTTP_AUTHORIZATION=f"Basic {self.auth}",
+            data={"token": token.token},
+        )
+        self.assertEqual(res.status_code, 200)
+        self.assertJSONEqual(
+            res.content.decode(),
+            {
+                "acr": ACR_AUTHENTIK_DEFAULT,
+                "sub": "bar",
+                "iss": "foo",
+                "active": True,
+                "client_id": other_provider.client_id,
+                "scope": " ".join(token.scope),
+            },
+        )

authentik/providers/oauth2/tests/test_revoke.py

@@ -46,7 +46,7 @@ def setUp(self) -> None:
 
     def test_revoke_refresh(self):
         """Test revoke"""
-        token: RefreshToken = RefreshToken.objects.create(
+        token = RefreshToken.objects.create(
             provider=self.provider,
             user=self.user,
             token=generate_id(),
@@ -69,7 +69,7 @@ def test_revoke_refresh(self):
 
     def test_revoke_access(self):
         """Test revoke"""
-        token: AccessToken = AccessToken.objects.create(
+        token = AccessToken.objects.create(
             provider=self.provider,
             user=self.user,
             token=generate_id(),
@@ -105,7 +105,19 @@ def test_revoke_invalid_auth(self):
         """Test revoke (invalid auth)"""
         res = self.client.post(
             reverse("authentik_providers_oauth2:token-revoke"),
-            HTTP_AUTHORIZATION="Basic fqewr",
+            HTTP_AUTHORIZATION="Basic aaa",
+            data={
+                "token": generate_id(),
+            },
+        )
+        self.assertEqual(res.status_code, 401)
+
+    def test_revoke_invalid_auth_secret(self):
+        """Test revoke (invalid secret)"""
+        invalid_auth = b64encode(f"{self.provider.client_id}:aaa".encode()).decode()
+        res = self.client.post(
+            reverse("authentik_providers_oauth2:token-revoke"),
+            HTTP_AUTHORIZATION=f"Basic {invalid_auth}",
             data={
                 "token": generate_id(),
             },
@@ -116,7 +128,7 @@ def test_revoke_public(self):
         """Test revoke public client"""
         self.provider.client_type = ClientTypes.PUBLIC
         self.provider.save()
-        token: AccessToken = AccessToken.objects.create(
+        token = AccessToken.objects.create(
             provider=self.provider,
             user=self.user,
             token=generate_id(),
@@ -220,3 +232,74 @@ def test_revoke_user_deactivated(self):
         self.assertEqual(AccessToken.objects.all().count(), 0)
         self.assertEqual(RefreshToken.objects.all().count(), 0)
         self.assertEqual(DeviceToken.objects.all().count(), 0)
+
+    def test_revoke_provider_fed(self):
+        """Test revoke with federation. self.provider is a confidential
+        client and other_provider is a public client."""
+        other_provider = OAuth2Provider.objects.create(
+            name=generate_id(),
+            authorization_flow=create_test_flow(),
+            redirect_uris=[RedirectURI(RedirectURIMatchingMode.STRICT, "")],
+            signing_key=create_test_cert(),
+            client_type=ClientTypes.PUBLIC,
+        )
+        Application.objects.create(name=generate_id(), slug=generate_id(), provider=other_provider)
+
+        other_provider.jwt_federation_providers.add(self.provider)
+
+        token = AccessToken.objects.create(
+            provider=other_provider,
+            user=self.user,
+            token=generate_id(),
+            auth_time=timezone.now(),
+            _scope="openid user profile",
+            _id_token=json.dumps(
+                asdict(
+                    IDToken("foo", "bar"),
+                )
+            ),
+        )
+        res = self.client.post(
+            reverse("authentik_providers_oauth2:token-revoke"),
+            HTTP_AUTHORIZATION=f"Basic {self.auth}",
+            data={"token": token.token},
+        )
+        self.assertEqual(res.status_code, 200)
+        self.assertJSONEqual(res.content.decode(), {})
+
+    def test_revoke_provider_fed_public(self):
+        """Test revoke with federation. self.provider is a public
+        client and other_provider is a public client."""
+        self.provider.client_type = ClientTypes.PUBLIC
+        self.provider.save()
+        other_provider = OAuth2Provider.objects.create(
+            name=generate_id(),
+            authorization_flow=create_test_flow(),
+            redirect_uris=[RedirectURI(RedirectURIMatchingMode.STRICT, "")],
+            signing_key=create_test_cert(),
+            client_type=ClientTypes.PUBLIC,
+        )
+        Application.objects.create(name=generate_id(), slug=generate_id(), provider=other_provider)
+
+        other_provider.jwt_federation_providers.add(self.provider)
+
+        token = AccessToken.objects.create(
+            provider=other_provider,
+            user=self.user,
+            token=generate_id(),
+            auth_time=timezone.now(),
+            _scope="openid user profile",
+            _id_token=json.dumps(
+                asdict(
+                    IDToken("foo", "bar"),
+                )
+            ),
+        )
+        auth_public = b64encode(f"{self.provider.client_id}:{generate_id()}".encode()).decode()
+        res = self.client.post(
+            reverse("authentik_providers_oauth2:token-revoke"),
+            HTTP_AUTHORIZATION=f"Basic {auth_public}",
+            data={"token": token.token},
+        )
+        self.assertEqual(res.status_code, 200)
+        self.assertTrue(AccessToken.objects.filter(token=token.token).exists())