authentik.core.tests.test_setup

View Source

  1from http import HTTPStatus
  2from os import environ
  3
  4from django.contrib.auth.hashers import make_password
  5from django.urls import reverse
  6
  7from authentik.blueprints.tests import apply_blueprint
  8from authentik.core.apps import Setup
  9from authentik.core.models import Token, TokenIntents, User
 10from authentik.flows.models import Flow
 11from authentik.flows.tests import FlowTestCase
 12from authentik.lib.generators import generate_id
 13from authentik.root.signals import post_startup, pre_startup
 14from authentik.tenants.flags import patch_flag
 15
 16
 17class TestSetup(FlowTestCase):
 18    def tearDown(self):
 19        environ.pop("AUTHENTIK_BOOTSTRAP_PASSWORD", None)
 20        environ.pop("AUTHENTIK_BOOTSTRAP_PASSWORD_HASH", None)
 21        environ.pop("AUTHENTIK_BOOTSTRAP_TOKEN", None)
 22
 23    @patch_flag(Setup, True)
 24    def test_setup(self):
 25        """Test existing instance"""
 26        res = self.client.get(reverse("authentik_core:root-redirect"))
 27        self.assertEqual(res.status_code, HTTPStatus.FOUND)
 28        self.assertRedirects(
 29            res,
 30            reverse("authentik_flows:default-authentication") + "?next=/",
 31            fetch_redirect_response=False,
 32        )
 33
 34        res = self.client.head(reverse("authentik_core:setup"))
 35        self.assertEqual(res.status_code, HTTPStatus.SERVICE_UNAVAILABLE)
 36
 37        res = self.client.get(reverse("authentik_core:setup"))
 38        self.assertEqual(res.status_code, HTTPStatus.FOUND)
 39        self.assertRedirects(
 40            res,
 41            reverse("authentik_core:root-redirect"),
 42            fetch_redirect_response=False,
 43        )
 44
 45    @patch_flag(Setup, False)
 46    def test_not_setup_no_flow(self):
 47        """Test case on initial startup; setup flag is not set and oobe flow does
 48        not exist yet"""
 49        Flow.objects.filter(slug="initial-setup").delete()
 50        res = self.client.get(reverse("authentik_core:root-redirect"))
 51        self.assertEqual(res.status_code, HTTPStatus.FOUND)
 52        self.assertRedirects(res, reverse("authentik_core:setup"), fetch_redirect_response=False)
 53        # Flow does not exist, hence 503
 54        res = self.client.get(reverse("authentik_core:setup"))
 55        self.assertEqual(res.status_code, HTTPStatus.SERVICE_UNAVAILABLE)
 56        res = self.client.head(reverse("authentik_core:setup"))
 57        self.assertEqual(res.status_code, HTTPStatus.SERVICE_UNAVAILABLE)
 58
 59    @patch_flag(Setup, False)
 60    @apply_blueprint("default/flow-oobe.yaml")
 61    def test_not_setup(self):
 62        """Test case for when worker comes up, and has created flow"""
 63        res = self.client.get(reverse("authentik_core:root-redirect"))
 64        self.assertEqual(res.status_code, HTTPStatus.FOUND)
 65        self.assertRedirects(res, reverse("authentik_core:setup"), fetch_redirect_response=False)
 66        # Flow does not exist, hence 503
 67        res = self.client.head(reverse("authentik_core:setup"))
 68        self.assertEqual(res.status_code, HTTPStatus.OK)
 69        res = self.client.get(reverse("authentik_core:setup"))
 70        self.assertEqual(res.status_code, HTTPStatus.FOUND)
 71        self.assertRedirects(
 72            res,
 73            reverse("authentik_core:if-flow", kwargs={"flow_slug": "initial-setup"}),
 74            fetch_redirect_response=False,
 75        )
 76
 77    @apply_blueprint("default/flow-oobe.yaml")
 78    @apply_blueprint("system/bootstrap.yaml")
 79    def test_setup_flow_full(self):
 80        """Test full setup flow"""
 81        Setup.set(False)
 82
 83        res = self.client.get(reverse("authentik_core:setup"))
 84        self.assertEqual(res.status_code, HTTPStatus.FOUND)
 85        self.assertRedirects(
 86            res,
 87            reverse("authentik_core:if-flow", kwargs={"flow_slug": "initial-setup"}),
 88            fetch_redirect_response=False,
 89        )
 90
 91        res = self.client.get(
 92            reverse("authentik_api:flow-executor", kwargs={"flow_slug": "initial-setup"}),
 93        )
 94        self.assertEqual(res.status_code, HTTPStatus.OK)
 95        self.assertStageResponse(res, component="ak-stage-prompt")
 96
 97        pw = generate_id()
 98        res = self.client.post(
 99            reverse("authentik_api:flow-executor", kwargs={"flow_slug": "initial-setup"}),
100            {
101                "email": f"{generate_id()}@t.goauthentik.io",
102                "password": pw,
103                "password_repeat": pw,
104                "component": "ak-stage-prompt",
105            },
106        )
107        self.assertEqual(res.status_code, HTTPStatus.FOUND)
108
109        res = self.client.get(
110            reverse("authentik_api:flow-executor", kwargs={"flow_slug": "initial-setup"}),
111        )
112        self.assertEqual(res.status_code, HTTPStatus.FOUND)
113
114        res = self.client.get(
115            reverse("authentik_api:flow-executor", kwargs={"flow_slug": "initial-setup"}),
116        )
117        self.assertEqual(res.status_code, HTTPStatus.FOUND)
118
119        res = self.client.get(
120            reverse("authentik_api:flow-executor", kwargs={"flow_slug": "initial-setup"}),
121        )
122        self.assertEqual(res.status_code, HTTPStatus.OK)
123
124        self.assertTrue(Setup.get())
125        user = User.objects.get(username="akadmin")
126        self.assertTrue(user.check_password(pw))
127
128    @patch_flag(Setup, False)
129    @apply_blueprint("default/flow-oobe.yaml")
130    @apply_blueprint("system/bootstrap.yaml")
131    def test_setup_flow_direct(self):
132        """Test setup flow, directly accessing the flow"""
133        res = self.client.get(
134            reverse("authentik_api:flow-executor", kwargs={"flow_slug": "initial-setup"})
135        )
136        self.assertStageResponse(
137            res,
138            component="ak-stage-access-denied",
139            error_message="Access the authentik setup by navigating to http://testserver/",
140        )
141
142    def test_setup_bootstrap_env(self):
143        """Test setup with env vars"""
144        User.objects.filter(username="akadmin").delete()
145        Setup.set(False)
146
147        environ["AUTHENTIK_BOOTSTRAP_PASSWORD"] = generate_id()
148        environ["AUTHENTIK_BOOTSTRAP_TOKEN"] = generate_id()
149        pre_startup.send(sender=self)
150        post_startup.send(sender=self)
151
152        self.assertTrue(Setup.get())
153        user = User.objects.get(username="akadmin")
154        self.assertTrue(user.check_password(environ["AUTHENTIK_BOOTSTRAP_PASSWORD"]))
155
156        token = Token.objects.filter(identifier="authentik-bootstrap-token").first()
157        self.assertEqual(token.intent, TokenIntents.INTENT_API)
158        self.assertEqual(token.key, environ["AUTHENTIK_BOOTSTRAP_TOKEN"])
159
160    def test_setup_bootstrap_env_password_hash(self):
161        """Test setup with password hash env var"""
162        User.objects.filter(username="akadmin").delete()
163        Setup.set(False)
164
165        password = generate_id()
166        password_hash = make_password(password)
167        environ["AUTHENTIK_BOOTSTRAP_PASSWORD_HASH"] = password_hash
168        pre_startup.send(sender=self)
169        post_startup.send(sender=self)
170
171        self.assertTrue(Setup.get())
172        user = User.objects.get(username="akadmin")
173        self.assertEqual(user.password, password_hash)
174        self.assertTrue(user.check_password(password))

class TestSetup(authentik.flows.tests.FlowTestCase): View Source

 18class TestSetup(FlowTestCase):
 19    def tearDown(self):
 20        environ.pop("AUTHENTIK_BOOTSTRAP_PASSWORD", None)
 21        environ.pop("AUTHENTIK_BOOTSTRAP_PASSWORD_HASH", None)
 22        environ.pop("AUTHENTIK_BOOTSTRAP_TOKEN", None)
 23
 24    @patch_flag(Setup, True)
 25    def test_setup(self):
 26        """Test existing instance"""
 27        res = self.client.get(reverse("authentik_core:root-redirect"))
 28        self.assertEqual(res.status_code, HTTPStatus.FOUND)
 29        self.assertRedirects(
 30            res,
 31            reverse("authentik_flows:default-authentication") + "?next=/",
 32            fetch_redirect_response=False,
 33        )
 34
 35        res = self.client.head(reverse("authentik_core:setup"))
 36        self.assertEqual(res.status_code, HTTPStatus.SERVICE_UNAVAILABLE)
 37
 38        res = self.client.get(reverse("authentik_core:setup"))
 39        self.assertEqual(res.status_code, HTTPStatus.FOUND)
 40        self.assertRedirects(
 41            res,
 42            reverse("authentik_core:root-redirect"),
 43            fetch_redirect_response=False,
 44        )
 45
 46    @patch_flag(Setup, False)
 47    def test_not_setup_no_flow(self):
 48        """Test case on initial startup; setup flag is not set and oobe flow does
 49        not exist yet"""
 50        Flow.objects.filter(slug="initial-setup").delete()
 51        res = self.client.get(reverse("authentik_core:root-redirect"))
 52        self.assertEqual(res.status_code, HTTPStatus.FOUND)
 53        self.assertRedirects(res, reverse("authentik_core:setup"), fetch_redirect_response=False)
 54        # Flow does not exist, hence 503
 55        res = self.client.get(reverse("authentik_core:setup"))
 56        self.assertEqual(res.status_code, HTTPStatus.SERVICE_UNAVAILABLE)
 57        res = self.client.head(reverse("authentik_core:setup"))
 58        self.assertEqual(res.status_code, HTTPStatus.SERVICE_UNAVAILABLE)
 59
 60    @patch_flag(Setup, False)
 61    @apply_blueprint("default/flow-oobe.yaml")
 62    def test_not_setup(self):
 63        """Test case for when worker comes up, and has created flow"""
 64        res = self.client.get(reverse("authentik_core:root-redirect"))
 65        self.assertEqual(res.status_code, HTTPStatus.FOUND)
 66        self.assertRedirects(res, reverse("authentik_core:setup"), fetch_redirect_response=False)
 67        # Flow does not exist, hence 503
 68        res = self.client.head(reverse("authentik_core:setup"))
 69        self.assertEqual(res.status_code, HTTPStatus.OK)
 70        res = self.client.get(reverse("authentik_core:setup"))
 71        self.assertEqual(res.status_code, HTTPStatus.FOUND)
 72        self.assertRedirects(
 73            res,
 74            reverse("authentik_core:if-flow", kwargs={"flow_slug": "initial-setup"}),
 75            fetch_redirect_response=False,
 76        )
 77
 78    @apply_blueprint("default/flow-oobe.yaml")
 79    @apply_blueprint("system/bootstrap.yaml")
 80    def test_setup_flow_full(self):
 81        """Test full setup flow"""
 82        Setup.set(False)
 83
 84        res = self.client.get(reverse("authentik_core:setup"))
 85        self.assertEqual(res.status_code, HTTPStatus.FOUND)
 86        self.assertRedirects(
 87            res,
 88            reverse("authentik_core:if-flow", kwargs={"flow_slug": "initial-setup"}),
 89            fetch_redirect_response=False,
 90        )
 91
 92        res = self.client.get(
 93            reverse("authentik_api:flow-executor", kwargs={"flow_slug": "initial-setup"}),
 94        )
 95        self.assertEqual(res.status_code, HTTPStatus.OK)
 96        self.assertStageResponse(res, component="ak-stage-prompt")
 97
 98        pw = generate_id()
 99        res = self.client.post(
100            reverse("authentik_api:flow-executor", kwargs={"flow_slug": "initial-setup"}),
101            {
102                "email": f"{generate_id()}@t.goauthentik.io",
103                "password": pw,
104                "password_repeat": pw,
105                "component": "ak-stage-prompt",
106            },
107        )
108        self.assertEqual(res.status_code, HTTPStatus.FOUND)
109
110        res = self.client.get(
111            reverse("authentik_api:flow-executor", kwargs={"flow_slug": "initial-setup"}),
112        )
113        self.assertEqual(res.status_code, HTTPStatus.FOUND)
114
115        res = self.client.get(
116            reverse("authentik_api:flow-executor", kwargs={"flow_slug": "initial-setup"}),
117        )
118        self.assertEqual(res.status_code, HTTPStatus.FOUND)
119
120        res = self.client.get(
121            reverse("authentik_api:flow-executor", kwargs={"flow_slug": "initial-setup"}),
122        )
123        self.assertEqual(res.status_code, HTTPStatus.OK)
124
125        self.assertTrue(Setup.get())
126        user = User.objects.get(username="akadmin")
127        self.assertTrue(user.check_password(pw))
128
129    @patch_flag(Setup, False)
130    @apply_blueprint("default/flow-oobe.yaml")
131    @apply_blueprint("system/bootstrap.yaml")
132    def test_setup_flow_direct(self):
133        """Test setup flow, directly accessing the flow"""
134        res = self.client.get(
135            reverse("authentik_api:flow-executor", kwargs={"flow_slug": "initial-setup"})
136        )
137        self.assertStageResponse(
138            res,
139            component="ak-stage-access-denied",
140            error_message="Access the authentik setup by navigating to http://testserver/",
141        )
142
143    def test_setup_bootstrap_env(self):
144        """Test setup with env vars"""
145        User.objects.filter(username="akadmin").delete()
146        Setup.set(False)
147
148        environ["AUTHENTIK_BOOTSTRAP_PASSWORD"] = generate_id()
149        environ["AUTHENTIK_BOOTSTRAP_TOKEN"] = generate_id()
150        pre_startup.send(sender=self)
151        post_startup.send(sender=self)
152
153        self.assertTrue(Setup.get())
154        user = User.objects.get(username="akadmin")
155        self.assertTrue(user.check_password(environ["AUTHENTIK_BOOTSTRAP_PASSWORD"]))
156
157        token = Token.objects.filter(identifier="authentik-bootstrap-token").first()
158        self.assertEqual(token.intent, TokenIntents.INTENT_API)
159        self.assertEqual(token.key, environ["AUTHENTIK_BOOTSTRAP_TOKEN"])
160
161    def test_setup_bootstrap_env_password_hash(self):
162        """Test setup with password hash env var"""
163        User.objects.filter(username="akadmin").delete()
164        Setup.set(False)
165
166        password = generate_id()
167        password_hash = make_password(password)
168        environ["AUTHENTIK_BOOTSTRAP_PASSWORD_HASH"] = password_hash
169        pre_startup.send(sender=self)
170        post_startup.send(sender=self)
171
172        self.assertTrue(Setup.get())
173        user = User.objects.get(username="akadmin")
174        self.assertEqual(user.password, password_hash)
175        self.assertTrue(user.check_password(password))

Helpers for testing flows and stages.

def tearDown(self): View Source

19    def tearDown(self):
20        environ.pop("AUTHENTIK_BOOTSTRAP_PASSWORD", None)
21        environ.pop("AUTHENTIK_BOOTSTRAP_PASSWORD_HASH", None)
22        environ.pop("AUTHENTIK_BOOTSTRAP_TOKEN", None)

Hook method for deconstructing the test fixture after testing it.

@patch_flag(Setup, True)

def test_setup(self): View Source

24    @patch_flag(Setup, True)
25    def test_setup(self):
26        """Test existing instance"""
27        res = self.client.get(reverse("authentik_core:root-redirect"))
28        self.assertEqual(res.status_code, HTTPStatus.FOUND)
29        self.assertRedirects(
30            res,
31            reverse("authentik_flows:default-authentication") + "?next=/",
32            fetch_redirect_response=False,
33        )
34
35        res = self.client.head(reverse("authentik_core:setup"))
36        self.assertEqual(res.status_code, HTTPStatus.SERVICE_UNAVAILABLE)
37
38        res = self.client.get(reverse("authentik_core:setup"))
39        self.assertEqual(res.status_code, HTTPStatus.FOUND)
40        self.assertRedirects(
41            res,
42            reverse("authentik_core:root-redirect"),
43            fetch_redirect_response=False,
44        )

Test existing instance

@patch_flag(Setup, False)

def test_not_setup_no_flow(self): View Source

46    @patch_flag(Setup, False)
47    def test_not_setup_no_flow(self):
48        """Test case on initial startup; setup flag is not set and oobe flow does
49        not exist yet"""
50        Flow.objects.filter(slug="initial-setup").delete()
51        res = self.client.get(reverse("authentik_core:root-redirect"))
52        self.assertEqual(res.status_code, HTTPStatus.FOUND)
53        self.assertRedirects(res, reverse("authentik_core:setup"), fetch_redirect_response=False)
54        # Flow does not exist, hence 503
55        res = self.client.get(reverse("authentik_core:setup"))
56        self.assertEqual(res.status_code, HTTPStatus.SERVICE_UNAVAILABLE)
57        res = self.client.head(reverse("authentik_core:setup"))
58        self.assertEqual(res.status_code, HTTPStatus.SERVICE_UNAVAILABLE)

Test case on initial startup; setup flag is not set and oobe flow does not exist yet

@patch_flag(Setup, False)

@apply_blueprint('default/flow-oobe.yaml')

def test_not_setup(self): View Source

60    @patch_flag(Setup, False)
61    @apply_blueprint("default/flow-oobe.yaml")
62    def test_not_setup(self):
63        """Test case for when worker comes up, and has created flow"""
64        res = self.client.get(reverse("authentik_core:root-redirect"))
65        self.assertEqual(res.status_code, HTTPStatus.FOUND)
66        self.assertRedirects(res, reverse("authentik_core:setup"), fetch_redirect_response=False)
67        # Flow does not exist, hence 503
68        res = self.client.head(reverse("authentik_core:setup"))
69        self.assertEqual(res.status_code, HTTPStatus.OK)
70        res = self.client.get(reverse("authentik_core:setup"))
71        self.assertEqual(res.status_code, HTTPStatus.FOUND)
72        self.assertRedirects(
73            res,
74            reverse("authentik_core:if-flow", kwargs={"flow_slug": "initial-setup"}),
75            fetch_redirect_response=False,
76        )

Test case for when worker comes up, and has created flow

@apply_blueprint('default/flow-oobe.yaml')

@apply_blueprint('system/bootstrap.yaml')

def test_setup_flow_full(self): View Source

 78    @apply_blueprint("default/flow-oobe.yaml")
 79    @apply_blueprint("system/bootstrap.yaml")
 80    def test_setup_flow_full(self):
 81        """Test full setup flow"""
 82        Setup.set(False)
 83
 84        res = self.client.get(reverse("authentik_core:setup"))
 85        self.assertEqual(res.status_code, HTTPStatus.FOUND)
 86        self.assertRedirects(
 87            res,
 88            reverse("authentik_core:if-flow", kwargs={"flow_slug": "initial-setup"}),
 89            fetch_redirect_response=False,
 90        )
 91
 92        res = self.client.get(
 93            reverse("authentik_api:flow-executor", kwargs={"flow_slug": "initial-setup"}),
 94        )
 95        self.assertEqual(res.status_code, HTTPStatus.OK)
 96        self.assertStageResponse(res, component="ak-stage-prompt")
 97
 98        pw = generate_id()
 99        res = self.client.post(
100            reverse("authentik_api:flow-executor", kwargs={"flow_slug": "initial-setup"}),
101            {
102                "email": f"{generate_id()}@t.goauthentik.io",
103                "password": pw,
104                "password_repeat": pw,
105                "component": "ak-stage-prompt",
106            },
107        )
108        self.assertEqual(res.status_code, HTTPStatus.FOUND)
109
110        res = self.client.get(
111            reverse("authentik_api:flow-executor", kwargs={"flow_slug": "initial-setup"}),
112        )
113        self.assertEqual(res.status_code, HTTPStatus.FOUND)
114
115        res = self.client.get(
116            reverse("authentik_api:flow-executor", kwargs={"flow_slug": "initial-setup"}),
117        )
118        self.assertEqual(res.status_code, HTTPStatus.FOUND)
119
120        res = self.client.get(
121            reverse("authentik_api:flow-executor", kwargs={"flow_slug": "initial-setup"}),
122        )
123        self.assertEqual(res.status_code, HTTPStatus.OK)
124
125        self.assertTrue(Setup.get())
126        user = User.objects.get(username="akadmin")
127        self.assertTrue(user.check_password(pw))

Test full setup flow

@patch_flag(Setup, False)

@apply_blueprint('default/flow-oobe.yaml')

@apply_blueprint('system/bootstrap.yaml')

def test_setup_flow_direct(self): View Source

129    @patch_flag(Setup, False)
130    @apply_blueprint("default/flow-oobe.yaml")
131    @apply_blueprint("system/bootstrap.yaml")
132    def test_setup_flow_direct(self):
133        """Test setup flow, directly accessing the flow"""
134        res = self.client.get(
135            reverse("authentik_api:flow-executor", kwargs={"flow_slug": "initial-setup"})
136        )
137        self.assertStageResponse(
138            res,
139            component="ak-stage-access-denied",
140            error_message="Access the authentik setup by navigating to http://testserver/",
141        )

Test setup flow, directly accessing the flow

def test_setup_bootstrap_env(self): View Source

143    def test_setup_bootstrap_env(self):
144        """Test setup with env vars"""
145        User.objects.filter(username="akadmin").delete()
146        Setup.set(False)
147
148        environ["AUTHENTIK_BOOTSTRAP_PASSWORD"] = generate_id()
149        environ["AUTHENTIK_BOOTSTRAP_TOKEN"] = generate_id()
150        pre_startup.send(sender=self)
151        post_startup.send(sender=self)
152
153        self.assertTrue(Setup.get())
154        user = User.objects.get(username="akadmin")
155        self.assertTrue(user.check_password(environ["AUTHENTIK_BOOTSTRAP_PASSWORD"]))
156
157        token = Token.objects.filter(identifier="authentik-bootstrap-token").first()
158        self.assertEqual(token.intent, TokenIntents.INTENT_API)
159        self.assertEqual(token.key, environ["AUTHENTIK_BOOTSTRAP_TOKEN"])

Test setup with env vars

def test_setup_bootstrap_env_password_hash(self): View Source

161    def test_setup_bootstrap_env_password_hash(self):
162        """Test setup with password hash env var"""
163        User.objects.filter(username="akadmin").delete()
164        Setup.set(False)
165
166        password = generate_id()
167        password_hash = make_password(password)
168        environ["AUTHENTIK_BOOTSTRAP_PASSWORD_HASH"] = password_hash
169        pre_startup.send(sender=self)
170        post_startup.send(sender=self)
171
172        self.assertTrue(Setup.get())
173        user = User.objects.get(username="akadmin")
174        self.assertEqual(user.password, password_hash)
175        self.assertTrue(user.check_password(password))

Test setup with password hash env var

Inherited Members

authentik.flows.tests.FlowTestCase: assertStageResponse; get_flow_plan; set_flow_plan; assertStageRedirects; assertFlowFinishes