authentik.stages.user_login.tests

  1"""login tests"""
  2
  3from time import sleep
  4from unittest.mock import patch
  5
  6from django.http import HttpRequest
  7from django.urls import reverse
  8from django.utils.timezone import now
  9
 10from authentik.blueprints.tests import apply_blueprint
 11from authentik.core.models import AuthenticatedSession, Session
 12from authentik.core.tests.utils import create_test_flow, create_test_user
 13from authentik.events.models import Event, EventAction
 14from authentik.events.utils import get_user
 15from authentik.flows.markers import StageMarker
 16from authentik.flows.models import FlowDesignation, FlowStageBinding
 17from authentik.flows.planner import PLAN_CONTEXT_PENDING_USER, FlowPlan
 18from authentik.flows.tests import FlowTestCase
 19from authentik.flows.tests.test_executor import TO_STAGE_RESPONSE_MOCK
 20from authentik.flows.views.executor import NEXT_ARG_NAME, SESSION_KEY_PLAN
 21from authentik.lib.generators import generate_id
 22from authentik.lib.utils.time import timedelta_from_string
 23from authentik.root.middleware import ClientIPMiddleware
 24from authentik.stages.user_login.middleware import (
 25    SESSION_KEY_BINDING_NET,
 26    BoundSessionMiddleware,
 27    SessionBindingBroken,
 28    logout_extra,
 29)
 30from authentik.stages.user_login.models import GeoIPBinding, NetworkBinding, UserLoginStage
 31
 32
 33class TestUserLoginStage(FlowTestCase):
 34    """Login tests"""
 35
 36    def setUp(self):
 37        super().setUp()
 38        self.user = create_test_user()
 39
 40        self.flow = create_test_flow(FlowDesignation.AUTHENTICATION)
 41        self.stage = UserLoginStage.objects.create(name="login")
 42        self.binding = FlowStageBinding.objects.create(target=self.flow, stage=self.stage, order=2)
 43
 44    def test_valid_get(self):
 45        """Test with a valid pending user and backend"""
 46        plan = FlowPlan(flow_pk=self.flow.pk.hex, bindings=[self.binding], markers=[StageMarker()])
 47        plan.context[PLAN_CONTEXT_PENDING_USER] = self.user
 48        session = self.client.session
 49        session[SESSION_KEY_PLAN] = plan
 50        session.save()
 51
 52        response = self.client.get(
 53            reverse("authentik_api:flow-executor", kwargs={"flow_slug": self.flow.slug})
 54        )
 55
 56        self.assertEqual(response.status_code, 200)
 57        self.assertStageRedirects(response, reverse("authentik_core:root-redirect"))
 58
 59    def test_valid_post(self):
 60        """Test with a valid pending user and backend"""
 61        plan = FlowPlan(flow_pk=self.flow.pk.hex, bindings=[self.binding], markers=[StageMarker()])
 62        plan.context[PLAN_CONTEXT_PENDING_USER] = self.user
 63        session = self.client.session
 64        session[SESSION_KEY_PLAN] = plan
 65        session.save()
 66
 67        response = self.client.post(
 68            reverse("authentik_api:flow-executor", kwargs={"flow_slug": self.flow.slug})
 69        )
 70
 71        self.assertEqual(response.status_code, 200)
 72        self.assertStageRedirects(response, reverse("authentik_core:root-redirect"))
 73
 74    def test_terminate_other_sessions(self):
 75        """Test terminate_other_sessions"""
 76        self.stage.terminate_other_sessions = True
 77        self.stage.save()
 78        plan = FlowPlan(flow_pk=self.flow.pk.hex, bindings=[self.binding], markers=[StageMarker()])
 79        plan.context[PLAN_CONTEXT_PENDING_USER] = self.user
 80        session = self.client.session
 81        session[SESSION_KEY_PLAN] = plan
 82        session.save()
 83
 84        key = generate_id()
 85        AuthenticatedSession.objects.create(
 86            session=Session.objects.create(
 87                session_key=key,
 88                last_ip=ClientIPMiddleware.default_ip,
 89            ),
 90            user=self.user,
 91        )
 92
 93        response = self.client.post(
 94            reverse("authentik_api:flow-executor", kwargs={"flow_slug": self.flow.slug})
 95        )
 96
 97        self.assertEqual(response.status_code, 200)
 98        self.assertStageRedirects(response, reverse("authentik_core:root-redirect"))
 99        self.assertFalse(AuthenticatedSession.objects.filter(session__session_key=key))
100        self.assertFalse(Session.objects.filter(session_key=key).exists())
101
102    def test_expiry(self):
103        """Test with expiry"""
104        self.stage.session_duration = "seconds=2"
105        self.stage.save()
106        plan = FlowPlan(flow_pk=self.flow.pk.hex, bindings=[self.binding], markers=[StageMarker()])
107        plan.context[PLAN_CONTEXT_PENDING_USER] = self.user
108        session = self.client.session
109        session[SESSION_KEY_PLAN] = plan
110        session.save()
111
112        before_request = now()
113        response = self.client.get(
114            reverse("authentik_api:flow-executor", kwargs={"flow_slug": self.flow.slug})
115        )
116        self.assertEqual(response.status_code, 200)
117        self.assertStageRedirects(response, reverse("authentik_core:root-redirect"))
118        self.assertNotEqual(list(self.client.session.keys()), [])
119        session_key = self.client.session.session_key
120        session = Session.objects.filter(session_key=session_key).first()
121        self.assertAlmostEqual(
122            session.expires.timestamp() - before_request.timestamp(),
123            timedelta_from_string(self.stage.session_duration).total_seconds(),
124            delta=1,
125        )
126        sleep(3)
127        self.client.session.clear_expired()
128        self.assertEqual(list(self.client.session.keys()), [])
129
130    def test_expiry_remember(self):
131        """Test with expiry"""
132        self.stage.session_duration = "seconds=2"
133        self.stage.remember_me_offset = "seconds=2"
134        self.stage.save()
135        plan = FlowPlan(flow_pk=self.flow.pk.hex, bindings=[self.binding], markers=[StageMarker()])
136        plan.context[PLAN_CONTEXT_PENDING_USER] = self.user
137        session = self.client.session
138        session[SESSION_KEY_PLAN] = plan
139        session.save()
140
141        response = self.client.get(
142            reverse("authentik_api:flow-executor", kwargs={"flow_slug": self.flow.slug}),
143        )
144        self.assertStageResponse(response, component="ak-stage-user-login")
145
146        response = self.client.post(
147            reverse("authentik_api:flow-executor", kwargs={"flow_slug": self.flow.slug}),
148            data={"remember_me": True},
149        )
150        _now = now().timestamp()
151        self.assertEqual(response.status_code, 200)
152        self.assertStageRedirects(response, reverse("authentik_core:root-redirect"))
153        self.assertNotEqual(list(self.client.session.keys()), [])
154        session_key = self.client.session.session_key
155        session = Session.objects.filter(session_key=session_key).first()
156        self.assertAlmostEqual(
157            session.expires.timestamp() - _now,
158            timedelta_from_string(self.stage.session_duration).total_seconds()
159            + timedelta_from_string(self.stage.remember_me_offset).total_seconds(),
160            delta=1,
161        )
162        sleep(5)
163        self.client.session.clear_expired()
164        self.assertEqual(list(self.client.session.keys()), [])
165
166    @patch(
167        "authentik.flows.views.executor.to_stage_response",
168        TO_STAGE_RESPONSE_MOCK,
169    )
170    def test_without_user(self):
171        """Test a plan without any pending user, resulting in a denied"""
172        plan = FlowPlan(flow_pk=self.flow.pk.hex, bindings=[self.binding], markers=[StageMarker()])
173        session = self.client.session
174        session[SESSION_KEY_PLAN] = plan
175        session.save()
176
177        response = self.client.get(
178            reverse("authentik_api:flow-executor", kwargs={"flow_slug": self.flow.slug})
179        )
180
181        self.assertStageResponse(
182            response,
183            self.flow,
184            component="ak-stage-access-denied",
185        )
186
187    @apply_blueprint("default/flow-default-user-settings-flow.yaml")
188    def test_inactive_account(self):
189        """Test with a valid pending user and backend"""
190        self.user.is_active = False
191        self.user.save()
192        plan = FlowPlan(flow_pk=self.flow.pk.hex, bindings=[self.binding], markers=[StageMarker()])
193        plan.context[PLAN_CONTEXT_PENDING_USER] = self.user
194        session = self.client.session
195        session[SESSION_KEY_PLAN] = plan
196        session.save()
197
198        response = self.client.get(
199            reverse("authentik_api:flow-executor", kwargs={"flow_slug": self.flow.slug})
200        )
201        self.assertEqual(response.status_code, 200)
202        self.assertStageResponse(
203            response, self.flow, component="ak-stage-access-denied", error_message="Unknown error"
204        )
205
206        # Check that API requests get rejected
207        response = self.client.get(reverse("authentik_api:application-list"))
208        self.assertEqual(response.status_code, 403)
209
210        # Check that flow requests requiring a user also get rejected
211        response = self.client.get(
212            reverse(
213                "authentik_api:flow-executor",
214                kwargs={"flow_slug": "default-user-settings-flow"},
215            )
216        )
217        self.assertStageResponse(
218            response,
219            self.flow,
220            component="ak-stage-access-denied",
221            error_message="Flow does not apply to current user.",
222        )
223
224    def test_binding_net_break_log(self):
225        """Test logout_extra with exception"""
226        # IPs from https://github.com/maxmind/MaxMind-DB/blob/main/source-data/GeoLite2-ASN-Test.json
227        for args, expect in [
228            [[NetworkBinding.BIND_ASN, "8.8.8.8", "8.8.8.8"], ["network.missing"]],
229            [[NetworkBinding.BIND_ASN, "1.0.0.1", "1.128.0.1"], ["network.asn"]],
230            [
231                [NetworkBinding.BIND_ASN_NETWORK, "12.81.96.1", "12.81.128.1"],
232                ["network.asn_network"],
233            ],
234            [[NetworkBinding.BIND_ASN_NETWORK_IP, "1.0.0.1", "1.0.0.2"], ["network.ip"]],
235        ]:
236            with self.subTest(args[0]):
237                with self.assertRaises(SessionBindingBroken) as cm:
238                    BoundSessionMiddleware.recheck_session_net(*args)
239                self.assertEqual(cm.exception.reason, expect[0])
240                # Ensure the request can be logged without throwing errors
241                self.client.force_login(self.user)
242                request = HttpRequest()
243                request.session = self.client.session
244                request.user = self.user
245                logout_extra(request, cm.exception)
246
247    def test_binding_geo_break_log(self):
248        """Test logout_extra with exception"""
249        # IPs from https://github.com/maxmind/MaxMind-DB/blob/main/source-data/GeoLite2-City-Test.json
250        for args, expect in [
251            [[GeoIPBinding.BIND_CONTINENT, "8.8.8.8", "8.8.8.8"], ["geoip.missing"]],
252            [[GeoIPBinding.BIND_CONTINENT, "2.125.160.216", "67.43.156.1"], ["geoip.continent"]],
253            [
254                [GeoIPBinding.BIND_CONTINENT_COUNTRY, "81.2.69.142", "89.160.20.112"],
255                ["geoip.country"],
256            ],
257            [
258                [GeoIPBinding.BIND_CONTINENT_COUNTRY_CITY, "2.125.160.216", "81.2.69.142"],
259                ["geoip.city"],
260            ],
261        ]:
262            with self.subTest(args[0]):
263                with self.assertRaises(SessionBindingBroken) as cm:
264                    BoundSessionMiddleware.recheck_session_geo(*args)
265                self.assertEqual(cm.exception.reason, expect[0])
266                # Ensure the request can be logged without throwing errors
267                self.client.force_login(self.user)
268                request = HttpRequest()
269                request.session = self.client.session
270                request.user = self.user
271                logout_extra(request, cm.exception)
272
273    def test_session_binding_broken(self):
274        """Test session binding"""
275        Event.objects.all().delete()
276        self.client.force_login(self.user)
277        session = self.client.session
278        session[Session.Keys.LAST_IP] = "192.0.2.1"
279        session[SESSION_KEY_BINDING_NET] = NetworkBinding.BIND_ASN_NETWORK_IP
280        session.save()
281
282        res = self.client.get(reverse("authentik_api:user-me"))
283        self.assertEqual(res.status_code, 302)
284        self.assertEqual(
285            res.url,
286            reverse(
287                "authentik_flows:default-authentication",
288            )
289            + f"?{NEXT_ARG_NAME}={reverse("authentik_api:user-me")}",
290        )
291        event = Event.objects.filter(action=EventAction.LOGOUT).first()
292        self.assertEqual(event.user, get_user(self.user))