fix(doc_parser): Add secure command execution with sandbox

Author: begoniezhao

docreader/parser/doc_parser.py

@@ -12,9 +12,93 @@
 logger = logging.getLogger(__name__)
 
 
+class SandboxExecutor:
+    """Sandbox executor for running commands with proxy configuration"""
+
+    def __init__(self, proxy: Optional[str] = None, default_timeout: int = 60):
+        """Initialize sandbox executor with configuration
+
+        Args:
+            proxy: Proxy URL to use for network access. If None, will use WEB_PROXY environment variable
+            default_timeout: Default timeout in seconds for command execution
+        """
+        # Get proxy from parameter, environment variable, or use default blocking proxy
+        # Use 'or None' to convert empty string to None, then apply default value
+        self.proxy = proxy or os.environ.get("WEB_PROXY") or "http://128.0.0.1:1"
+        self.default_timeout = default_timeout
+
+    def execute_in_sandbox(self, cmd: List[str]) -> tuple:
+        """Execute command in sandbox with proxy configuration
+
+        Args:
+            cmd: Command to execute
+
+        Returns:
+            Tuple of (stdout, stderr, returncode)
+        """
+        # Try different sandbox methods in order of preference
+        sandbox_methods = [
+            self._execute_with_proxy,
+        ]
+
+        for method in sandbox_methods:
+            try:
+                return method(cmd)
+            except Exception as e:
+                logger.warning(f"Sandbox method {method.__name__} failed: {e}")
+                continue
+
+        raise RuntimeError("All sandbox methods failed")
+
+    def _execute_with_proxy(self, cmd: List[str]) -> tuple:
+        """Execute command with proxy configuration
+
+        Args:
+            cmd: Command to execute
+
+        Returns:
+            Tuple of (stdout, stderr, returncode)
+        """
+        # Set up environment with proxy configuration
+        env = os.environ.copy()
+        if self.proxy:
+            env["http_proxy"] = self.proxy
+            env["https_proxy"] = self.proxy
+            env["HTTP_PROXY"] = self.proxy
+            env["HTTPS_PROXY"] = self.proxy
+
+        logger.info(f"Executing command with proxy: {' '.join(cmd)}")
+        if self.proxy:
+            logger.info(f"Using proxy: {self.proxy}")
+
+        process = subprocess.Popen(
+            cmd,
+            stdout=subprocess.PIPE,
+            stderr=subprocess.PIPE,
+            env=env,
+        )
+
+        try:
+            stdout, stderr = process.communicate(timeout=self.default_timeout)
+            return stdout, stderr, process.returncode
+        except subprocess.TimeoutExpired:
+            process.kill()
+            raise RuntimeError(
+                f"Command execution timeout after {self.default_timeout} seconds"
+            )
+
+
+logger = logging.getLogger(__name__)
+
+
 class DocParser(Docx2Parser):
     """DOC document parser"""
 
+    def __init__(self, *args, **kwargs):
+        """Initialize DOC parser with sandbox executor"""
+        super().__init__(*args, **kwargs)
+        self.sandbox_executor = SandboxExecutor()
+
     def parse_into_text(self, content: bytes) -> Document:
         logger.info(f"Parsing DOC document, content size: {len(content)} bytes")
 
@@ -25,7 +109,8 @@ def parse_into_text(self, content: bytes) -> Document:
             # try using antiword to extract text
             self._parse_with_antiword,
             # 3. If antiword extraction fails, use textract
-            self._parse_with_textract,
+            # NOTE: _parse_with_textract is disabled due to SSRF vulnerability
+            # self._parse_with_textract,
         ]
 
         # Save byte content as a temporary file
@@ -61,14 +146,13 @@ def _parse_with_antiword(self, temp_file_path: str) -> Document:
         if not antiword_path:
             raise RuntimeError("antiword not found in PATH")
 
-        # Use antiword to extract text directly
-        process = subprocess.Popen(
-            [antiword_path, temp_file_path],
-            stdout=subprocess.PIPE,
-            stderr=subprocess.PIPE,
-        )
-        stdout, stderr = process.communicate()
-        if process.returncode != 0:
+        # Use antiword to extract text directly in sandbox
+        cmd = [antiword_path, temp_file_path]
+        logger.info("Executing antiword in sandbox with proxy configuration")
+
+        stdout, stderr, returncode = self.sandbox_executor.execute_in_sandbox(cmd)
+
+        if returncode != 0:
             raise RuntimeError(
                 f"antiword extraction failed: {stderr.decode('utf-8', errors='ignore')}"
             )
@@ -114,13 +198,12 @@ def _try_convert_doc_to_docx(self, doc_path: str) -> Optional[bytes]:
                 temp_dir,
                 doc_path,
             ]
-            logger.info(f"Running command: {' '.join(cmd)}")
-            process = subprocess.Popen(
-                cmd, stdout=subprocess.PIPE, stderr=subprocess.PIPE
-            )
-            stdout, stderr = process.communicate()
+            logger.info(f"Running command in sandbox: {' '.join(cmd)}")
+
+            # Execute in sandbox with proxy configuration
+            stdout, stderr, returncode = self.sandbox_executor.execute_in_sandbox(cmd)
 
-            if process.returncode != 0:
+            if returncode != 0:
                 logger.warning(
                     f"Error converting DOC to DOCX: {stderr.decode('utf-8')}"
                 )