fix(sync): block path traversal in storage-metadata write (GHSA-833q, 0.60.3)

.claude-plugin/marketplace.json

@@ -10,7 +10,7 @@
   "plugins": [
     {
       "name": "kbagent",
-      "version": "0.60.2",
+      "version": "0.60.3",
       "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.2",
+  "version": "0.60.3",
   "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.2"
+version = "0.60.3"
 description = "AI-friendly CLI for managing Keboola projects"
 readme = "README.md"
 requires-python = ">=3.12"

src/keboola_agent_cli/changelog.py

@@ -24,6 +24,19 @@
 
 # Ordered newest-first.  Each value is a list of brief one-line descriptions.
 CHANGELOG: dict[str, list[str]] = {
+    "0.60.3": [
+        "Security: `kbagent sync pull` now sanitizes the API-supplied bucket id and table name "
+        "before using them as filesystem paths when writing storage metadata + samples, and asserts "
+        "the resolved path stays inside the sync workspace. `_write_storage_metadata` previously used "
+        "the table `name` verbatim (and `bucket_id.replace('.', '-')`, which neutralizes `..` but not "
+        "`/` or an absolute path), so a malicious or compromised Storage API response with a table "
+        "named like `../../../../etc/cron.d/evil` could write attacker-controlled JSON outside the "
+        "workspace. The config-write path already had this defense (`sanitize_path_segment` + "
+        "`_ensure_within_branch`); the storage-metadata and samples writers now mirror it via "
+        "`sanitize_path_segment(...)` plus a new `_ensure_path_within` containment check. Behavior is "
+        "unchanged for legitimate data: the `in.c-foo` -> `in-c-foo` bucket-directory convention and "
+        "the `<table>.json` filename are preserved. Private advisory GHSA-833q-c5wv-26r7.",
+    ],
     "0.60.2": [
         "Security: scheduled `ai_agent` tasks (claude/codex/gemini spawned by `kbagent serve`) no "
         "longer inherit the manage (super-admin) or master tokens from the serve process "

src/keboola_agent_cli/services/sync_service.py

@@ -56,7 +56,7 @@
     load_manifest,
     save_manifest,
 )
-from ..sync.naming import config_path, config_row_path, sanitize_name
+from ..sync.naming import config_path, config_row_path, sanitize_name, sanitize_path_segment
 from ._encryption import (
     apply_encrypted_to_local,
     encrypt_secrets_in_config,
@@ -167,6 +167,28 @@ def _ensure_within_branch(
         )
 
 
+def _ensure_path_within(base_dir: Path, target: Path, what: str) -> None:
+    """Reject a write whose path escapes *base_dir* (defense-in-depth).
+
+    Mirrors :func:`_ensure_within_branch` for non-config writes (storage
+    metadata + samples) whose path segments derive from API-controlled bucket
+    ids / table names (GHSA-833q-c5wv-26r7). Raises ConfigError on escape so a
+    malformed or compromised Storage response cannot write outside the sync
+    workspace.
+    """
+    try:
+        base_resolved = base_dir.resolve()
+        target_resolved = target.resolve()
+    except OSError as exc:
+        raise ConfigError(f"Cannot resolve sync path: {exc}") from exc
+    if not target_resolved.is_relative_to(base_resolved):
+        raise ConfigError(
+            f"Storage path escapes sync workspace ({what}). Refusing to write "
+            f"outside '{base_resolved}'. This indicates a malformed or compromised "
+            f"API response or a path-sanitization regression."
+        )
+
+
 def scan_synced_plaintext_secrets(
     project_root: Path, manifest: Manifest | None = None
 ) -> list[dict[str, Any]]:
@@ -3097,9 +3119,13 @@ def _write_storage_metadata(
         tables_written = 0
         tables_dir = storage_dir / "tables"
         for bucket_id, bucket_tables in tables_by_bucket.items():
-            # Sanitize bucket_id for filesystem (replace dots with dashes)
-            safe_bucket = bucket_id.replace(".", "-")
+            # Sanitize bucket_id for filesystem. sanitize_path_segment first
+            # kills traversal (`/`, `..`, absolute paths); the trailing replace
+            # keeps the legacy `in.c-foo` -> `in-c-foo` directory naming for
+            # legitimate ids (GHSA-833q-c5wv-26r7).
+            safe_bucket = sanitize_path_segment(bucket_id).replace(".", "-")
             bucket_dir = tables_dir / safe_bucket
+            _ensure_path_within(storage_dir, bucket_dir, f"bucket_id={bucket_id!r}")
             bucket_dir.mkdir(parents=True, exist_ok=True)
 
             for t in bucket_tables:
@@ -3120,7 +3146,13 @@ def _write_storage_metadata(
                     "metadata": t.get("metadata", []),
                     "column_metadata": t.get("columnMetadata", {}),
                 }
-                table_file = bucket_dir / f"{table_name}.json"
+                # The table name comes from the API; sanitize it for the
+                # filename and assert containment so a crafted name cannot
+                # escape the bucket dir (GHSA-833q-c5wv-26r7). The original
+                # name stays verbatim in the metadata body above.
+                safe_table = sanitize_path_segment(table_name)
+                table_file = bucket_dir / f"{safe_table}.json"
+                _ensure_path_within(storage_dir, table_file, f"table={table_name!r}")
                 table_file.write_text(
                     json.dumps(table_meta, indent=2, ensure_ascii=False),
                     encoding="utf-8",
@@ -3133,14 +3165,17 @@ def _write_storage_metadata(
             samples_dir = storage_dir / STORAGE_SAMPLES_DIR_NAME
             for table_id, csv_data in samples.items():
                 # table_id format: "in.c-bucket.table" -> samples/in-c-bucket/table/
+                # Every segment derives from the API table_id; sanitize each and
+                # assert containment so a crafted id cannot escape (GHSA-833q).
                 parts = table_id.split(".", 2)
                 if len(parts) >= 3:
-                    safe_bucket = f"{parts[0]}-{parts[1]}"
-                    table_name = parts[2]
+                    safe_bucket = sanitize_path_segment(f"{parts[0]}-{parts[1]}")
+                    safe_table = sanitize_path_segment(parts[2])
                 else:
-                    safe_bucket = table_id.replace(".", "-")
-                    table_name = "data"
-                sample_dir = samples_dir / safe_bucket / table_name
+                    safe_bucket = sanitize_path_segment(table_id.replace(".", "-"))
+                    safe_table = "data"
+                sample_dir = samples_dir / safe_bucket / safe_table
+                _ensure_path_within(storage_dir, sample_dir, f"table_id={table_id!r}")
                 sample_dir.mkdir(parents=True, exist_ok=True)
 
                 # Mask encrypted columns in CSV

tests/test_sync_storage_jobs.py

@@ -608,6 +608,102 @@ def test_samples_written_to_correct_path(self, tmp_config_dir: Path, tmp_path: P
         assert '"Alice"' in sample_file.read_text(encoding="utf-8")
 
 
+class TestWriteStorageMetadataPathTraversal:
+    """GHSA-833q-c5wv-26r7: API-controlled bucket ids / table names must not
+    escape the sync workspace when storage metadata is written."""
+
+    def _make_svc(self, tmp_config_dir: Path) -> SyncService:
+        store = setup_single_project(tmp_config_dir)
+        return SyncService(config_store=store)
+
+    def test_malicious_table_name_stays_inside_workspace(
+        self, tmp_config_dir: Path, tmp_path: Path
+    ) -> None:
+        svc = self._make_svc(tmp_config_dir)
+        project_root = tmp_path / "project"
+        project_root.mkdir()
+        tables = [
+            {
+                "id": "in.c-data.evil",
+                "name": "../../../../evil",
+                "bucket": {"id": "in.c-data"},
+                "columns": [],
+            }
+        ]
+
+        svc._write_storage_metadata(project_root, [], tables, {})
+
+        # The traversal target above the workspace must NOT be created; the
+        # metadata lands safely inside under a sanitized filename instead.
+        assert not (tmp_path / "evil.json").exists()
+        storage_dir = project_root / STORAGE_DIR_NAME
+        written = list(storage_dir.rglob("*.json"))
+        assert written, "table metadata should still be written (sanitized)"
+        for p in written:
+            assert p.resolve().is_relative_to(storage_dir.resolve())
+
+    def test_malicious_bucket_id_stays_inside_workspace(
+        self, tmp_config_dir: Path, tmp_path: Path
+    ) -> None:
+        svc = self._make_svc(tmp_config_dir)
+        project_root = tmp_path / "project"
+        project_root.mkdir()
+        tables = [
+            {
+                "id": "x.evil",
+                "name": "t",
+                "bucket": {"id": "../../../../tmp/pwned"},
+                "columns": [],
+            }
+        ]
+
+        svc._write_storage_metadata(project_root, [], tables, {})
+
+        storage_dir = project_root / STORAGE_DIR_NAME
+        written = list(storage_dir.rglob("*.json"))
+        assert written, "table metadata should still be written (sanitized)"
+        for p in written:
+            assert p.resolve().is_relative_to(storage_dir.resolve())
+        assert not (tmp_path / "tmp" / "pwned").exists()
+
+    def test_malicious_sample_id_stays_inside_workspace(
+        self, tmp_config_dir: Path, tmp_path: Path
+    ) -> None:
+        svc = self._make_svc(tmp_config_dir)
+        project_root = tmp_path / "project"
+        project_root.mkdir()
+        samples = {"in.c-data.../../../../etc/evil": "col\nval\n"}
+
+        svc._write_storage_metadata(project_root, [], [], samples)
+
+        storage_dir = project_root / STORAGE_DIR_NAME
+        written = list(storage_dir.rglob("sample.csv"))
+        assert written, "sample should still be written (sanitized)"
+        for p in written:
+            assert p.resolve().is_relative_to(storage_dir.resolve())
+
+    def test_legitimate_names_preserve_dir_convention(
+        self, tmp_config_dir: Path, tmp_path: Path
+    ) -> None:
+        # Regression: the fix must keep `in.c-foo` -> `in-c-foo` and the plain
+        # `<table>.json` filename for legitimate data (no behavior change).
+        svc = self._make_svc(tmp_config_dir)
+        project_root = tmp_path / "project"
+        project_root.mkdir()
+        tables = [
+            {
+                "id": "in.c-data.users",
+                "name": "users",
+                "bucket": {"id": "in.c-data"},
+                "columns": [],
+            }
+        ]
+
+        svc._write_storage_metadata(project_root, [], tables, {})
+
+        assert (project_root / STORAGE_DIR_NAME / "tables" / "in-c-data" / "users.json").exists()
+
+
 # ===================================================================
 # 4. SyncService tests - _write_per_config_jobs()
 # ===================================================================