mirror of
https://github.com/ansible/awx.git
synced 2026-01-22 06:58:06 -03:30
Merge pull request #2017 from AlanCoding/1969_job_orphans
Org admin access to orphaned jobs
This commit is contained in:
commit
b8524e9832
@ -877,6 +877,16 @@ class JobTemplateAccess(BaseAccess):
|
||||
return self.user in obj.admin_role
|
||||
|
||||
class JobAccess(BaseAccess):
|
||||
'''
|
||||
I can see jobs when:
|
||||
- I am a superuser.
|
||||
- I can see its job template
|
||||
- I am an admin or auditor of the organization which contains its inventory
|
||||
- I am an admin or auditor of the organization which contains its project
|
||||
I can delete jobs when:
|
||||
- I am an admin of the organization which contains its inventory
|
||||
- I am an admin of the organization which contains its project
|
||||
'''
|
||||
|
||||
model = Job
|
||||
|
||||
@ -888,10 +898,20 @@ class JobAccess(BaseAccess):
|
||||
if self.user.is_superuser:
|
||||
return qs.all()
|
||||
|
||||
return qs.filter(
|
||||
qs_jt = qs.filter(
|
||||
job_template__in=JobTemplate.accessible_objects(self.user, 'read_role')
|
||||
)
|
||||
|
||||
org_access_qs = Organization.objects.filter(
|
||||
Q(admin_role__members=self.user) | Q(auditor_role__members=self.user))
|
||||
if not org_access_qs.exists():
|
||||
return qs_jt
|
||||
|
||||
return qs.filter(
|
||||
Q(job_template__in=JobTemplate.accessible_objects(self.user, 'read_role')) |
|
||||
Q(inventory__organization__in=org_access_qs) |
|
||||
Q(project__organization__in=org_access_qs)).distinct()
|
||||
|
||||
def can_add(self, data):
|
||||
if not data or '_method' in data: # So the browseable API will work?
|
||||
return True
|
||||
@ -920,7 +940,11 @@ class JobAccess(BaseAccess):
|
||||
|
||||
@check_superuser
|
||||
def can_delete(self, obj):
|
||||
return self.user in obj.inventory.admin_role
|
||||
if obj.inventory is not None and self.user in obj.inventory.organization.admin_role:
|
||||
return True
|
||||
if obj.project is not None and self.user in obj.project.organization.admin_role:
|
||||
return True
|
||||
return False
|
||||
|
||||
def can_start(self, obj):
|
||||
self.check_license()
|
||||
|
||||
@ -215,6 +215,13 @@ def org_admin(user, organization):
|
||||
organization.member_role.members.add(ret)
|
||||
return ret
|
||||
|
||||
@pytest.fixture
|
||||
def org_auditor(user, organization):
|
||||
ret = user('org-auditor', False)
|
||||
organization.auditor_role.members.add(ret)
|
||||
organization.member_role.members.add(ret)
|
||||
return ret
|
||||
|
||||
@pytest.fixture
|
||||
def org_member(user, organization):
|
||||
ret = user('org-member', False)
|
||||
|
||||
72
awx/main/tests/functional/test_rbac_job.py
Normal file
72
awx/main/tests/functional/test_rbac_job.py
Normal file
@ -0,0 +1,72 @@
|
||||
import pytest
|
||||
|
||||
from awx.main.access import JobAccess
|
||||
from awx.main.models import Job
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def normal_job(deploy_jobtemplate):
|
||||
return Job.objects.create(
|
||||
job_template=deploy_jobtemplate,
|
||||
project=deploy_jobtemplate.project,
|
||||
inventory=deploy_jobtemplate.inventory
|
||||
)
|
||||
|
||||
# Read permissions testing
|
||||
@pytest.mark.django_db
|
||||
def test_superuser_sees_orphans(normal_job, admin_user):
|
||||
normal_job.job_template = None
|
||||
access = JobAccess(admin_user)
|
||||
assert access.can_read(normal_job)
|
||||
|
||||
@pytest.mark.django_db
|
||||
def test_org_member_does_not_see_orphans(normal_job, org_member, project):
|
||||
normal_job.job_template = None
|
||||
# Check that privledged access to project still does not grant access
|
||||
project.admin_role.members.add(org_member)
|
||||
access = JobAccess(org_member)
|
||||
assert not access.can_read(normal_job)
|
||||
|
||||
@pytest.mark.django_db
|
||||
def test_org_admin_sees_orphans(normal_job, org_admin):
|
||||
normal_job.job_template = None
|
||||
access = JobAccess(org_admin)
|
||||
assert access.can_read(normal_job)
|
||||
|
||||
@pytest.mark.django_db
|
||||
def test_org_auditor_sees_orphans(normal_job, org_auditor):
|
||||
normal_job.job_template = None
|
||||
access = JobAccess(org_auditor)
|
||||
assert access.can_read(normal_job)
|
||||
|
||||
# Delete permissions testing
|
||||
@pytest.mark.django_db
|
||||
def test_JT_admin_delete_denied(normal_job, rando):
|
||||
normal_job.job_template.admin_role.members.add(rando)
|
||||
access = JobAccess(rando)
|
||||
assert not access.can_delete(normal_job)
|
||||
|
||||
@pytest.mark.django_db
|
||||
def test_inventory_admin_delete_denied(normal_job, rando):
|
||||
normal_job.job_template.inventory.admin_role.members.add(rando)
|
||||
access = JobAccess(rando)
|
||||
assert not access.can_delete(normal_job)
|
||||
|
||||
@pytest.mark.django_db
|
||||
def test_null_related_delete_denied(normal_job, rando):
|
||||
normal_job.project = None
|
||||
normal_job.inventory = None
|
||||
access = JobAccess(rando)
|
||||
assert not access.can_delete(normal_job)
|
||||
|
||||
@pytest.mark.django_db
|
||||
def test_inventory_org_admin_delete_allowed(normal_job, org_admin):
|
||||
normal_job.project = None # do this so we test job->inventory->org->admin connection
|
||||
access = JobAccess(org_admin)
|
||||
assert access.can_delete(normal_job)
|
||||
|
||||
@pytest.mark.django_db
|
||||
def test_project_org_admin_delete_allowed(normal_job, org_admin):
|
||||
normal_job.inventory = None # do this so we test job->project->org->admin connection
|
||||
access = JobAccess(org_admin)
|
||||
assert access.can_delete(normal_job)
|
||||
Loading…
x
Reference in New Issue
Block a user