Repository: ansible/awx
File: awx/main/access.py (Python)

Code before:
        'admin_role',
        'execute_role',
        'read_role',
    )

    def filtered_queryset(self):
        return self.model.accessible_objects(self.user, 'read_role')

    @check_superuser
    def can_add(self, data):
        """
        a user can create a job template if they are a superuser, an org admin
        of any org that the project is a member, or if they have user or team
        based permissions tying the project to the inventory source for the
        given action as well as the 'create' deploy permission.
        Users who are able to create deploy jobs can also run normal and check (dry run) jobs.
        """
        if not data:  # So the browseable API will work
            return Organization.accessible_objects(self.user, 'workflow_admin_role').exists()

        if not self.check_related('organization', Organization, data, role_field='workflow_admin_role', mandatory=True):
            if data.get('organization', None) is None:
                self.messages['workflow_job_template'] = [_('An organization is required to create a workflow job template for normal user')]
                return False

            self.messages['workflow_job_template'] = [_('You do not have workflow_admin_role to the organization')]
            return False

        if not self.check_related('inventory', Inventory, data, role_field='use_role'):
            self.messages['workflow_job_template'] = [_('You do not have use_role to the inventory')]
            return False

        if not self.check_related('execution_environment', ExecutionEnvironment, data, role_field='read_role'):
            self.messages['workflow_job_template'] = [_('You do not have read_role to the execution environment')]
            return False

        return True

    def can_copy(self, obj):
        if self.save_messages:
            missing_ujt = []
            missing_credentials = []
            missing_inventories = []
            qs = obj.workflow_job_template_nodes
            qs = qs.prefetch_related('unified_job_template', 'inventory__use_role', 'credentials__use_role')
            for node in qs.all():
                if node.inventory and self.user not in node.inventory.use_role:
                    missing_inventories.append(node.inventory.name)
                for cred in node.credentials.all():
                    if self.user not in cred.use_role:
                        missing_credentials.append(cred.name)

Review comment (security):
This isn't true any more. `workflow_admin_role` is a role. The core idea of the new system (DAB RBAC) is that it enumerates permissions of a role. So access control only considers permissions, never roles. You get permissions from roles, but policy evaluators should never know what role gave you the permission.

I also don't think that we need this, and I would say it would be more _consistent_ right now to give an unqualified 403 "you do not have permission to perform this action" error message, considering that it was the "add" permission lacked for the "add" action.

Code after:
        'admin_role',
        'execute_role',
        'read_role',
    )

    def filtered_queryset(self):
        return self.model.accessible_objects(self.user, 'read_role')

    @check_superuser
    def can_add(self, data):
        """
        a user can create a job template if they are a superuser, an org admin
        of any org that the project is a member, or if they have user or team
        based permissions tying the project to the inventory source for the
        given action as well as the 'create' deploy permission.
        Users who are able to create deploy jobs can also run normal and check (dry run) jobs.
        """
        if not data:  # So the browseable API will work
            return Organization.accessible_objects(self.user, 'workflow_admin_role').exists()

        if not self.check_related('organization', Organization, data, role_field='workflow_admin_role', mandatory=True):
            if data.get('organization', None) is None:
                self.messages['organization'] = [_('An organization is required to create a workflow job template for normal user')]
            return False

        if not self.check_related('inventory', Inventory, data, role_field='use_role'):
            self.messages['inventory'] = [_('You do not have use_role to the inventory')]
            return False

        if not self.check_related('execution_environment', ExecutionEnvironment, data, role_field='read_role'):
            self.messages['execution_environment'] = [_('You do not have read_role to the execution environment')]
            return False

        return True

    def can_copy(self, obj):
        if self.save_messages:
            missing_ujt = []
            missing_credentials = []
            missing_inventories = []
            qs = obj.workflow_job_template_nodes
            qs = qs.prefetch_related('unified_job_template', 'inventory__use_role', 'credentials__use_role')
            for node in qs.all():
                if node.inventory and self.user not in node.inventory.use_role:
                    missing_inventories.append(node.inventory.name)
                for cred in node.credentials.all():
                    if self.user not in cred.use_role:
                        missing_credentials.append(cred.name)
                ujt = node.unified_job_template
                if ujt and not self.user.can_access(UnifiedJobTemplate, 'start', ujt, validate_license=False):
                    missing_ujt.append(ujt.name)