Fix incorrect RBAC permissions in object creation tasks

fixes: #5683
https://access.redhat.com/security/cve/cve-2024-7143

CHANGES/5683.bugfix

@@ -0,0 +1,2 @@
+Fixed RBAC permissions being incorrectly assigned in tasks that create objects.
+https://access.redhat.com/security/cve/cve-2024-7143

pulpcore/app/models/task.py

@@ -13,6 +13,7 @@
 from django.core.serializers.json import DjangoJSONEncoder
 from django.db import connection, models
 from django.utils import timezone
+from django_lifecycle import hook, AFTER_CREATE
 
 from pulpcore.app.models import (
     AutoAddObjPermsMixin,
@@ -163,6 +164,11 @@ def current():
         """
         return current_task.get()
 
+    @hook(AFTER_CREATE)
+    def add_role_dispatcher(self):
+        """Set the "core.task_user_dispatcher" role for the current user after creation."""
+        self.add_roles_for_object_creator("core.task_user_dispatcher")
+
     def set_running(self):
         """
         Set this Task to the running state, save it, and log output in warning cases.

pulpcore/app/viewsets/task.py

@@ -145,6 +145,10 @@ class TaskViewSet(
             ],
         },
         "core.task_viewer": ["core.view_task"],
+        # This is a special role to designate the user who dispatched the task, it is assigned to
+        # the user on the object-level at task creation. It is not meant to be edited or manually
+        # added/removed from users.
+        "core.task_user_dispatcher": ["core.add_task"],
     }
 
     def get_serializer(self, *args, **kwargs):
@@ -193,6 +197,11 @@ def get_queryset(self):
                 ),
                 Prefetch("child_tasks", queryset=Task.objects.only("pk")),
             )
+        if self.action in ("add_role", "remove_role"):
+            if "core.task_user_dispatcher" == self.request.data.get("role"):
+                raise ValidationError(
+                    _("core.task_user_dispatcher can not be added/removed from a task.")
+                )
         return qs
 
     @extend_schema(

pulpcore/tasking/_util.py

@@ -97,7 +97,26 @@ def perform_task(task_pk, task_working_dir_rel_path):
     connection.connection = None
     # enc_args and enc_kwargs are deferred by default but we actually want them
     task = Task.objects.defer(None).select_related("pulp_domain").get(pk=task_pk)
-    user = get_users_with_perms(task, with_group_users=False).first()
+    # These queries were specifically constructed and ordered this way to ensure we have the highest
+    # chance of getting the user who dispatched the task since we don't have a user relation on the
+    # task model. The second query acts as a fallback to provide ZDU support. Future changes will
+    # require to keep these around till a breaking change release is planned (3.70 the earliest).
+    user = (
+        get_users_with_perms(
+            task,
+            only_with_perms_in=["core.add_task"],
+            with_group_users=False,
+            include_model_permissions=False,
+            include_domain_permissions=False,
+        ).first()
+        or get_users_with_perms(
+            task,
+            only_with_perms_in=["core.manage_roles_task"],
+            with_group_users=False,
+            include_model_permissions=False,
+            include_domain_permissions=False,
+        ).first()
+    )
     # Isolate from the parent asyncio.
     asyncio.set_event_loop(asyncio.new_event_loop())
     # Set current contexts

pulpcore/tests/functional/api/test_tasking.py

@@ -379,6 +379,43 @@ def test_emmiting_unblocked_task_telemetry(
     ]
 
 
+@pytest.mark.parallel
+def test_correct_task_ownership(
+    dispatch_task, pulpcore_bindings, gen_user, file_repository_factory
+):
+    """Test that tasks get the correct ownership when dispatched."""
+    alice = gen_user(model_roles=["core.task_viewer"])
+    bob = gen_user(model_roles=["file.filerepository_creator"])
+
+    with alice:
+        atask_href = dispatch_task("pulpcore.app.tasks.test.sleep", args=(0,))
+    aroles = pulpcore_bindings.UsersRolesApi.list(alice.user.pulp_href)
+    assert aroles.count == 3
+    roles = {r.role: r.content_object for r in aroles.results}
+    correct_roles = {
+        "core.task_owner": atask_href,
+        "core.task_user_dispatcher": atask_href,
+        "core.task_viewer": None,
+    }
+    assert roles == correct_roles
+
+    with bob:
+        btask_href = dispatch_task("pulpcore.app.tasks.test.sleep", args=(0,))
+        repo = file_repository_factory()
+    aroles = pulpcore_bindings.UsersRolesApi.list(alice.user.pulp_href)
+    assert aroles.count == 3
+    broles = pulpcore_bindings.UsersRolesApi.list(bob.user.pulp_href)
+    assert broles.count == 4
+    roles = {r.role: r.content_object for r in broles.results}
+    correct_roles = {
+        "core.task_owner": btask_href,
+        "core.task_user_dispatcher": btask_href,
+        "file.filerepository_owner": repo.pulp_href,
+        "file.filerepository_creator": None,
+    }
+    assert roles == correct_roles
+
+
 @pytest.fixture
 def task_group(dispatch_task_group, monitor_task_group):
     """Fixture containing a finished Task Group."""