fix(storage): block path traversal in file-download (GHSA-6px9-99p6-7j7g, 0.60.1)

.claude-plugin/marketplace.json

@@ -10,7 +10,7 @@
   "plugins": [
     {
       "name": "kbagent",
-      "version": "0.60.0",
+      "version": "0.60.1",
       "source": "./plugins/kbagent",
       "description": "AI-friendly interface to Keboola Connection projects — explore configs, jobs, lineage, call MCP tools, manage dev branches, and debug SQL in workspaces",
       "category": "development"

plugins/kbagent/.claude-plugin/plugin.json

@@ -1,6 +1,6 @@
 {
   "name": "kbagent",
-  "version": "0.60.0",
+  "version": "0.60.1",
   "description": "AI-friendly interface to Keboola Connection projects — explore configs, jobs, lineage, call MCP tools, manage dev branches, and debug SQL in workspaces",
   "author": {
     "name": "Keboola",

pyproject.toml

@@ -1,6 +1,6 @@
 [project]
 name = "keboola-agent-cli"
-version = "0.60.0"
+version = "0.60.1"
 description = "AI-friendly CLI for managing Keboola projects"
 readme = "README.md"
 requires-python = ">=3.12"

src/keboola_agent_cli/changelog.py

@@ -24,6 +24,18 @@
 
 # Ordered newest-first.  Each value is a list of brief one-line descriptions.
 CHANGELOG: dict[str, list[str]] = {
+    "0.60.1": [
+        "Security: `kbagent storage file-download` now contains the API-supplied file name under "
+        "the target directory, refusing path-traversal escapes. When `--output` was omitted, the "
+        "downloaded bytes were written to the server-provided `name` verbatim -- so a malicious or "
+        "compromised Storage API response with a name like `../../../../.zshrc` (or an absolute "
+        "path) could overwrite an arbitrary file on the user's machine with attacker-controlled "
+        "content. Leading separators are now stripped (an absolute name can no longer override the "
+        "target), legitimate nested subpaths are preserved, and the resolved path is asserted to "
+        "stay within the chosen directory (CWD, or the `--output` directory) -- otherwise the "
+        "download is rejected with `INVALID_ARGUMENT`. Reported via private advisory "
+        "GHSA-6px9-99p6-7j7g.",
+    ],
     "0.60.0": [
         "New (#353): kbagent installs and self-updates from a prebuilt wheel attached to each "
         "GitHub release instead of building from `git+` source. `uv tool install git+...` "

src/keboola_agent_cli/services/storage_service.py

@@ -44,6 +44,31 @@ def _detect_legacy_branch_storage(client: Any, branch_id: int | None) -> bool:
         return False
 
 
+def _safe_download_target(base: Path, server_name: str) -> Path:
+    """Contain an API-supplied file name under ``base``.
+
+    The Storage API controls the file ``name``; using it verbatim as a write
+    path lets a malicious or compromised response escape the user's chosen
+    directory (``../../etc/...`` or an absolute path) and overwrite arbitrary
+    files with attacker-controlled bytes. We strip leading separators so an
+    absolute name cannot override ``base``, preserve legitimate nested
+    subpaths, and assert the resolved path stays within ``base``.
+    """
+    cleaned = server_name.lstrip("/\\").strip() or "download"
+    candidate = (base / cleaned).resolve()
+    if not candidate.is_relative_to(base.resolve()):
+        raise KeboolaApiError(
+            message=(
+                f"Refusing to write outside the target directory: the "
+                f"server-provided file name {server_name!r} escapes {base.resolve()}"
+            ),
+            status_code=400,
+            error_code=ErrorCode.INVALID_ARGUMENT,
+            retryable=False,
+        )
+    return candidate
+
+
 # "name:TYPE" or "name:TYPE(length)" -- type is pass-through to the Keboola
 # Storage API, which validates type/length combinations per backend and
 # returns clear errors (e.g. "'10' is not valid length for INTEGER"). This
@@ -1753,7 +1778,13 @@ def download_file(
             is_parquet = is_sliced and file_name.endswith(".parquet")
 
             if is_parquet:
-                effective_output = output_path or f"{file_name}.d"
+                # --output is the user's own choice (trusted); without it the
+                # slice dir is derived from the API-controlled name, so contain
+                # it under CWD to block path traversal.
+                if output_path:
+                    effective_output = output_path
+                else:
+                    effective_output = str(_safe_download_target(Path.cwd(), f"{file_name}.d"))
                 slice_info = client.download_sliced_file_to_dir(file_detail, effective_output)
                 result: dict[str, Any] = {
                     "project_alias": alias,
@@ -1767,11 +1798,18 @@ def download_file(
                 }
                 return result
 
-            effective_output = output_path or file_name
             if output_path and Path(output_path).is_dir():
                 # Caller passed a directory (e.g. the REST file-download endpoint);
-                # save inside it under the file's own name instead of clobbering it.
-                effective_output = str(Path(output_path) / file_name)
+                # save inside it under the file's own name. The name comes from the
+                # API, so contain it under the directory to block path traversal.
+                effective_output = str(_safe_download_target(Path(output_path), file_name))
+            elif output_path:
+                # Explicit --output file path: the user's own choice (trusted).
+                effective_output = output_path
+            else:
+                # No --output: the API-controlled name becomes the path; contain
+                # it under CWD so a malicious name (../../, absolute) cannot escape.
+                effective_output = str(_safe_download_target(Path.cwd(), file_name))
             if is_sliced:
                 bytes_written = client.download_sliced_file(file_detail, effective_output)
             else:

tests/test_storage_files.py

@@ -1565,3 +1565,84 @@ def test_keep_slices_on_non_sliced_export_errors(self, tmp_path: Path) -> None:
         assert exc_info.value.error_code == "NOT_SLICED"
         # Crucially: the sliced download path must not have been reached
         fake_client.download_sliced_file_to_dir.assert_not_called()
+
+
+class TestDownloadFilePathTraversal:
+    """Security (GHSA-6px9-99p6-7j7g): an API-supplied file name must never
+    escape the user's target directory when writing the downloaded bytes."""
+
+    def test_rejects_traversal_name_no_output(self, tmp_path: Path, monkeypatch) -> None:
+        monkeypatch.chdir(tmp_path)
+        store = _make_store(tmp_path)
+        mock_client = MagicMock()
+        mock_client.get_file_info.return_value = {
+            **SAMPLE_FILE,
+            "name": "../../../../etc/cron.d/evil",
+            "isSliced": False,
+        }
+        service = _make_service(store, mock_client)
+
+        with pytest.raises(KeboolaApiError) as exc_info:
+            service.download_file(alias="test", file_id=12345)
+
+        assert exc_info.value.error_code == "INVALID_ARGUMENT"
+        # The escape is refused before any bytes are fetched/written.
+        mock_client.download_file.assert_not_called()
+
+    def test_neutralizes_absolute_name(self, tmp_path: Path, monkeypatch) -> None:
+        monkeypatch.chdir(tmp_path)
+        store = _make_store(tmp_path)
+        mock_client = MagicMock()
+        mock_client.get_file_info.return_value = {
+            **SAMPLE_FILE,
+            "name": "/etc/passwd",
+            "isSliced": False,
+        }
+        mock_client.download_file.return_value = 10
+        service = _make_service(store, mock_client)
+
+        service.download_file(alias="test", file_id=12345)
+
+        # Absolute name is stripped to a relative path contained under CWD,
+        # never written to the real /etc/passwd.
+        called_path = Path(mock_client.download_file.call_args[0][1])
+        assert called_path.is_relative_to(Path.cwd())
+        assert called_path.name == "passwd"
+
+    def test_preserves_legitimate_nested_name(self, tmp_path: Path, monkeypatch) -> None:
+        monkeypatch.chdir(tmp_path)
+        store = _make_store(tmp_path)
+        mock_client = MagicMock()
+        mock_client.get_file_info.return_value = {
+            **SAMPLE_FILE,
+            "name": "exports/2026/report.csv",
+            "isSliced": False,
+        }
+        mock_client.download_file.return_value = 10
+        service = _make_service(store, mock_client)
+
+        service.download_file(alias="test", file_id=12345)
+
+        # A benign nested name still creates the subpath -- behavior preserved.
+        called_path = Path(mock_client.download_file.call_args[0][1])
+        assert called_path.is_relative_to(Path.cwd())
+        assert called_path.parts[-3:] == ("exports", "2026", "report.csv")
+
+    def test_rejects_traversal_into_output_directory(self, tmp_path: Path, monkeypatch) -> None:
+        monkeypatch.chdir(tmp_path)
+        outdir = tmp_path / "downloads"
+        outdir.mkdir()
+        store = _make_store(tmp_path)
+        mock_client = MagicMock()
+        mock_client.get_file_info.return_value = {
+            **SAMPLE_FILE,
+            "name": "../../escape.txt",
+            "isSliced": False,
+        }
+        service = _make_service(store, mock_client)
+
+        with pytest.raises(KeboolaApiError) as exc_info:
+            service.download_file(alias="test", file_id=12345, output_path=str(outdir))
+
+        assert exc_info.value.error_code == "INVALID_ARGUMENT"
+        mock_client.download_file.assert_not_called()