authentik.policies.tests.test_process

  1"""policy process tests"""
  2
  3from django.contrib.auth.models import AnonymousUser
  4from django.core.cache import cache
  5from django.test import RequestFactory, TestCase
  6from django.urls import resolve, reverse
  7from django.views.debug import SafeExceptionReporterFilter
  8from guardian.shortcuts import get_anonymous_user
  9
 10from authentik.core.models import Application, Group, User
 11from authentik.events.models import Event, EventAction
 12from authentik.lib.generators import generate_id
 13from authentik.policies.dummy.models import DummyPolicy
 14from authentik.policies.expression.models import ExpressionPolicy
 15from authentik.policies.models import Policy, PolicyBinding
 16from authentik.policies.process import PolicyProcess
 17from authentik.policies.types import CACHE_PREFIX, PolicyRequest
 18
 19
 20def clear_policy_cache():
 21    """Ensure no policy-related keys are still cached"""
 22    keys = cache.keys(f"{CACHE_PREFIX}*")
 23    cache.delete(keys)
 24
 25
 26class TestPolicyProcess(TestCase):
 27    """Policy Process tests"""
 28
 29    def setUp(self):
 30        clear_policy_cache()
 31        self.factory = RequestFactory()
 32        self.user = User.objects.create_user(username=generate_id())
 33
 34    def test_group_passing(self):
 35        """Test binding to group"""
 36        group = Group.objects.create(name=generate_id())
 37        group.users.add(self.user)
 38        binding = PolicyBinding(group=group)
 39
 40        request = PolicyRequest(self.user)
 41        response = PolicyProcess(binding, request, None).execute()
 42        self.assertEqual(response.passing, True)
 43
 44    def test_group_negative(self):
 45        """Test binding to group"""
 46        group = Group.objects.create(name=generate_id())
 47        binding = PolicyBinding(group=group)
 48
 49        request = PolicyRequest(self.user)
 50        response = PolicyProcess(binding, request, None).execute()
 51        self.assertEqual(response.passing, False)
 52
 53    def test_user_passing(self):
 54        """Test binding to user"""
 55        binding = PolicyBinding(user=self.user)
 56
 57        request = PolicyRequest(self.user)
 58        response = PolicyProcess(binding, request, None).execute()
 59        self.assertEqual(response.passing, True)
 60
 61    def test_user_negative(self):
 62        """Test binding to user"""
 63        binding = PolicyBinding(user=get_anonymous_user())
 64
 65        request = PolicyRequest(self.user)
 66        response = PolicyProcess(binding, request, None).execute()
 67        self.assertEqual(response.passing, False)
 68
 69    def test_empty(self):
 70        """Test binding to user"""
 71        binding = PolicyBinding()
 72
 73        request = PolicyRequest(self.user)
 74        response = PolicyProcess(binding, request, None).execute()
 75        self.assertEqual(response.passing, False)
 76
 77    def test_invalid(self):
 78        """Test Process with invalid arguments"""
 79        policy = DummyPolicy.objects.create(result=True, wait_min=0, wait_max=1)
 80        binding = PolicyBinding(policy=policy)
 81        with self.assertRaises(ValueError):
 82            PolicyProcess(binding, None, None)  # type: ignore
 83
 84    def test_true(self):
 85        """Test policy execution"""
 86        policy = DummyPolicy.objects.create(result=True, wait_min=0, wait_max=1)
 87        binding = PolicyBinding(policy=policy)
 88
 89        request = PolicyRequest(self.user)
 90        response = PolicyProcess(binding, request, None).execute()
 91        self.assertEqual(response.passing, True)
 92        self.assertEqual(response.messages, ("dummy",))
 93
 94    def test_false(self):
 95        """Test policy execution"""
 96        policy = DummyPolicy.objects.create(result=False, wait_min=0, wait_max=1)
 97        binding = PolicyBinding(policy=policy)
 98
 99        request = PolicyRequest(self.user)
100        response = PolicyProcess(binding, request, None).execute()
101        self.assertEqual(response.passing, False)
102        self.assertEqual(response.messages, ("dummy",))
103
104    def test_negate(self):
105        """Test policy execution"""
106        policy = DummyPolicy.objects.create(result=False, wait_min=0, wait_max=1)
107        binding = PolicyBinding(policy=policy, negate=True)
108
109        request = PolicyRequest(self.user)
110        response = PolicyProcess(binding, request, None).execute()
111        self.assertEqual(response.passing, True)
112        self.assertEqual(response.messages, ("dummy",))
113
114    def test_exception(self):
115        """Test policy execution"""
116        policy = Policy.objects.create(name=generate_id())
117        binding = PolicyBinding(
118            policy=policy, target=Application.objects.create(name=generate_id())
119        )
120
121        request = PolicyRequest(self.user)
122        response = PolicyProcess(binding, request, None).execute()
123        self.assertEqual(response.passing, False)
124
125    def test_execution_logging(self):
126        """Test policy execution creates event"""
127        policy = DummyPolicy.objects.create(
128            name=generate_id(),
129            result=False,
130            wait_min=0,
131            wait_max=1,
132            execution_logging=True,
133        )
134        binding = PolicyBinding(
135            policy=policy, target=Application.objects.create(name=generate_id())
136        )
137
138        http_request = self.factory.get(reverse("authentik_api:user-impersonate-end"))
139        http_request.user = self.user
140        http_request.resolver_match = resolve(reverse("authentik_api:user-impersonate-end"))
141
142        password = generate_id()
143        request = PolicyRequest(self.user)
144        request.set_http_request(http_request)
145        request.context = {
146            "complex": {
147                "dict": {"foo": "bar"},
148                "list": ["foo", "bar"],
149                "tuple": ("foo", "bar"),
150                "set": {"foo", "bar"},
151                "password": password,
152            }
153        }
154        response = PolicyProcess(binding, request, None).execute()
155        self.assertEqual(response.passing, False)
156        self.assertEqual(response.messages, ("dummy",))
157
158        events = Event.objects.filter(
159            action=EventAction.POLICY_EXECUTION,
160            context__policy_uuid=policy.policy_uuid.hex,
161        )
162        self.assertTrue(events.exists())
163        self.assertEqual(len(events), 1)
164        event = events.first()
165        self.assertEqual(event.user["username"], self.user.username)
166        self.assertEqual(event.context["result"]["passing"], False)
167        self.assertEqual(event.context["result"]["messages"], ["dummy"])
168        self.assertEqual(event.client_ip, "127.0.0.1")
169        # Python sets don't preserve order when converted to list,
170        # so ensure we sort the converted set
171        event.context["request"]["context"]["complex"]["set"].sort()
172        self.assertEqual(
173            event.context["request"]["context"],
174            {
175                "complex": {
176                    "set": [
177                        "bar",
178                        "foo",
179                    ],
180                    "dict": {"foo": "bar"},
181                    "list": ["foo", "bar"],
182                    "tuple": ["foo", "bar"],
183                    "password": SafeExceptionReporterFilter.cleansed_substitute,
184                }
185            },
186        )
187
188    def test_execution_logging_anonymous(self):
189        """Test policy execution creates event with anonymous user"""
190        policy = DummyPolicy.objects.create(
191            name=generate_id(),
192            result=False,
193            wait_min=0,
194            wait_max=1,
195            execution_logging=True,
196        )
197        binding = PolicyBinding(
198            policy=policy, target=Application.objects.create(name=generate_id())
199        )
200
201        user = AnonymousUser()
202
203        http_request = self.factory.get("/")
204        http_request.user = user
205
206        request = PolicyRequest(user)
207        request.set_http_request(http_request)
208        response = PolicyProcess(binding, request, None).execute()
209        self.assertEqual(response.passing, False)
210        self.assertEqual(response.messages, ("dummy",))
211
212        events = Event.objects.filter(
213            action=EventAction.POLICY_EXECUTION,
214            context__policy_uuid=policy.policy_uuid.hex,
215        )
216        self.assertTrue(events.exists())
217        self.assertEqual(len(events), 1)
218        event = events.first()
219        self.assertEqual(event.user["username"], "AnonymousUser")
220        self.assertEqual(event.context["result"]["passing"], False)
221        self.assertEqual(event.context["result"]["messages"], ["dummy"])
222        self.assertEqual(event.client_ip, "127.0.0.1")
223
224    def test_raises(self):
225        """Test policy that raises error"""
226        policy_raises = ExpressionPolicy.objects.create(name=generate_id(), expression="{{ 0/0 }}")
227        binding = PolicyBinding(
228            policy=policy_raises, target=Application.objects.create(name=generate_id())
229        )
230
231        request = PolicyRequest(self.user)
232        response = PolicyProcess(binding, request, None).execute()
233        self.assertEqual(response.passing, False)
234        self.assertEqual(response.messages, ("division by zero",))
235
236        events = Event.objects.filter(
237            action=EventAction.POLICY_EXCEPTION,
238            context__policy_uuid=policy_raises.policy_uuid.hex,
239        )
240        self.assertTrue(events.exists())
241        self.assertEqual(len(events), 1)
242        event = events.first()
243        self.assertEqual(event.user["username"], self.user.username)
244        self.assertIn("Policy failed to execute", event.context["message"])