add oidc token test

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

authentik/enterprise/providers/ssf/tests/test_stream.py

@@ -1,15 +1,21 @@
+import json
+from dataclasses import asdict
+
 from django.urls import reverse
+from django.utils import timezone
 from rest_framework.test import APITestCase
 
 from authentik.core.models import Application
-from authentik.core.tests.utils import create_test_cert
+from authentik.core.tests.utils import create_test_admin_user, create_test_cert, create_test_flow
 from authentik.enterprise.providers.ssf.models import (
     SSFEventStatus,
     SSFProvider,
     Stream,
     StreamEvent,
 )
 from authentik.lib.generators import generate_id
+from authentik.providers.oauth2.id_token import IDToken
+from authentik.providers.oauth2.models import AccessToken, OAuth2Provider
 
 
 class TestStream(APITestCase):
@@ -21,15 +27,15 @@ def setUp(self):
             backchannel_application=self.application,
         )
 
-    def test_stream_add(self):
-        """test stream add"""
+    def test_stream_add_token(self):
+        """test stream add (token auth)"""
         res = self.client.post(
             reverse(
                 "authentik_providers_ssf:stream",
                 kwargs={"application_slug": self.application.slug},
             ),
             data={
-                "iss": "https://screw-fotos-bracelets-longitude.trycloudflare.com/.well-known/ssf-configuration/abm-ssf/5",
+                "iss": "https://authentik.company/.well-known/ssf-configuration/foo/5",
                 "aud": ["https://app.authentik.company"],
                 "delivery": {
                     "method": "https://schemas.openid.net/secevent/risc/delivery-method/push",
@@ -54,6 +60,59 @@ def test_stream_add(self):
             {"https://schemas.openid.net/secevent/ssf/event-type/verification": {"state": None}},
         )
 
+    def test_stream_add_oidc(self):
+        """test stream add (oidc auth)"""
+        provider = OAuth2Provider.objects.create(
+            name=generate_id(),
+            authorization_flow=create_test_flow(),
+        )
+        self.application.provider = provider
+        self.application.save()
+        user = create_test_admin_user()
+        token = AccessToken.objects.create(
+            provider=provider,
+            user=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_ssf:stream",
+                kwargs={"application_slug": self.application.slug},
+            ),
+            data={
+                "iss": "https://authentik.company/.well-known/ssf-configuration/foo/5",
+                "aud": ["https://app.authentik.company"],
+                "delivery": {
+                    "method": "https://schemas.openid.net/secevent/risc/delivery-method/push",
+                    "endpoint_url": "https://app.authentik.company",
+                },
+                "events_requested": [
+                    "https://schemas.openid.net/secevent/caep/event-type/credential-change",
+                    "https://schemas.openid.net/secevent/caep/event-type/session-revoked",
+                ],
+                "format": "iss_sub",
+            },
+            HTTP_AUTHORIZATION=f"Bearer {token.token}",
+        )
+        self.assertEqual(res.status_code, 201)
+        stream = Stream.objects.filter(provider=self.provider).first()
+        self.assertIsNotNone(stream)
+        event = StreamEvent.objects.filter(stream=stream).first()
+        self.assertIsNotNone(event)
+        self.assertEqual(event.status, SSFEventStatus.PENDING_FAILED)
+        self.assertEqual(
+            event.payload["events"],
+            {"https://schemas.openid.net/secevent/ssf/event-type/verification": {"state": None}},
+        )
+
     def test_stream_delete(self):
         """delete stream"""
         stream = Stream.objects.create(provider=self.provider)

authentik/enterprise/providers/ssf/views/configuration.py

@@ -9,6 +9,8 @@
 
 
 class ConfigurationView(SSFView):
+    """SSF configuration endpoint"""
+
     permission_classes = [AllowAny]
 
     def get_authenticators(self):

authentik/enterprise/providers/ssf/views/jwks.py

@@ -9,6 +9,7 @@
 
 
 class JWKSview(View):
+    """SSF JWKS endpoint, similar to the OAuth2 provider's endpoint"""
 
     def get(self, request: HttpRequest, application_slug: str) -> HttpResponse:
         """Show JWK Key data for Provider"""