Sync to 2 different DAB team member roles

galaxy_ng/app/signals/handlers.py

@@ -15,6 +15,7 @@
 from django.db.models import CharField, Value
 from django.db.models.functions import Concat
 from django.contrib.contenttypes.models import ContentType
+from django.contrib.auth.models import Group
 from django.conf import settings
 from rest_framework.exceptions import ValidationError
 from django.apps import apps
@@ -121,6 +122,7 @@ def _update_metadata():
 # ___ DAB RBAC ___
 
 TEAM_MEMBER_ROLE = 'Galaxy Team Member'
+SHARED_TEAM_ROLE = 'Team Member'
 
 
 def create_managed_roles(*args, **kwargs) -> None:
@@ -495,7 +497,7 @@ def copy_dab_user_role_assignment(sender, instance, created, **kwargs):
  if rbac_signal_in_progress():
  return
  with dab_rbac_signals():
- if instance.role_definition.name == TEAM_MEMBER_ROLE and \
+ if instance.role_definition.name in (TEAM_MEMBER_ROLE, SHARED_TEAM_ROLE) and \
  isinstance(instance, RoleUserAssignment):
  instance.content_object.group.user_set.add(instance.user)
  return
@@ -508,12 +510,18 @@ def delete_dab_user_role_assignment(sender, instance, **kwargs):
  if rbac_signal_in_progress():
  return
  with dab_rbac_signals():
- if instance.role_definition.name == TEAM_MEMBER_ROLE and \
+ if instance.role_definition.name in (TEAM_MEMBER_ROLE, SHARED_TEAM_ROLE) and \
  isinstance(instance, RoleUserAssignment):
+ if instance.role_definition.name == TEAM_MEMBER_ROLE:
+ dup_name = SHARED_TEAM_ROLE
+ else:
+ dup_name = TEAM_MEMBER_ROLE
  # If the assignment does not have a content_object then it may be a global group role
  # this type of role is not compatible with DAB RBAC and what we do is still TBD
  if instance.content_object:
- instance.content_object.group.user_set.remove(instance.user)
+ # Only remove assignment if other role does not grant this, otherwise leave it
+ if not RoleUserAssignment.objects.filter(role_definition__name=dup_name, user=instance.user, object_id=instance.object_id).exists():
+ instance.content_object.group.user_set.remove(instance.user)
  return
  _unapply_dab_assignment(instance)
 
@@ -545,39 +553,41 @@ def copy_dab_group_to_role(instance, action, model, pk_set, reverse, **kwargs):
  return
 
  member_rd = RoleDefinition.objects.get(name=TEAM_MEMBER_ROLE)
+ shared_member_rd = RoleDefinition.objects.get(name=SHARED_TEAM_ROLE)
  if reverse:
- # NOTE: for we might prefer to use pk_set
- # but it appears to be incorrectly empty on the reverse signal,
- # the models itself on post_ actions seems good so we use that
- team = Team.objects.get(group_id=instance.pk)
+ groups = [instance]
+ else:
+ if action == 'post_clear':
+ qs = RoleUserAssignment.objects.filter(role_definition=member_rd, user=instance)
+ groups = [assignment.content_object for assignment in qs]
+ else:
+ groups = Group.objects.filter(pk__in=pk_set)
+
+ for group in groups:
+ team = Team.objects.get(group_id=group.pk)
  current_dab_members = set(
  assignment.user for assignment in RoleUserAssignment.objects.filter(
  role_definition=member_rd, object_id=team.pk
  )
  )
- desired_members = set(instance.user_set.all())
+ current_dab_shared_members = set(
+ assignment.user for assignment in RoleUserAssignment.objects.filter(
+ role_definition=member_rd, object_id=team.pk
+ )
+ )
+ current_pulp_members = set(group.user_set.all())
+ not_allowed = current_dab_shared_members - current_pulp_members
+ if not_allowed:
+ # TODO: this looks non-viable with current JWT data for tests, maybe make into a log?
+ raise ValidationError(f'Can not remove team members from resource provider: {", ".join([u.username for u in not_allowed])}')
+ desired_members = current_pulp_members - current_dab_shared_members
  users_to_add = desired_members - current_dab_members
  users_to_remove = current_dab_members - desired_members
  with dab_rbac_signals():
  for user in users_to_add:
  member_rd.give_permission(user, team)
  for user in users_to_remove:
  member_rd.remove_permission(user, team)
- return
-
- with dab_rbac_signals():
- if action == 'post_add':
- for group_id in pk_set:
- team = Team.objects.get(group_id=group_id)
- member_rd.give_permission(instance, team)
- elif action == 'post_remove':
- for group_id in pk_set:
- team = Team.objects.get(group_id=group_id)
- member_rd.remove_permission(instance, team)
- elif action == 'post_clear':
- qs = RoleUserAssignment.objects.filter(role_definition=member_rd, user=instance)
- for assignment in qs:
- member_rd.remove_permission(instance, assignment.content_object)
 
 
 m2m_changed.connect(copy_dab_group_to_role, sender=User.groups.through)

galaxy_ng/tests/integration/dab/test_dab_rbac_contract.py

@@ -594,7 +594,7 @@ def _rf(user_id, group_id, expected=True):
 
  assignment_r = gc.get(
  "_ui/v2/role_user_assignments/",
- params={"user": user['id'], "content_type__model": "team"},
+ params={"user": user['id'], "content_type__model": "team", "role_definition__name": "Galaxy Team Member"},
  )
  group_ids = [assignment["object_id"] for assignment in assignment_r["results"]]
  if expected: