authentik.enterprise.tests.test_read_only

  1"""read only tests"""
  2
  3from datetime import timedelta
  4from unittest.mock import MagicMock, patch
  5
  6from django.urls import reverse
  7from django.utils.timezone import now
  8
  9from authentik.core.tests.utils import create_test_admin_user, create_test_flow, create_test_user
 10from authentik.enterprise.license import LicenseKey
 11from authentik.enterprise.models import (
 12    THRESHOLD_READ_ONLY_WEEKS,
 13    License,
 14    LicenseUsage,
 15    LicenseUsageStatus,
 16)
 17from authentik.enterprise.tests.test_license import expiry_valid
 18from authentik.flows.models import (
 19    FlowDesignation,
 20    FlowStageBinding,
 21)
 22from authentik.flows.tests import FlowTestCase
 23from authentik.lib.generators import generate_id
 24from authentik.stages.identification.models import IdentificationStage, UserFields
 25from authentik.stages.password import BACKEND_INBUILT
 26from authentik.stages.password.models import PasswordStage
 27from authentik.stages.user_login.models import UserLoginStage
 28
 29
 30class TestReadOnly(FlowTestCase):
 31    """Test read_only"""
 32
 33    @patch(
 34        "authentik.enterprise.license.LicenseKey.validate",
 35        MagicMock(
 36            return_value=LicenseKey(
 37                aud="",
 38                exp=expiry_valid,
 39                name=generate_id(),
 40                internal_users=100,
 41                external_users=100,
 42            )
 43        ),
 44    )
 45    @patch(
 46        "authentik.enterprise.license.LicenseKey.get_internal_user_count",
 47        MagicMock(return_value=1000),
 48    )
 49    @patch(
 50        "authentik.enterprise.license.LicenseKey.get_external_user_count",
 51        MagicMock(return_value=1000),
 52    )
 53    @patch(
 54        "authentik.enterprise.license.LicenseKey.record_usage",
 55        MagicMock(),
 56    )
 57    def test_login(self):
 58        """Test flow, ensure login is still possible with read only mode"""
 59        License.objects.create(key=generate_id())
 60        usage = LicenseUsage.objects.create(
 61            internal_user_count=100,
 62            external_user_count=100,
 63            status=LicenseUsageStatus.VALID,
 64        )
 65        usage.record_date = now() - timedelta(weeks=THRESHOLD_READ_ONLY_WEEKS + 1)
 66        usage.save(update_fields=["record_date"])
 67
 68        flow = create_test_flow(
 69            FlowDesignation.AUTHENTICATION,
 70        )
 71
 72        ident_stage = IdentificationStage.objects.create(
 73            name=generate_id(),
 74            user_fields=[UserFields.E_MAIL],
 75            pretend_user_exists=False,
 76        )
 77        FlowStageBinding.objects.create(
 78            target=flow,
 79            stage=ident_stage,
 80            order=0,
 81        )
 82        password_stage = PasswordStage.objects.create(
 83            name=generate_id(), backends=[BACKEND_INBUILT]
 84        )
 85        FlowStageBinding.objects.create(
 86            target=flow,
 87            stage=password_stage,
 88            order=1,
 89        )
 90        login_stage = UserLoginStage.objects.create(
 91            name=generate_id(),
 92        )
 93        FlowStageBinding.objects.create(
 94            target=flow,
 95            stage=login_stage,
 96            order=2,
 97        )
 98
 99        user = create_test_user()
100
101        exec_url = reverse("authentik_api:flow-executor", kwargs={"flow_slug": flow.slug})
102        response = self.client.get(exec_url)
103        self.assertStageResponse(
104            response,
105            flow,
106            component="ak-stage-identification",
107            password_fields=False,
108            primary_action="Log in",
109            sources=[],
110            show_source_labels=False,
111            user_fields=[UserFields.E_MAIL],
112        )
113        response = self.client.post(exec_url, {"uid_field": user.email}, follow=True)
114        self.assertStageResponse(response, flow, component="ak-stage-password")
115        response = self.client.post(exec_url, {"password": user.username}, follow=True)
116        self.assertStageRedirects(response, reverse("authentik_core:root-redirect"))
117
118    @patch(
119        "authentik.enterprise.license.LicenseKey.validate",
120        MagicMock(
121            return_value=LicenseKey(
122                aud="",
123                exp=expiry_valid,
124                name=generate_id(),
125                internal_users=100,
126                external_users=100,
127            )
128        ),
129    )
130    @patch(
131        "authentik.enterprise.license.LicenseKey.get_internal_user_count",
132        MagicMock(return_value=1000),
133    )
134    @patch(
135        "authentik.enterprise.license.LicenseKey.get_external_user_count",
136        MagicMock(return_value=1000),
137    )
138    @patch(
139        "authentik.enterprise.license.LicenseKey.record_usage",
140        MagicMock(),
141    )
142    def test_manage_licenses(self):
143        """Test that managing licenses is still possible"""
144        license = License.objects.create(key=generate_id())
145        usage = LicenseUsage.objects.create(
146            internal_user_count=100,
147            external_user_count=100,
148            status=LicenseUsageStatus.VALID,
149        )
150        usage.record_date = now() - timedelta(weeks=THRESHOLD_READ_ONLY_WEEKS + 1)
151        usage.save(update_fields=["record_date"])
152
153        admin = create_test_admin_user()
154        self.client.force_login(admin)
155
156        # Reading is always allowed
157        response = self.client.get(reverse("authentik_api:license-list"))
158        self.assertEqual(response.status_code, 200)
159
160        # Writing should also be allowed
161        response = self.client.patch(
162            reverse("authentik_api:license-detail", kwargs={"pk": license.pk})
163        )
164        self.assertEqual(response.status_code, 200)
165
166    @patch(
167        "authentik.enterprise.license.LicenseKey.validate",
168        MagicMock(
169            return_value=LicenseKey(
170                aud="",
171                exp=expiry_valid,
172                name=generate_id(),
173                internal_users=100,
174                external_users=100,
175            )
176        ),
177    )
178    @patch(
179        "authentik.enterprise.license.LicenseKey.get_internal_user_count",
180        MagicMock(return_value=1000),
181    )
182    @patch(
183        "authentik.enterprise.license.LicenseKey.get_external_user_count",
184        MagicMock(return_value=1000),
185    )
186    @patch(
187        "authentik.enterprise.license.LicenseKey.record_usage",
188        MagicMock(),
189    )
190    def test_manage_flows(self):
191        """Test flow"""
192        License.objects.create(key=generate_id())
193        usage = LicenseUsage.objects.create(
194            internal_user_count=100,
195            external_user_count=100,
196            status=LicenseUsageStatus.VALID,
197        )
198        usage.record_date = now() - timedelta(weeks=THRESHOLD_READ_ONLY_WEEKS + 1)
199        usage.save(update_fields=["record_date"])
200
201        admin = create_test_admin_user()
202        self.client.force_login(admin)
203
204        # Read only is still allowed
205        response = self.client.get(reverse("authentik_api:flow-list"))
206        self.assertEqual(response.status_code, 200)
207
208        flow = create_test_flow()
209        # Writing is not
210        response = self.client.patch(
211            reverse("authentik_api:flow-detail", kwargs={"slug": flow.slug})
212        )
213        self.assertJSONEqual(
214            response.content,
215            {"detail": "Request denied due to expired/invalid license.", "code": "denied_license"},
216        )
217        self.assertEqual(response.status_code, 400)
218
219    @patch(
220        "authentik.enterprise.license.LicenseKey.validate",
221        MagicMock(
222            return_value=LicenseKey(
223                aud="",
224                exp=expiry_valid,
225                name=generate_id(),
226                internal_users=100,
227                external_users=100,
228            )
229        ),
230    )
231    @patch(
232        "authentik.enterprise.license.LicenseKey.get_internal_user_count",
233        MagicMock(return_value=1000),
234    )
235    @patch(
236        "authentik.enterprise.license.LicenseKey.get_external_user_count",
237        MagicMock(return_value=1000),
238    )
239    @patch(
240        "authentik.enterprise.license.LicenseKey.record_usage",
241        MagicMock(),
242    )
243    def test_manage_users(self):
244        """Test that managing users is still possible"""
245        License.objects.create(key=generate_id())
246        usage = LicenseUsage.objects.create(
247            internal_user_count=100,
248            external_user_count=100,
249            status=LicenseUsageStatus.VALID,
250        )
251        usage.record_date = now() - timedelta(weeks=THRESHOLD_READ_ONLY_WEEKS + 1)
252        usage.save(update_fields=["record_date"])
253
254        admin = create_test_admin_user()
255        self.client.force_login(admin)
256
257        # Reading is always allowed
258        response = self.client.get(reverse("authentik_api:user-list"))
259        self.assertEqual(response.status_code, 200)
260
261        # Writing should also be allowed
262        response = self.client.patch(reverse("authentik_api:user-detail", kwargs={"pk": admin.pk}))
263        self.assertEqual(response.status_code, 200)