KC-1468,1473,1474: Protect SailPoint config record from Service Mode and fix ${VARNAME} expansion bypass

keepercommander/service/commands/integrations/sailpoint/command_hook.py

@@ -81,7 +81,7 @@ def before_command(self, params: KeeperParams, command: str) -> Tuple[str, Optio
 
         mutation = SailPointCommandParser.parse_identity_mutation(command)
         if mutation:
-            err = self._first_scim_identity_error(params, mutation.emails)
+            err = self._first_scim_identity_error(params, mutation.identifiers)
             if err:
                 return self._reject(command, err, 403)
         return command, None

keepercommander/service/commands/integrations/sailpoint/command_parse.py

@@ -63,7 +63,7 @@ def is_grant(self) -> bool:
 class ParsedIdentityMutation:
     """enterprise-user identity change (role/team/node) — used for SCIM coexistence."""
 
-    emails: List[str] = field(default_factory=list)
+    identifiers: List[str] = field(default_factory=list)
     has_role_change: bool = False
     has_team_change: bool = False
     has_node_change: bool = False
@@ -134,14 +134,15 @@ def parse_identity_mutation(cls, command: str) -> Optional[ParsedIdentityMutatio
         if not parsed:
             return None
         ns, _unknown = parsed
-        emails = [e for e in (ns.email or []) if isinstance(e, str) and '@' in e]
+        # No '@' filter here (unlike parse_invite/parse_transfer): the SCIM guard must see IDs and '@all' too.
+        identifiers = [e for e in (ns.email or []) if isinstance(e, str) and e.strip()]
         has_role = bool(ns.add_role or ns.remove_role)
         has_team = bool(ns.add_team or ns.remove_team)
         has_node = bool(ns.node)
-        if not (has_role or has_team or has_node) or not emails:
+        if not (has_role or has_team or has_node) or not identifiers:
             return None
         return ParsedIdentityMutation(
-            emails=emails,
+            identifiers=identifiers,
             has_role_change=has_role,
             has_team_change=has_team,
             has_node_change=has_node,

keepercommander/service/commands/integrations/sailpoint/scim_guard.py

@@ -13,11 +13,13 @@
 
 from __future__ import annotations
 
-from typing import Iterable, Optional, Set
+from typing import Any, Dict, Iterable, Optional, Set
 
 from ..... import api
 from .....params import KeeperParams
 
+_ALL_PSEUDO_USER = '@all'
+
 
 class SailPointScimGuard:
     """Detect SCIM-managed users/nodes and block identity stomps."""
@@ -40,15 +42,21 @@ def _scim_node_ids(params: KeeperParams) -> Set[int]:
         return nodes
 
     @staticmethod
-    def _node_ancestors(params: KeeperParams, node_id: int) -> Iterable[int]:
-        by_id = {int(n['node_id']): n for n in params.enterprise.get('nodes') or []}
+    def _node_parents(params: KeeperParams) -> Dict[int, Optional[int]]:
+        return {
+            int(n['node_id']): (int(n['parent_id']) if n.get('parent_id') else None)
+            for n in params.enterprise.get('nodes') or []
+        }
+
+    @staticmethod
+    def _walk_ancestors(node_id: int, parents: Dict[int, Optional[int]]) -> Iterable[int]:
         current: Optional[int] = node_id
         seen = set()
         while current and current not in seen:
             yield current
             seen.add(current)
-            parent = by_id.get(current, {}).get('parent_id')
-            current = int(parent) if parent and int(parent) != current else None
+            parent = parents.get(current)
+            current = parent if parent != current else None
 
     @classmethod
     def is_scim_managed_node(cls, params: KeeperParams, node_id: Optional[int]) -> bool:
@@ -58,35 +66,82 @@ def is_scim_managed_node(cls, params: KeeperParams, node_id: Optional[int]) -> b
         scim_nodes = cls._scim_node_ids(params)
         if not scim_nodes:
             return False
-        return any(n in scim_nodes for n in cls._node_ancestors(params, int(node_id)))
+        parents = cls._node_parents(params)
+        return any(n in scim_nodes for n in cls._walk_ancestors(int(node_id), parents))
+
+    @staticmethod
+    def _normalize(value: Any) -> str:
+        return str(value).strip().lower() if isinstance(value, str) else ''
 
     @classmethod
-    def find_user(cls, params: KeeperParams, email: str):
+    def _build_user_lookup(cls, params: KeeperParams) -> Dict[str, Any]:
+        """Mirrors EnterpriseUserCommand's own lookup: id, username, then aliases."""
+        lookup: Dict[str, Any] = {}
+        for user in params.enterprise.get('users') or []:
+            if 'enterprise_user_id' in user:
+                lookup[str(user['enterprise_user_id']).strip()] = user
+            username = cls._normalize(user.get('username'))
+            if username:
+                lookup[username] = user
+        for alias in params.enterprise.get('user_aliases') or []:
+            username = cls._normalize(alias.get('username'))
+            if username and username not in lookup:
+                user_id = str(alias.get('enterprise_user_id')).strip()
+                if user_id in lookup:
+                    lookup[username] = lookup[user_id]
+        return lookup
+
+    @classmethod
+    def _is_all_pseudo_user(cls, identifier: str) -> bool:
+        return cls._normalize(identifier) == _ALL_PSEUDO_USER
+
+    @classmethod
+    def find_user(cls, params: KeeperParams, identifier: str) -> Optional[Dict[str, Any]]:
         cls.ensure_enterprise(params)
         if not params.enterprise:
             return None
-        target = email.strip().lower()
-        return next(
-            (
-                user for user in params.enterprise.get('users') or []
-                if (user.get('username') or '').lower() == target
-            ),
-            None,
-        )
+        target = cls._normalize(identifier)
+        if not target:
+            return None
+        return cls._build_user_lookup(params).get(target)
 
     @classmethod
-    def is_scim_managed_user(cls, params: KeeperParams, email: str) -> bool:
-        user = cls.find_user(params, email)
+    def is_scim_managed_user(cls, params: KeeperParams, identifier: str) -> bool:
+        user = cls.find_user(params, identifier)
         if not user:
             return False
         return cls.is_scim_managed_node(params, user.get('node_id'))
 
     @classmethod
-    def identity_change_error(cls, params: KeeperParams, email: str) -> Optional[str]:
-        if cls.is_scim_managed_user(params, email):
+    def is_scim_managed_identifier(cls, params: KeeperParams, identifier: str) -> bool:
+        """Like is_scim_managed_user, but '@all' matches if any user is SCIM-managed."""
+        if not cls._is_all_pseudo_user(identifier):
+            return cls.is_scim_managed_user(params, identifier)
+        cls.ensure_enterprise(params)
+        if not params.enterprise:
+            return False
+        scim_nodes = cls._scim_node_ids(params)
+        if not scim_nodes:
+            return False
+        parents = cls._node_parents(params)
+        return any(
+            any(n in scim_nodes for n in cls._walk_ancestors(int(user['node_id']), parents))
+            for user in params.enterprise.get('users') or []
+            if user.get('node_id')
+        )
+
+    @classmethod
+    def identity_change_error(cls, params: KeeperParams, identifier: str) -> Optional[str]:
+        if not cls.is_scim_managed_identifier(params, identifier):
+            return None
+        if cls._is_all_pseudo_user(identifier):
             return (
-                f'User {email} is managed by an existing SCIM provider. '
+                '@all includes one or more users managed by an existing SCIM provider. '
                 'SailPoint may only change folder/record (and admin) entitlements; '
-                'node/team/role identity changes are not allowed.'
+                'node/team/role identity changes are not allowed for those users.'
             )
-        return None
+        return (
+            f'User {identifier} is managed by an existing SCIM provider. '
+            'SailPoint may only change folder/record (and admin) entitlements; '
+            'node/team/role identity changes are not allowed.'
+        )

keepercommander/service/commands/integrations/sailpoint/service.py

@@ -25,14 +25,7 @@
 
 
 class SailPointService:
-    """
-    Entry point for SailPoint Service Mode.
-
-    Pending entitlements and SailPoint settings live on a dedicated vault
-    config record (``Commander Service Mode SailPoint Config``). Runtime
-    resolves that UID from ``SAILPOINT_RECORD`` (compose env) or
-    ``params.sailpoint_record_uid``. Docker config stays on ``COMMANDER_RECORD``.
-    """
+    """Entry point for SailPoint Service Mode; settings/pending entitlements live on the record pinned by SAILPOINT_RECORD."""
 
     PARAMS_ATTR = PARAMS_ATTR
 
@@ -50,6 +43,16 @@ def record_has_marker(cls, params: KeeperParams, record_uid: str) -> bool:
             for field in record.custom
         )
 
+    @classmethod
+    def _record_has_marker_safe(cls, params: KeeperParams, uid: str) -> bool:
+        """record_has_marker, but never raises; startup-only -- handle_command/after_command
+        must call record_has_marker directly so a check failure denies the request instead of silently skipping it."""
+        try:
+            return cls.record_has_marker(params, uid)
+        except Exception as e:
+            logger.warning(f'SailPoint: marker check failed for record {uid}: {e}')
+            return False
+
     @classmethod
     def record_uid(cls, params: Optional[KeeperParams] = None) -> Optional[str]:
         if params is not None:
@@ -68,22 +71,13 @@ def bind_params(cls, params: KeeperParams, record_uid: Optional[str] = None) ->
 
     @classmethod
     def maybe_enable(cls, params: KeeperParams, args) -> None:
-        """
-        Bind params and sanitize the Service Mode command allowlist when the
-        SailPoint config record has the integration marker.
-
-        Callers must gate on ``SAILPOINT_RECORD`` before invoking this.
-        """
+        """Bind params and sanitize the allowlist when the config record has the marker; callers must gate on SAILPOINT_RECORD first."""
         uid = cls.record_uid(params)
-        try:
-            if not cls.record_has_marker(params, uid):
-                logger.warning(
-                    f'{SAILPOINT_RECORD_ENV}={uid} is set but record is missing '
-                    f'{SAILPOINT_MARKER_FIELD}; SailPoint mode not enabled'
-                )
-                return
-        except Exception as e:
-            logger.warning(f'SailPoint marker check failed; mode not enabled: {e}')
+        if not cls._record_has_marker_safe(params, uid):
+            logger.warning(
+                f'{SAILPOINT_RECORD_ENV}={uid} is set but record is missing '
+                f'{SAILPOINT_MARKER_FIELD}; SailPoint mode not enabled'
+            )
             return
 
         cls.bind_params(params, uid)
@@ -98,25 +92,17 @@ def maybe_enable(cls, params: KeeperParams, args) -> None:
 
     @classmethod
     def start_background_services(cls) -> None:
-        """
-        Start the entitlement poller when SailPoint is enabled.
-
-        Callers must gate on ``SAILPOINT_RECORD`` before invoking this.
-        """
+        """Start the entitlement poller; callers must gate on SAILPOINT_RECORD before invoking this."""
         from ....core.globals import get_current_params
         params = get_current_params()
         if not params:
             logger.warning('SailPoint poller not started: Keeper params not loaded')
             return
         uid = cls.record_uid(params)
-        try:
-            if not cls.record_has_marker(params, uid):
-                logger.warning(
-                    f'SailPoint poller not started: record {uid} missing {SAILPOINT_MARKER_FIELD}'
-                )
-                return
-        except Exception as e:
-            logger.warning(f'SailPoint poller not started: marker check failed: {e}')
+        if not cls._record_has_marker_safe(params, uid):
+            logger.warning(
+                f'SailPoint poller not started: record {uid} missing {SAILPOINT_MARKER_FIELD}'
+            )
             return
         cls.bind_params(params, uid)
         try:
@@ -129,15 +115,12 @@ def start_background_services(cls) -> None:
     def handle_command(
         cls, params: KeeperParams, command: str
     ) -> Tuple[str, Optional[Tuple[Any, int]]]:
-        """
-        Prepare a SailPoint Service Mode command.
-
-        Returns ``(command_to_run, short_circuit)``. Callers must gate on
-        ``SAILPOINT_RECORD`` before invoking this.
-        """
+        """Prepare a SailPoint command, returning (command_to_run, short_circuit); callers must gate on SAILPOINT_RECORD first."""
         cls.bind_params(params)
         uid = cls.record_uid(params)
+        # Not swallowed here -- a check failure must deny the request, not silently skip gating.
         if not cls.record_has_marker(params, uid):
+            logger.debug(f'SailPoint: record {uid} has no active marker; skipping all command gating for this request')
             return command, None
         return SailPointCommandHook(uid).before_command(params, command)
 
@@ -146,6 +129,8 @@ def after_command(cls, params: KeeperParams, command: str, success: bool = True)
         """Callers must gate on ``SAILPOINT_RECORD`` before invoking this."""
         cls.bind_params(params)
         uid = cls.record_uid(params)
+        # Not swallowed here either -- callers turn a raised exception into an explicit 500.
         if not cls.record_has_marker(params, uid):
+            logger.debug(f'SailPoint: record {uid} has no active marker; skipping pending-entitlement queue for this request')
             return
         SailPointCommandHook(uid).after_command(params, command, success)

keepercommander/service/util/command_util.py

@@ -219,10 +219,15 @@ def blocked(error):
             if protected_command_error:
                 return blocked(protected_command_error)
 
-            sailpoint_enabled = bool((os.environ.get('SAILPOINT_RECORD') or '').strip())
+            sailpoint_uid = (os.environ.get('SAILPOINT_RECORD') or '').strip()
+            sailpoint_enabled = bool(sailpoint_uid)
 
-            with hide_from_record_cache(params, protected_uids), \
-                 hide_from_folder_cache(params, protected_folder_uids):
+            # Only SailPoint's own record is exempt from handle_command's guard; every other
+            # protected record (and the folder cache, which has no exemption) stays hidden throughout.
+            handle_command_uids = {uid: title for uid, title in protected_uids.items() if uid != sailpoint_uid}
+
+            with hide_from_folder_cache(params, protected_folder_uids), \
+                    hide_from_record_cache(params, handle_command_uids):
                 if sailpoint_enabled:
                     from ..commands.integrations.sailpoint.service import SailPointService
                     command, sailpoint_response = SailPointService.handle_command(params, command)
@@ -231,7 +236,8 @@ def blocked(error):
                         response = CommandExecutor.encrypt_response(response)
                         return response, status_code
 
-                return_value, printed_output, log_output = CommandExecutor.capture_output_and_logs(params, command)
+                with hide_from_record_cache(params, protected_uids):
+                    return_value, printed_output, log_output = CommandExecutor.capture_output_and_logs(params, command)
             response = return_value if return_value else printed_output
 
             # Debug logging with sanitization
@@ -247,7 +253,10 @@ def blocked(error):
 
             if status_code == 200 and sailpoint_enabled:
                 try:
-                    SailPointService.after_command(params, command, success=True)
+                    # Same rule as the pre-dispatch phase: every protected record except
+                    # SailPoint's own stays hidden here too.
+                    with hide_from_record_cache(params, handle_command_uids):
+                        SailPointService.after_command(params, command, success=True)
                 except Exception as e:
                     logger.error(f'SailPoint post-process failed: {sanitize_debug_data(str(e))}')
                     err = {

keepercommander/service/util/protected_records.py

@@ -16,6 +16,7 @@
 import contextlib
 import os
 from collections import UserDict
+from collections.abc import MutableMapping
 from typing import Dict, FrozenSet, Iterable, Optional, Set, Tuple
 
 # Each integration's own setup pins its config record's UID here, regardless of its title. Extend when a new integration gets an always-hidden record.
@@ -25,6 +26,7 @@
     'SLACK_RECORD': '<Slack config record>',
     'TEAMS_RECORD': '<Teams config record>',
     'GCHAT_RECORD': '<GChat config record>',
+    'SAILPOINT_RECORD': '<SailPoint config record>',
 }
 
 # uid-keyed caches resolve_single_record/load_pam_record fall back to when a UID isn't in record_cache.
@@ -70,6 +72,7 @@ def _protected_titles() -> Tuple[str, ...]:
     from ..commands.terraform_app_setup import TerraformSetupConstants
     from ..commands.integrations.slack_app_setup import SlackAppSetupCommand
     from ..commands.integrations.teams_app_setup import TeamsAppSetupCommand
+    from ..commands.integrations.sailpoint_app_setup import SailPointAppSetupCommand
     from ..docker.models import DockerSetupConstants, GChatConstants
     return (
         *SERVICE_CONFIG_RECORD_TITLES,
@@ -78,6 +81,7 @@ def _protected_titles() -> Tuple[str, ...]:
         GChatConstants.DEFAULT_RECORD_NAME,
         SlackAppSetupCommand().get_default_record_name(),
         TeamsAppSetupCommand().get_default_record_name(),
+        SailPointAppSetupCommand().get_default_record_name(),
     )
 
 
@@ -206,7 +210,7 @@ def hide_from_record_cache(params, protected_uids: Dict[str, str]):
     saved_entries = {}
     for attr in _GUARDED_CACHE_ATTRS:
         source = getattr(params, attr, None)
-        if not isinstance(source, dict):
+        if not isinstance(source, MutableMapping):
             continue
         original_caches[attr] = source
         saved_entries[attr] = {uid: source[uid] for uid in protected_uid_set if uid in source}
@@ -283,7 +287,7 @@ def hide_from_folder_cache(params, protected_folder_uids: Set[str]):
 
         for attr in _GUARDED_FOLDER_CACHE_ATTRS:
             source = getattr(params, attr, None)
-            if not isinstance(source, dict):
+            if not isinstance(source, MutableMapping):
                 continue
             original_caches[attr] = source
             saved_entries[attr] = {uid: source[uid] for uid in protected_uid_set if uid in source}