authentik.policies.tests.test_engine

  1"""policy engine tests"""
  2
  3from django.core.cache import cache
  4from django.db import connections
  5from django.test import TestCase
  6from django.test.utils import CaptureQueriesContext
  7
  8from authentik.core.models import Group
  9from authentik.core.tests.utils import create_test_user
 10from authentik.lib.generators import generate_id
 11from authentik.policies.dummy.models import DummyPolicy
 12from authentik.policies.engine import PolicyEngine
 13from authentik.policies.exceptions import PolicyEngineException
 14from authentik.policies.expression.models import ExpressionPolicy
 15from authentik.policies.models import Policy, PolicyBinding, PolicyBindingModel, PolicyEngineMode
 16from authentik.policies.tests.test_process import clear_policy_cache
 17from authentik.policies.types import CACHE_PREFIX
 18
 19
 20class TestPolicyEngine(TestCase):
 21    """PolicyEngine tests"""
 22
 23    def setUp(self):
 24        clear_policy_cache()
 25        self.user = create_test_user()
 26        self.policy_false = DummyPolicy.objects.create(
 27            name=generate_id(), result=False, wait_min=0, wait_max=1
 28        )
 29        self.policy_true = DummyPolicy.objects.create(
 30            name=generate_id(), result=True, wait_min=0, wait_max=1
 31        )
 32        self.policy_wrong_type = Policy.objects.create(name=generate_id())
 33        self.policy_raises = ExpressionPolicy.objects.create(
 34            name=generate_id(), expression="{{ 0/0 }}"
 35        )
 36        self.group_member = Group.objects.create(name=generate_id())
 37        self.user.groups.add(self.group_member)
 38        self.group_non_member = Group.objects.create(name=generate_id())
 39
 40    def test_engine_empty(self):
 41        """Ensure empty policy list passes"""
 42        pbm = PolicyBindingModel.objects.create()
 43        engine = PolicyEngine(pbm, self.user)
 44        result = engine.build().result
 45        self.assertEqual(result.passing, True)
 46        self.assertEqual(result.messages, ())
 47
 48    def test_engine_simple(self):
 49        """Ensure simplest use-case"""
 50        pbm = PolicyBindingModel.objects.create()
 51        PolicyBinding.objects.create(target=pbm, policy=self.policy_true, order=0)
 52        engine = PolicyEngine(pbm, self.user)
 53        result = engine.build().result
 54        self.assertEqual(result.passing, True)
 55        self.assertEqual(result.messages, ("dummy",))
 56
 57    def test_engine_mode_all_dyn(self):
 58        """Ensure all policies passes with AND mode (false and true -> false)"""
 59        pbm = PolicyBindingModel.objects.create(policy_engine_mode=PolicyEngineMode.MODE_ALL)
 60        PolicyBinding.objects.create(target=pbm, policy=self.policy_false, order=0)
 61        PolicyBinding.objects.create(target=pbm, policy=self.policy_true, order=1)
 62        engine = PolicyEngine(pbm, self.user)
 63        result = engine.build().result
 64        self.assertEqual(result.passing, False)
 65        self.assertEqual(
 66            result.messages,
 67            (
 68                "dummy",
 69                "dummy",
 70            ),
 71        )
 72
 73    def test_engine_mode_any_dyn(self):
 74        """Ensure all policies passes with OR mode (false and true -> true)"""
 75        pbm = PolicyBindingModel.objects.create(policy_engine_mode=PolicyEngineMode.MODE_ANY)
 76        PolicyBinding.objects.create(target=pbm, policy=self.policy_false, order=0)
 77        PolicyBinding.objects.create(target=pbm, policy=self.policy_true, order=1)
 78        engine = PolicyEngine(pbm, self.user)
 79        result = engine.build().result
 80        self.assertEqual(result.passing, True)
 81        self.assertEqual(
 82            result.messages,
 83            (
 84                "dummy",
 85                "dummy",
 86            ),
 87        )
 88
 89    def test_engine_mode_all_static(self):
 90        """Ensure all policies passes with OR mode (false and true -> true)"""
 91        pbm = PolicyBindingModel.objects.create(policy_engine_mode=PolicyEngineMode.MODE_ALL)
 92        PolicyBinding.objects.create(target=pbm, group=self.group_member, order=0)
 93        PolicyBinding.objects.create(target=pbm, group=self.group_non_member, order=1)
 94        engine = PolicyEngine(pbm, self.user)
 95        result = engine.build().result
 96        self.assertEqual(result.passing, False)
 97        self.assertEqual(result.messages, ())
 98
 99    def test_engine_mode_any_static(self):
100        """Ensure all policies passes with OR mode (false and true -> true)"""
101        pbm = PolicyBindingModel.objects.create(policy_engine_mode=PolicyEngineMode.MODE_ANY)
102        PolicyBinding.objects.create(target=pbm, group=self.group_member, order=0)
103        PolicyBinding.objects.create(target=pbm, group=self.group_non_member, order=1)
104        engine = PolicyEngine(pbm, self.user)
105        result = engine.build().result
106        self.assertEqual(result.passing, True)
107        self.assertEqual(result.messages, ())
108
109    def test_engine_negate(self):
110        """Test negate flag"""
111        pbm = PolicyBindingModel.objects.create()
112        PolicyBinding.objects.create(target=pbm, policy=self.policy_true, negate=True, order=0)
113        engine = PolicyEngine(pbm, self.user)
114        result = engine.build().result
115        self.assertEqual(result.passing, False)
116        self.assertEqual(result.messages, ("dummy",))
117
118    def test_engine_policy_error(self):
119        """Test policy raising an error flag"""
120        pbm = PolicyBindingModel.objects.create()
121        PolicyBinding.objects.create(target=pbm, policy=self.policy_raises, order=0)
122        engine = PolicyEngine(pbm, self.user)
123        result = engine.build().result
124        self.assertEqual(result.passing, False)
125        self.assertEqual(result.messages, ("division by zero",))
126
127    def test_engine_policy_error_failure(self):
128        """Test policy raising an error flag"""
129        pbm = PolicyBindingModel.objects.create()
130        PolicyBinding.objects.create(
131            target=pbm, policy=self.policy_raises, order=0, failure_result=True
132        )
133        engine = PolicyEngine(pbm, self.user)
134        result = engine.build().result
135        self.assertEqual(result.passing, True)
136        self.assertEqual(result.messages, ("division by zero",))
137
138    def test_engine_policy_type(self):
139        """Test invalid policy type"""
140        pbm = PolicyBindingModel.objects.create()
141        PolicyBinding.objects.create(target=pbm, policy=self.policy_wrong_type, order=0)
142        with self.assertRaises(PolicyEngineException):
143            engine = PolicyEngine(pbm, self.user)
144            engine.build()
145
146    def test_engine_cache(self):
147        """Ensure empty policy list passes"""
148        pbm = PolicyBindingModel.objects.create()
149        binding = PolicyBinding.objects.create(target=pbm, policy=self.policy_false, order=0)
150        engine = PolicyEngine(pbm, self.user)
151        self.assertEqual(len(cache.keys(f"{CACHE_PREFIX}{binding.policy_binding_uuid.hex}*")), 0)
152        self.assertEqual(engine.build().passing, False)
153        self.assertEqual(len(cache.keys(f"{CACHE_PREFIX}{binding.policy_binding_uuid.hex}*")), 1)
154        self.assertEqual(engine.build().passing, False)
155        self.assertEqual(len(cache.keys(f"{CACHE_PREFIX}{binding.policy_binding_uuid.hex}*")), 1)
156
157    def test_engine_static_bindings(self):
158        """Test static bindings"""
159        group_a = Group.objects.create(name=generate_id())
160        group_b = Group.objects.create(name=generate_id())
161        group_b.users.add(self.user)
162        user = create_test_user()
163
164        for case in [
165            {
166                "message": "Group, not member",
167                "binding_args": {"group": group_a},
168                "passing": False,
169            },
170            {
171                "message": "Group, member",
172                "binding_args": {"group": group_b},
173                "passing": True,
174            },
175            {
176                "message": "User, other",
177                "binding_args": {"user": user},
178                "passing": False,
179            },
180            {
181                "message": "User, same",
182                "binding_args": {"user": self.user},
183                "passing": True,
184            },
185        ]:
186            with self.subTest():
187                pbm = PolicyBindingModel.objects.create()
188                for x in range(1000):
189                    PolicyBinding.objects.create(target=pbm, order=x, **case["binding_args"])
190                engine = PolicyEngine(pbm, self.user)
191                engine.use_cache = False
192                with CaptureQueriesContext(connections["default"]) as ctx:
193                    engine.build()
194                self.assertLess(ctx.final_queries, 1000)
195                self.assertEqual(engine.result.passing, case["passing"])
196
197    def test_engine_group_complex(self):
198        """Test more complex group setups"""
199        group_a = Group.objects.create(name=generate_id())
200        group_b = Group.objects.create(name=generate_id())
201        group_b.parents.add(group_a)
202        user = create_test_user()
203        group_b.users.add(user)
204        pbm = PolicyBindingModel.objects.create()
205        PolicyBinding.objects.create(target=pbm, order=0, group=group_a)
206        engine = PolicyEngine(pbm, user)
207        engine.use_cache = False
208        with CaptureQueriesContext(connections["default"]) as ctx:
209            engine.build()
210        self.assertLess(ctx.final_queries, 1000)
211        self.assertTrue(engine.result.passing)