Python: fix(python): check the full download path in SessionsPythonTool

python/.pre-commit-config.yaml

@@ -21,13 +21,13 @@ repos:
         name: Check Valid Python Samples
         types: ["python"]
   - repo: https://github.com/nbQA-dev/nbQA
-    rev: 1.8.5
+    rev: 1.10.0
     hooks:
       - id: nbqa-check-ast
         name: Check Valid Python Notebooks
         types: ["jupyter"]
   - repo: https://github.com/asottile/pyupgrade
-    rev: v3.17.0
+    rev: v3.21.2
     hooks:
     - id: pyupgrade
       args: [--py311-plus]

python/semantic_kernel/core_plugins/sessions_python_tool/sessions_python_plugin.py

@@ -230,6 +230,9 @@ def _validate_local_path_for_upload(self, local_file_path: str) -> str:
     def _validate_local_path_for_download(self, local_file_path: str) -> str:
         """Validate local path is within allowed download directories (optional protection).
 
+        The whole path is resolved, including the file name, so a symbolic link is judged by
+        the location it points to and not by where the link itself lives.
+
         Args:
             local_file_path: The path to validate.
 
@@ -243,21 +246,18 @@ def _validate_local_path_for_download(self, local_file_path: str) -> str:
         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)
+        canonical_path = os.path.realpath(local_file_path)
 
         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:
+                common = os.path.commonpath([allowed_canonical, canonical_path])
+                if common == allowed_canonical and canonical_path != allowed_canonical:
                     return canonical_path
             except ValueError:
-                continue
+                continue  # Different drives on Windows
 
-        logger.warning(f"Download denied for path: {local_file_path}")
+        logger.warning(f"Download denied for path: {local_file_path} (resolved: {canonical_path})")
         raise FunctionExecutionException(
             f"Access denied: '{local_file_path}' is not within allowed download directories."
         )

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, mock_open, patch
+from unittest.mock import AsyncMock, Mock, mock_open, patch
 
 import httpx
 import pytest
@@ -1032,6 +1032,70 @@ async def async_return(result):
         mock_get.assert_awaited_once()
 
 
+def _symlink_or_skip(link, target):
+    try:
+        link.symlink_to(target)
+    except (OSError, NotImplementedError):
+        pytest.skip("Symbolic links cannot be created in this environment.")
+
+
+async def _download_to(local_file_path, allowed_dir):
+    plugin = SessionsPythonTool(
+        auth_callback=lambda: "sample_token",
+        allowed_download_directories={str(allowed_dir)},
+    )
+    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)
+    with (
+        patch(
+            "semantic_kernel.core_plugins.sessions_python_tool.sessions_python_plugin.SessionsPythonTool._ensure_auth_token",
+            return_value="test_token",
+        ),
+        patch("httpx.AsyncClient.get", new=AsyncMock(return_value=mock_response)),
+    ):
+        await plugin.download_file(remote_file_name="remote.txt", local_file_path=str(local_file_path))
+
+
+@pytest.mark.parametrize("target_exists", [True, False])
+async def test_download_file_denied_through_symlink_to_outside(
+    aca_python_sessions_unit_test_env, tmp_path, target_exists
+):
+    """Test that a symlink inside an allowed directory cannot redirect the write outside it."""
+    allowed_dir = tmp_path / "allowed"
+    allowed_dir.mkdir()
+    outside_dir = tmp_path / "outside"
+    outside_dir.mkdir()
+    outside_file = outside_dir / "outside.txt"
+    if target_exists:
+        outside_file.write_bytes(b"original")
+    link = allowed_dir / "link.txt"
+    _symlink_or_skip(link, outside_file)
+
+    with pytest.raises(FunctionExecutionException, match="Access denied"):
+        await _download_to(link, allowed_dir)
+
+    if target_exists:
+        assert outside_file.read_bytes() == b"original"
+    else:
+        assert not outside_file.exists()
+
+
+async def test_download_file_allowed_through_symlink_within_allowed_directory(
+    aca_python_sessions_unit_test_env, tmp_path
+):
+    """Test that a symlink whose target stays inside the allowed directory is still usable."""
+    allowed_dir = tmp_path / "allowed"
+    allowed_dir.mkdir()
+    target = allowed_dir / "target.txt"
+    link = allowed_dir / "link.txt"
+    _symlink_or_skip(link, target)
+
+    await _download_to(link, allowed_dir)
+
+    assert target.read_bytes() == b"file data"
+    assert link.is_symlink()
+
+
 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(