Fix RBAC query bug in checking if user can run ad hoc commands. Fixes https://trello.com/c/fl36Hvh3

awx/main/access.py

@@ -278,7 +278,7 @@ class InventoryAccess(BaseAccess):
 
     model = Inventory
 
-    def get_queryset(self, allowed=None):
+    def get_queryset(self, allowed=None, ad_hoc=None):
         allowed = allowed or PERMISSION_TYPES_ALLOWING_INVENTORY_READ
         qs = Inventory.objects.filter(active=True).distinct()
         qs = qs.select_related('created_by', 'modified_by', 'organization')
@@ -286,21 +286,27 @@ class InventoryAccess(BaseAccess):
             return qs
         qs = qs.filter(organization__active=True)
         admin_of = qs.filter(organization__admins__in=[self.user]).distinct()
-        has_user_perms = qs.filter(
+        has_user_kw = dict(
             permissions__user__in=[self.user],
             permissions__permission_type__in=allowed,
             permissions__active=True,
-        ).distinct()
+        )
-        has_team_perms = qs.filter(
+        if ad_hoc is not None:
+            has_user_kw['permissions__run_ad_hoc_commands'] = ad_hoc
+        has_user_perms = qs.filter(**has_user_kw).distinct()
+        has_team_kw = dict(
             permissions__team__users__in=[self.user],
             permissions__team__active=True,
             permissions__permission_type__in=allowed,
             permissions__active=True,
-        ).distinct()
+        )
+        if ad_hoc is not None:
+            has_team_kw['permissions__run_ad_hoc_commands'] = ad_hoc
+        has_team_perms = qs.filter(**has_team_kw).distinct()
         return admin_of | has_user_perms | has_team_perms
 
-    def has_permission_types(self, obj, allowed):
+    def has_permission_types(self, obj, allowed, ad_hoc=None):
-        return bool(obj and self.get_queryset(allowed).filter(pk=obj.pk).exists())
+        return bool(obj and self.get_queryset(allowed, ad_hoc).filter(pk=obj.pk).exists())
 
     def can_read(self, obj):
         return self.has_permission_types(obj, PERMISSION_TYPES_ALLOWING_INVENTORY_READ)
@@ -347,16 +353,7 @@ class InventoryAccess(BaseAccess):
         return self.can_admin(obj, None)
 
     def can_run_ad_hoc_commands(self, obj):
-        qs = self.get_queryset(PERMISSION_TYPES_ALLOWING_INVENTORY_READ)
+        return self.has_permission_types(obj, PERMISSION_TYPES_ALLOWING_INVENTORY_READ, True)
-        if not obj or not qs.filter(pk=obj.pk).exists():
-            return False
-        if self.user.is_superuser:
-            return True
-        if self.user in obj.organization.admins.all():
-            return True
-        if qs.filter(pk=obj.pk, permissions__permission_type__in=PERMISSION_TYPES_ALLOWING_INVENTORY_READ, permissions__run_ad_hoc_commands=True).exists():
-            return True
-        return False
 
 class HostAccess(BaseAccess):
     '''

awx/main/tests/ad_hoc.py

@@ -952,6 +952,19 @@ class AdHocCommandApiTest(BaseAdHocCommandTest):
             self.patch(url, {}, expect=401)
             self.delete(url, expect=401)
 
+        # Create another unrelated inventory permission with run_ad_hoc_commands
+        # set; this tests an edge case in the RBAC query where we'll return
+        # can_run_ad_hoc_commands = True when we shouldn't.
+        nobody_perm_url = reverse('api:user_permissions_list', args=(self.nobody_django_user.pk,))
+        nobody_perm_data = {
+            'name': 'Allow Nobody to Read Inventory',
+            'inventory': self.inventory.pk,
+            'permission_type': 'read',
+            'run_ad_hoc_commands': True,
+        }
+        with self.current_user('admin'):
+            response = self.post(nobody_perm_url, nobody_perm_data, expect=201)
+
         # Create a credential for the other user and explicitly give other
         # user admin permission on the inventory (still not allowed to run ad
         # hoc commands; can get the list but can't see any items).
@@ -968,9 +981,9 @@ class AdHocCommandApiTest(BaseAdHocCommandTest):
         with self.current_user('other'):
             response = self.get(url, expect=200)
             self.assertEqual(response['count'], 0)
-            self.run_test_ad_hoc_command(url=url, inventory=None, credential=other_cred.pk, expect=403)
             response = self.get(inventory_url, expect=200)
             self.assertFalse(response['can_run_ad_hoc_commands'])
+            self.run_test_ad_hoc_command(url=url, inventory=None, credential=other_cred.pk, expect=403)
 
         # Update permission to allow other user to run ad hoc commands.  Can
         # only see his own ad hoc commands (because of credential permission).