Merge pull request RedHatInsights#1172 from RedHatInsights/dual_write…

…_for_role

Dual Write: Generate replication events for Role endpoints

deploy/rbac-clowdapp.yml

@@ -220,7 +220,8 @@ objects:
             value: ${SA_NAME}
           - name: RELATION_API_SERVER
             value: ${RELATION_API_SERVER}
-
+          - name: REPLICATION_TO_RELATION_ENABLED
+            value: ${REPLICATION_TO_RELATION_ENABLED}
     - name: scheduler-service
       minReplicas: ${{MIN_SCHEDULER_REPLICAS}}
       metadata:
@@ -291,6 +292,8 @@ objects:
               name: ${GLITCHTIP_SECRET}
               key: dsn
               optional: true
+        - name: PRINCIPAL_USER_DOMAIN
+          value: ${PRINCIPAL_USER_DOMAIN}
         - name: PRINCIPAL_CLEANUP_DELETION_ENABLED_UMB
           value: ${PRINCIPAL_CLEANUP_DELETION_ENABLED_UMB}
         - name: UMB_JOB_ENABLED
@@ -459,8 +462,14 @@ objects:
             value: ${GROUP_SEEDING_ENABLED}
           - name: BYPASS_BOP_VERIFICATION
             value: ${BYPASS_BOP_VERIFICATION}
+          - name: REPLICATION_TO_RELATION_ENABLED
+            value: ${REPLICATION_TO_RELATION_ENABLED}
           - name: ROLE_CREATE_ALLOW_LIST
             value: ${ROLE_CREATE_ALLOW_LIST}
+          - name: V2_MIGRATION_APP_EXCLUDE_LIST
+            value: ${V2_MIGRATION_APP_EXCLUDE_LIST}
+          - name: V2_MIGRATION_RESOURCE_EXCLUDE_LIST
+            value: ${V2_MIGRATION_RESOURCE_EXCLUDE_LIST}
           - name: RBAC_DESTRUCTIVE_API_ENABLED_UNTIL
             value: ${RBAC_DESTRUCTIVE_API_ENABLED_UNTIL}
           - name: RBAC_DESTRUCTIVE_SEEDING_ENABLED_UNTIL
@@ -739,6 +748,12 @@ parameters:
 - description: Application allow list for role creation in RBAC
   name: ROLE_CREATE_ALLOW_LIST
   value: cost-management,remediations,inventory,drift,policies,advisor,vulnerability,compliance,automation-analytics,notifications,patch,integrations,ros,staleness,config-manager,idmsvc
+- description: Application exclude list for v2 migration (all permissions)
+  name: V2_MIGRATION_APP_EXCLUDE_LIST
+  value: approval
+- description: Resources (by namespace:name) exclude list for v2 migration (resource definitions only)
+  name: V2_MIGRATION_RESOURCE_EXCLUDE_LIST
+  value: rbac:workspace
 - description: Timestamp expiration allowance on destructive actions through the internal RBAC API
   name: RBAC_DESTRUCTIVE_API_ENABLED_UNTIL
   value: ''
@@ -899,6 +914,14 @@ parameters:
   value: '10'
 - name: IT_TOKEN_JKWS_CACHE_LIFETIME
   value: '28800'
+- name: PRINCIPAL_USER_DOMAIN
+  description: >
+    Kessel requires principal IDs to be qualified by a domain,
+    in order to future proof integration of identities from multiple issuers.
+    RBAC currently expects all principals to either come from itself (cross-account),
+    or from a single identity infrastructure domain (identity header, SSO, BOP).
+    This defines that single domain.
+  value: 'redhat.com'
 - name: PRINCIPAL_CLEANUP_DELETION_ENABLED_UMB
   description: Allow cleanup job to delete principals via messages from UMB
   value: 'False'
@@ -920,6 +943,9 @@ parameters:
 - name: RELATION_API_SERVER
   description: The gRPC API server to use for the relation
   value: "localhost:9000"
+- name: REPLICATION_TO_RELATION_ENABLED
+  description: Enable replication to Relation API
+  value: "False"
 - name: V2_APIS_ENABLED
   description: Flag to explicitly enable v2 API endpoints
 - name: READ_ONLY_API_MODE

rbac/internal/views.py

@@ -16,11 +16,13 @@
 #
 
 """View for internal tenant management."""
+
 import json
 import logging
 
 import requests
 from core.utils import destructive_ok
+from django.conf import settings
 from django.db import transaction
 from django.db.migrations.recorder import MigrationRecorder
 from django.http import HttpResponse
@@ -469,27 +471,28 @@ def ocm_performance(request):
     return HttpResponse('Invalid method, only "POST" is allowed.', status=405)
 
 
-def get_param_list(request, param_name):
+def get_param_list(request, param_name, default: list = []):
     """Get a list of params from a request."""
     params = request.GET.get(param_name, [])
     if params:
-        params = params.split(",")
-    return params
+        return params.split(",")
+    else:
+        return default
 
 
 def data_migration(request):
     """View method for running migrations from V1 to V2 spiceDB schema.
 
-    POST /_private/api/utils/data_migration/?exclude_apps=cost_management,rbac&orgs=id_1,id_2&write_db=True
+    POST /_private/api/utils/data_migration/?exclude_apps=cost_management,rbac&orgs=id_1,id_2&write_relationships=True
     """
     if request.method != "POST":
         return HttpResponse('Invalid method, only "POST" is allowed.', status=405)
     logger.info("Running V1 data migration.")
 
     args = {
-        "exclude_apps": get_param_list(request, "exclude_apps"),
+        "exclude_apps": get_param_list(request, "exclude_apps", default=settings.V2_MIGRATION_APP_EXCLUDE_LIST),
         "orgs": get_param_list(request, "orgs"),
-        "write_db": request.GET.get("write_db", "False") == "True",
+        "write_relationships": request.GET.get("write_relationships", "False") == "True",
     }
     migrate_data_in_worker.delay(args)
     return HttpResponse("Data migration from V1 to V2 are running in a background worker.", status=202)

rbac/management/group/definer.py

@@ -17,13 +17,17 @@
 
 """Handler for system defined group."""
 import logging
+from typing import Union
 from uuid import uuid4
 
 from django.db import transaction
-from django.db.models import Q
 from django.db.models.query import QuerySet
 from django.utils.translation import gettext as _
 from management.group.model import Group
+from management.group.relation_api_dual_write_group_handler import (
+    RelationApiDualWriteGroupHandler,
+    ReplicationEventType,
+)
 from management.notifications.notification_handlers import (
     group_flag_change_notification_handler,
     group_role_change_notification_handler,
@@ -35,6 +39,7 @@
 
 from api.models import Tenant
 
+
 logger = logging.getLogger(__name__)  # pylint: disable=invalid-name
 
 
@@ -106,15 +111,11 @@ def clone_default_group_in_public_schema(group, tenant):
     return group
 
 
+@transaction.atomic
 def add_roles(group, roles_or_role_ids, tenant, user=None):
     """Process list of roles and add them to the group."""
-    if not isinstance(roles_or_role_ids, QuerySet):
-        # If given an iterable of UUIDs, get the corresponding objects
-        roles = Role.objects.filter(uuid__in=roles_or_role_ids)
-    else:
-        roles = roles_or_role_ids
+    roles = _roles_by_query_or_ids(roles_or_role_ids)
     group_name = group.name
-    role_names = list(roles.values_list("name", flat=True))
 
     group, created = Group.objects.get_or_create(name=group_name, tenant=tenant)
     system_policy_name = "System Policy for Group {}".format(group.uuid)
@@ -125,10 +126,14 @@ def add_roles(group, roles_or_role_ids, tenant, user=None):
     if system_policy_created:
         logger.info(f"Created new system policy for tenant {tenant.org_id}.")
 
-    roles = Role.objects.filter(
-        Q(tenant=tenant) | Q(tenant=Tenant.objects.get(tenant_name="public")), name__in=role_names
-    )
-    for role in roles:
+    system_roles = roles.filter(tenant=Tenant.objects.get(tenant_name="public"))
+
+    # Custom roles are locked to prevent resources from being added/removed concurrently,
+    # in the case that the Roles had _no_ resources specified to begin with.
+    # This should not be necessary for system roles.
+    custom_roles = roles.filter(tenant=tenant).select_for_update()
+
+    for role in [*system_roles, *custom_roles]:
         # Only Organization administrators are allowed to add the role with RBAC permission
         # higher than "read" into a group.
         for access in role.access.all():
@@ -144,31 +149,41 @@ def add_roles(group, roles_or_role_ids, tenant, user=None):
                     "into groups."
                 )
                 raise serializers.ValidationError({key: _(message)})
+
         # Only add the role if it was not attached
-        if not system_policy.roles.filter(pk=role.pk).exists():
-            system_policy.roles.add(role)
+        if system_policy.roles.filter(pk=role.pk).exists():
+            continue
 
-            # Send notifications
-            group_role_change_notification_handler(user, group, role, "added")
+        system_policy.roles.add(role)
 
+        dual_write_handler = RelationApiDualWriteGroupHandler(group, ReplicationEventType.ASSIGN_ROLE, [])
+        dual_write_handler.replicate_added_role(role)
+        # Send notifications
+        group_role_change_notification_handler(user, group, role, "added")
 
+
+@transaction.atomic
 def remove_roles(group, roles_or_role_ids, tenant, user=None):
     """Process list of roles and remove them from the group."""
-    if not isinstance(roles_or_role_ids, QuerySet):
-        # If given an iterable of UUIDs, get the corresponding objects
-        roles = Role.objects.filter(uuid__in=roles_or_role_ids)
-    else:
-        roles = roles_or_role_ids
-    role_names = list(roles.values_list("name", flat=True))
-
+    roles = _roles_by_query_or_ids(roles_or_role_ids)
     group = Group.objects.get(name=group.name, tenant=tenant)
-    roles = group.roles().filter(name__in=role_names)
+    system_roles = roles.filter(tenant=Tenant.objects.get(tenant_name="public"))
+
+    # Custom roles are locked to prevent resources from being added/removed concurrently,
+    # in the case that the Roles had _no_ resources specified to begin with.
+    # This should not be necessary for system roles.
+    custom_roles = roles.filter(tenant=tenant).select_for_update()
+
     for policy in group.policies.all():
-        # Only remove the role if it was attached
-        for role in roles:
+        for role in [*system_roles, *custom_roles]:
+            # Only remove the role if it was attached
             if policy.roles.filter(pk=role.pk).exists():
                 policy.roles.remove(role)
                 logger.info(f"Removing role {role} from group {group.name} for tenant {tenant.org_id}.")
+
+                dual_write_handler = RelationApiDualWriteGroupHandler(group, ReplicationEventType.UNASSIGN_ROLE, [])
+                dual_write_handler.replicate_removed_role(role)
+
                 # Send notifications
                 group_role_change_notification_handler(user, group, role, "removed")
 
@@ -182,3 +197,16 @@ def update_group_roles(group, roleset, tenant):
     role_ids = list(roleset.values_list("uuid", flat=True))
     roles_to_remove = group.roles().exclude(uuid__in=role_ids)
     remove_roles(group, roles_to_remove, tenant)
+
+
+def _roles_by_query_or_ids(roles_or_role_ids: Union[QuerySet[Role], list[str]]) -> QuerySet[Role]:
+    if not isinstance(roles_or_role_ids, QuerySet):
+        # If given an iterable of UUIDs, get the corresponding objects
+        return Role.objects.filter(uuid__in=roles_or_role_ids)
+    else:
+        # Given a queryset, so because it may not be efficient (e.g. query on non indexed field)
+        # keep prior behavior of querying once to get names, then use names (indexed) as base query
+        # for further queries.
+        # It MAY be faster to avoid this extra query, but this maintains prior behavior.
+        role_names = list(roles_or_role_ids.values_list("name", flat=True))
+        return Role.objects.filter(name__in=role_names)