diff --git a/python/samples/getting_started_with_agents/chat_completion/step12_chat_completion_agent_code_interpreter.py b/python/samples/getting_started_with_agents/chat_completion/step12_chat_completion_agent_code_interpreter.py index 754220ce9509..9b02b4312a9a 100644 --- a/python/samples/getting_started_with_agents/chat_completion/step12_chat_completion_agent_code_interpreter.py +++ b/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 diff --git a/python/semantic_kernel/core_plugins/sessions_python_tool/sessions_python_plugin.py b/python/semantic_kernel/core_plugins/sessions_python_tool/sessions_python_plugin.py index db3c3ca6fea8..cf9edc4e0be1 100644 --- a/python/semantic_kernel/core_plugins/sessions_python_tool/sessions_python_plugin.py +++ b/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 @@ def __init__( 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 @@ def __init__( 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 @@ def _build_url_with_version(self, base_url, endpoint, params): 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 @@ async def upload_file( 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 @@ async def upload_file( ) 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 @@ async def download_file( 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 @@ async def download_file( ) 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 diff --git a/python/tests/unit/core_plugins/test_sessions_python_plugin.py b/python/tests/unit/core_plugins/test_sessions_python_plugin.py index ebd6709f3d78..a6382846b18f 100644 --- a/python/tests/unit/core_plugins/test_sessions_python_plugin.py +++ b/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 async_return(result): 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 async_return(result): @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 async_return(result): 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 async_raise_http_error(*args, **kwargs): 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 async_raise_http_error(*args, **kwargs): 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 async_return(result): ) 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 async_raise_http_error(*args, **kwargs): @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 async_return(result): 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 mock_auth_callback(): 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 token_cb(): 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