KC-1505, KC-1506: Fix Service Mode filesystem bypasses

keepercommander/commands/base.py

@@ -709,6 +709,14 @@ def normalize_output_param(args: str) -> str:
     return args
 
 
+def tokenize_command_args(args, envvars=None):
+    """Tokenize command arguments using the same transforms as Command.execute_args."""
+    args = '' if args is None else args
+    if envvars is not None:
+        args = expand_cmd_args(args, envvars)
+    return shlex.split(normalize_output_param(args))
+
+
 def as_boolean(value, default=None):  # type: (Any, Optional[bool]) -> bool
     if isinstance(value, bool):
         return value
@@ -765,14 +773,13 @@ def execute_args(self, params, args, **kwargs):
                 alias_cmd = d.get('command')
                 if alias_cmd and alias_cmd != parser.prog:
                     parser.prog = alias_cmd
-                args = expand_cmd_args(args, envvars)
-                args = normalize_output_param(args)
+                command_tokens = tokenize_command_args(args, envvars)
                 if self.support_extra_parameters():
-                    opts, extra_args = parser.parse_known_args(shlex.split(args))
+                    opts, extra_args = parser.parse_known_args(command_tokens)
                     if extra_args:
                         self.extra_parameters = ' '.join(extra_args)
                 else:
-                    opts = parser.parse_args(shlex.split(args))
+                    opts = parser.parse_args(command_tokens)
                 d.update(opts.__dict__)
 
             return self.execute(params, **d)

keepercommander/service/util/command_util.py

@@ -270,21 +270,40 @@ def execute(cls, command: str, temp_files: Optional[list] = None) -> Tuple[Any,
             command = ensure_record_add_json_format(html.unescape(command))
 
             try:
-                command_tokens = Verifycommand.tokenize_service_command(command)
+                # Check the raw command before environment expansion: Service
+                # Mode explicitly forbids this syntax, and expansion must not
+                # erase evidence of it before the restriction check.
+                raw_command_tokens = Verifycommand.tokenize_service_command(command)
             except ValueError:
                 return {"status": "error", "error": "Invalid command syntax"}, 400
 
-            # This request's own FILEDATA directory - the only paths Service
-            # Mode will treat as safe, not the whole shared OS temp root.
-            request_temp_dir = os.path.dirname(temp_files[0]) if temp_files else None
+            blocked_command = raw_command_tokens[0] if raw_command_tokens else ''
 
             def blocked(error):
                 logger.warning(
-                    f"Service Mode blocked command '{command_tokens[0] if command_tokens else ''}': {error}"
+                    f"Service Mode blocked command '{blocked_command}': {error}"
                 )
                 return {"status": "error", "error": error}, 403
 
-            # Same tokens the CLI will run — do not use raw HTTP split(" ")
+            expansion_error = Verifycommand.validate_service_mode_env_var_expansion_command(
+                raw_command_tokens
+            )
+            if expansion_error:
+                return blocked(expansion_error)
+
+            try:
+                command_tokens = Verifycommand.tokenize_service_command(
+                    command, params.environment_variables
+                )
+            except ValueError:
+                return {"status": "error", "error": "Invalid command syntax"}, 400
+
+            # This request's own FILEDATA directory - the only paths Service
+            # Mode will treat as safe, not the whole shared OS temp root.
+            request_temp_dir = os.path.dirname(temp_files[0]) if temp_files else None
+
+            # Apply every Service Mode restriction to the same tokens the
+            # CLI's Command.execute_args parser will receive.
             service_mode_error = Verifycommand.validate_service_mode_restrictions(
                 command_tokens, request_temp_dir
             )

keepercommander/service/util/request_validation.py

@@ -16,6 +16,7 @@
 import tempfile
 import os
 import json
+import shlex
 import shutil
 from ..decorators.logging import logger, sanitize_command_fields
 
@@ -82,19 +83,35 @@ def process_file_data(request_data: Dict[str, Any], command: str) -> Tuple[str,
                     temp_file_path = temp_file.name
                     temp_files.append(temp_file_path)
 
-                    # Replace FILEDATA placeholder with temporary file path, but only when it's used as a file parameter
-                    # Supports: PAM import (--filename), Import (positional), Enterprise-push (positional)
-                    # Callable replacement -- a plain string would let a Windows path's backslashes be parsed as regex backreferences.
-                    file_param_pattern = r'(--filename[=\s]+["\']?)FILEDATA(["\']?)'
+                    # Replace the entire FILEDATA token with a quoted path.
+                    # Reusing the caller's quotes would break when the
+                    # generated path itself contains a single quote.
+                    file_param_pattern = (
+                        r'(--filename[=\s]+)(?P<quote>["\']?)FILEDATA(?P=quote)'
+                    )
+
+                    def replace_filename_placeholder(match):
+                        prefix = match.group(1)
+                        return f'{prefix}{shlex.quote(temp_file_path)}'
+
                     processed_command = re.sub(
                         file_param_pattern,
-                        lambda m: f'{m.group(1)}{temp_file_path}{m.group(2)}',
+                        replace_filename_placeholder,
                         processed_command,
                     )
 
-                    # Also handle standalone FILEDATA (not in quotes)
-                    standalone_pattern = r'\bFILEDATA\b(?!["\'])'
-                    processed_command = re.sub(standalone_pattern, lambda _: temp_file_path, processed_command)
+                    # Also handle positional FILEDATA (including a quoted
+                    # placeholder), replacing its original quoting entirely.
+                    standalone_pattern = r'(?P<quote>["\']?)\bFILEDATA\b(?P=quote)'
+
+                    def replace_standalone_placeholder(match):
+                        return shlex.quote(temp_file_path)
+
+                    processed_command = re.sub(
+                        standalone_pattern,
+                        replace_standalone_placeholder,
+                        processed_command,
+                    )
 
                     logger.debug(f"Created temporary file {temp_file_path} for filedata")
 

keepercommander/service/util/verified_command.py

@@ -2,7 +2,6 @@
 import contextlib
 import io
 import os
-import shlex
 
 
 class Verifycommand:
@@ -99,16 +98,20 @@ class Verifycommand:
     _HOST_FS_MSG = (
         'Local filesystem access is not permitted through Service Mode'
     )
+    _AUTOMATOR_DATA_EXPANSION_MSG = (
+        'Automator file and environment expansions are not permitted through Service Mode'
+    )
 
     @staticmethod
     def validate_service_mode_restrictions(command_tokens, request_temp_dir=None):
-        """Run Service Mode bans on executor tokens (shlex); error string or None."""
+        """Run Service Mode bans over the exact tokens used by the command parser."""
         if not command_tokens:
             return None
 
         for validator in (
             Verifycommand.validate_service_mode_double_dash,
             Verifycommand.validate_service_mode_env_var_expansion_command,
+            Verifycommand.validate_service_mode_automator_file_access_command,
             Verifycommand.validate_service_mode_legacy_command,
             Verifycommand.validate_nsf_convert_command,
             Verifycommand.validate_service_mode_pam_tunnel_command,
@@ -127,9 +130,10 @@ def validate_service_mode_restrictions(command_tokens, request_temp_dir=None):
         return None
 
     @staticmethod
-    def tokenize_service_command(command):
-        """Tokenize a command using the executor's backslash handling."""
-        return shlex.split(command.replace('\\', '\\\\'))
+    def tokenize_service_command(command, envvars=None):
+        """Tokenize with the same argument transforms used by Command.execute_args."""
+        from ...commands.base import tokenize_command_args
+        return tokenize_command_args(command, envvars)
 
     @staticmethod
     def validate_nsf_convert_command(command_tokens, request_temp_dir=None):
@@ -186,6 +190,62 @@ def validate_service_mode_env_var_expansion_command(command_tokens, request_temp
             return Verifycommand._ENV_VAR_EXPANSION_MSG
         return None
 
+    @staticmethod
+    def validate_service_mode_automator_file_access_command(command_tokens, request_temp_dir=None):
+        """Block Automator operations that read local files or environment values."""
+        if not command_tokens:
+            return None
+
+        command = command_tokens[0].lower()
+        if command == 'automator-certificate':
+            return Verifycommand._HOST_FS_MSG
+
+        if command != 'automator' or len(command_tokens) < 2:
+            return None
+
+        action = command_tokens[1].lower()
+        if action == 'certificate':
+            # This is the registered group form in current Commander releases.
+            return Verifycommand._HOST_FS_MSG
+
+        if action != 'edit':
+            return None
+
+        # --set supports both ``--set KEY=VALUE`` and ``--set=KEY=VALUE``;
+        # argparse also accepts the unambiguous ``--se`` abbreviation.
+        # Extract only those settings instead of matching unrelated Automator
+        # arguments such as --url.
+        settings = []
+        i = 2
+        while i < len(command_tokens):
+            token = command_tokens[i]
+            option, separator, inline_value = token.partition('=')
+            if option in ('--set', '--se'):
+                if separator:
+                    settings.append(inline_value)
+                elif i + 1 < len(command_tokens):
+                    settings.append(command_tokens[i + 1])
+                    i += 1
+            i += 1
+
+        for setting in settings:
+            separator = setting.find('=')
+            if separator <= 1:
+                # AutomatorEditCommand uses the same pos > 1 condition and
+                # returns before resolving expansions for malformed settings.
+                continue
+            value = setting[separator + 1:].strip()
+            if not (value.startswith('${') and value.endswith('}')):
+                continue
+            expansion_type, colon, source = value[2:-1].partition(':')
+            if (
+                    colon
+                    and expansion_type.lower() in ('file', 'base64', 'env')
+                    and source
+                    and '}' not in source):
+                return Verifycommand._AUTOMATOR_DATA_EXPANSION_MSG
+        return None
+
     @staticmethod
     def validate_service_mode_legacy_command(command_tokens, request_temp_dir=None):
         """Block legacy commands in Service Mode; error or None."""
@@ -254,7 +314,7 @@ def validate_service_mode_record_file_attachment_command(command_tokens, request
 
     @staticmethod
     def _is_record_file_attachment_arg(token):
-        """True when parse_field would treat this token as a file attachment field."""
+        """True when parse_field would treat this executor token as a file field."""
         if not token or '=' not in token:
             return False
         # Mirror RecordEditMixin.parse_field field-name normalization.