authentik.stages.invitation.tests

  1"""invitation tests"""
  2
  3from datetime import timedelta
  4from unittest.mock import MagicMock, patch
  5
  6from django.urls import reverse
  7from django.utils.http import urlencode
  8from django.utils.timezone import now
  9from guardian.shortcuts import get_anonymous_user
 10from rest_framework.test import APITestCase
 11
 12from authentik.blueprints.v1.importer import SERIALIZER_CONTEXT_BLUEPRINT
 13from authentik.core.tests.utils import create_test_admin_user, create_test_flow
 14from authentik.flows.markers import StageMarker
 15from authentik.flows.models import FlowDesignation, FlowStageBinding
 16from authentik.flows.planner import PLAN_CONTEXT_PENDING_USER, FlowPlan
 17from authentik.flows.tests import FlowTestCase
 18from authentik.flows.tests.test_executor import TO_STAGE_RESPONSE_MOCK
 19from authentik.flows.views.executor import SESSION_KEY_PLAN
 20from authentik.stages.invitation.api import InvitationSerializer
 21from authentik.stages.invitation.models import Invitation, InvitationStage
 22from authentik.stages.invitation.stage import (
 23    PLAN_CONTEXT_INVITATION_TOKEN,
 24    PLAN_CONTEXT_PROMPT,
 25    QS_INVITATION_TOKEN_KEY,
 26)
 27from authentik.stages.password import BACKEND_INBUILT
 28from authentik.stages.password.stage import PLAN_CONTEXT_AUTHENTICATION_BACKEND
 29
 30
 31class TestInvitationStage(FlowTestCase):
 32    """Login tests"""
 33
 34    def setUp(self):
 35        super().setUp()
 36        self.user = create_test_admin_user()
 37        self.flow = create_test_flow(FlowDesignation.AUTHENTICATION)
 38        self.stage = InvitationStage.objects.create(name="invitation")
 39        self.binding = FlowStageBinding.objects.create(target=self.flow, stage=self.stage, order=2)
 40
 41    @patch(
 42        "authentik.flows.views.executor.to_stage_response",
 43        TO_STAGE_RESPONSE_MOCK,
 44    )
 45    def test_without_invitation_fail(self):
 46        """Test without any invitation, continue_flow_without_invitation not set."""
 47        plan = FlowPlan(flow_pk=self.flow.pk.hex, bindings=[self.binding], markers=[StageMarker()])
 48        plan.context[PLAN_CONTEXT_PENDING_USER] = self.user
 49        plan.context[PLAN_CONTEXT_AUTHENTICATION_BACKEND] = BACKEND_INBUILT
 50        session = self.client.session
 51        session[SESSION_KEY_PLAN] = plan
 52        session.save()
 53
 54        response = self.client.get(
 55            reverse("authentik_api:flow-executor", kwargs={"flow_slug": self.flow.slug})
 56        )
 57        self.assertStageResponse(
 58            response,
 59            flow=self.flow,
 60            component="ak-stage-access-denied",
 61        )
 62
 63    def test_without_invitation_continue(self):
 64        """Test without any invitation, continue_flow_without_invitation is set."""
 65        self.stage.continue_flow_without_invitation = True
 66        self.stage.save()
 67        plan = FlowPlan(flow_pk=self.flow.pk.hex, bindings=[self.binding], markers=[StageMarker()])
 68        plan.context[PLAN_CONTEXT_PENDING_USER] = self.user
 69        plan.context[PLAN_CONTEXT_AUTHENTICATION_BACKEND] = BACKEND_INBUILT
 70        session = self.client.session
 71        session[SESSION_KEY_PLAN] = plan
 72        session.save()
 73
 74        response = self.client.get(
 75            reverse("authentik_api:flow-executor", kwargs={"flow_slug": self.flow.slug})
 76        )
 77
 78        self.assertEqual(response.status_code, 200)
 79        self.assertStageRedirects(response, reverse("authentik_core:root-redirect"))
 80
 81        self.stage.continue_flow_without_invitation = False
 82        self.stage.save()
 83
 84    def test_with_invitation_expired(self):
 85        """Test with invitation, expired"""
 86        plan = FlowPlan(flow_pk=self.flow.pk.hex, bindings=[self.binding], markers=[StageMarker()])
 87        session = self.client.session
 88        session[SESSION_KEY_PLAN] = plan
 89        session.save()
 90
 91        data = {"foo": "bar"}
 92        invite = Invitation.objects.create(
 93            created_by=get_anonymous_user(),
 94            fixed_data=data,
 95            expires=now() - timedelta(hours=1),
 96        )
 97
 98        base_url = reverse("authentik_api:flow-executor", kwargs={"flow_slug": self.flow.slug})
 99        args = urlencode({QS_INVITATION_TOKEN_KEY: invite.pk.hex})
100        response = self.client.get(base_url + f"?query={args}")
101
102        self.assertEqual(response.status_code, 200)
103        self.assertStageResponse(
104            response,
105            flow=self.flow,
106            component="ak-stage-access-denied",
107        )
108
109    def test_with_invitation_get(self):
110        """Test with invitation, check data in session"""
111        plan = FlowPlan(flow_pk=self.flow.pk.hex, bindings=[self.binding], markers=[StageMarker()])
112        session = self.client.session
113        session[SESSION_KEY_PLAN] = plan
114        session.save()
115
116        data = {"foo": "bar"}
117        invite = Invitation.objects.create(created_by=get_anonymous_user(), fixed_data=data)
118
119        with patch("authentik.flows.views.executor.FlowExecutorView.cancel", MagicMock()):
120            base_url = reverse("authentik_api:flow-executor", kwargs={"flow_slug": self.flow.slug})
121            args = urlencode({QS_INVITATION_TOKEN_KEY: invite.pk.hex})
122            response = self.client.get(base_url + f"?query={args}")
123
124        session = self.client.session
125        plan: FlowPlan = session[SESSION_KEY_PLAN]
126        self.assertEqual(plan.context[PLAN_CONTEXT_PROMPT], data)
127
128        self.assertEqual(response.status_code, 200)
129        self.assertStageRedirects(response, reverse("authentik_core:root-redirect"))
130
131    def test_invalid_flow(self):
132        """Test with invitation, invalid flow limit"""
133        invalid_flow = create_test_flow(FlowDesignation.ENROLLMENT)
134        plan = FlowPlan(flow_pk=self.flow.pk.hex, bindings=[self.binding], markers=[StageMarker()])
135        session = self.client.session
136        session[SESSION_KEY_PLAN] = plan
137        session.save()
138
139        data = {"foo": "bar"}
140        invite = Invitation.objects.create(
141            created_by=get_anonymous_user(), fixed_data=data, flow=invalid_flow
142        )
143
144        with patch("authentik.flows.views.executor.FlowExecutorView.cancel", MagicMock()):
145            base_url = reverse("authentik_api:flow-executor", kwargs={"flow_slug": self.flow.slug})
146            args = urlencode({QS_INVITATION_TOKEN_KEY: invite.pk.hex})
147            response = self.client.get(base_url + f"?query={args}")
148
149        session = self.client.session
150        plan: FlowPlan = session[SESSION_KEY_PLAN]
151
152        self.assertStageResponse(
153            response,
154            flow=self.flow,
155            component="ak-stage-access-denied",
156        )
157
158    def test_with_invitation_prompt_data(self):
159        """Test with invitation, check data in session"""
160        data = {"foo": "bar"}
161        invite = Invitation.objects.create(
162            created_by=get_anonymous_user(), fixed_data=data, single_use=True
163        )
164
165        plan = FlowPlan(flow_pk=self.flow.pk.hex, bindings=[self.binding], markers=[StageMarker()])
166        plan.context[PLAN_CONTEXT_PROMPT] = {PLAN_CONTEXT_INVITATION_TOKEN: invite.pk.hex}
167        session = self.client.session
168        session[SESSION_KEY_PLAN] = plan
169        session.save()
170
171        with patch("authentik.flows.views.executor.FlowExecutorView.cancel", MagicMock()):
172            base_url = reverse("authentik_api:flow-executor", kwargs={"flow_slug": self.flow.slug})
173            response = self.client.get(base_url, follow=True)
174
175        session = self.client.session
176        plan: FlowPlan = session[SESSION_KEY_PLAN]
177        self.assertEqual(
178            plan.context[PLAN_CONTEXT_PROMPT], data | plan.context[PLAN_CONTEXT_PROMPT]
179        )
180
181        self.assertEqual(response.status_code, 200)
182        self.assertStageRedirects(response, reverse("authentik_core:root-redirect"))
183        self.assertFalse(Invitation.objects.filter(pk=invite.pk))
184
185
186class TestInvitationsAPI(APITestCase):
187    """Test Invitations API"""
188
189    def setUp(self) -> None:
190        super().setUp()
191        self.user = create_test_admin_user()
192        self.client.force_login(self.user)
193
194    def test_invite_create(self):
195        """Test Invitations creation endpoint"""
196        response = self.client.post(
197            reverse("authentik_api:invitation-list"),
198            {"name": "test-token", "fixed_data": {}},
199            format="json",
200        )
201        self.assertEqual(response.status_code, 201)
202        self.assertEqual(Invitation.objects.first().created_by, self.user)
203
204    def test_invite_create_blueprint_context(self):
205        """Test Invitations creation via blueprint context"""
206
207        flow = create_test_flow(FlowDesignation.ENROLLMENT)
208        data = {
209            "name": "test-blueprint-invitation",
210            "flow": flow.pk.hex,
211            "single_use": True,
212            "fixed_data": {"email": "test@example.com"},
213        }
214        serializer = InvitationSerializer(data=data, context={SERIALIZER_CONTEXT_BLUEPRINT: True})
215        self.assertTrue(serializer.is_valid())
216        invitation = serializer.save()
217        self.assertEqual(invitation.created_by, get_anonymous_user())
218        self.assertEqual(invitation.name, "test-blueprint-invitation")
219        self.assertEqual(invitation.fixed_data, {"email": "test@example.com"})
220
221    def test_send_email_no_addresses(self):
222        """Test send_email endpoint with no email addresses"""
223        flow = create_test_flow(FlowDesignation.ENROLLMENT)
224        invite = Invitation.objects.create(
225            name="test-invite",
226            created_by=self.user,
227            flow=flow,
228        )
229
230        response = self.client.post(
231            reverse("authentik_api:invitation-send-email", kwargs={"pk": invite.pk}),
232            {"email_addresses": []},
233            format="json",
234        )
235        self.assertEqual(response.status_code, 400)
236        self.assertIn("error", response.data)
237
238    def test_send_email_no_flow(self):
239        """Test send_email endpoint with invitation without flow"""
240        invite = Invitation.objects.create(
241            name="test-invite-no-flow",
242            created_by=self.user,
243            flow=None,
244        )
245
246        response = self.client.post(
247            reverse("authentik_api:invitation-send-email", kwargs={"pk": invite.pk}),
248            {"email_addresses": ["test@example.com"]},
249            format="json",
250        )
251        self.assertEqual(response.status_code, 400)
252        self.assertIn("error", response.data)
253
254    @patch("authentik.stages.invitation.api.BaseEvaluator.expr_send_email")
255    def test_send_email_success(self, mock_send_email: MagicMock):
256        """Test send_email endpoint successfully queues emails"""
257        flow = create_test_flow(FlowDesignation.ENROLLMENT)
258        invite = Invitation.objects.create(
259            name="test-invite",
260            created_by=self.user,
261            flow=flow,
262        )
263
264        response = self.client.post(
265            reverse("authentik_api:invitation-send-email", kwargs={"pk": invite.pk}),
266            {
267                "email_addresses": ["user1@example.com", "user2@example.com"],
268                "template": "email/invitation.html",
269            },
270            format="json",
271        )
272        self.assertEqual(response.status_code, 204)
273        self.assertEqual(mock_send_email.call_count, 2)
274
275    @patch("authentik.stages.invitation.api.BaseEvaluator.expr_send_email")
276    def test_send_email_with_cc_bcc(self, mock_send_email: MagicMock):
277        """Test send_email endpoint with CC and BCC addresses"""
278        flow = create_test_flow(FlowDesignation.ENROLLMENT)
279        invite = Invitation.objects.create(
280            name="test-invite",
281            created_by=self.user,
282            flow=flow,
283        )
284
285        response = self.client.post(
286            reverse("authentik_api:invitation-send-email", kwargs={"pk": invite.pk}),
287            {
288                "email_addresses": ["user@example.com"],
289                "cc_addresses": ["cc@example.com"],
290                "bcc_addresses": ["bcc@example.com"],
291                "template": "email/invitation.html",
292            },
293            format="json",
294        )
295        self.assertEqual(response.status_code, 204)
296        mock_send_email.assert_called_once()
297        call_kwargs = mock_send_email.call_args.kwargs
298        self.assertEqual(call_kwargs["cc"], ["cc@example.com"])
299        self.assertEqual(call_kwargs["bcc"], ["bcc@example.com"])
300
301    @patch("authentik.stages.invitation.api.BaseEvaluator.expr_send_email")
302    def test_send_email_context(self, mock_send_email: MagicMock):
303        """Test send_email endpoint passes correct context to email"""
304        flow = create_test_flow(FlowDesignation.ENROLLMENT)
305        invite = Invitation.objects.create(
306            name="test-invite",
307            created_by=self.user,
308            flow=flow,
309        )
310
311        response = self.client.post(
312            reverse("authentik_api:invitation-send-email", kwargs={"pk": invite.pk}),
313            {"email_addresses": ["user@example.com"]},
314            format="json",
315        )
316        self.assertEqual(response.status_code, 204)
317        mock_send_email.assert_called_once()
318        call_kwargs = mock_send_email.call_args.kwargs
319        self.assertIn("url", call_kwargs["context"])
320        self.assertIn(str(invite.pk), call_kwargs["context"]["url"])
321        self.assertIn(flow.slug, call_kwargs["context"]["url"])