perf: remove path validation delay

markitdown_mcp/server.py

@@ -47,15 +47,15 @@
 def with_timeout(timeout_seconds: int = 30) -> Any:
     """Decorator to add timeout protection to functions using threading."""
 
     def decorator(func: Any) -> Any:
         @functools.wraps(func)
         def wrapper(*args: Any, **kwargs: Any) -> Any:
             import threading
 
             result: list[Any] = [None]
             exception: list[Exception | None] = [None]
 
             def target() -> None:
                 try:
                     result[0] = func(*args, **kwargs)
                 except Exception as e:
@@ -130,7 +130,7 @@
         SecurityError: If XML contains dangerous constructs
     """
     try:
         with Path(file_path).open(encoding="utf-8", errors="ignore") as f:
             content = f.read()
 
         # Check for dangerous XML patterns
@@ -181,7 +181,7 @@
         SecurityError: If JSON is too deeply nested or complex
     """
     try:
         with Path(file_path).open(encoding="utf-8", errors="ignore") as f:
             content = f.read()
 
         # Check file size first
@@ -198,7 +198,7 @@
             raise SecurityError("Security violation: JSON recursion depth limit exceeded") from e
 
         # Check nesting depth
         def check_depth(obj: Any, current_depth: int = 0, max_depth: int = 30) -> None:
             if current_depth > max_depth:
                 raise SecurityError("Security violation: JSON recursion depth limit exceeded")
 
@@ -240,7 +240,7 @@
             raise SecurityError("Security violation: CSV file too large")
 
         # Analyze CSV structure
         with Path(file_path).open(encoding="utf-8", errors="ignore") as f:
             # Read first few lines to check structure
             sample = f.read(1024 * 1024)  # 1MB sample
 
@@ -337,7 +337,7 @@
     """
 
     @functools.wraps(func)
     def wrapper(*args: Any, **kwargs: Any) -> Any:
         start_time = time.time()
         try:
             result = func(*args, **kwargs)
@@ -439,7 +439,7 @@
 
 
 @with_timeout(30)  # type: ignore[misc]
 def safe_convert_with_limits(markitdown_instance: MarkItDown, file_path: str) -> Any:
     """Safely convert a file with timeout and recursion protection.
 
     Args:
@@ -467,7 +467,7 @@
         # Check if file might contain binary data in text format
         file_path_obj = Path(validated_file_path)
         if file_path_obj.exists():
             with Path(validated_file_path).open("rb") as f:
                 data = f.read(1024)  # Read first 1KB to check
 
             # If it's a text file but contains significant binary content
@@ -538,7 +538,6 @@
                 Path(sanitized_file_path).unlink(missing_ok=True)
 
 
-@normalize_timing
 def validate_and_sanitize_path(
     file_path: str, allowed_dirs: list[str] | None = None
 ) -> tuple[Path, bool]:
@@ -886,7 +885,7 @@
                 id=request.id, error={"code": -32603, "message": f"Internal error: {e!s}"}
             )
 
     async def convert_file_tool(
         self, request_id: JSONRPCId, arguments: dict[str, Any]
     ) -> MCPResponse:
         """Convert a single file to Markdown."""
@@ -1057,7 +1056,7 @@
             },
         )
 
     async def convert_directory_tool(
         self, request_id: JSONRPCId, arguments: dict[str, Any]
     ) -> MCPResponse:
         """Convert all supported files in a directory."""
@@ -1159,10 +1158,10 @@
                         markdown_content = result.text_content
 
                         # Write file asynchronously
                         def write_file(
                             path: str = output_path, content: str = markdown_content
                         ) -> None:
                             with Path(path).open("w", encoding="utf-8") as f:
                                 f.write(content)
 
                         await asyncio.get_event_loop().run_in_executor(None, write_file)
@@ -1250,7 +1249,7 @@
                     if response.error is not None:
                         response_dict["error"] = response.error
 
                     print(json.dumps(response_dict), flush=True)
 
                 except json.JSONDecodeError as e:
                     logger.error(f"Invalid JSON received: {e}")

tests/unit/test_additional_coverage.py

@@ -12,16 +12,17 @@
 
 from markitdown_mcp.server import (
     MarkItDownMCPServer,
-    MCPRequest,
     SecurityError,
-    validate_xml_security,
-    validate_json_security,
     extract_text_from_binary,
+    normalize_timing,
+    safe_convert_with_limits,
     sanitize_unicode_text,
-    with_timeout,
+    secure_compare,
     validate_base64,
-    safe_convert_with_limits,
     validate_file_content_security,
+    validate_json_security,
+    validate_xml_security,
+    with_timeout,
 )
 
 
@@ -177,7 +178,7 @@
     def test_timeout_decorator_success(self):
         """Test timeout decorator with successful operation."""
         @with_timeout(timeout_seconds=1)
         def quick_operation():
             return "success"
 
         result = quick_operation()
@@ -186,7 +187,7 @@
     def test_timeout_decorator_exception(self):
         """Test timeout decorator when function raises exception."""
         @with_timeout(timeout_seconds=1)
         def failing_operation():
             raise ValueError("Test error")
 
         with pytest.raises(ValueError, match="Test error"):
@@ -195,12 +196,41 @@
     def test_timeout_decorator_no_timeout(self):
         """Test timeout decorator with no timeout specified."""
         @with_timeout()
         def operation():
             return "no timeout"
 
         result = operation()
         assert result == "no timeout"
 
+    def test_secure_compare(self):
+        """Test constant-time comparison helper."""
+        assert secure_compare("same", "same") is True
+        assert secure_compare("same", "different") is False
+
+    def test_normalize_timing_success(self):
+        """Test timing normalization decorator returns successful results."""
+        @normalize_timing
+        def operation():
+            return "normalized"
+
+        start_time = time.time()
+        result = operation()
+
+        assert result == "normalized"
+        assert time.time() - start_time >= 0.05
+
+    def test_normalize_timing_exception(self):
+        """Test timing normalization decorator re-raises exceptions."""
+        @normalize_timing
+        def operation():
+            raise ValueError("normalized error")
+
+        start_time = time.time()
+        with pytest.raises(ValueError, match="normalized error"):
+            operation()
+
+        assert time.time() - start_time >= 0.05
+
     def test_validate_base64_valid(self):
         """Test base64 validation with valid data."""
         import base64
@@ -320,4 +350,4 @@
             result = validate_file_content_security(temp_path)
             assert result == temp_path  # Should return original for non-special files
         finally:
-            Path(temp_path).unlink(missing_ok=True)
+            Path(temp_path).unlink(missing_ok=True)