feat: create ph logs for permissions changes

kobo/apps/audit_log/audit_actions.py

@@ -3,6 +3,7 @@
 
 class AuditAction(models.TextChoices):
     ADD_MEDIA = 'add-media'
+    ALLOW_ANONYMOUS_SUBMISSIONS = 'allow-anonymous-submissions'
     ARCHIVE = 'archive'
     AUTH = 'auth'
     CONNECT_PROJECT = 'connect-project'
@@ -12,19 +13,25 @@ class AuditAction(models.TextChoices):
     DELETE_SERVICE = 'delete-service'
     DEPLOY = 'deploy'
     DISABLE_SHARING = 'disable-sharing'
+    DISALLOW_ANONYMOUS_SUBMISSIONS = 'disallow-anonymous-submissions'
     DISCONNECT_PROJECT = 'disconnect-project'
     ENABLE_SHARING = 'enable-sharing'
     EXPORT = 'export'
     IN_TRASH = 'in-trash'
     MODIFY_IMPORTED_FIELDS = 'modify-imported-fields'
     MODIFY_SERVICE = 'modify-service'
     MODIFY_SHARING = 'modify_sharing'
+    MODIFY_USER_PERMISSIONS = 'modify-user-permissions'
     PUT_BACK = 'put-back'
     REDEPLOY = 'redeploy'
     REGISTER_SERVICE = 'register-service'
     REMOVE = 'remove'
     REPLACE_FORM = 'replace-form'
+    SHARE_FORM_PUBLICLY = 'share-form-publicly'
+    SHARE_DATA_PUBLICLY = 'share-data-publicly'
     UNARCHIVE = 'unarchive'
+    UNSHARE_FORM_PUBLICLY = 'unshare-form-publicly'
+    UNSHARE_DATA_PUBLICLY = 'unshare-data-publicly'
     UPDATE = 'update'
     UPDATE_CONTENT = 'update-content'
     UPDATE_NAME = 'update-name'

kobo/apps/audit_log/models.py

@@ -1,3 +1,5 @@
+import copy
+
 import jsonschema
 from django.conf import settings
 from django.db import models
@@ -19,6 +21,10 @@
     ACCESS_LOG_SUBMISSION_AUTH_TYPE,
     ACCESS_LOG_SUBMISSION_GROUP_AUTH_TYPE,
     ACCESS_LOG_UNKNOWN_AUTH_TYPE,
+    PERM_ADD_SUBMISSIONS,
+    PERM_VIEW_ASSET,
+    PERM_VIEW_SUBMISSIONS,
+    PROJECT_HISTORY_LOG_PERMISSION_SUBTYPE,
     PROJECT_HISTORY_LOG_PROJECT_SUBTYPE,
 )
 from kpi.fields.kpi_uid import UUID_LENGTH
@@ -30,6 +36,15 @@
 ADDED = 'added'
 REMOVED = 'removed'
 
+ANONYMOUS_USER_PERMISSION_ACTIONS = {
+    (PERM_VIEW_ASSET, True): AuditAction.SHARE_FORM_PUBLICLY,
+    (PERM_VIEW_SUBMISSIONS, True): AuditAction.SHARE_DATA_PUBLICLY,
+    (PERM_ADD_SUBMISSIONS, True): AuditAction.ALLOW_ANONYMOUS_SUBMISSIONS,
+    (PERM_VIEW_ASSET, False): AuditAction.UNSHARE_FORM_PUBLICLY,
+    (PERM_VIEW_SUBMISSIONS, False): AuditAction.UNSHARE_DATA_PUBLICLY,
+    (PERM_ADD_SUBMISSIONS, False): AuditAction.DISALLOW_ANONYMOUS_SUBMISSIONS,
+}
+
 
 class AuditType(models.TextChoices):
     ACCESS = 'access'
@@ -247,10 +262,7 @@ def create_from_request(
         elif is_loginas:
             # third option: loginas
             auth_type = ACCESS_LOG_LOGINAS_AUTH_TYPE
-        elif (
-            hasattr(logged_in_user, 'backend')
-            and logged_in_user.backend is not None
-        ):
+        elif hasattr(logged_in_user, 'backend') and logged_in_user.backend is not None:
             # fourth option: the backend that authenticated the user
             auth_type = logged_in_user.backend
         else:
@@ -336,6 +348,9 @@ def create_from_request(cls, request):
             'asset-export-list': cls.create_from_export_request,
             'exporttask-list': cls.create_from_v1_export,
             'asset-bulk': cls.create_from_bulk_request,
+            'asset-permission-assignment-bulk-assignments': cls.create_from_permissions_request,  # noqa
+            'asset-permission-assignment-detail': cls.create_from_permissions_request,
+            'asset-permission-assignment-list': cls.create_from_permissions_request,
         }
         url_name = request.resolver_match.url_name
         method = url_name_to_action.get(url_name, None)
@@ -456,7 +471,7 @@ def create_from_detail_request(cls, request):
             'log_subtype': PROJECT_HISTORY_LOG_PROJECT_SUBTYPE,
             'ip_address': get_client_ip(request),
             'source': get_human_readable_client_user_agent(request),
-            'latest_version_uid': updated_data['latest_version.uid']
+            'latest_version_uid': updated_data['latest_version.uid'],
         }
 
         changed_field_to_action_map = {
@@ -668,3 +683,118 @@ def create_from_v1_export(cls, request):
                 'source': get_human_readable_client_user_agent(request),
             },
         )
+
+    @classmethod
+    def create_from_permissions_request(cls, request):
+        logs = []
+        initial_data = getattr(request, 'initial_data', None)
+        updated_data = getattr(request, 'updated_data', None)
+        asset_uid = request.resolver_match.kwargs['parent_lookup_asset']
+        source_data = updated_data if updated_data else initial_data
+        if source_data is None:
+            # there was an error on the request, ignore
+            return
+        asset_id = source_data['asset.id']
+        # these will be dicts of username: [permissions] (as set or list)
+        permissions_added = getattr(request, 'permissions_added', {})
+        permissions_removed = getattr(request, 'permissions_removed', {})
+        partial_permissions_added = getattr(request, 'partial_permissions_added', {})
+        # basic metadata for all PH logs
+        base_metadata = {
+            'ip_address': get_client_ip(request),
+            'source': get_human_readable_client_user_agent(request),
+            'asset_uid': asset_uid,
+            'log_subtype': PROJECT_HISTORY_LOG_PERMISSION_SUBTYPE,
+        }
+        # we'll be bulk creating logs instead of using .create, so we have to set
+        # all fields manually
+        log_base = {
+            'user': request.user,
+            'object_id': asset_id,
+            'app_label': Asset._meta.app_label,
+            'model_name': Asset._meta.model_name,
+            'log_type': AuditType.PROJECT_HISTORY,
+            'user_uid': request.user.extra_details.uid,
+        }
+        # get all users whose permissions changed
+        for username in {
+            *permissions_added,
+            *permissions_removed,
+            *partial_permissions_added,
+        }:
+            user_permissions_added = permissions_added.get(username, {})
+            user_permissions_removed = permissions_removed.get(username, {})
+            user_partial_permissions_added = partial_permissions_added.get(username, [])
+            if username == 'AnonymousUser':
+                cls.handle_anonymous_user_permissions(
+                    user_permissions_added,
+                    user_permissions_removed,
+                    user_partial_permissions_added,
+                    base_metadata,
+                    log_base,
+                    logs,
+                )
+                continue
+            metadata = copy.deepcopy(base_metadata)
+            metadata['permissions'] = {
+                'username': username,
+                REMOVED: list(user_permissions_removed),
+                ADDED: list(user_permissions_added) + user_partial_permissions_added,
+            }
+            logs.append(
+                ProjectHistoryLog(
+                    **log_base,
+                    action=AuditAction.MODIFY_USER_PERMISSIONS,
+                    metadata=metadata,
+                )
+            )
+        ProjectHistoryLog.objects.bulk_create(logs)
+
+    @classmethod
+    def handle_anonymous_user_permissions(
+        cls,
+        perms_added,
+        perms_removed,
+        partial_perms_added,
+        base_metadata,
+        log_base,
+        logs,
+    ):
+        # go through all the usual anonymous user permissions and create
+        # logs if they were changed
+        # remove each permission as it is logged so we can see if there
+        # are any unusual ones left over
+        for combination, action in ANONYMOUS_USER_PERMISSION_ACTIONS.items():
+            # ANONYMOUS_USER_PERMIISSION_ACTIONS has tuples as keys
+            permission, was_added = combination
+            list_to_update = perms_added if was_added else perms_removed
+            if permission in list_to_update:
+                list_to_update.discard(permission)
+                logs.append(
+                    ProjectHistoryLog(
+                        **log_base,
+                        action=action,
+                        metadata=base_metadata,
+                    )
+                )
+
+        # this shouldn't happen, but if anonymous users are granted other permissions,
+        # we want to know
+        if (
+            len(perms_removed) > 0
+            or len(perms_added) > 0
+            or len(partial_perms_added) > 0
+        ):
+            metadata = copy.deepcopy(base_metadata)
+            metadata['permissions'] = {
+                'username': 'AnonymousUser',
+                ADDED: list(perms_added) + partial_perms_added,
+                REMOVED: list(perms_removed),
+            }
+            logs.append(
+                ProjectHistoryLog(
+                    **log_base,
+                    metadata=metadata,
+                    action=AuditAction.MODIFY_USER_PERMISSIONS,
+                )
+            )

kobo/apps/audit_log/signals.py

@@ -1,11 +1,20 @@
+from collections import defaultdict
+
 from celery.signals import task_success
 from django.contrib.auth.signals import user_logged_in
 from django.dispatch import receiver
+from django_userforeignkey.request import get_current_request
 
-from kpi.models import ImportTask
+from kpi.constants import ASSET_TYPE_SURVEY, PERM_PARTIAL_SUBMISSIONS
+from kpi.models import Asset, ImportTask
 from kpi.tasks import import_in_background
 from kpi.utils.log import logging
-
+from kpi.utils.object_permission import (
+    post_assign_partial_perm,
+    post_assign_perm,
+    post_remove_partial_perms,
+    post_remove_perm,
+)
 from .models import AccessLog, ProjectHistoryLog
 
 
@@ -24,3 +33,61 @@ def create_access_log(sender, user, **kwargs):
 def create_ph_log_for_import(sender, result, **kwargs):
     task = ImportTask.objects.get(uid=result)
     ProjectHistoryLog.create_from_import_task(task)
+
+
+def initialize_permission_lists_if_necessary(request):
+    if getattr(request, 'permissions_added', None) is None:
+        request.permissions_added = defaultdict(set)
+    if getattr(request, 'permissions_removed', None) is None:
+        request.permissions_removed = defaultdict(set)
+    if getattr(request, 'partial_permissions_added', None) is None:
+        request.partial_permissions_added = defaultdict(list)
+
+
+def initialize_request():
+    request = get_current_request()
+    if request is None:
+        return None
+    initialize_permission_lists_if_necessary(request)
+    return request
+
+
+@receiver(post_assign_perm, sender=Asset)
+def add_assigned_perms(sender, instance, user, codename, deny, **kwargs):
+    request = initialize_request()
+    if not request or instance.asset_type != ASSET_TYPE_SURVEY or deny:
+        return
+    request.permissions_added[user.username].add(codename)
+    request.permissions_removed[user.username].discard(codename)
+
+
+@receiver(post_remove_perm, sender=Asset)
+def add_removed_perms(sender, instance, user, codename, **kwargs):
+    request = initialize_request()
+    if not request or instance.asset_type != ASSET_TYPE_SURVEY:
+        return
+    request.permissions_removed[user.username].add(codename)
+    request.permissions_added[user.username].discard(codename)
+
+
+@receiver(post_assign_partial_perm, sender=Asset)
+def add_assigned_partial_perms(sender, instance, user, perms, **kwargs):
+    request = initialize_request()
+    if not request or instance.asset_type != ASSET_TYPE_SURVEY:
+        return
+    perms_as_list_of_dicts = [{'code': k, 'filters': v} for k, v in perms.items()]
+    # partial permissions are replaced rather than added
+    request.partial_permissions_added[user.username] = perms_as_list_of_dicts
+
+
+@receiver(post_remove_partial_perms, sender=Asset)
+def remove_partial_perms(sender, instance, user, **kwargs):
+    request = initialize_request()
+    if not request or instance.asset_type != ASSET_TYPE_SURVEY:
+        return
+    request.partial_permissions_added[user.username] = []
+    # in case we somehow deleted partial permissions
+    # without deleting 'partial_submissions', take care of that now since
+    # we can't have one without the other
+    request.permissions_added[user.username].discard(PERM_PARTIAL_SUBMISSIONS)
+    request.permissions_removed[user.username].add(PERM_PARTIAL_SUBMISSIONS)

kobo/apps/audit_log/tests/test_models.py

@@ -13,6 +13,8 @@
 from kobo.apps.audit_log.models import (
     ACCESS_LOG_LOGINAS_AUTH_TYPE,
     ACCESS_LOG_UNKNOWN_AUTH_TYPE,
+    ADDED,
+    REMOVED,
     AccessLog,
     AuditLog,
     AuditType,
@@ -644,3 +646,37 @@ def test_create_from_import_task_with_name_change(self):
                 'name': {'old': old_name, 'new': 'new_name'},
             },
         )
+
+    def test_create_from_unexpected_anonymous_permissions(self):
+        # Normal anonymous permissions tested elsewhere
+        # This test is for if somehow other permissions are assigned
+        factory = RequestFactory()
+        request = factory.post('/')
+        request.user = User.objects.get(username='someuser')
+        request.resolver_match = Mock()
+        request.resolver_match.kwargs = {'parent_lookup_asset': 'a12345'}
+        request.updated_data = {
+            'asset.id': 1,
+        }
+        request.permissions_added = {
+            # these permissions are not allowed for anonymous users,
+            # pretend something went wrong/changed and they were assigned anyway
+            'AnonymousUser': {'discover_asset', 'validate_submissions'}
+        }
+        ProjectHistoryLog.create_from_permissions_request(
+            request,
+        )
+        self.assertEqual(ProjectHistoryLog.objects.count(), 1)
+        log = ProjectHistoryLog.objects.first()
+        self.assertEqual(log.object_id, 1)
+        # should create a regular 'MODIFY_USER_PERMISSIONS' log
+        self.assertEqual(log.action, AuditAction.MODIFY_USER_PERMISSIONS)
+
+        self.assertDictEqual(
+            log.metadata['permissions'],
+            {
+                ADDED: ['discover_asset', 'validate_submissions'],
+                REMOVED: [],
+                'username': 'AnonymousUser',
+            },
+        )