authentik.events.tests.test_event

  1"""event tests"""
  2
  3from urllib.parse import urlencode
  4
  5from django.contrib.contenttypes.models import ContentType
  6from django.test import RequestFactory, TestCase
  7from django.views.debug import SafeExceptionReporterFilter
  8from guardian.shortcuts import get_anonymous_user
  9
 10from authentik.brands.models import Brand
 11from authentik.core.models import Group, User
 12from authentik.core.tests.utils import create_test_user
 13from authentik.events.models import Event
 14from authentik.flows.planner import PLAN_CONTEXT_PENDING_USER, FlowPlan
 15from authentik.flows.views.executor import QS_QUERY, SESSION_KEY_PLAN
 16from authentik.lib.generators import generate_id
 17from authentik.policies.dummy.models import DummyPolicy
 18
 19
 20class TestEvents(TestCase):
 21    """Test Event"""
 22
 23    def setUp(self) -> None:
 24        self.factory = RequestFactory()
 25
 26    def test_new_with_model(self):
 27        """Create a new Event passing a model as kwarg"""
 28        test_model = Group.objects.create(name="test")
 29        event = Event.new("unittest", test={"model": test_model})
 30        event.save()  # We save to ensure nothing is un-saveable
 31        model_content_type = ContentType.objects.get_for_model(test_model)
 32        self.assertEqual(
 33            event.context.get("test").get("model").get("app"),
 34            model_content_type.app_label,
 35        )
 36
 37    def test_new_with_user(self):
 38        """Create a new Event passing a user as kwarg"""
 39        event = Event.new("unittest", test={"model": get_anonymous_user()})
 40        event.save()  # We save to ensure nothing is un-saveable
 41        self.assertEqual(
 42            event.context.get("test").get("model").get("username"),
 43            get_anonymous_user().username,
 44        )
 45
 46    def test_new_with_uuid_model(self):
 47        """Create a new Event passing a model (with UUID PK) as kwarg"""
 48        temp_model = DummyPolicy.objects.create(name="test", result=True)
 49        event = Event.new("unittest", model=temp_model)
 50        event.save()  # We save to ensure nothing is un-saveable
 51        model_content_type = ContentType.objects.get_for_model(temp_model)
 52        self.assertEqual(event.context.get("model").get("app"), model_content_type.app_label)
 53        self.assertEqual(event.context.get("model").get("pk"), temp_model.pk.hex)
 54
 55    def test_from_http_basic(self):
 56        """Test plain from_http"""
 57        event = Event.new("unittest").from_http(self.factory.get("/"))
 58        self.assertEqual(
 59            event.context,
 60            {
 61                "http_request": {
 62                    "args": {},
 63                    "method": "GET",
 64                    "path": "/",
 65                    "user_agent": "",
 66                }
 67            },
 68        )
 69
 70    def test_from_http_clean_querystring(self):
 71        """Test cleansing query string"""
 72        token = generate_id()
 73        request = self.factory.get(f"/?token={token}")
 74        event = Event.new("unittest").from_http(request)
 75        self.assertEqual(
 76            event.context,
 77            {
 78                "http_request": {
 79                    "args": {"token": SafeExceptionReporterFilter.cleansed_substitute},
 80                    "method": "GET",
 81                    "path": "/",
 82                    "user_agent": "",
 83                }
 84            },
 85        )
 86
 87    def test_from_http_clean_querystring_flow(self):
 88        """Test cleansing query string (nested query string like flow executor)"""
 89        token = generate_id()
 90        nested_qs = {"token": token}
 91        request = self.factory.get(f"/?{QS_QUERY}={urlencode(nested_qs)}")
 92        event = Event.new("unittest").from_http(request)
 93        self.assertEqual(
 94            event.context,
 95            {
 96                "http_request": {
 97                    "args": {"token": SafeExceptionReporterFilter.cleansed_substitute},
 98                    "method": "GET",
 99                    "path": "/",
100                    "user_agent": "",
101                }
102            },
103        )
104
105    def test_from_http_brand(self):
106        """Test from_http brand"""
107        # Test brand
108        request = self.factory.get("/")
109        brand = Brand(domain="test-brand")
110        request.brand = brand
111        event = Event.new("unittest").from_http(request)
112        self.assertEqual(
113            event.brand,
114            {
115                "app": "authentik_brands",
116                "model_name": "brand",
117                "name": "Brand test-brand",
118                "pk": brand.pk.hex,
119            },
120        )
121
122    def test_from_http_flow_pending_user(self):
123        """Test request from flow request with a pending user"""
124        user = create_test_user()
125
126        session = self.client.session
127        plan = FlowPlan(generate_id())
128        plan.context[PLAN_CONTEXT_PENDING_USER] = user
129        session[SESSION_KEY_PLAN] = plan
130        session.save()
131
132        request = self.factory.get("/")
133        request.session = session
134        request.user = user
135
136        event = Event.new("unittest").from_http(request)
137        self.assertEqual(
138            event.user,
139            {
140                "email": user.email,
141                "pk": user.pk,
142                "username": user.username,
143            },
144        )
145
146    def test_from_http_flow_pending_user_anon(self):
147        """Test request from flow request with a pending user"""
148        user = create_test_user()
149        anon = get_anonymous_user()
150
151        session = self.client.session
152        plan = FlowPlan(generate_id())
153        plan.context[PLAN_CONTEXT_PENDING_USER] = user
154        session[SESSION_KEY_PLAN] = plan
155        session.save()
156
157        request = self.factory.get("/")
158        request.session = session
159        request.user = anon
160
161        event = Event.new("unittest").from_http(request)
162        self.assertEqual(
163            event.user,
164            {
165                "authenticated_as": {
166                    "pk": anon.pk,
167                    "is_anonymous": True,
168                    "username": "AnonymousUser",
169                    "email": "",
170                },
171                "email": user.email,
172                "pk": user.pk,
173                "username": user.username,
174            },
175        )
176
177    def test_from_http_flow_pending_user_fake(self):
178        """Test request from flow request with a pending user"""
179        user = User(
180            username=generate_id(),
181            email=generate_id(),
182        )
183        anon = get_anonymous_user()
184
185        session = self.client.session
186        plan = FlowPlan(generate_id())
187        plan.context[PLAN_CONTEXT_PENDING_USER] = user
188        session[SESSION_KEY_PLAN] = plan
189        session.save()
190
191        request = self.factory.get("/")
192        request.session = session
193        request.user = anon
194
195        event = Event.new("unittest").from_http(request)
196        self.assertEqual(
197            event.user,
198            {
199                "authenticated_as": {
200                    "pk": anon.pk,
201                    "is_anonymous": True,
202                    "username": "AnonymousUser",
203                    "email": "",
204                },
205                "email": user.email,
206                "pk": user.pk,
207                "username": user.username,
208            },
209        )
210
211    def test_invalid_string(self):
212        """Test creating an event with invalid unicode string data"""
213        event = Event.new("unittest", foo="foo bar \u0000 baz")
214        event.save()
215        self.assertEqual(event.context["foo"], "foo bar  baz")