Fix to ensure org auditors can see team credentials

#3081
This commit is contained in:
Akita Noek 2016-08-10 16:58:39 -04:00
parent e55de3d073
commit d181aefddf
6 changed files with 141 additions and 1 deletions

View File

@ -0,0 +1,35 @@
# -*- coding: utf-8 -*-
from __future__ import unicode_literals
from django.db import migrations, models
import awx.main.fields
class Migration(migrations.Migration):
dependencies = [
('main', '0029_v302_add_ask_skip_tags'),
]
operations = [
migrations.AddField(
model_name='credential',
name='teams',
field=models.ManyToManyField(related_name='credentials', to='main.Team', blank=True),
),
migrations.AddField(
model_name='team',
name='auditor_role',
field=awx.main.fields.ImplicitRoleField(related_name='+', parent_role=b'organization.auditor_role', to='main.Role', null=b'True'),
),
migrations.AlterField(
model_name='credential',
name='read_role',
field=awx.main.fields.ImplicitRoleField(related_name='+', parent_role=[b'singleton:system_auditor', b'organization.auditor_role', b'teams.auditor_role', b'use_role', b'admin_role'], to='main.Role', null=b'True'),
),
migrations.AlterField(
model_name='team',
name='read_role',
field=awx.main.fields.ImplicitRoleField(related_name='+', parent_role=[b'auditor_role', b'member_role'], to='main.Role', null=b'True'),
),
]

View File

@ -0,0 +1,44 @@
# -*- coding: utf-8 -*-
from __future__ import unicode_literals
from awx.main.migrations import _rbac as rbac
from awx.main.migrations import _migration_utils as migration_utils
from django.db.models import Q
from django.db import migrations
def synchronize_role_changes(apps, schema_editor):
# The implicit parent roles have been updated for Credential.read_role and
# Team.read_role so these saves will pickup those changes and fix things up.
Team = apps.get_model('main', 'Team')
Credential = apps.get_model('main', 'Credential')
for credential in Credential.objects.iterator():
credential.save()
for team in Team.objects.iterator():
team.save()
def populate_credential_teams_field(apps, schema_editor):
Team = apps.get_model('main', 'Team')
Credential = apps.get_model('main', 'Credential')
for credential in Credential.objects.iterator():
teams_qs = Team.objects.filter(
Q(member_role__children=credential.use_role) |
Q(member_role__children=credential.admin_role)
)
for team in teams_qs.iterator():
credential.teams.add(team)
class Migration(migrations.Migration):
dependencies = [
('main', '0030_audit_team_credential_changes'),
]
operations = [
migrations.RunPython(migration_utils.set_current_apps_for_migrations),
migrations.RunPython(synchronize_role_changes),
migrations.RunPython(populate_credential_teams_field),
migrations.RunPython(rbac.rebuild_role_hierarchy),
]

View File

@ -87,6 +87,11 @@ class Credential(PasswordFieldsModel, CommonModelNameNotUnique, ResourceMixin):
on_delete=models.CASCADE,
related_name='credentials',
)
teams = models.ManyToManyField(
'Team',
blank=True,
related_name='credentials',
)
kind = models.CharField(
max_length=32,
choices=KIND_CHOICES,
@ -226,6 +231,7 @@ class Credential(PasswordFieldsModel, CommonModelNameNotUnique, ResourceMixin):
read_role = ImplicitRoleField(parent_role=[
'singleton:' + ROLE_SINGLETON_SYSTEM_AUDITOR,
'organization.auditor_role',
'teams.auditor_role',
'use_role',
'admin_role',
])

View File

@ -107,8 +107,11 @@ class Team(CommonModelNameNotUnique, ResourceMixin):
member_role = ImplicitRoleField(
parent_role='admin_role',
)
auditor_role = ImplicitRoleField(
parent_role='organization.auditor_role',
)
read_role = ImplicitRoleField(
parent_role=['organization.auditor_role', 'member_role'],
parent_role=['auditor_role', 'member_role'],
)
def get_absolute_url(self):

View File

@ -9,6 +9,7 @@ import json
# Django
from django.db.models.signals import post_save, pre_delete, post_delete, m2m_changed
from django.db.models import Q
from django.dispatch import receiver
# Django-CRUM
@ -120,6 +121,39 @@ def rebuild_role_ancestor_list(reverse, model, instance, pk_set, action, **kwarg
else:
model.rebuild_role_ancestor_list([], [instance.id])
def link_credentials_to_teams(reverse, model, instance, pk_set, action, **kwargs):
'When a team is granted access to a credential, add the team to the credential teams m2m'
# If reverse is true, then we got here by something like
# team.member_role.children.add(credential.(use|admin)_role)
# else
# credential.(use|admin)_role.parents.add(team.member_role)
if action in ['post_add', 'post_remove']:
if reverse:
teams = [co for co in [instance.content_object] if isinstance(co, Team)]
credentials = [role.content_object for role in Role.objects.filter(id__in=pk_set).all()
if isinstance(role.content_object, Credential) and role.role_field != 'read_role'] # exclude read role to prevent signal looping
else:
credentials = [co for co in [instance.content_object] if isinstance(co, Credential) and instance.role_field != 'read_role']
teams = [role.content_object for role in Role.objects.filter(id__in=pk_set).all()
if isinstance(role.content_object, Team)]
if not teams or not credentials:
return
if action == 'post_add':
for credential in credentials:
for team in teams:
credential.teams.add(team)
if action == 'post_remove':
for credential in credentials:
for team in teams:
if not Team.objects.filter(Q(member_role__children=credential.use_role) | Q(member_role__children=credential.admin_role), id=team.id).exists():
credential.teams.remove(team)
credential.save()
def sync_superuser_status_to_rbac(instance, **kwargs):
'When the is_superuser flag is changed on a user, reflect that in the membership of the System Admnistrator role'
if instance.is_superuser:
@ -209,6 +243,7 @@ post_save.connect(emit_update_inventory_on_created_or_deleted, sender=Job)
post_delete.connect(emit_update_inventory_on_created_or_deleted, sender=Job)
post_save.connect(emit_job_event_detail, sender=JobEvent)
post_save.connect(emit_ad_hoc_command_event_detail, sender=AdHocCommandEvent)
m2m_changed.connect(link_credentials_to_teams, Role.parents.through)
m2m_changed.connect(rebuild_role_ancestor_list, Role.parents.through)
m2m_changed.connect(org_admin_edit_members, Role.members.through)
m2m_changed.connect(rbac_activity_stream, Role.members.through)

View File

@ -113,6 +113,23 @@ def test_create_team_credential_by_team_member_xfail(post, team, alice, team_mem
assert response.status_code == 403
@pytest.mark.django_db
def test_team_credential_visibility_by_org_admins(team, credential, organization, user):
org_auditor = user('org_auditor')
organization.auditor_role.members.add(org_auditor)
assert org_auditor not in credential.read_role
team.member_role.children.add(credential.use_role)
assert org_auditor in credential.read_role
team.member_role.children.remove(credential.use_role)
assert org_auditor not in credential.read_role
credential.use_role.parents.add(team.member_role)
assert org_auditor in credential.read_role
credential.use_role.parents.remove(team.member_role)
assert org_auditor not in credential.read_role
#
# organization credentials