fix(servicenow): add compact role-based DLS to reduce ACL memory

app/connectors_service/connectors/sources/servicenow/datasource.py

@@ -55,6 +55,7 @@
     "change_request": ["admin", "sn_change_read", "itil"],
 }
 ACLS_QUERY = "sys_security_acl.operation=read^sys_security_acl.name={table_name}"
+PUBLIC_ROLE_NAME = "public"
 
 
 def _prefix_email(email):
@@ -69,6 +70,10 @@ def _prefix_user_id(user_id):
     return prefix_identity("user_id", user_id)
 
 
+def _prefix_role_id(role_id):
+    return prefix_identity("role_id", role_id)
+
+
 class EndSignal(Enum):
     SERVICE = "SERVICE_TASK_FINISHED"
     RECORD = "RECORD_TASK_FINISHED"
@@ -97,6 +102,8 @@ def __init__(self, configuration):
 
         self.servicenow_mapping = {}
         self.invalid_services = []
+        # Lazy per-sync cache of sys_user_role: (by_name, by_sys_id)
+        self._roles_maps = None
 
         self.task_count = 0
         self.queue = MemQueue(maxmemsize=QUEUE_MEM_SIZE, refresh_timeout=120)
@@ -176,6 +183,15 @@ def get_default_configuration(cls):
                 "type": "bool",
                 "value": False,
             },
+            "expand_role_members": {
+                "depends_on": [{"field": "use_document_level_security", "value": True}],
+                "display": "toggle",
+                "label": "Expand role members",
+                "order": 9,
+                "tooltip": "When enabled, ServiceNow role members are written individually onto each document's access control list. Disable this for large tenants to store compact role tokens on documents instead, and resolve membership during access control syncs. Changing this setting requires a full content sync and access control sync.",
+                "type": "bool",
+                "value": True,
+            },
         }
 
     def _dls_enabled(self):
@@ -192,25 +208,53 @@ def _dls_enabled(self):
 
         return self.configuration["use_document_level_security"]
 
-    async def _user_access_control_doc(self, user):
+    def _expand_role_members(self):
+        """Whether roles are expanded into individual users on document ACLs.
+
+        Default True preserves legacy behavior. When False (compact mode), documents
+        receive role_id tokens and membership is resolved on identity docs.
+        """
+        return self.configuration.get("expand_role_members", True)
+
+    async def _user_access_control_doc(self, user, role_ids=None):
         user_id = user.get("_id", "")
         user_name = user.get("user_name", "")
         user_email = user.get("email", "")
 
         _prefixed_user_id = _prefix_user_id(user_id=user_id)
         _prefixed_user_name = _prefix_username(user=user_name)
         _prefixed_email = _prefix_email(email=user_email)
-        return {
-            "_id": user_id,
-            "identity": {
+        access_control = [
+            _prefixed_user_id,
+            _prefixed_user_name,
+            _prefixed_email,
+        ]
+        # role_ids is only set in compact mode; omit the field in legacy mode.
+        if role_ids is not None:
+            role_ids_list = [role_id for role_id in role_ids if role_id]
+            prefixed_role_ids = sorted(
+                prefixed
+                for role_id in role_ids_list
+                if (prefixed := _prefix_role_id(role_id)) is not None
+            )
+            identity = {
                 "user_id": _prefixed_user_id,
                 "display_name": _prefixed_user_name,
                 "email": _prefixed_email,
-            },
+                "role_ids": prefixed_role_ids,
+            }
+            access_control.extend(prefixed_role_ids)
+        else:
+            identity = {
+                "user_id": _prefixed_user_id,
+                "display_name": _prefixed_user_name,
+                "email": _prefixed_email,
+            }
+        return {
+            "_id": user_id,
+            "identity": identity,
             "created_at": user.get("_timestamp"),
-        } | es_access_control_query(
-            access_control=[_prefixed_user_id, _prefixed_user_name, _prefixed_email]
-        )
+        } | es_access_control_query(access_control=access_control)
 
     async def _fetch_all_users(self):
         self._logger.debug("Fetching all users.")
@@ -227,19 +271,59 @@ async def _fetch_users_by_roles(self, role):
         ):
             yield user
 
+    async def _fetch_user_roles_map(self):
+        """Build a map of user sys_id -> set of role sys_ids from sys_user_has_role.
+
+        API fetches are already paginated via ``_table_data_generator``
+        (``TABLE_FETCH_SIZE``). Only the user→roles map is held in memory for the
+        ACL sync: O(role assignments), not O(users × documents). For large tenants
+        that is tens of MB, versus the GB-scale per-document ACL expansion this
+        compact mode replaces.
+        """
+        user_roles = {}
+        async for assignment in self._table_data_generator(
+            service_name="sys_user_has_role", params={}
+        ):
+            user_id = (assignment.get("user") or {}).get("value")
+            role_id = (assignment.get("role") or {}).get("value")
+            if not user_id or not role_id:
+                self._logger.debug(
+                    "Skipping sys_user_has_role row with missing user or role reference"
+                )
+                continue
+            user_roles.setdefault(user_id, set()).add(role_id)
+        return user_roles
+
     async def get_access_control(self):
         if not self._dls_enabled():
             self._logger.warning("DLS is not enabled. Skipping")
             return
 
+        user_roles = None
+        if not self._expand_role_members():
+            self._logger.info(
+                "Enriching identity docs with role memberships from sys_user_has_role"
+            )
+            user_roles = await self._fetch_user_roles_map()
+
         async for user in self._fetch_all_users():
-            yield await self._user_access_control_doc(user=user)
+            if user_roles is None:
+                yield await self._user_access_control_doc(user=user)
+            else:
+                user_id = user.get("_id") or user.get("sys_id")
+                yield await self._user_access_control_doc(
+                    user=user, role_ids=user_roles.get(user_id, set())
+                )
 
     def _decorate_with_access_control(self, document, access_control):
-        if self._dls_enabled():
-            document[ACCESS_CONTROL] = list(
-                set(document.get(ACCESS_CONTROL, []) + access_control)
-            )
+        if not self._dls_enabled():
+            return document
+        if access_control is None:
+            # Public table: omit field so DLS must_not exists grants access.
+            return document
+        document[ACCESS_CONTROL] = sorted(
+            set(document.get(ACCESS_CONTROL, []) + access_control)
+        )
         return document
 
     async def _remote_validation(self):
@@ -411,25 +495,120 @@ async def _fetch_table_data(self, batched_apis, table_access_control):
         finally:
             await self.queue.put(EndSignal.RECORD)
 
-    async def _fetch_access_controls(self, table_name):
-        access_control, user_roles, roles = [], [], {}
-        if table_name in DEFAULT_SERVICE_NAMES.keys():
-            async for role in self._table_data_generator(
-                service_name="sys_user_role", params={}
-            ):
-                roles[role.get("name")] = role.get("sys_id")
+    async def _get_roles_maps(self):
+        """Return cached (by_name, by_sys_id) maps from sys_user_role for this sync."""
+        if self._roles_maps is not None:
+            return self._roles_maps
+        by_name, by_sys_id = {}, {}
+        async for role in self._table_data_generator(
+            service_name="sys_user_role", params={}
+        ):
+            name = role.get("name")
+            sys_id = role.get("sys_id")
+            if name and sys_id:
+                by_name[name] = sys_id
+                by_sys_id[sys_id] = name
+        self._roles_maps = (by_name, by_sys_id)
+        return self._roles_maps
+
+    async def _table_read_role_sys_ids(self, table_name):
+        """Return role sys_ids that grant read on a custom (non-default) table."""
+        self._logger.info(f"Fetching roles of {table_name} with read operation.")
+        acl_params = {
+            "sys_security_acl.operation": "read",
+            "sys_security_acl.name": table_name,
+            "sys_security_acl.script": "",
+            "sys_security_acl.condition": "",
+        }
+        role_sys_ids = []
+        async for acl in self._table_data_generator(
+            service_name="sys_security_acl_role", params=acl_params
+        ):
+            role_sys_id = (acl.get("sys_user_role") or {}).get("value")
+            if role_sys_id:
+                role_sys_ids.append(role_sys_id)
+        return role_sys_ids
+
+    async def _fetch_access_controls_compact(self, table_name):
+        """Return role_id tokens for a table, or None when the table is public."""
+        if table_name in DEFAULT_SERVICE_NAMES:
+            roles_by_name, _ = await self._get_roles_maps()
+            role_sys_ids = []
+            for role_name in DEFAULT_SERVICE_NAMES.get(table_name, []):
+                if role_name.lower() == PUBLIC_ROLE_NAME:
+                    self._logger.info(
+                        f"Compact DLS: table {table_name} has public read role — "
+                        "omitting _allow_access_control on content documents"
+                    )
+                    return None
+                role_sys_id = roles_by_name.get(role_name)
+                if role_sys_id:
+                    role_sys_ids.append(role_sys_id)
+                else:
+                    self._logger.debug(
+                        f"Role '{role_name}' not found while building compact ACL "
+                        f"for table {table_name}"
+                    )
+            return self._finalize_compact_access_control(table_name, role_sys_ids)
+
+        # Resolve table ACL roles first so we skip loading sys_user_role when empty.
+        role_sys_ids = await self._table_read_role_sys_ids(table_name)
+        if not role_sys_ids:
+            self._logger.info(
+                f"Compact DLS: no read roles for table {table_name}; "
+                "omitting _allow_access_control (treating as world-readable)"
+            )
+            return None
+
+        _, roles_by_sys_id = await self._get_roles_maps()
+        resolved = []
+        for role_sys_id in role_sys_ids:
+            role_name = roles_by_sys_id.get(role_sys_id)
+            if not role_name:
+                self._logger.warning(
+                    f"Skipping unknown role sys_id {role_sys_id} for table {table_name}"
+                )
+                continue
+            if role_name.lower() == PUBLIC_ROLE_NAME:
+                self._logger.info(
+                    f"Compact DLS: table {table_name} has public read role — "
+                    "omitting _allow_access_control on content documents"
+                )
+                return None
+            resolved.append(role_sys_id)
+        return self._finalize_compact_access_control(table_name, resolved)
+
+    def _finalize_compact_access_control(self, table_name, role_sys_ids):
+        compact_acl = sorted(
+            prefixed
+            for role_id in role_sys_ids
+            if role_id and (prefixed := _prefix_role_id(role_id)) is not None
+        )
+        if not compact_acl:
+            self._logger.info(
+                f"Compact DLS: no read roles for table {table_name}; "
+                "omitting _allow_access_control (treating as world-readable)"
+            )
+            return None
+        return compact_acl
 
+    async def _fetch_access_controls_legacy(self, table_name):
+        """Expand role members into individual user_id tokens (legacy behavior)."""
+        access_control, user_roles = [], []
+        roles_by_name, roles_by_sys_id = await self._get_roles_maps()
+        if table_name in DEFAULT_SERVICE_NAMES.keys():
             for role in DEFAULT_SERVICE_NAMES.get(table_name, []):
-                async for user in self._fetch_users_by_roles(roles[role]):
+                role_sys_id = roles_by_name.get(role)
+                if not role_sys_id:
+                    self._logger.warning(
+                        f"Skipping unknown role '{role}' for table {table_name}"
+                    )
+                    continue
+                async for user in self._fetch_users_by_roles(role_sys_id):
                     access_control.append(
                         _prefix_user_id(user_id=user.get("user", {}).get("value"))
                     )
         else:
-            async for role in self._table_data_generator(
-                service_name="sys_user_role", params={}
-            ):
-                roles[role.get("sys_id")] = role.get("name")
-
             self._logger.info(f"Fetching roles of {table_name} with read operation.")
             acl_params = {
                 "sys_security_acl.operation": "read",
@@ -443,7 +622,13 @@ async def _fetch_access_controls(self, table_name):
                 user_roles.append(acl.get("sys_user_role", {}).get("value"))
 
             for role in user_roles:
-                if roles.get(role).lower() == "public":
+                role_name = roles_by_sys_id.get(role)
+                if not role_name:
+                    self._logger.warning(
+                        f"Skipping unknown role sys_id {role} for table {table_name}"
+                    )
+                    continue
+                if role_name.lower() == PUBLIC_ROLE_NAME:
                     self._logger.info(
                         f"Found public role in {table_name}, Fetching all users."
                     )
@@ -456,7 +641,12 @@ async def _fetch_access_controls(self, table_name):
                     access_control.append(
                         _prefix_user_id(user_id=user.get("user", {}).get("value"))
                     )
-        return list(set(access_control))
+        return sorted(set(access_control))
+
+    async def _fetch_access_controls(self, table_name):
+        if self._expand_role_members():
+            return await self._fetch_access_controls_legacy(table_name)
+        return await self._fetch_access_controls_compact(table_name)
 
     async def _get_batched_apis(self, service_name, params):
         table_length = await self.servicenow_client.get_table_length(
@@ -546,14 +736,29 @@ async def get_docs(self, filtering=None):
                         advanced_rules_index + TABLE_BATCH_SIZE
                     )  # noqa
                 ]
-                filter_apis = await self.servicenow_client.get_filter_apis(
-                    rules=batched_advanced_rules, mapping=servicenow_mapping
-                )
+                rules_by_service = {}
+                for rule in batched_advanced_rules:
+                    rules_by_service.setdefault(rule["service"], []).append(rule)
+
+                for service_label, service_rules in rules_by_service.items():
+                    table_name = servicenow_mapping[service_label]
+                    table_access_control = None
+                    if self._dls_enabled():
+                        table_access_control = await self._fetch_access_controls(
+                            table_name=table_name
+                        )
+                    filter_apis = await self.servicenow_client.get_filter_apis(
+                        rules=service_rules, mapping=servicenow_mapping
+                    )
 
-                await self.fetchers.put(
-                    partial(self._fetch_table_data, filter_apis, [])
-                )
-                self.task_count += 1
+                    await self.fetchers.put(
+                        partial(
+                            self._fetch_table_data,
+                            filter_apis,
+                            table_access_control,
+                        )
+                    )
+                    self.task_count += 1
 
         else:
             if (
@@ -569,7 +774,7 @@ async def get_docs(self, filtering=None):
             for service_name in (
                 self.servicenow_mapping.values() or DEFAULT_SERVICE_NAMES.keys()
             ):
-                table_access_control = []
+                table_access_control = None
                 if self._dls_enabled():
                     table_access_control = await self._fetch_access_controls(
                         table_name=service_name