fix: address PR review comments

- Add path traversal sanitization in save_bulk_data_file and download_rosbags_for_fault
- Add SSRF protection: reject absolute URLs in download_bulk_data
- Handle all non-2xx responses in get_bulk_data_info (not just 404)
- Remove unused BulkDataDescriptor model

Author: bburda

src/ros2_medkit_mcp/client.py

@@ -1000,10 +1000,10 @@ async def get_bulk_data_info(self, bulk_data_uri: str) -> dict[str, Any]:
         client = await self._ensure_client()
         response = await client.head(bulk_data_uri)
 
-        if response.status_code == 404:
+        if not response.is_success:
             raise SovdClientError(
-                message=f"Bulk data not found: {bulk_data_uri}",
-                status_code=404,
+                message=f"Bulk data not found: {bulk_data_uri} (HTTP {response.status_code})",
+                status_code=response.status_code,
             )
 
         headers = response.headers
@@ -1027,11 +1027,18 @@ async def download_bulk_data(self, bulk_data_uri: str) -> tuple[bytes, str | Non
         """Download a bulk-data file.
 
         Args:
-            bulk_data_uri: Full bulk-data URI path.
+            bulk_data_uri: Relative bulk-data URI path (must start with /).
 
         Returns:
             Tuple of (file_content, filename).
+
+        Raises:
+            ValueError: If the URI is an absolute URL (SSRF protection).
         """
+        # SSRF protection: reject absolute URLs - only allow relative paths
+        if bulk_data_uri.startswith(("http://", "https://", "//")):
+            raise ValueError(f"Absolute URLs not allowed for bulk data download: {bulk_data_uri}")
+
         client = await self._ensure_client()
         response = await client.get(bulk_data_uri, timeout=httpx.Timeout(300.0))
 

src/ros2_medkit_mcp/mcp_app.py

@@ -419,7 +419,7 @@ def save_bulk_data_file(
     Returns:
         Formatted TextContent list with download result.
     """
-    output_path = Path(output_dir)
+    output_path = Path(output_dir).resolve()
     output_path.mkdir(parents=True, exist_ok=True)
 
     # Generate filename if not provided
@@ -431,7 +431,16 @@ def save_bulk_data_file(
         if "." not in filename:
             filename += ".mcap"
 
-    file_path = output_path / filename
+    # Sanitize filename to prevent path traversal
+    safe_filename = Path(filename).name
+    if not safe_filename:
+        safe_filename = "download.mcap"
+
+    file_path = (output_path / safe_filename).resolve()
+    # Ensure the resolved path is still within output_dir
+    if not str(file_path).startswith(str(output_path)):
+        raise ValueError(f"Path traversal detected in filename: {filename}")
+
     file_path.write_bytes(content)
 
     size_mb = len(content) / (1024 * 1024)
@@ -507,7 +516,7 @@ async def download_rosbags_for_fault(
             )
         ]
 
-    output_path = Path(output_dir)
+    output_path = Path(output_dir).resolve()
     output_path.mkdir(parents=True, exist_ok=True)
 
     downloaded: list[str] = []
@@ -527,7 +536,13 @@ async def download_rosbags_for_fault(
             if not filename:
                 filename = f"{snap_id}.mcap"
 
-            file_path = output_path / filename
+            # Sanitize filename to prevent path traversal
+            safe_filename = Path(filename).name or f"{snap_id}.mcap"
+            file_path = (output_path / safe_filename).resolve()
+            if not str(file_path).startswith(str(output_path)):
+                errors.append(f"  - {snap_id}: Path traversal detected in filename")
+                continue
+
             file_path.write_bytes(content)
 
             size_mb = len(content) / (1024 * 1024)

src/ros2_medkit_mcp/models.py

@@ -521,37 +521,6 @@ class FreezeFrameSnapshot(BaseModel):
     model_config = {"populate_by_name": True, "extra": "allow"}
 
 
-class BulkDataDescriptor(BaseModel):
-    """Descriptor for bulk data (rosbag) with download URI."""
-
-    id: str = Field(..., description="Bulk data identifier")
-    category: str | None = Field(
-        default=None,
-        description="Data category (e.g., 'rosbag', 'snapshot')",
-    )
-    bulk_data_uri: str = Field(
-        ...,
-        alias="bulkDataUri",
-        description="URI to download the bulk data file",
-    )
-    file_size: int | None = Field(
-        default=None,
-        alias="fileSize",
-        description="File size in bytes",
-    )
-    is_available: bool = Field(
-        default=True,
-        alias="isAvailable",
-        description="Whether the file is available for download",
-    )
-    timestamp: str | None = Field(
-        default=None,
-        description="ISO 8601 timestamp when the data was captured",
-    )
-
-    model_config = {"populate_by_name": True, "extra": "allow"}
-
-
 class RosbagSnapshot(BaseModel):
     """Rosbag snapshot with bulk data download URI."""