Python: Encode and validate session download filenames

python/semantic_kernel/core_plugins/sessions_python_tool/sessions_python_plugin.py

@@ -7,6 +7,7 @@
 from collections.abc import Awaitable, Callable
 from io import BytesIO
 from typing import Annotated, Any
+from urllib.parse import quote
 
 from azure.core.credentials import TokenCredential
 from httpx import AsyncClient, HTTPStatusError
@@ -424,16 +425,21 @@ async def download_file(
         """Download a file from the session pool.
 
         Args:
-            remote_file_name: The name of the file to download, relative to `/mnt/data`.
+            remote_file_name: The unescaped name of the file to download, relative to `/mnt/data`.
+                Must not contain `.` or `..` path segments.
             local_file_path: The path to save the downloaded file to. Should include the extension.
                 If not provided, the file is returned as a BytesIO object.
 
         Returns:
             BytesIO | None: The file content as BytesIO if no local_file_path provided, otherwise None.
 
         Raises:
-            FunctionExecutionException: If local_file_path is not in allowed directories.
+            FunctionExecutionException: If remote_file_name contains `.` or `..` path segments,
+                or local_file_path is not in allowed directories.
         """
+        if any(segment in (".", "..") for segment in remote_file_name.split("/")):
+            raise FunctionExecutionException("The remote file name must not contain '.' or '..' path segments.")
+
         auth_token = await self._ensure_auth_token()
         self.http_client.headers.update({
             "Authorization": f"Bearer {auth_token}",
@@ -442,7 +448,7 @@ async def download_file(
 
         url = self._build_url_with_version(
             base_url=str(self.pool_management_endpoint),
-            endpoint=f"files/content/{remote_file_name}",
+            endpoint=f"files/content/{quote(remote_file_name, safe='/')}",
             params={"identifier": self.settings.session_id},
         )
 

python/tests/unit/core_plugins/test_sessions_python_plugin.py

@@ -1,6 +1,6 @@
 # Copyright (c) Microsoft. All rights reserved.
 
-from unittest.mock import mock_open, patch
+from unittest.mock import Mock, mock_open, patch
 
 import httpx
 import pytest
@@ -10,6 +10,7 @@
     SESSIONS_API_VERSION,
     SessionsPythonTool,
 )
+from semantic_kernel.core_plugins.sessions_python_tool.sessions_python_settings import SessionsPythonSettings
 from semantic_kernel.core_plugins.sessions_python_tool.sessions_remote_file_metadata import SessionsRemoteFileMetadata
 from semantic_kernel.exceptions.function_exceptions import FunctionExecutionException, FunctionInitializationError
 from semantic_kernel.kernel import Kernel
@@ -116,6 +117,7 @@ def test_it_can_be_imported(kernel: Kernel, aca_python_sessions_unit_test_env):
     assert kernel.add_plugin(plugin=plugin, plugin_name="PythonCodeInterpreter")
     assert kernel.get_plugin(plugin_name="PythonCodeInterpreter") is not None
     assert kernel.get_plugin(plugin_name="PythonCodeInterpreter").name == "PythonCodeInterpreter"
+    assert "download_file" not in kernel.get_plugin(plugin_name="PythonCodeInterpreter").functions
 
 
 @patch("httpx.AsyncClient.post")
@@ -577,6 +579,72 @@ async def mock_auth_callback():
         mock_get.assert_awaited_once()
 
 
+@pytest.mark.parametrize(
+    "remote_file_name, encoded_file_name",
+    [
+        ("report.txt", "report.txt"),
+        ("nested/report.txt", "nested/report.txt"),
+        ("report..txt", "report..txt"),
+        ("report with spaces.txt", "report%20with%20spaces.txt"),
+        ("r\u00e9sum\u00e9.txt", "r%C3%A9sum%C3%A9.txt"),
+        ("report?part.txt", "report%3Fpart.txt"),
+        ("report#part.txt", "report%23part.txt"),
+        ("report%23part.txt", "report%2523part.txt"),
+        ("nested%2Freport.txt", "nested%252Freport.txt"),
+        ("%2e%2e/report.txt", "%252e%252e/report.txt"),
+        (
+            "report.txt?identifier=other&api-version=invalid#",
+            "report.txt%3Fidentifier%3Dother%26api-version%3Dinvalid%23",
+        ),
+    ],
+)
+async def test_download_file_encodes_filename(remote_file_name, encoded_file_name, aca_python_sessions_unit_test_env):
+    def handle_request(request: httpx.Request) -> httpx.Response:
+        assert request.method == "GET"
+        assert request.url.host == "test.endpoint"
+        assert request.url.raw_path == (
+            f"/files/content/{encoded_file_name}?identifier=session-fixed&api-version={SESSIONS_API_VERSION}".encode()
+        )
+        return httpx.Response(200, content=b"file data", request=request)
+
+    async with httpx.AsyncClient(transport=httpx.MockTransport(handle_request), trust_env=False) as http_client:
+        plugin = SessionsPythonTool(
+            auth_callback=auth_callback_test,
+            settings=SessionsPythonSettings(session_id="session-fixed"),
+            http_client=http_client,
+        )
+        buffer = await plugin.download_file(remote_file_name=remote_file_name)
+        assert buffer is not None
+        assert buffer.read() == b"file data"
+
+
+@pytest.mark.parametrize(
+    "remote_file_name",
+    [
+        ".",
+        "..",
+        "../report.txt",
+        "../../report.txt",
+        "./report.txt",
+        "nested/./report.txt",
+        "nested/../report.txt",
+        "nested/..",
+    ],
+)
+async def test_download_file_rejects_dot_segments(remote_file_name, aca_python_sessions_unit_test_env):
+    auth_callback = Mock(return_value="sample_token")
+    handle_request = Mock(return_value=httpx.Response(200, content=b"file data"))
+
+    async with httpx.AsyncClient(transport=httpx.MockTransport(handle_request), trust_env=False) as http_client:
+        plugin = SessionsPythonTool(auth_callback=auth_callback, http_client=http_client)
+
+        with pytest.raises(FunctionExecutionException, match="path segments"):
+            await plugin.download_file(remote_file_name=remote_file_name)
+
+    auth_callback.assert_not_called()
+    handle_request.assert_not_called()
+
+
 @patch("httpx.AsyncClient.get")
 async def test_download_file_throws_exception(mock_get, aca_python_sessions_unit_test_env):
     """Test throwing exception during download file."""