KC-1430: Deny host-path I/O in Service Mode

keepercommander/service/api/command.py

@@ -99,7 +99,7 @@ def execute_command_direct(**kwargs) -> Tuple[Union[Response, bytes], int]:
                 )
                 return response_data
 
-            response, status_code = CommandExecutor.execute(processed_command)
+            response, status_code = CommandExecutor.execute(processed_command, temp_files=temp_files)
 
             # If we get a busy response, add v1-specific message
             if (isinstance(response, dict) and

keepercommander/service/core/request_queue.py

@@ -325,7 +325,7 @@ def _process_request(self, request: QueuedRequest):
 
         try:
             # Execute the command using existing CommandExecutor
-            result, status_code = CommandExecutor.execute(request.command)
+            result, status_code = CommandExecutor.execute(request.command, temp_files=request.temp_files)
 
             # Mark as completed
             request.status = RequestStatus.COMPLETED

keepercommander/service/util/command_util.py

@@ -148,13 +148,12 @@ def _finalize_parsed_response(cls, response: Any) -> Tuple[Any, int]:
         return response, status_code
 
     @classmethod
-    def execute(cls, command: str) -> Tuple[Any, int]:
+    def execute(cls, command: str, temp_files: Optional[list] = None) -> Tuple[Any, int]:
         logger.debug(f"Executing command: {sanitize_command_fields(command)}")
-
         validation_error = cls.validate_command(command)
         if validation_error:
             return validation_error
-        
+
         from ..core.globals import ensure_params_loaded
         try:
             params = ensure_params_loaded()
@@ -169,11 +168,19 @@ def execute(cls, command: str) -> Tuple[Any, int]:
             except ValueError:
                 command_tokens = command.split()
 
+            # 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
+
             # Same tokens the CLI will run — do not use raw HTTP split(" ")
             service_mode_error = Verifycommand.validate_service_mode_restrictions(
-                command_tokens
+                command_tokens, request_temp_dir
             )
             if service_mode_error:
+                logger.warning(
+                    f"Service Mode blocked command '{command_tokens[0] if command_tokens else ''}': "
+                    f"{service_mode_error}"
+                )
                 return {"status": "error", "error": service_mode_error}, 403
 
             force_error = Verifycommand.validate_enterprise_user_add_role_force(

keepercommander/service/util/request_validation.py

@@ -15,6 +15,7 @@
 import tempfile
 import os
 import json
+import shutil
 from ..decorators.logging import logger, sanitize_command_fields
 
 
@@ -71,9 +72,11 @@ def process_file_data(request_data: Dict[str, Any], command: str) -> Tuple[str,
                 logger.warning("filedata must be a JSON object or array")
                 return processed_command, temp_files
 
+            request_temp_dir = None
             try:
-                # Create temporary file with the filedata content
-                with tempfile.NamedTemporaryFile(mode='w', suffix='.json', delete=False, encoding='utf-8') as temp_file:
+                request_temp_dir = tempfile.mkdtemp(prefix='keeper_svc_')
+                with tempfile.NamedTemporaryFile(mode='w', suffix='.json', delete=False,
+                                                  encoding='utf-8', dir=request_temp_dir) as temp_file:
                     json.dump(filedata, temp_file, indent=2)
                     temp_file_path = temp_file.name
                     temp_files.append(temp_file_path)
@@ -91,30 +94,41 @@ def process_file_data(request_data: Dict[str, Any], command: str) -> Tuple[str,
 
             except Exception as e:
                 logger.error(f"Error creating temporary file for filedata: {e}")
-                # Clean up any created temp files
-                for temp_path in temp_files:
+                if request_temp_dir:
                     try:
-                        os.unlink(temp_path)
-                    except Exception:
-                        pass
+                        shutil.rmtree(request_temp_dir)
+                        logger.debug(f"Cleaned up request temp directory: {request_temp_dir}")
+                    except Exception as cleanup_error:
+                        logger.warning(f"Failed to clean up request temp directory {request_temp_dir}: {cleanup_error}")
                 return command, []
 
         return processed_command, temp_files
 
     @staticmethod
     def cleanup_temp_files(temp_files: list) -> None:
-        """Clean up temporary files.
-        
+        """Clean up temporary files and their parent per-request directories.
+
         Args:
             temp_files: List of temporary file paths to clean up
         """
+        parent_dirs = set()
         for temp_path in temp_files:
             try:
                 if os.path.exists(temp_path):
                     os.unlink(temp_path)
                     logger.debug(f"Cleaned up temporary file: {temp_path}")
             except Exception as e:
                 logger.warning(f"Failed to clean up temporary file {temp_path}: {e}")
+            parent_dirs.add(os.path.dirname(temp_path))
+
+        # Remove parent directories (each file has one dedicated per-request directory)
+        for parent_dir in parent_dirs:
+            try:
+                if os.path.exists(parent_dir):
+                    shutil.rmtree(parent_dir)
+                    logger.debug(f"Cleaned up request temp directory: {parent_dir}")
+            except Exception as e:
+                logger.warning(f"Failed to clean up request temp directory {parent_dir}: {e}")
 
     @staticmethod
     def validate_request_json() -> Optional[Tuple]: