fix: validate example file names (#3066)

Co-authored-by: wolfkill <27876473+wolfkill@users.noreply.github.com>

packages/dbgpt-app/src/dbgpt_app/openapi/api_v1/examples_api.py

@@ -13,6 +13,7 @@ Example files are resolved in the following order:
 import logging
 import os
 import shutil
+from pathlib import PurePosixPath, PureWindowsPath
 from typing import Optional
 
 from fastapi import APIRouter, Body, Depends
@@ -61,6 +62,23 @@ EXAMPLE_FILES = {
 }
 
 
+def _validate_example_filename(filename: str) -> str:
+    if "\x00" in filename:
+        raise ValueError("example filename must not contain null bytes")
+
+    posix_path = PurePosixPath(filename)
+    windows_path = PureWindowsPath(filename)
+    if (
+        posix_path.is_absolute()
+        or windows_path.is_absolute()
+        or len(posix_path.parts) != 1
+        or len(windows_path.parts) != 1
+        or filename in {"", ".", ".."}
+    ):
+        raise ValueError("example filename must be a plain file name")
+    return filename
+
+
 def _resolve_example_source(example: dict) -> Optional[str]:
     """Return the absolute path to an example file, or *None* if not found.
 
@@ -126,7 +144,12 @@ async def use_example_file(
         upload_dir = os.path.join(base_dir, "python_uploads", user_id)
         os.makedirs(upload_dir, exist_ok=True)
 
-        target_path = os.path.join(upload_dir, example["name"])
+        try:
+            target_name = _validate_example_filename(example["name"])
+        except ValueError as exc:
+            return Result.failed(msg=str(exc))
+
+        target_path = os.path.join(upload_dir, target_name)
         shutil.copy2(source_path, target_path)
 
         abs_path = os.path.abspath(target_path)

packages/dbgpt-app/src/dbgpt_app/tests/test_examples_api.py

@@ -0,0 +1,75 @@
+import pytest
+
+from dbgpt_serve.utils.auth import UserRequest
+
+
+def _configure_example(tmp_path, monkeypatch, name: str):
+    from dbgpt_app.openapi.api_v1 import examples_api
+
+    source_path = tmp_path / "source.csv"
+    source_path.write_text("value\n1\n", encoding="utf-8")
+    monkeypatch.chdir(tmp_path)
+    monkeypatch.setitem(
+        examples_api.EXAMPLE_FILES,
+        "test_example",
+        {
+            "source_path": "source.csv",
+            "builtin_path": "source.csv",
+            "name": name,
+        },
+    )
+    monkeypatch.setattr(
+        examples_api, "_resolve_example_source", lambda _: str(source_path)
+    )
+    return examples_api
+
+
+@pytest.mark.asyncio
+async def test_use_example_file_rejects_traversal_name(tmp_path, monkeypatch):
+    examples_api = _configure_example(tmp_path, monkeypatch, "../../outside.csv")
+
+    result = await examples_api.use_example_file(
+        "test_example", UserRequest(user_id="alice")
+    )
+
+    assert result.success is False
+    assert not (tmp_path / "outside.csv").exists()
+
+
+@pytest.mark.asyncio
+async def test_use_example_file_rejects_absolute_name(tmp_path, monkeypatch):
+    outside_path = tmp_path / "outside.csv"
+    examples_api = _configure_example(tmp_path, monkeypatch, str(outside_path))
+
+    result = await examples_api.use_example_file(
+        "test_example", UserRequest(user_id="alice")
+    )
+
+    assert result.success is False
+    assert not outside_path.exists()
+
+
+@pytest.mark.asyncio
+async def test_use_example_file_rejects_windows_path_separator(tmp_path, monkeypatch):
+    examples_api = _configure_example(tmp_path, monkeypatch, r"..\outside.csv")
+
+    result = await examples_api.use_example_file(
+        "test_example", UserRequest(user_id="alice")
+    )
+
+    assert result.success is False
+    assert not (tmp_path / "python_uploads" / "alice" / r"..\outside.csv").exists()
+
+
+@pytest.mark.asyncio
+async def test_use_example_file_accepts_plain_name(tmp_path, monkeypatch):
+    examples_api = _configure_example(tmp_path, monkeypatch, "report.csv")
+
+    result = await examples_api.use_example_file(
+        "test_example", UserRequest(user_id="alice")
+    )
+
+    target_path = tmp_path / "python_uploads" / "alice" / "report.csv"
+    assert result.success is True
+    assert result.data == str(target_path)
+    assert target_path.read_text(encoding="utf-8") == "value\n1\n"