diff --git a/pyrit/prompt_target/azure_blob_storage_target.py b/pyrit/prompt_target/azure_blob_storage_target.py index 5cb6b89886..d0cbe13dec 100644 --- a/pyrit/prompt_target/azure_blob_storage_target.py +++ b/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 .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 diff --git a/tests/unit/prompt_target/target/test_prompt_target_azure_blob_storage.py b/tests/unit/prompt_target/target/test_prompt_target_azure_blob_storage.py index c5d15ff5fc..4feb350bb4 100644 --- a/tests/unit/prompt_target/target/test_prompt_target_azure_blob_storage.py +++ b/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()