FIX: sanitize metadata file name in azure_blob_storage_target

pyrit/prompt_target/azure_blob_storage_target.py

@@ -2,6 +2,7 @@
 # Licensed under the MIT license.
 
 import logging
+import posixpath
 from enum import Enum
 from typing import ClassVar
 from urllib.parse import urlparse
@@ -251,7 +252,7 @@ async def _send_prompt_to_target_async(self, *, normalized_conversation: list[Me
         # default file name is <conversation_id>.txt, but can be overridden by prompt metadata
         file_name = f"{request.conversation_id}.txt"
         if request.prompt_metadata.get("file_name"):
-            file_name = str(request.prompt_metadata["file_name"])
+            file_name = self._sanitize_file_name(str(request.prompt_metadata["file_name"]))
 
         data = str.encode(request.converted_value)
         blob_url = self._container_url + "/" + file_name
@@ -263,3 +264,24 @@ async def _send_prompt_to_target_async(self, *, normalized_conversation: list[Me
         )
 
         return [response]
+
+    @staticmethod
+    def _sanitize_file_name(file_name: str) -> str:
+        """
+        Validate that *file_name* is a bare leaf name with no path components.
+
+        Guards against path traversal (e.g. ``../../other/x``) that would let a
+        caller-supplied ``file_name`` escape the configured container prefix.
+
+        Args:
+            file_name (str): The caller-supplied file name from prompt metadata.
+
+        Returns:
+            str: The validated file name, unchanged.
+
+        Raises:
+            ValueError: If *file_name* contains path separators or traversal segments.
+        """
+        if file_name != posixpath.basename(file_name) or file_name in (".", "..") or "\\" in file_name:
+            raise ValueError(f"Invalid file_name '{file_name}': must be a bare filename without path separators.")
+        return file_name

tests/unit/prompt_target/target/test_prompt_target_azure_blob_storage.py

@@ -240,3 +240,35 @@ async def test_upload_blob_async_raises_when_client_async_none(azure_blob_storag
                 await azure_blob_storage_target._upload_blob_async(
                     file_name="test.txt", data=b"hello", content_type="text/plain"
                 )
+
+
+@pytest.mark.parametrize(
+    "file_name",
+    ["../../admin/stolen.txt", "sub/dir/file.txt", "..", ".", "dir\\file.txt", "/abs.txt"],
+)
+def test_sanitize_file_name_rejects_path_traversal(file_name):
+    """Caller-supplied file_name with path components must be rejected (CWE-22 hardening)."""
+    with pytest.raises(ValueError, match="bare filename"):
+        AzureBlobStorageTarget._sanitize_file_name(file_name)
+
+
+def test_sanitize_file_name_allows_bare_name():
+    """A plain leaf file name passes through unchanged."""
+    assert AzureBlobStorageTarget._sanitize_file_name("results.txt") == "results.txt"
+
+
+async def test_send_prompt_async_rejects_traversal_file_name(
+    azure_blob_storage_target: AzureBlobStorageTarget,
+    sample_entries: MutableSequence[MessagePiece],
+):
+    """A traversal file_name in prompt metadata is refused before any upload."""
+    message_piece = sample_entries[0]
+    message_piece.converted_value = "Test content"
+    message_piece.prompt_metadata = {"file_name": "../../admin/stolen.txt"}
+    request = Message(message_pieces=[message_piece])
+
+    with patch.object(AzureBlobStorageTarget, "_upload_blob_async", new_callable=AsyncMock) as mock_upload:
+        with pytest.raises(ValueError, match="bare filename"):
+            await azure_blob_storage_target.send_prompt_async(message=request)
+
+    mock_upload.assert_not_called()