Python: fix(python): reject dot-segment remote paths in SessionsPythonTool.upload_file

python/semantic_kernel/core_plugins/sessions_python_tool/sessions_python_plugin.py

@@ -165,11 +165,17 @@ def _construct_remote_file_path(self, remote_file_path: str) -> str:
         """Construct the remote file path.
 
         Args:
-            remote_file_path (str): The remote file path.
+            remote_file_path (str): The remote file path. Must not contain `.` or `..`
+                path segments.
 
         Returns:
             str: The remote file path.
+
+        Raises:
+            FunctionExecutionException: If remote_file_path contains `.` or `..` path segments.
         """
+        if any(segment in (".", "..") for segment in remote_file_path.split("/")):
+            raise FunctionExecutionException("The remote file path must not contain '.' or '..' path segments.")
         if not remote_file_path.startswith("/mnt/data/"):
             remote_file_path = f"/mnt/data/{remote_file_path}"
         return remote_file_path
@@ -340,14 +346,16 @@ async def upload_file(
 
         Args:
             remote_file_path (str): The path to the file in the session.
+                Must not contain `.` or `..` path segments.
             local_file_path (str): The path to the file on the local machine.
                 Must be within allowed_upload_directories.
 
         Returns:
             RemoteFileMetadata: The metadata of the uploaded file.
 
         Raises:
-            FunctionExecutionException: If local_file_path is not provided or not in allowed directories.
+            FunctionExecutionException: If remote_file_path contains `.` or `..` path segments,
+                or local_file_path is not provided or not in allowed directories.
         """
         if not local_file_path:
             raise FunctionExecutionException("Please provide a local file path to upload.")

python/tests/unit/core_plugins/test_sessions_python_plugin.py

@@ -426,6 +426,99 @@ async def test_upload_file_fail_with_no_local_path(aca_python_sessions_unit_test
         )
 
 
+@pytest.mark.parametrize(
+    "remote_file_path",
+    [
+        ".",
+        "..",
+        "../uploaded_test.txt",
+        "../../uploaded_test.txt",
+        "./uploaded_test.txt",
+        "nested/./uploaded_test.txt",
+        "nested/../uploaded_test.txt",
+        "nested/..",
+        "/mnt/data/../../uploaded_test.txt",
+    ],
+)
+async def test_upload_file_rejects_dot_segments(remote_file_path, aca_python_sessions_unit_test_env, tmp_path):
+    """upload_file must reject the same dot segments download_file rejects."""
+    test_file = tmp_path / "file.py"
+    test_file.write_text("print('hello, world~')")
+
+    plugin = SessionsPythonTool(
+        auth_callback=lambda: "sample_token",
+        enable_dangerous_file_uploads=True,
+        allowed_upload_directories={str(tmp_path)},
+    )
+
+    with pytest.raises(FunctionExecutionException, match="path segments"):
+        await plugin.upload_file(local_file_path=str(test_file), remote_file_path=remote_file_path)
+
+
+@pytest.mark.parametrize(
+    "remote_file_path, expected_remote_file_path",
+    [
+        ("uploaded_test.txt", "/mnt/data/uploaded_test.txt"),
+        ("/mnt/data/input.py", "/mnt/data/input.py"),
+        ("nested/dir/input.py", "/mnt/data/nested/dir/input.py"),
+    ],
+)
+async def test_upload_file_allows_ordinary_remote_paths(
+    remote_file_path, expected_remote_file_path, aca_python_sessions_unit_test_env, tmp_path
+):
+    """Ordinary names, including nested paths, must still upload unchanged."""
+
+    async def async_return(result):
+        return result
+
+    test_file = tmp_path / "file.py"
+    test_file.write_text("print('hello, world~')")
+
+    with patch(
+        "semantic_kernel.core_plugins.sessions_python_tool.sessions_python_plugin.SessionsPythonTool._ensure_auth_token",
+        return_value="test_token",
+    ):
+        mock_request = httpx.Request(method="POST", url="https://example.com/files/upload?identifier=None")
+        mock_response = httpx.Response(
+            status_code=200,
+            json={"$id": "1", "value": []},
+            request=mock_request,
+        )
+        mock_post = Mock(side_effect=lambda *args, **kwargs: async_return(mock_response))
+
+        mock_get_request = httpx.Request(method="GET", url="https://example.com/files?identifier=None")
+        mock_get_response = httpx.Response(
+            status_code=200,
+            json={
+                "$id": "1",
+                "value": [
+                    {
+                        "$id": "2",
+                        "properties": {
+                            "$id": "3",
+                            "filename": expected_remote_file_path,
+                            "size": 456,
+                            "lastModifiedTime": "2024-07-02T19:29:23.4369699Z",
+                        },
+                    },
+                ],
+            },
+            request=mock_get_request,
+        )
+        mock_get = Mock(side_effect=lambda *args, **kwargs: async_return(mock_get_response))
+
+        with patch("httpx.AsyncClient.post", new=mock_post), patch("httpx.AsyncClient.get", new=mock_get):
+            plugin = SessionsPythonTool(
+                auth_callback=lambda: "sample_token",
+                enable_dangerous_file_uploads=True,
+                allowed_upload_directories={str(tmp_path)},
+            )
+
+            result = await plugin.upload_file(local_file_path=str(test_file), remote_file_path=remote_file_path)
+
+    assert result.filename == expected_remote_file_path
+
+
 @patch("httpx.AsyncClient.get")
 async def test_list_files(mock_get, aca_python_sessions_unit_test_env):
     """Test list_files function."""