Python: Add directory allowlist configuration for SessionsPythonTool file (#13467)

### Motivation and Context

When `SessionsPythonTool` is used with an AI model, providing explicit
directory boundaries for file operations gives developers fine-grained
control over which parts of the filesystem the plugin can access. This
follows the pattern established by `HttpPlugin.allowed_domains`.

- Add `allowed_upload_directories` parameter to control which local
directories can be used for file uploads
- Add `allowed_download_directories` parameter for optional restrictions
on file download destinations
- Implement path canonicalization to resolve symlinks and normalize
paths

Usage:

```python
tool = SessionsPythonTool(
    credential=credential,
    allowed_upload_directories={"/app/data/uploads", "/app/user_files"},
    allowed_download_directories={"/app/data/downloads"},  # optional
)
```


<!-- Thank you for your contribution to the semantic-kernel repo!
Please help reviewers and future users, providing the following
information:
  1. Why is this change required?
  2. What problem does it solve?
  3. What scenario does it contribute to?
  4. If it fixes an open issue, please link to the issue here.
-->

### Description

- `upload_file` now requires `allowed_upload_directories` to be
configured (deny-by-default)
- `download_file` optionally supports `allowed_download_directories`
(permissive-by-default, since it's not exposed as a kernel function)
- Both parameters accept `set[str]` or `list[str]` for convenience
- Paths are canonicalized using `os.path.realpath()` before validation

<!-- Describe your changes, the overall approach, the underlying design.
These notes will help understanding how your code works. Thanks! -->

### Contribution Checklist

<!-- Before submitting this PR, please make sure: -->

- [X] The code builds clean without any errors or warnings
- [X] The PR follows the [SK Contribution
Guidelines](https://github.com/microsoft/semantic-kernel/blob/main/CONTRIBUTING.md)
and the [pre-submission formatting
script](https://github.com/microsoft/semantic-kernel/blob/main/CONTRIBUTING.md#development-scripts)
raises no violations
- [X] All unit tests pass, and I have added new tests where possible
- [ ] I didn't break anyone 😄

python/samples/getting_started_with_agents/chat_completion/step12_chat_completion_agent_code_interpreter.py

@@ -29,8 +29,15 @@ async def handle_intermediate_steps(message: ChatMessageContent) -> None:
 async def main():
     credential = AzureCliCredential()
 
+    # Define the resources directory for file uploads
+    resources_dir = os.path.join(os.path.dirname(os.path.dirname(os.path.realpath(__file__))), "resources")
+
     # 1. Create the python code interpreter tool using the SessionsPythonTool
-    python_code_interpreter = SessionsPythonTool(credential=credential)
+    # allowed_upload_directories restricts which local directories can be accessed for uploads
+    python_code_interpreter = SessionsPythonTool(
+        credential=credential,
+        allowed_upload_directories=[resources_dir],
+    )
 
     # 2. Create the agent
     agent = ChatCompletionAgent(
@@ -41,7 +48,7 @@ async def main():
     )
 
     # 3. Upload a CSV file to the session
-    csv_file_path = os.path.join(os.path.dirname(os.path.dirname(os.path.realpath(__file__))), "resources", "sales.csv")
+    csv_file_path = os.path.join(resources_dir, "sales.csv")
     file_metadata = await python_code_interpreter.upload_file(local_file_path=csv_file_path)
 
     # 4. Invoke the agent for a response to a task

python/semantic_kernel/core_plugins/sessions_python_tool/sessions_python_plugin.py

@@ -38,6 +38,10 @@ class SessionsPythonTool(KernelBaseModel):
     settings: SessionsPythonSettings
     auth_callback: Callable[..., Any | Awaitable[Any]]
     http_client: AsyncClient
+    allowed_upload_directories: set[str] | None = None
+    """Allowed local directories for file uploads. If None, upload_file is disabled (deny-by-default)."""
+    allowed_download_directories: set[str] | None = None
+    """Allowed local directories for file downloads. If None, all paths are allowed (permissive-by-default)."""
 
     def __init__(
         self,
@@ -48,9 +52,28 @@ class SessionsPythonTool(KernelBaseModel):
         env_file_path: str | None = None,
         token_endpoint: str | None = None,
         credential: TokenCredential | None = None,
+        allowed_upload_directories: set[str] | list[str] | None = None,
+        allowed_download_directories: set[str] | list[str] | None = None,
         **kwargs,
     ):
-        """Initializes a new instance of the SessionsPythonTool class."""
+        """Initializes a new instance of the SessionsPythonTool class.
+
+        Args:
+            auth_callback: Callback to retrieve authentication token.
+            pool_management_endpoint: The ACA pool management endpoint URL.
+            settings: Python session settings.
+            http_client: HTTP client for making requests.
+            env_file_path: Path to .env file.
+            token_endpoint: Token endpoint for authentication.
+            credential: Azure credential for authentication.
+            allowed_upload_directories: Set or list of allowed directories for file uploads.
+                If None, upload_file will be disabled (deny-by-default).
+                Empty set/list means no directories are allowed (all uploads denied).
+            allowed_download_directories: Set or list of allowed directories for file downloads.
+                If None, all paths are allowed (permissive-by-default).
+                Empty set/list means no directories are allowed (all local downloads denied).
+            kwargs: Additional keyword arguments.
+        """
         try:
             aca_settings = ACASessionsSettings(
                 env_file_path=env_file_path,
@@ -70,11 +93,19 @@ class SessionsPythonTool(KernelBaseModel):
         if auth_callback is None:
             auth_callback = self._default_auth_callback(aca_settings, credential)
 
+        # Convert lists to sets and filter out empty strings (which resolve to CWD)
+        upload_dirs = {d for d in allowed_upload_directories if d} if allowed_upload_directories is not None else None
+        download_dirs = (
+            {d for d in allowed_download_directories if d} if allowed_download_directories is not None else None
+        )
+
         super().__init__(
             pool_management_endpoint=aca_settings.pool_management_endpoint,
             settings=settings,
             auth_callback=auth_callback,
             http_client=http_client,
+            allowed_upload_directories=upload_dirs,
+            allowed_download_directories=download_dirs,
             **kwargs,
         )
 
@@ -145,6 +176,74 @@ class SessionsPythonTool(KernelBaseModel):
             endpoint = endpoint[:-1]
         return f"{base_url}{endpoint}?{query_string}"
 
+    def _validate_local_path_for_upload(self, local_file_path: str) -> str:
+        """Validate local path is within allowed upload directories.
+
+        Args:
+            local_file_path: The path to validate.
+
+        Returns:
+            str: The canonicalized absolute path.
+
+        Raises:
+            FunctionExecutionException: If the path is not within allowed directories.
+        """
+        if self.allowed_upload_directories is None:
+            raise FunctionExecutionException(
+                "File upload is disabled. Configure 'allowed_upload_directories' to enable."
+            )
+
+        canonical_path = os.path.realpath(local_file_path)
+
+        for allowed_dir in self.allowed_upload_directories:
+            allowed_canonical = os.path.realpath(allowed_dir)
+            try:
+                common = os.path.commonpath([allowed_canonical, canonical_path])
+                if common == allowed_canonical:
+                    return canonical_path
+            except ValueError:
+                continue  # Different drives on Windows
+
+        logger.warning(f"Upload denied for path: {local_file_path} (resolved: {canonical_path})")
+        raise FunctionExecutionException(
+            f"Access denied: '{local_file_path}' is not within allowed upload directories."
+        )
+
+    def _validate_local_path_for_download(self, local_file_path: str) -> str:
+        """Validate local path is within allowed download directories (optional protection).
+
+        Args:
+            local_file_path: The path to validate.
+
+        Returns:
+            str: The canonicalized absolute path.
+
+        Raises:
+            FunctionExecutionException: If allowed_download_directories is set and path is not within.
+        """
+        # Permissive by default - if no restrictions configured, allow all paths
+        if self.allowed_download_directories is None:
+            return os.path.realpath(local_file_path)
+
+        parent_dir = os.path.dirname(local_file_path) or "."
+        canonical_parent = os.path.realpath(parent_dir)
+        filename = os.path.basename(local_file_path)
+        canonical_path = os.path.join(canonical_parent, filename)
+
+        for allowed_dir in self.allowed_download_directories:
+            allowed_canonical = os.path.realpath(allowed_dir)
+            try:
+                common = os.path.commonpath([allowed_canonical, canonical_parent])
+                if common == allowed_canonical:
+                    return canonical_path
+            except ValueError:
+                continue
+
+        logger.warning(f"Download denied for path: {local_file_path}")
+        raise FunctionExecutionException(
+            f"Access denied: '{local_file_path}' is not within allowed download directories."
+        )
+
     # endregion
 
     # region Kernel Functions
@@ -230,17 +329,21 @@ class SessionsPythonTool(KernelBaseModel):
         Args:
             remote_file_path (str): The path to the file in the session.
             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.
+            FunctionExecutionException: If 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.")
 
-        remote_file_path = self._construct_remote_file_path(remote_file_path or os.path.basename(local_file_path))
+        # Validate path is in allowed directories (deny-by-default)
+        validated_path = self._validate_local_path_for_upload(local_file_path)
+
+        remote_file_path = self._construct_remote_file_path(remote_file_path or os.path.basename(validated_path))
 
         auth_token = await self._ensure_auth_token()
         self.http_client.headers.update({
@@ -255,7 +358,7 @@ class SessionsPythonTool(KernelBaseModel):
         )
 
         try:
-            with open(local_file_path, "rb") as data:
+            with open(validated_path, "rb") as data:
                 files = {"file": (remote_file_path, data, "application/octet-stream")}
                 response = await self.http_client.post(url=url, files=files)
                 response.raise_for_status()
@@ -312,10 +415,14 @@ class SessionsPythonTool(KernelBaseModel):
         Args:
             remote_file_name: The name of the file to download, relative to `/mnt/data`.
             local_file_path: The path to save the downloaded file to. Should include the extension.
-                If not provided, the file is returned as a BufferedReader.
+                If not provided, the file is returned as a BytesIO object.
+                If allowed_download_directories is configured, must be within those directories.
 
         Returns:
-            BufferedReader: The data of the downloaded file.
+            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 (when configured).
         """
         auth_token = await self._ensure_auth_token()
         self.http_client.headers.update({
@@ -335,7 +442,9 @@ class SessionsPythonTool(KernelBaseModel):
             )
             response.raise_for_status()
             if local_file_path:
-                with open(local_file_path, "wb") as f:
+                # Validate path is in allowed directories (optional, permissive by default)
+                validated_path = self._validate_local_path_for_download(local_file_path)
+                with open(validated_path, "wb") as f:
                     f.write(response.content)
                 return None
 

python/tests/unit/core_plugins/test_sessions_python_plugin.py

@@ -183,18 +183,19 @@ async def test_empty_call_to_container_fails_raises_exception(aca_python_session
 
 @patch("httpx.AsyncClient.get")
 @patch("httpx.AsyncClient.post")
-async def test_upload_file_with_local_path(mock_post, mock_get, aca_python_sessions_unit_test_env):
+async def test_upload_file_with_local_path(mock_post, mock_get, aca_python_sessions_unit_test_env, tmp_path):
     """Test upload_file when providing a local file path."""
 
     async def async_return(result):
         return result
 
-    with (
-        patch(
-            "semantic_kernel.core_plugins.sessions_python_tool.sessions_python_plugin.SessionsPythonTool._ensure_auth_token",
-            return_value="test_token",
-        ),
-        patch("builtins.open", mock_open(read_data=b"file data")),
+    # Create a real file in a temp directory for the test
+    test_file = tmp_path / "hello.py"
+    test_file.write_bytes(b"file data")
+
+    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(
@@ -231,9 +232,10 @@ async def test_upload_file_with_local_path(mock_post, mock_get, aca_python_sessi
         plugin = SessionsPythonTool(
             auth_callback=lambda: "sample_token",
             env_file_path="test.env",
+            allowed_upload_directories={str(tmp_path)},
         )
 
-        result = await plugin.upload_file(local_file_path="hello.py", remote_file_path="hello.py")
+        result = await plugin.upload_file(local_file_path=str(test_file), remote_file_path="hello.py")
         assert result.filename == "hello.py"
         assert result.size_in_bytes == 123
         assert result.full_path == "/mnt/data/hello.py"
@@ -242,18 +244,21 @@ async def test_upload_file_with_local_path(mock_post, mock_get, aca_python_sessi
 
 @patch("httpx.AsyncClient.get")
 @patch("httpx.AsyncClient.post")
-async def test_upload_file_with_local_path_and_no_remote(mock_post, mock_get, aca_python_sessions_unit_test_env):
+async def test_upload_file_with_local_path_and_no_remote(
+    mock_post, mock_get, aca_python_sessions_unit_test_env, tmp_path
+):
     """Test upload_file when providing a local file path."""
 
     async def async_return(result):
         return result
 
-    with (
-        patch(
-            "semantic_kernel.core_plugins.sessions_python_tool.sessions_python_plugin.SessionsPythonTool._ensure_auth_token",
-            return_value="test_token",
-        ),
-        patch("builtins.open", mock_open(read_data=b"file data")),
+    # Create a real file in a temp directory for the test
+    test_file = tmp_path / "hello.py"
+    test_file.write_bytes(b"file data")
+
+    with patch(
+        "semantic_kernel.core_plugins.sessions_python_tool.sessions_python_plugin.SessionsPythonTool._ensure_auth_token",
+        return_value="test_token",
     ):
         mock_post_request = httpx.Request(method="POST", url="https://example.com/files/upload?identifier=None")
         mock_post_response = httpx.Response(
@@ -290,16 +295,17 @@ async def test_upload_file_with_local_path_and_no_remote(mock_post, mock_get, ac
         plugin = SessionsPythonTool(
             auth_callback=lambda: "sample_token",
             env_file_path="test.env",
+            allowed_upload_directories={str(tmp_path)},
         )
 
-        result = await plugin.upload_file(local_file_path="hello.py")
+        result = await plugin.upload_file(local_file_path=str(test_file))
         assert result.filename == "hello.py"
         assert result.size_in_bytes == 123
         mock_post.assert_awaited_once()
 
 
 @patch("httpx.AsyncClient.post")
-async def test_upload_file_throws_exception(mock_post, aca_python_sessions_unit_test_env):
+async def test_upload_file_throws_exception(mock_post, aca_python_sessions_unit_test_env, tmp_path):
     """Test throwing exception during file upload."""
 
     async def async_raise_http_error(*args, **kwargs):
@@ -307,32 +313,34 @@ async def test_upload_file_throws_exception(mock_post, aca_python_sessions_unit_
         mock_response = httpx.Response(status_code=500, request=mock_request)
         raise HTTPStatusError("Server Error", request=mock_request, response=mock_response)
 
-    with (
-        patch(
-            "semantic_kernel.core_plugins.sessions_python_tool.sessions_python_plugin.SessionsPythonTool._ensure_auth_token",
-            return_value="test_token",
-        ),
-        patch("builtins.open", mock_open(read_data=b"file data")),
+    # Create a real file in a temp directory for the test
+    test_file = tmp_path / "hello.py"
+    test_file.write_bytes(b"file data")
+
+    with patch(
+        "semantic_kernel.core_plugins.sessions_python_tool.sessions_python_plugin.SessionsPythonTool._ensure_auth_token",
+        return_value="test_token",
     ):
         mock_post.side_effect = async_raise_http_error
 
         plugin = SessionsPythonTool(
             auth_callback=lambda: "sample_token",
             env_file_path="test.env",
+            allowed_upload_directories={str(tmp_path)},
         )
 
         with pytest.raises(
             FunctionExecutionException, match="Upload failed with status code 500 and error: Internal Server Error"
         ):
-            await plugin.upload_file(local_file_path="hello.py")
+            await plugin.upload_file(local_file_path=str(test_file))
         mock_post.assert_awaited_once()
 
 
 @pytest.mark.parametrize(
-    "local_file_path, input_remote_file_path, expected_remote_file_path",
+    "input_remote_file_path, expected_remote_file_path",
     [
-        ("./file.py", "uploaded_test.txt", "/mnt/data/uploaded_test.txt"),
-        ("./file.py", "/mnt/data/input.py", "/mnt/data/input.py"),
+        ("uploaded_test.txt", "/mnt/data/uploaded_test.txt"),
+        ("/mnt/data/input.py", "/mnt/data/input.py"),
     ],
 )
 @patch("httpx.AsyncClient.get")
@@ -340,22 +348,23 @@ async def test_upload_file_throws_exception(mock_post, aca_python_sessions_unit_
 async def test_upload_file_with_buffer(
     mock_post,
     mock_get,
-    local_file_path,
     input_remote_file_path,
     expected_remote_file_path,
     aca_python_sessions_unit_test_env,
+    tmp_path,
 ):
     """Test upload_file when providing file data as a BufferedReader."""
 
     async def async_return(result):
         return result
 
-    with (
-        patch(
-            "semantic_kernel.core_plugins.sessions_python_tool.sessions_python_plugin.SessionsPythonTool._ensure_auth_token",
-            return_value="test_token",
-        ),
-        patch("builtins.open", mock_open(read_data="print('hello, world~')")),
+    # Create a real file in a temp directory for the test
+    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(
@@ -389,9 +398,12 @@ async def test_upload_file_with_buffer(
         )
         mock_get.return_value = await async_return(mock_get_response)
 
-        plugin = SessionsPythonTool(auth_callback=lambda: "sample_token")
+        plugin = SessionsPythonTool(
+            auth_callback=lambda: "sample_token",
+            allowed_upload_directories={str(tmp_path)},
+        )
 
-        result = await plugin.upload_file(local_file_path=local_file_path, remote_file_path=input_remote_file_path)
+        result = await plugin.upload_file(local_file_path=str(test_file), remote_file_path=input_remote_file_path)
         assert result.filename == expected_remote_file_path
         assert result.size_in_bytes == 456
         mock_post.assert_awaited_once()
@@ -491,7 +503,7 @@ async def test_list_files_throws_exception(mock_get, aca_python_sessions_unit_te
 
 
 @patch("httpx.AsyncClient.get")
-async def test_download_file_to_local(mock_get, aca_python_sessions_unit_test_env):
+async def test_download_file_to_local(mock_get, aca_python_sessions_unit_test_env, tmp_path):
     """Test download_file when saving to a local file path."""
 
     async def async_return(result):
@@ -500,6 +512,8 @@ async def test_download_file_to_local(mock_get, aca_python_sessions_unit_test_en
     async def mock_auth_callback():
         return "test_token"
 
+    local_file = tmp_path / "local_test.txt"
+
     with (
         patch(
             "semantic_kernel.core_plugins.sessions_python_tool.sessions_python_plugin.SessionsPythonTool._ensure_auth_token",
@@ -520,9 +534,10 @@ async def test_download_file_to_local(mock_get, aca_python_sessions_unit_test_en
             env_file_path="test.env",
         )
 
-        await plugin.download_file(remote_file_name="remote_test.txt", local_file_path="local_test.txt")
+        await plugin.download_file(remote_file_name="remote_test.txt", local_file_path=str(local_file))
         mock_get.assert_awaited_once()
-        mock_file.assert_called_once_with("local_test.txt", "wb")
+        # Path is canonicalized via os.path.realpath()
+        mock_file.assert_called_once_with(str(local_file), "wb")
         mock_file().write.assert_called_once_with(b"file data")
 
 
@@ -646,3 +661,232 @@ async def test_auth_token_fail(aca_python_sessions_unit_test_env):
 def test_full_path(filename, expected_full_path):
     metadata = SessionsRemoteFileMetadata(filename=filename, size_in_bytes=123)
     assert metadata.full_path == expected_full_path
+
+
+# region Tests for File Operations
+
+
+async def test_upload_file_denied_when_no_allowed_directories(aca_python_sessions_unit_test_env):
+    """Test that upload_file raises an exception when allowed_upload_directories is not configured."""
+    plugin = SessionsPythonTool(auth_callback=lambda: "sample_token")
+
+    with pytest.raises(FunctionExecutionException, match="File upload is disabled"):
+        await plugin.upload_file(local_file_path="/some/path/file.txt")
+
+
+async def test_upload_file_denied_outside_allowed_directories(aca_python_sessions_unit_test_env, tmp_path):
+    """Test that upload_file raises an exception when path is outside allowed directories."""
+    allowed_dir = tmp_path / "allowed"
+    allowed_dir.mkdir()
+
+    plugin = SessionsPythonTool(
+        auth_callback=lambda: "sample_token",
+        allowed_upload_directories={str(allowed_dir)},
+    )
+
+    with pytest.raises(FunctionExecutionException, match="Access denied"):
+        await plugin.upload_file(local_file_path="/etc/passwd")
+
+
+async def test_upload_file_path_traversal_blocked(aca_python_sessions_unit_test_env, tmp_path):
+    """Test that path traversal attacks are blocked."""
+    allowed_dir = tmp_path / "allowed"
+    allowed_dir.mkdir()
+
+    plugin = SessionsPythonTool(
+        auth_callback=lambda: "sample_token",
+        allowed_upload_directories={str(allowed_dir)},
+    )
+
+    # Attempt path traversal
+    traversal_path = str(allowed_dir / ".." / ".." / "etc" / "passwd")
+
+    with pytest.raises(FunctionExecutionException, match="Access denied"):
+        await plugin.upload_file(local_file_path=traversal_path)
+
+
+@patch("httpx.AsyncClient.get")
+@patch("httpx.AsyncClient.post")
+async def test_upload_file_succeeds_within_allowed_directory(
+    mock_post, mock_get, aca_python_sessions_unit_test_env, tmp_path
+):
+    """Test that upload_file succeeds when path is within allowed directories."""
+
+    async def async_return(result):
+        return result
+
+    allowed_dir = tmp_path / "allowed"
+    allowed_dir.mkdir()
+    test_file = allowed_dir / "test.txt"
+    test_file.write_text("test content")
+
+    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.return_value = await 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": "test.txt",
+                            "size": 12,
+                            "lastModifiedTime": "2024-07-02T19:29:23.4369699Z",
+                        },
+                    },
+                ],
+            },
+            request=mock_get_request,
+        )
+        mock_get.return_value = await async_return(mock_get_response)
+
+        plugin = SessionsPythonTool(
+            auth_callback=lambda: "sample_token",
+            allowed_upload_directories={str(allowed_dir)},
+        )
+
+        result = await plugin.upload_file(local_file_path=str(test_file))
+        assert result.filename == "test.txt"
+        mock_post.assert_awaited_once()
+
+
+@patch("httpx.AsyncClient.get")
+async def test_download_file_works_without_allowed_directories(mock_get, aca_python_sessions_unit_test_env, tmp_path):
+    """Test that download_file works without restrictions when allowed_download_directories is None."""
+
+    async def async_return(result):
+        return result
+
+    local_file = tmp_path / "downloaded.txt"
+
+    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="GET", url="https://example.com/files/content/remote.txt")
+        mock_response = httpx.Response(status_code=200, content=b"file data", request=mock_request)
+        mock_get.return_value = await async_return(mock_response)
+
+        # No allowed_download_directories configured - should work (permissive by default)
+        plugin = SessionsPythonTool(auth_callback=lambda: "sample_token")
+
+        await plugin.download_file(remote_file_name="remote.txt", local_file_path=str(local_file))
+        assert local_file.read_bytes() == b"file data"
+        mock_get.assert_awaited_once()
+
+
+async def test_download_file_denied_outside_allowed_directories(aca_python_sessions_unit_test_env, tmp_path):
+    """Test that download_file raises an exception when path is outside allowed directories."""
+    allowed_dir = tmp_path / "allowed"
+    allowed_dir.mkdir()
+    outside_dir = tmp_path / "outside"
+    outside_dir.mkdir()
+
+    plugin = SessionsPythonTool(
+        auth_callback=lambda: "sample_token",
+        allowed_download_directories={str(allowed_dir)},
+    )
+
+    with (
+        patch(
+            "semantic_kernel.core_plugins.sessions_python_tool.sessions_python_plugin.SessionsPythonTool._ensure_auth_token",
+            return_value="test_token",
+        ),
+        patch("httpx.AsyncClient.get") as mock_get,
+    ):
+
+        async def async_return(result):
+            return result
+
+        mock_request = httpx.Request(method="GET", url="https://example.com/files/content/remote.txt")
+        mock_response = httpx.Response(status_code=200, content=b"file data", request=mock_request)
+        mock_get.return_value = await async_return(mock_response)
+
+        with pytest.raises(FunctionExecutionException, match="Access denied"):
+            await plugin.download_file(
+                remote_file_name="remote.txt",
+                local_file_path=str(outside_dir / "output.txt"),
+            )
+
+
+@patch("httpx.AsyncClient.get")
+async def test_download_file_succeeds_within_allowed_directory(mock_get, aca_python_sessions_unit_test_env, tmp_path):
+    """Test that download_file succeeds when path is within allowed directories."""
+
+    async def async_return(result):
+        return result
+
+    allowed_dir = tmp_path / "allowed"
+    allowed_dir.mkdir()
+    local_file = allowed_dir / "downloaded.txt"
+
+    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="GET", url="https://example.com/files/content/remote.txt")
+        mock_response = httpx.Response(status_code=200, content=b"file data", request=mock_request)
+        mock_get.return_value = await async_return(mock_response)
+
+        plugin = SessionsPythonTool(
+            auth_callback=lambda: "sample_token",
+            allowed_download_directories={str(allowed_dir)},
+        )
+
+        await plugin.download_file(remote_file_name="remote.txt", local_file_path=str(local_file))
+        assert local_file.read_bytes() == b"file data"
+        mock_get.assert_awaited_once()
+
+
+def test_allowed_directories_accepts_list(aca_python_sessions_unit_test_env):
+    """Test that allowed directories can be passed as a list and are converted to a set."""
+    plugin = SessionsPythonTool(
+        auth_callback=lambda: "sample_token",
+        allowed_upload_directories=["/path/one", "/path/two"],
+        allowed_download_directories=["/path/three"],
+    )
+
+    assert plugin.allowed_upload_directories == {"/path/one", "/path/two"}
+    assert plugin.allowed_download_directories == {"/path/three"}
+
+
+async def test_empty_set_denies_all_uploads(aca_python_sessions_unit_test_env, tmp_path):
+    """Test that an empty set for allowed_upload_directories denies all uploads."""
+    test_file = tmp_path / "test.txt"
+    test_file.write_text("content")
+
+    plugin = SessionsPythonTool(
+        auth_callback=lambda: "sample_token",
+        allowed_upload_directories=set(),  # Empty set - deny all
+    )
+
+    with pytest.raises(FunctionExecutionException, match="Access denied"):
+        await plugin.upload_file(local_file_path=str(test_file))
+
+
+def test_empty_strings_filtered_from_allowed_directories(aca_python_sessions_unit_test_env):
+    """Test that empty strings are filtered out from allowed directories."""
+    plugin = SessionsPythonTool(
+        auth_callback=lambda: "sample_token",
+        allowed_upload_directories=["", "/valid/path", ""],
+        allowed_download_directories=["", ""],
+    )
+
+    assert plugin.allowed_upload_directories == {"/valid/path"}
+    assert plugin.allowed_download_directories == set()  # All empty strings filtered out
+
+
+# endregion