Create access logs from requests

kobo/apps/audit_log/models.py

@@ -1,10 +1,22 @@
+import logging
+
+from django.conf import settings
+from django.contrib.auth import get_user_model
+from django.contrib.auth.models import AnonymousUser
 from django.contrib.contenttypes.models import ContentType
 from django.db import models
-from django.conf import settings
 from django.utils.timezone import now
 
+from kobo.apps.openrosa.libs.utils.viewer_tools import (
+    get_client_ip,
+    get_human_readable_client_user_agent,
+)
 from kpi.fields.kpi_uid import UUID_LENGTH
 
+KOBO_AUTH_APP_LABEL = 'kobo_auth'
+LOGINAS_AUTH_TYPE = 'django-loginas'
+UNKNOWN_AUTH_TYPE = 'Unknown'
+
 
 class AuditAction(models.TextChoices):
 
@@ -60,3 +72,51 @@ def save(
             using=using,
             update_fields=update_fields,
         )
+
+    @staticmethod
+    def create_auth_log_from_request(request, authentication_type=None):
+        logged_in_user = request.user
+
+        # django-loginas will keep the superuser as the _cached_user while request.user is set to the new one
+        initial_user = request._cached_user
+        is_loginas_url = request.resolver_match.url_name == 'loginas-user-login'
+        # a regular login may have an anonymous user as _cached_user, ignore that
+        user_changed = initial_user.is_authenticated and initial_user.id != logged_in_user.id
+        is_loginas = is_loginas_url and user_changed
+        if authentication_type and authentication_type != '':
+            # authentication_type parameter has precedence
+            auth_type = authentication_type
+        elif is_loginas:
+            # second option: loginas
+            auth_type = LOGINAS_AUTH_TYPE
+        elif hasattr(logged_in_user, 'backend') and logged_in_user.backend is not None:
+            # third option: the backend that authenticated the user
+            auth_type = logged_in_user.backend
+        else:
+            # default: unknown
+            auth_type = UNKNOWN_AUTH_TYPE
+
+        # gather information about the source of the request
+        ip = get_client_ip(request)
+        source = get_human_readable_client_user_agent(request)
+        metadata = {
+            'ip_address': ip,
+            'source': source,
+            'auth_type': auth_type,
+        }
+
+        # add extra information if needed for django-loginas
+        if is_loginas:
+            metadata['initial_user_uid'] = initial_user.extra_details.uid
+            metadata['initial_user_username'] = initial_user.username
+        audit_log = AuditLog(
+            user=logged_in_user,
+            app_label=KOBO_AUTH_APP_LABEL,
+            model_name=get_user_model(),
+            object_id=logged_in_user.id,
+            user_uid=logged_in_user.extra_details.uid,
+            action=AuditAction.AUTH,
+            metadata=metadata,
+        )
+        audit_log.save()
+        return audit_log

kobo/apps/audit_log/tests/test_models.py

@@ -0,0 +1,121 @@
+from unittest.mock import patch
+
+from django.contrib.auth import get_user_model
+from django.contrib.auth.models import AnonymousUser
+from django.test.client import RequestFactory
+from django.urls import resolve
+
+from kobo.apps.audit_log.models import (
+    KOBO_AUTH_APP_LABEL,
+    LOGINAS_AUTH_TYPE,
+    UNKNOWN_AUTH_TYPE,
+    AuditAction,
+    AuditLog,
+)
+from kpi.tests.base_test_case import BaseTestCase
+
+
+@patch('kobo.apps.audit_log.models.get_human_readable_client_user_agent', return_value='source')
+@patch('kobo.apps.audit_log.models.get_client_ip', return_value='127.0.0.1')
+class AuditLogTestCase(BaseTestCase):
+
+    @classmethod
+    def setUpClass(cls):
+        super().setUpClass()
+        cls.SUPER_USER = get_user_model().objects.create_user(
+            'user', '[email protected]', 'userpass'
+        )
+        cls.SUPER_USER.is_super = True
+        cls.SUPER_USER.backend = 'django.contrib.auth.backends.ModelBackend'
+        cls.SUPER_USER.save()
+
+    def _create_request(self, url: str, cached_user, new_user):
+        factory = RequestFactory()
+        request = factory.post(url)
+        request.user = new_user
+        request._cached_user = cached_user
+        request.resolver_match = resolve(url)
+        return request
+
+    def _check_common_fields(self, audit_log: AuditLog, user):
+        self.assertEqual(audit_log.user.id, user.id)
+        self.assertEqual(audit_log.app_label, KOBO_AUTH_APP_LABEL)
+        self.assertEqual(audit_log.model_name, get_user_model())
+        self.assertEqual(audit_log.object_id, user.id)
+        self.assertEqual(audit_log.user_uid, user.extra_details.uid)
+        self.assertEqual(audit_log.action, AuditAction.AUTH)
+
+    def test_basic_create_auth_log_from_request(self, patched_ip, patched_source):
+        request = self._create_request(
+            '/accounts/login/', AnonymousUser(), AuditLogTestCase.SUPER_USER
+        )
+        log: AuditLog = AuditLog.create_auth_log_from_request(request)
+        self._check_common_fields(log, AuditLogTestCase.SUPER_USER)
+        self.assertDictEqual(
+            log.metadata,
+            {
+                'ip_address': '127.0.0.1',
+                'source': 'source',
+                'auth_type': AuditLogTestCase.SUPER_USER.backend,
+            },
+        )
+
+    def test_create_auth_log_from_loginas_request(self, patched_ip, patched_source):
+        second_user = get_user_model().objects.create_user(
+            'second_user', '[email protected]', 'pass'
+        )
+        second_user.save()
+        request = self._create_request(
+            f'/admin/login/user/{second_user.id}/',
+            AuditLogTestCase.SUPER_USER,
+            second_user,
+        )
+        log: AuditLog = AuditLog.create_auth_log_from_request(request)
+        self._check_common_fields(log, second_user)
+        self.assertDictEqual(
+            log.metadata,
+            {
+                'ip_address': '127.0.0.1',
+                'source': 'source',
+                'auth_type': LOGINAS_AUTH_TYPE,
+                'initial_user_uid': AuditLogTestCase.SUPER_USER.extra_details.uid,
+                'initial_user_username': AuditLogTestCase.SUPER_USER.username,
+            },
+        )
+
+    def test_create_auth_log_with_different_auth_type(self, patched_ip, patched_source):
+        request = self._create_request(
+            '/api/v2/assets/', AnonymousUser(), AuditLogTestCase.SUPER_USER
+        )
+        log: AuditLog = AuditLog.create_auth_log_from_request(
+            request, authentication_type='Token'
+        )
+        self._check_common_fields(log, AuditLogTestCase.SUPER_USER)
+        self.assertDictEqual(
+            log.metadata,
+            {
+                'ip_address': '127.0.0.1',
+                'source': 'source',
+                'auth_type': 'Token',
+            },
+        )
+
+    def test_create_auth_log_unknown_authenticator(self, patched_ip, patched_source):
+        # no backend attached to the user object
+        second_user = get_user_model().objects.create_user(
+            'second_user', '[email protected]', 'pass'
+        )
+        second_user.save()
+        request = self._create_request(
+            f'/api/v2/assets/', AuditLogTestCase.SUPER_USER, second_user
+        )
+        log: AuditLog = AuditLog.create_auth_log_from_request(request)
+        self._check_common_fields(log, second_user)
+        self.assertDictEqual(
+            log.metadata,
+            {
+                'ip_address': '127.0.0.1',
+                'source': 'source',
+                'auth_type': UNKNOWN_AUTH_TYPE,
+            },
+        )