diff --git a/authentik/events/utils.py b/authentik/events/utils.py index 29974c693..a8d288cfc 100644 --- a/authentik/events/utils.py +++ b/authentik/events/utils.py @@ -85,7 +85,7 @@ def sanitize_dict(source: dict[Any, Any]) -> dict[Any, Any]: value = asdict(value) if isinstance(value, dict): final_dict[key] = sanitize_dict(value) - elif isinstance(value, User): + elif isinstance(value, (User, AnonymousUser)): final_dict[key] = sanitize_dict(get_user(value)) elif isinstance(value, models.Model): final_dict[key] = sanitize_dict(model_to_dict(value)) diff --git a/authentik/policies/tests/test_process.py b/authentik/policies/tests/test_process.py index 84b616471..9c6505618 100644 --- a/authentik/policies/tests/test_process.py +++ b/authentik/policies/tests/test_process.py @@ -1,4 +1,5 @@ """policy process tests""" +from django.contrib.auth.models import AnonymousUser from django.core.cache import cache from django.test import RequestFactory, TestCase from guardian.shortcuts import get_anonymous_user @@ -149,6 +150,38 @@ class TestPolicyProcess(TestCase): self.assertEqual(event.context["result"]["messages"], ["dummy"]) self.assertEqual(event.client_ip, "127.0.0.1") + def test_execution_logging_anonymous(self): + """Test policy execution creates event with anonymous user""" + policy = DummyPolicy.objects.create( + result=False, wait_min=0, wait_max=1, execution_logging=True + ) + binding = PolicyBinding( + policy=policy, target=Application.objects.create(name="test") + ) + + user = AnonymousUser() + + http_request = self.factory.get("/") + http_request.user = user + + request = PolicyRequest(user) + request.set_http_request(http_request) + response = PolicyProcess(binding, request, None).execute() + self.assertEqual(response.passing, False) + self.assertEqual(response.messages, ("dummy",)) + + events = Event.objects.filter( + action=EventAction.POLICY_EXECUTION, + context__policy_uuid=policy.policy_uuid.hex, + ) + self.assertTrue(events.exists()) + self.assertEqual(len(events), 1) + event = events.first() + self.assertEqual(event.user["username"], "AnonymousUser") + self.assertEqual(event.context["result"]["passing"], False) + self.assertEqual(event.context["result"]["messages"], ["dummy"]) + self.assertEqual(event.client_ip, "127.0.0.1") + def test_raises(self): """Test policy that raises error""" policy_raises = ExpressionPolicy.objects.create(