authentik.blueprints.tests.test_v1

  1"""Test blueprints v1"""
  2
  3from os import chmod, environ, unlink, write
  4from tempfile import mkstemp
  5
  6from django.test import TransactionTestCase
  7
  8from authentik.blueprints.tests import apply_blueprint
  9from authentik.blueprints.v1.exporter import FlowExporter
 10from authentik.blueprints.v1.importer import Importer, transaction_rollback
 11from authentik.core.models import Group
 12from authentik.flows.models import Flow, FlowDesignation, FlowStageBinding
 13from authentik.lib.generators import generate_id
 14from authentik.lib.tests.utils import load_fixture
 15from authentik.policies.expression.models import ExpressionPolicy
 16from authentik.policies.models import PolicyBinding
 17from authentik.sources.oauth.models import OAuthSource
 18from authentik.stages.prompt.models import FieldTypes, Prompt, PromptStage
 19from authentik.stages.user_login.models import UserLoginStage
 20
 21
 22class TestBlueprintsV1(TransactionTestCase):
 23    """Test Blueprints"""
 24
 25    def test_blueprint_invalid_format(self):
 26        """Test blueprint with invalid format"""
 27        importer = Importer.from_string('{"version": 3}')
 28        self.assertFalse(importer.validate()[0])
 29        importer = Importer.from_string(
 30            '{"version": 1,"entries":[{"identifiers":{},"attrs":{},'
 31            '"model": "authentik_core.User"}]}'
 32        )
 33        self.assertFalse(importer.validate()[0])
 34        importer = Importer.from_string(
 35            '{"version": 1, "entries": [{"attrs": {"name": "test"}, '
 36            '"identifiers": {}, '
 37            '"model": "authentik_core.Group"}]}'
 38        )
 39        self.assertFalse(importer.validate()[0])
 40
 41    def test_validated_import_dict_identifiers(self):
 42        """Test importing blueprints with dict identifiers."""
 43        Group.objects.filter(name__istartswith="test").delete()
 44
 45        Group.objects.create(
 46            name="test1",
 47            attributes={
 48                "key": ["value"],
 49                "other_key": ["a_value", "other_value"],
 50            },
 51        )
 52        Group.objects.create(
 53            name="test2",
 54            attributes={
 55                "key": ["value"],
 56                "other_key": ["diff_value", "other_diff_value"],
 57            },
 58        )
 59
 60        importer = Importer.from_string(
 61            '{"version": 1, "entries": [{"attrs": {"name": "test999", "attributes": '
 62            '{"key": ["updated_value"]}}, "identifiers": {"attributes": {"other_key": '
 63            '["other_value"]}}, "model": "authentik_core.Group"}]}'
 64        )
 65        self.assertTrue(importer.validate()[0])
 66        self.assertTrue(importer.apply())
 67        self.assertTrue(
 68            Group.objects.filter(
 69                name="test2",
 70                attributes={
 71                    "key": ["value"],
 72                    "other_key": ["diff_value", "other_diff_value"],
 73                },
 74            )
 75        )
 76        self.assertTrue(
 77            Group.objects.filter(
 78                name="test999",
 79                # All attributes used as identifiers are kept and merged with the
 80                # new attributes declared in the blueprint
 81                attributes={"key": ["updated_value"], "other_key": ["other_value"]},
 82            )
 83        )
 84        self.assertFalse(Group.objects.filter(name="test1"))
 85
 86    def test_export_validate_import(self):
 87        """Test export and validate it"""
 88        flow_slug = generate_id()
 89        with transaction_rollback():
 90            login_stage = UserLoginStage.objects.create(name=generate_id())
 91
 92            flow = Flow.objects.create(
 93                slug=flow_slug,
 94                designation=FlowDesignation.AUTHENTICATION,
 95                name=generate_id(),
 96                title=generate_id(),
 97            )
 98            FlowStageBinding.objects.update_or_create(
 99                target=flow,
100                stage=login_stage,
101                order=0,
102            )
103
104            exporter = FlowExporter(flow)
105            export = exporter.export()
106            self.assertEqual(len(export.entries), 3)
107            export_yaml = exporter.export_to_string()
108
109        importer = Importer.from_string(export_yaml)
110        self.assertTrue(importer.validate()[0])
111        self.assertTrue(importer.apply())
112
113        self.assertTrue(Flow.objects.filter(slug=flow_slug).exists())
114
115    def test_export_validate_import_re_import(self):
116        """Test export and import it twice"""
117        count_initial = Prompt.objects.filter(field_key="username").count()
118
119        importer = Importer.from_string(load_fixture("fixtures/static_prompt_export.yaml"))
120        self.assertTrue(importer.validate()[0])
121        self.assertTrue(importer.apply())
122
123        count_before = Prompt.objects.filter(field_key="username").count()
124        self.assertEqual(count_initial + 1, count_before)
125
126        importer = Importer.from_string(load_fixture("fixtures/static_prompt_export.yaml"))
127        self.assertTrue(importer.apply())
128
129        self.assertEqual(Prompt.objects.filter(field_key="username").count(), count_before)
130
131    @apply_blueprint("system/providers-oauth2.yaml")
132    def test_import_yaml_tags(self):
133        """Test some yaml tags"""
134        ExpressionPolicy.objects.filter(name="foo-bar-baz-qux").delete()
135        Group.objects.filter(name="test").delete()
136        environ["foo"] = generate_id()
137        file, file_name = mkstemp()
138        write(file, b"foo")
139        _, file_default_name = mkstemp()
140        chmod(file_default_name, 0o000)  # Remove all permissions so we can't read the file
141        importer = Importer.from_string(
142            load_fixture(
143                "fixtures/tags.yaml",
144                file_name=file_name,
145                file_default_name=file_default_name,
146            ),
147            {"bar": "baz"},
148        )
149        self.assertTrue(importer.validate()[0])
150        self.assertTrue(importer.apply())
151        policy = ExpressionPolicy.objects.filter(name="foo-bar-baz-qux").first()
152        self.assertTrue(policy)
153        group = Group.objects.filter(name="test").first()
154        self.assertIsNotNone(group)
155        self.assertEqual(
156            group.attributes,
157            {
158                "policy_pk1": str(policy.pk) + "-suffix",
159                "policy_pk2": str(policy.pk) + "-suffix",
160                "boolAnd": True,
161                "boolNand": False,
162                "boolOr": True,
163                "boolNor": False,
164                "boolXor": True,
165                "boolXnor": False,
166                "boolComplex": True,
167                "if_true_complex": {
168                    "dictionary": {
169                        "with": {"keys": "and_values"},
170                        "and_nested_custom_tags": "foo-bar",
171                    }
172                },
173                "if_false_complex": ["list", "with", "items", "foo-bar"],
174                "if_true_simple": True,
175                "if_short": True,
176                "if_false_simple": 2,
177                "enumerate_mapping_to_mapping": {
178                    "prefix-key1": "other-prefix-value",
179                    "prefix-key2": "other-prefix-2",
180                },
181                "enumerate_mapping_to_sequence": [
182                    "prefixed-pair-key1-value",
183                    "prefixed-pair-key2-2",
184                ],
185                "enumerate_sequence_to_sequence": [
186                    "prefixed-items-0-foo",
187                    "prefixed-items-1-bar",
188                ],
189                "enumerate_sequence_to_mapping": {"index: 0": "foo", "index: 1": "bar"},
190                "nested_complex_enumeration": {
191                    "0": {
192                        "key1": [
193                            ["prefixed-f", "prefixed-o", "prefixed-o"],
194                            {
195                                "outer_value": "foo",
196                                "outer_index": 0,
197                                "middle_value": "value",
198                                "middle_index": "key1",
199                            },
200                        ],
201                        "key2": [
202                            ["prefixed-f", "prefixed-o", "prefixed-o"],
203                            {
204                                "outer_value": "foo",
205                                "outer_index": 0,
206                                "middle_value": 2,
207                                "middle_index": "key2",
208                            },
209                        ],
210                    },
211                    "1": {
212                        "key1": [
213                            ["prefixed-b", "prefixed-a", "prefixed-r"],
214                            {
215                                "outer_value": "bar",
216                                "outer_index": 1,
217                                "middle_value": "value",
218                                "middle_index": "key1",
219                            },
220                        ],
221                        "key2": [
222                            ["prefixed-b", "prefixed-a", "prefixed-r"],
223                            {
224                                "outer_value": "bar",
225                                "outer_index": 1,
226                                "middle_value": 2,
227                                "middle_index": "key2",
228                            },
229                        ],
230                    },
231                },
232                "nested_context": "context-nested-value",
233                "env_null": None,
234                "file_content": "foo",
235                "file_default": "default",
236                "file_non_existent": None,
237                "json_parse": {"foo": "bar"},
238                "at_index_sequence": "foo",
239                "at_index_sequence_default": "non existent",
240                "at_index_mapping": 2,
241                "at_index_mapping_default": "non existent",
242                "find_object": "goauthentik.io/providers/oauth2/scope-openid",
243            },
244        )
245        self.assertTrue(
246            OAuthSource.objects.filter(
247                slug="test",
248                consumer_key=environ["foo"],
249            )
250        )
251        unlink(file_name)
252        unlink(file_default_name)
253
254    def test_export_validate_import_policies(self):
255        """Test export and validate it"""
256        flow_slug = generate_id()
257        stage_name = generate_id()
258        with transaction_rollback():
259            flow_policy = ExpressionPolicy.objects.create(
260                name=generate_id(),
261                expression="return True",
262            )
263            flow = Flow.objects.create(
264                slug=flow_slug,
265                designation=FlowDesignation.AUTHENTICATION,
266                name=generate_id(),
267                title=generate_id(),
268            )
269            PolicyBinding.objects.create(policy=flow_policy, target=flow, order=0)
270
271            user_login = UserLoginStage.objects.create(name=stage_name)
272            fsb = FlowStageBinding.objects.create(target=flow, stage=user_login, order=0)
273            PolicyBinding.objects.create(policy=flow_policy, target=fsb, order=0)
274
275            exporter = FlowExporter(flow)
276            export_yaml = exporter.export_to_string()
277
278        importer = Importer.from_string(export_yaml)
279        self.assertTrue(importer.validate()[0])
280        self.assertTrue(importer.apply())
281        self.assertTrue(UserLoginStage.objects.filter(name=stage_name).exists())
282        self.assertTrue(Flow.objects.filter(slug=flow_slug).exists())
283
284    def test_export_validate_import_prompt(self):
285        """Test export and validate it"""
286        with transaction_rollback():
287            # First stage fields
288            username_prompt = Prompt.objects.create(
289                name=generate_id(),
290                field_key="username",
291                label="Username",
292                order=0,
293                type=FieldTypes.TEXT,
294            )
295            password = Prompt.objects.create(
296                name=generate_id(),
297                field_key="password",
298                label="Password",
299                order=1,
300                type=FieldTypes.PASSWORD,
301            )
302            password_repeat = Prompt.objects.create(
303                name=generate_id(),
304                field_key="password_repeat",
305                label="Password (repeat)",
306                order=2,
307                type=FieldTypes.PASSWORD,
308            )
309
310            # Stages
311            first_stage = PromptStage.objects.create(name=generate_id())
312            first_stage.fields.set([username_prompt, password, password_repeat])
313            first_stage.save()
314
315            flow = Flow.objects.create(
316                name=generate_id(),
317                slug=generate_id(),
318                designation=FlowDesignation.ENROLLMENT,
319                title=generate_id(),
320            )
321
322            FlowStageBinding.objects.create(target=flow, stage=first_stage, order=0)
323
324            exporter = FlowExporter(flow)
325            export_yaml = exporter.export_to_string()
326
327        importer = Importer.from_string(export_yaml)
328
329        self.assertTrue(importer.validate()[0])
330        self.assertTrue(importer.apply())