authentik.enterprise.providers.ssf.tests.test_signals

  1from uuid import uuid4
  2
  3from django.urls import reverse
  4from rest_framework.test import APITestCase
  5
  6from authentik.core.models import Application, Group
  7from authentik.core.tests.utils import (
  8    create_test_cert,
  9    create_test_user,
 10)
 11from authentik.enterprise.providers.ssf.models import (
 12    EventTypes,
 13    SSFEventStatus,
 14    SSFProvider,
 15    Stream,
 16    StreamEvent,
 17)
 18from authentik.lib.generators import generate_id
 19from authentik.policies.models import PolicyBinding
 20from authentik.stages.authenticator_webauthn.models import WebAuthnDevice
 21
 22
 23class TestSignals(APITestCase):
 24    """Test individual SSF Signals"""
 25
 26    def setUp(self):
 27        self.application = Application.objects.create(name=generate_id(), slug=generate_id())
 28        self.provider = SSFProvider.objects.create(
 29            name=generate_id(),
 30            signing_key=create_test_cert(),
 31            backchannel_application=self.application,
 32        )
 33        res = self.client.post(
 34            reverse(
 35                "authentik_providers_ssf:stream",
 36                kwargs={"application_slug": self.application.slug},
 37            ),
 38            data={
 39                "iss": "https://authentik.company/.well-known/ssf-configuration/foo/5",
 40                "aud": ["https://app.authentik.company"],
 41                "delivery": {
 42                    "method": "https://schemas.openid.net/secevent/risc/delivery-method/push",
 43                    "endpoint_url": "https://app.authentik.company",
 44                },
 45                "events_requested": [
 46                    "https://schemas.openid.net/secevent/caep/event-type/credential-change",
 47                    "https://schemas.openid.net/secevent/caep/event-type/session-revoked",
 48                ],
 49                "format": "iss_sub",
 50            },
 51            HTTP_AUTHORIZATION=f"Bearer {self.provider.token.key}",
 52        )
 53        self.assertEqual(res.status_code, 201, res.content)
 54
 55    def test_signal_logout(self):
 56        """Test user logout"""
 57        user = create_test_user()
 58        self.client.force_login(user)
 59        self.client.logout()
 60
 61        stream = Stream.objects.filter(provider=self.provider).first()
 62        self.assertIsNotNone(stream)
 63        event = StreamEvent.objects.filter(stream=stream).first()
 64        self.assertIsNotNone(event)
 65        self.assertEqual(event.status, SSFEventStatus.PENDING_FAILED)
 66        event_payload = event.payload["events"][
 67            "https://schemas.openid.net/secevent/caep/event-type/session-revoked"
 68        ]
 69        self.assertEqual(event_payload["initiating_entity"], "user")
 70        self.assertEqual(event.payload["sub_id"]["format"], "complex")
 71        self.assertEqual(event.payload["sub_id"]["session"]["format"], "opaque")
 72        self.assertEqual(event.payload["sub_id"]["user"]["format"], "email")
 73        self.assertEqual(event.payload["sub_id"]["user"]["email"], user.email)
 74
 75    def test_signal_password_change(self):
 76        """Test user password change"""
 77        user = create_test_user()
 78        self.client.force_login(user)
 79        user.set_password(generate_id())
 80        user.save()
 81
 82        stream = Stream.objects.filter(provider=self.provider).first()
 83        self.assertIsNotNone(stream)
 84        event = StreamEvent.objects.filter(stream=stream).first()
 85        self.assertIsNotNone(event)
 86        self.assertEqual(event.status, SSFEventStatus.PENDING_FAILED)
 87        event_payload = event.payload["events"][
 88            "https://schemas.openid.net/secevent/caep/event-type/credential-change"
 89        ]
 90        self.assertEqual(event_payload["change_type"], "update")
 91        self.assertEqual(event_payload["credential_type"], "password")
 92        self.assertEqual(event.payload["sub_id"]["format"], "complex")
 93        self.assertEqual(event.payload["sub_id"]["user"]["format"], "email")
 94        self.assertEqual(event.payload["sub_id"]["user"]["email"], user.email)
 95
 96    def test_signal_authenticator_added(self):
 97        """Test authenticator creation signal"""
 98        user = create_test_user()
 99        self.client.force_login(user)
100        dev = WebAuthnDevice.objects.create(
101            user=user,
102            name=generate_id(),
103            credential_id=generate_id(),
104            public_key=generate_id(),
105            aaguid=str(uuid4()),
106        )
107
108        stream = Stream.objects.filter(provider=self.provider).first()
109        self.assertIsNotNone(stream)
110        event = StreamEvent.objects.filter(stream=stream).exclude().first()
111        self.assertIsNotNone(event)
112        self.assertEqual(event.status, SSFEventStatus.PENDING_FAILED)
113        event_payload = event.payload["events"][
114            "https://schemas.openid.net/secevent/caep/event-type/credential-change"
115        ]
116        self.assertEqual(event_payload["change_type"], "create")
117        self.assertEqual(event_payload["fido2_aaguid"], dev.aaguid)
118        self.assertEqual(event_payload["friendly_name"], dev.name)
119        self.assertEqual(event_payload["credential_type"], "fido-u2f")
120        self.assertEqual(event.payload["sub_id"]["format"], "complex")
121        self.assertEqual(event.payload["sub_id"]["user"]["format"], "email")
122        self.assertEqual(event.payload["sub_id"]["user"]["email"], user.email)
123
124    def test_signal_authenticator_deleted(self):
125        """Test authenticator deletion signal"""
126        user = create_test_user()
127        self.client.force_login(user)
128        dev = WebAuthnDevice.objects.create(
129            user=user,
130            name=generate_id(),
131            credential_id=generate_id(),
132            public_key=generate_id(),
133            aaguid=str(uuid4()),
134        )
135        dev.delete()
136
137        stream = Stream.objects.filter(provider=self.provider).first()
138        self.assertIsNotNone(stream)
139        event = StreamEvent.objects.filter(stream=stream).exclude().first()
140        self.assertIsNotNone(event)
141        self.assertEqual(event.status, SSFEventStatus.PENDING_FAILED)
142        event_payload = event.payload["events"][
143            "https://schemas.openid.net/secevent/caep/event-type/credential-change"
144        ]
145        self.assertEqual(event_payload["change_type"], "delete")
146        self.assertEqual(event_payload["fido2_aaguid"], dev.aaguid)
147        self.assertEqual(event_payload["friendly_name"], dev.name)
148        self.assertEqual(event_payload["credential_type"], "fido-u2f")
149        self.assertEqual(event.payload["sub_id"]["format"], "complex")
150        self.assertEqual(event.payload["sub_id"]["user"]["format"], "email")
151        self.assertEqual(event.payload["sub_id"]["user"]["email"], user.email)
152
153    def test_signal_policy_ignore(self):
154        """Test event not being created for user that doesn't have access to the application"""
155        PolicyBinding.objects.create(
156            target=self.application, group=Group.objects.create(name=generate_id()), order=0
157        )
158        user = create_test_user()
159        self.client.force_login(user)
160        user.set_password(generate_id())
161        user.save()
162
163        stream = Stream.objects.filter(provider=self.provider).first()
164        self.assertIsNotNone(stream)
165        event = StreamEvent.objects.filter(
166            stream=stream, type=EventTypes.CAEP_CREDENTIAL_CHANGE
167        ).first()
168        self.assertIsNone(event)