authentik.enterprise.stages.account_lockdown.tests.test_api

  1"""Test Users Account Lockdown API"""
  2
  3from json import loads
  4from unittest.mock import MagicMock, patch
  5from urllib.parse import urlparse
  6
  7from django.urls import reverse
  8from rest_framework.test import APITestCase
  9
 10from authentik.core.tests.utils import (
 11    create_test_brand,
 12    create_test_flow,
 13    create_test_user,
 14)
 15from authentik.enterprise.stages.account_lockdown.models import AccountLockdownStage
 16from authentik.flows.models import FlowDesignation, FlowStageBinding
 17from authentik.flows.planner import PLAN_CONTEXT_PENDING_USER
 18from authentik.flows.views.executor import SESSION_KEY_PLAN
 19from authentik.lib.generators import generate_id
 20
 21# Patch for enterprise license check
 22patch_license = patch(
 23    "authentik.enterprise.models.LicenseUsageStatus.is_valid",
 24    MagicMock(return_value=True),
 25)
 26
 27
 28@patch_license
 29class AccountLockdownAPITestCase(APITestCase):
 30    """Shared helpers for account lockdown API tests."""
 31
 32    def setUp(self) -> None:
 33        self.lockdown_flow = create_test_flow(FlowDesignation.STAGE_CONFIGURATION)
 34        self.lockdown_stage = AccountLockdownStage.objects.create(name=generate_id())
 35        FlowStageBinding.objects.create(
 36            target=self.lockdown_flow,
 37            stage=self.lockdown_stage,
 38            order=0,
 39        )
 40        self.brand = create_test_brand()
 41        self.brand.flow_lockdown = self.lockdown_flow
 42        self.brand.save()
 43
 44    def create_user_with_email(self):
 45        """Create a regular user with a unique email address."""
 46        user = create_test_user()
 47        user.email = f"{generate_id()}@test.com"
 48        user.save()
 49        return user
 50
 51    def assert_redirect_targets(self, response, user):
 52        """Assert that a response contains a pre-planned lockdown flow link for a user."""
 53        self.assertEqual(response.status_code, 200)
 54        body = loads(response.content)
 55        self.assertIn(self.lockdown_flow.slug, body["link"])
 56        self.assertEqual(urlparse(body["link"]).query, "")
 57        plan = self.client.session[SESSION_KEY_PLAN]
 58        self.assertEqual(plan.context[PLAN_CONTEXT_PENDING_USER].pk, user.pk)
 59
 60    def assert_no_flow_configured(self, response):
 61        """Assert that the API reports a missing lockdown flow."""
 62        self.assertEqual(response.status_code, 400)
 63        body = loads(response.content)
 64        self.assertIn("No lockdown flow configured", body["non_field_errors"][0])
 65
 66
 67@patch_license
 68class TestUsersAccountLockdownAPI(AccountLockdownAPITestCase):
 69    """Test Users Account Lockdown API"""
 70
 71    def setUp(self) -> None:
 72        super().setUp()
 73        self.actor = create_test_user()
 74        self.user = self.create_user_with_email()
 75
 76    def test_account_lockdown_with_change_user_returns_redirect(self):
 77        """Test that account lockdown allows users with change_user permission."""
 78        self.actor.assign_perms_to_managed_role("authentik_core.change_user", self.user)
 79        self.client.force_login(self.actor)
 80
 81        response = self.client.post(
 82            reverse("authentik_api:user-account-lockdown"),
 83            data={"user": self.user.pk},
 84            format="json",
 85        )
 86
 87        self.assert_redirect_targets(response, self.user)
 88
 89    def test_account_lockdown_no_flow_configured(self):
 90        """Test account lockdown when no flow is configured"""
 91        self.brand.flow_lockdown = None
 92        self.brand.save()
 93        self.actor.assign_perms_to_managed_role("authentik_core.change_user", self.user)
 94        self.client.force_login(self.actor)
 95
 96        response = self.client.post(
 97            reverse("authentik_api:user-account-lockdown"),
 98            data={"user": self.user.pk},
 99            format="json",
100        )
101
102        self.assert_no_flow_configured(response)
103
104    def test_account_lockdown_unauthenticated(self):
105        """Test account lockdown requires authentication"""
106        response = self.client.post(
107            reverse("authentik_api:user-account-lockdown"),
108            data={"user": self.user.pk},
109            format="json",
110        )
111
112        self.assertEqual(response.status_code, 403)
113
114    def test_account_lockdown_without_change_user_denied(self):
115        """Test account lockdown denies users without change_user permission."""
116        self.client.force_login(self.actor)
117
118        response = self.client.post(
119            reverse("authentik_api:user-account-lockdown"),
120            data={"user": self.user.pk},
121            format="json",
122        )
123
124        self.assertEqual(response.status_code, 403)
125
126    def test_account_lockdown_self_returns_redirect(self):
127        """Test successful self-service account lockdown returns a direct redirect."""
128        self.client.force_login(self.user)
129
130        response = self.client.post(
131            reverse("authentik_api:user-account-lockdown"),
132            data={},
133            format="json",
134        )
135
136        self.assert_redirect_targets(response, self.user)
137
138    def test_account_lockdown_self_target_without_change_user_returns_redirect(self):
139        """Test self-service does not require change_user permission."""
140        self.client.force_login(self.user)
141
142        response = self.client.post(
143            reverse("authentik_api:user-account-lockdown"),
144            data={"user": self.user.pk},
145            format="json",
146        )
147
148        self.assert_redirect_targets(response, self.user)