KC-1458: Restrict app-setup command lists to their own allowlist

keepercommander/service/commands/create_service.py

@@ -106,6 +106,10 @@ def execute(self, params: KeeperParams, **kwargs) -> None:
                 from .integrations.sailpoint.service import SailPointService
                 SailPointService.maybe_enable(params, args)
 
+            # Re-sanitize commands against each integration's own allowlist in case docker-compose.yml is stale/hand-edited.
+            from .integrations.runtime_policy import apply_runtime_command_policy
+            apply_runtime_command_policy(args)
+
             from .integrations.vault_metadata import get_existing_api_key, write_service_metadata
             existing_api_key = (
                 get_existing_api_key(params, args.update_vault_record)

keepercommander/service/commands/integrations/command_policy.py

@@ -0,0 +1,41 @@
+#  _  __
+# | |/ /___ ___ _ __  ___ _ _ ®
+# | ' </ -_) -_) '_ \/ -_) '_|
+# |_|\_\___\___| .__/\___|_|
+#              |_|
+#
+# Keeper Commander
+# Copyright 2026 Keeper Security Inc.
+# Contact: commander@keepersecurity.com
+#
+
+"""Shared command-list sanitization for Service Mode *-app-setup commands."""
+
+from __future__ import annotations
+
+from typing import Iterable
+
+
+def sanitize_commands(commands: str, allowed: Iterable[str], banned: Iterable[str] = ()) -> str:
+    """Normalizes `commands` to exactly the allowed set minus banned; cannot narrow below the allowlist, only add missing/drop extra entries."""
+    allowed = [c.strip() for c in allowed if c and c.strip()]
+    allowed_set = {c.lower() for c in allowed}
+    banned_set = {c.strip().lower() for c in banned}
+    filtered = [
+        cmd for raw in (commands or '').split(',')
+        if (cmd := raw.strip())
+        and (key := cmd.lower()) not in banned_set
+        and key in allowed_set
+    ]
+    # Input order first, then any missing required allowlist entries.
+    by_key = {cmd.lower(): cmd for cmd in filtered}
+    for cmd in allowed:
+        key = cmd.strip().lower()
+        if key not in banned_set and key not in by_key:
+            by_key[key] = cmd.strip()
+    return ','.join(by_key.values())
+
+
+def default_allowlist(allowed: Iterable[str], banned: Iterable[str] = ()) -> str:
+    allowed = list(allowed)
+    return sanitize_commands(','.join(allowed), allowed, banned)

keepercommander/service/commands/integrations/integration_setup_base.py

@@ -30,6 +30,7 @@
 )
 from .approvals_setup import ApprovalsChannelProfile, collect_approvals_config, is_valid_keeper_uid
 from .approvals_sync import merge_approvals_custom_fields, run_approvals_sync_down
+from .command_policy import sanitize_commands
 
 UUID_PATTERN = re.compile(
     r'^[0-9a-fA-F]{8}-[0-9a-fA-F]{4}-[0-9a-fA-F]{4}-[0-9a-fA-F]{4}-[0-9a-fA-F]{12}$'
@@ -121,6 +122,15 @@ def get_service_commands(self) -> str:
             commands = f'{commands},{self.get_command_name()}'
         return commands
 
+    def get_banned_commands(self) -> tuple:
+        """Commands to always strip from this integration's command list; override per integration if needed."""
+        return ()
+
+    def sanitize_service_commands(self, commands: str) -> str:
+        """Restrict `commands` to this integration's own get_service_commands() list."""
+        allowed = self.get_service_commands().split(',')
+        return sanitize_commands(commands, allowed, self.get_banned_commands())
+
     # -- Parser (auto-built from name, cached per subclass) ----------
 
     def get_parser(self):
@@ -456,10 +466,13 @@ def _update_docker_compose(self, setup_result: SetupResult,
                 )
 
         try:
+            cfg = asdict(service_config)
+            cfg['commands'] = self.sanitize_service_commands(cfg.get('commands') or '')
             builder = DockerComposeBuilder(
-                setup_result, asdict(service_config),
+                setup_result, cfg,
                 commander_service_name=self.get_commander_service_name(),
-                commander_container_name=self.get_commander_container_name()
+                commander_container_name=self.get_commander_container_name(),
+                commander_environment={self.get_record_env_key(): record_uid},
             )
             yaml_content = builder.add_integration_service(
                 service_name=service_name,

keepercommander/service/commands/integrations/runtime_policy.py

@@ -0,0 +1,68 @@
+#  _  __
+# | |/ /___ ___ _ __  ___ _ _ ®
+# | ' </ -_) -_) '_ \/ -_) '_|
+# |_|\_\___\___| .__/\___|_|
+#              |_|
+#
+# Keeper Commander
+# Copyright 2026 Keeper Security Inc.
+# Contact: commander@keepersecurity.com
+#
+
+"""Container-start command-list enforcement for *-app-setup integrations."""
+
+from __future__ import annotations
+
+import os
+
+from ...util.exceptions import ValidationError
+
+
+def apply_runtime_command_policy(args) -> None:
+    if getattr(args, 'commands', None) is None:
+        return
+
+    sanitizers = _integration_sanitizers()
+    matched = [env_key for env_key in sanitizers if (os.environ.get(env_key) or '').strip()]
+    if not matched:
+        return
+    if len(matched) > 1:
+        raise ValidationError(
+            f'Multiple integration env vars are set ({", ".join(matched)}); '
+            f'remove all but one before starting the service.'
+        )
+
+    env_key = matched[0]
+    cleaned = sanitizers[env_key](args.commands)
+    if cleaned != args.commands:
+        print(
+            f'Service Mode ({env_key}): removed commands outside the integration '
+            f'allowlist before service-create.\n  Was: {args.commands}\n  Now: {cleaned}'
+        )
+    args.commands = cleaned
+
+
+def _integration_sanitizers():
+    """Maps each integration's env var to its own sanitize_service_commands(); SailPoint is included too as a fallback in case SailPointService.maybe_enable() skipped sanitizing."""
+    from ..terraform_app_setup import TerraformSetupConstants
+    from .command_policy import sanitize_commands
+    from .gchat_app_setup import GChatAppSetupCommand
+    from .sailpoint_app_setup import SailPointAppSetupCommand
+    from .slack_app_setup import SlackAppSetupCommand
+    from .teams_app_setup import TeamsAppSetupCommand
+    from ...decorators.min_commander_version import TERRAFORM_DOCKER_ENV
+
+    slack = SlackAppSetupCommand()
+    teams = TeamsAppSetupCommand()
+    gchat = GChatAppSetupCommand()
+    sailpoint = SailPointAppSetupCommand()
+
+    return {
+        slack.get_record_env_key(): slack.sanitize_service_commands,
+        teams.get_record_env_key(): teams.sanitize_service_commands,
+        gchat.get_record_env_key(): gchat.sanitize_service_commands,
+        sailpoint.get_record_env_key(): sailpoint.sanitize_service_commands,
+        TERRAFORM_DOCKER_ENV: lambda commands: sanitize_commands(
+            commands, TerraformSetupConstants.SERVICE_COMMANDS_LIST
+        ),
+    }

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

@@ -17,6 +17,7 @@
 from typing import Any, Optional, Tuple
 
 from .....utils import is_email
+from .. import command_policy as _shared_policy
 from .command_parse import SailPointCommandParser
 from .constants import SAILPOINT_ALLOWED_COMMANDS, SAILPOINT_BANNED_COMMANDS
 
@@ -50,25 +51,13 @@ def sanitize(cls, commands: str) -> str:
         commands are not dropped when the input list is a partial or older
         compose allowlist.
         """
-        allowed = {c.strip().lower() for c in SAILPOINT_ALLOWED_COMMANDS}
-        banned = {c.lower() for c in SAILPOINT_BANNED_COMMANDS}
-        filtered = [
-            cmd for raw in (commands or '').split(',')
-            if (cmd := raw.strip())
-            and (key := cmd.lower()) not in banned
-            and key in allowed
-        ]
-        # Input order first, then any missing required allowlist entries.
-        by_key = {cmd.lower(): cmd for cmd in filtered}
-        for cmd in SAILPOINT_ALLOWED_COMMANDS:
-            key = cmd.lower()
-            if key not in banned and key not in by_key:
-                by_key[key] = cmd
-        return ','.join(by_key.values())
+        return _shared_policy.sanitize_commands(
+            commands, SAILPOINT_ALLOWED_COMMANDS, SAILPOINT_BANNED_COMMANDS
+        )
 
     @classmethod
     def default_allowlist(cls) -> str:
-        return cls.sanitize(','.join(SAILPOINT_ALLOWED_COMMANDS))
+        return _shared_policy.default_allowlist(SAILPOINT_ALLOWED_COMMANDS, SAILPOINT_BANNED_COMMANDS)
 
     @classmethod
     def validate_enterprise_user(cls, command: str) -> Optional[str]:

keepercommander/service/commands/integrations/sailpoint_app_setup.py

@@ -33,6 +33,7 @@
     MIN_POLL_INTERVAL_SECONDS,
     PENDING_ENTITLEMENTS_FIELD,
     POLL_INTERVAL_FIELD,
+    SAILPOINT_BANNED_COMMANDS,
     SAILPOINT_MARKER_FIELD,
     SAILPOINT_RECORD_ENV,
     TRANSFER_TARGET_EMAIL_FIELD,
@@ -60,6 +61,9 @@ def get_record_env_key(self) -> str:
     def get_service_commands(self) -> str:
         return SailPointCommandPolicy.default_allowlist()
 
+    def get_banned_commands(self) -> tuple:
+        return tuple(SAILPOINT_BANNED_COMMANDS)
+
     def collect_integration_config(self, params, transfer_target_default: str = ''):
         print(f"\n{bcolors.BOLD}SHARE ENTITLEMENTS:{bcolors.ENDC}")
         print(f"  Control which share entitlements SailPoint may manage via Service Mode")
@@ -186,7 +190,7 @@ def _update_docker_compose(self, setup_result, service_config, record_uid, confi
 
         try:
             cfg = asdict(service_config)
-            cfg['commands'] = SailPointCommandPolicy.sanitize(cfg.get('commands') or '')
+            cfg['commands'] = self.sanitize_service_commands(cfg.get('commands') or '')
             builder = DockerComposeBuilder(
                 setup_result,
                 cfg,

keepercommander/service/commands/terraform_app_setup.py

@@ -26,6 +26,7 @@
 )
 from ..decorators.min_commander_version import TERRAFORM_DOCKER_ENV
 from ..util.exceptions import ValidationError
+from .integrations.command_policy import sanitize_commands
 from .service_docker_setup import ServiceDockerSetupCommand
 
 
@@ -47,7 +48,7 @@ class TerraformSetupConstants:
         'share-folder', 'rmdir', 'rndir', 'mkdir', 'epm', 'scim', 'mv', 'pam',
         'secrets-manager', 'ln', 'share-record',
         'nsf-mkdir', 'nsf-get', 'nsf-rmdir', 'nsf-record-add', 'nsf-record-update',
-        'nsf-rm', 'nsf-rndir', 'nsf-share-folder', 'nsf-share-record', 'nsf-ln',
+        'nsf-rm', 'nsf-rndir', 'nsf-share-folder', 'nsf-share-record', 'nsf-ln', 'nsf-move',
     )
     SERVICE_COMMANDS = ','.join(SERVICE_COMMANDS_LIST)
 
@@ -113,14 +114,15 @@ def execute(self, params, **kwargs):
 
     def _validate_terraform_commands(self, params) -> str:
         try:
-            return RuntimeServiceConfig().validate_command_list(
+            validated = RuntimeServiceConfig().validate_command_list(
                 TerraformSetupConstants.SERVICE_COMMANDS, params
             )
         except ValidationError as e:
             raise CommandError(
                 self.get_parser().prog,
                 f'Terraform command allowlist validation failed: {e}',
             )
+        return sanitize_commands(validated, TerraformSetupConstants.SERVICE_COMMANDS_LIST)
 
     def _get_commands_config(self, params) -> str:
         return self._validate_terraform_commands(params)

unit-tests/service/test_command_policy.py

@@ -0,0 +1,59 @@
+#  _  __
+# | |/ /___ ___ _ __  ___ _ _ ®
+# | ' </ -_) -_) '_ \/ -_) '_|
+# |_|\_\___\___| .__/\___|_|
+#              |_|
+#
+# Keeper Commander
+# Copyright 2026 Keeper Security Inc.
+# Contact: commander@keepersecurity.com
+#
+
+import unittest
+
+from keepercommander.service.commands.integrations.command_policy import (
+    default_allowlist,
+    sanitize_commands,
+)
+
+
+class TestSanitizeCommands(unittest.TestCase):
+    def test_strips_commands_outside_allowlist(self):
+        result = sanitize_commands('search,get,malicious-command', allowed=['search', 'get'])
+        self.assertEqual(set(result.split(',')), {'search', 'get'})
+
+    def test_drops_banned_even_if_in_allowlist_and_input(self):
+        result = sanitize_commands(
+            'search,get,rm', allowed=['search', 'get', 'rm'], banned=['rm']
+        )
+        self.assertEqual(set(result.split(',')), {'search', 'get'})
+
+    def test_force_includes_missing_allowed_entries(self):
+        result = sanitize_commands('search', allowed=['search', 'get'])
+        self.assertEqual(set(result.split(',')), {'search', 'get'})
+
+    def test_empty_input_returns_full_allowlist_minus_banned(self):
+        result = sanitize_commands('', allowed=['search', 'get', 'rm'], banned=['rm'])
+        self.assertEqual(set(result.split(',')), {'search', 'get'})
+
+    def test_case_insensitive_matching_preserves_original_casing(self):
+        result = sanitize_commands('Search,GET', allowed=['search', 'get'])
+        self.assertEqual(set(result.split(',')), {'Search', 'GET'})
+
+    def test_whitespace_and_empty_tokens_are_ignored(self):
+        result = sanitize_commands(' search , , get ', allowed=['search', 'get'])
+        self.assertEqual(set(result.split(',')), {'search', 'get'})
+
+
+class TestDefaultAllowlist(unittest.TestCase):
+    def test_returns_full_allowlist_minus_banned(self):
+        result = default_allowlist(['search', 'get', 'rm'], banned=['rm'])
+        self.assertEqual(set(result.split(',')), {'search', 'get'})
+
+    def test_no_banned_returns_full_allowlist(self):
+        result = default_allowlist(['search', 'get'])
+        self.assertEqual(set(result.split(',')), {'search', 'get'})
+
+
+if __name__ == '__main__':
+    unittest.main()

unit-tests/service/test_integration_setup_base_command_sanitize.py

@@ -0,0 +1,70 @@
+#  _  __
+# | |/ /___ ___ _ __  ___ _ _ ®
+# | ' </ -_) -_) '_ \/ -_) '_|
+# |_|\_\___\___| .__/\___|_|
+#              |_|
+#
+# Keeper Commander
+# Copyright 2026 Keeper Security Inc.
+# Contact: commander@keepersecurity.com
+#
+
+import os
+import re
+import tempfile
+import unittest
+
+from keepercommander.service.commands.integrations.gchat_app_setup import GChatAppSetupCommand
+from keepercommander.service.docker.models import ServiceConfig, SetupResult
+
+_COMMAND_LIST_RE = re.compile(r"-c '([^']*)'")
+
+
+class TestIntegrationSetupBaseCommandSanitize(unittest.TestCase):
+    """A tampered/injected commands string must be confined to the integration's own get_service_commands() list."""
+
+    def _make_service_config(self, commands):
+        return ServiceConfig(
+            port=8900,
+            commands=commands,
+            queue_enabled=True,
+            ngrok_enabled=False,
+            ngrok_auth_token='',
+            ngrok_custom_domain='',
+            cloudflare_enabled=False,
+            cloudflare_tunnel_token='',
+            cloudflare_custom_domain='',
+        )
+
+    def test_update_docker_compose_strips_commands_outside_allowlist(self):
+        cmd = GChatAppSetupCommand()
+        allowed = set(cmd.get_service_commands().split(','))
+
+        setup_result = SetupResult(
+            folder_uid='folder-uid', folder_name='folder', app_uid='app-uid',
+            app_name='app', record_uid='setup-record-uid', b64_config='cfg',
+        )
+        tampered_commands = cmd.get_service_commands() + ',malicious-command,rm'
+        service_config = self._make_service_config(tampered_commands)
+
+        cwd = os.getcwd()
+        with tempfile.TemporaryDirectory() as tmp_dir:
+            os.chdir(tmp_dir)
+            try:
+                cmd._update_docker_compose(setup_result, service_config, 'integration-record-uid')
+                with open(os.path.join(tmp_dir, 'docker-compose.yml')) as f:
+                    yaml_content = f.read()
+            finally:
+                os.chdir(cwd)
+
+        match = _COMMAND_LIST_RE.search(yaml_content)
+        self.assertIsNotNone(match)
+        written_commands = set(match.group(1).split(','))
+
+        self.assertNotIn('malicious-command', written_commands)
+        self.assertNotIn('rm', written_commands)
+        self.assertEqual(written_commands, allowed)
+
+
+if __name__ == '__main__':
+    unittest.main()