fix(datasource): store Excel uploads under a sanitised name #1448

backend/apps/datasource/api/datasource.py

@@ -25,7 +25,7 @@
 from common.audit.schemas.logger_decorator import LogConfig, system_log
 from common.core.config import settings
 from common.core.deps import SessionDep, CurrentUser, Trans
-from common.utils.utils import SQLBotLogUtil
+from common.utils.utils import SQLBotLogUtil, safe_upload_name
 from ..crud.datasource import get_datasource_list, check_status, create_ds, update_ds, delete_ds, getTables, getFields, \
     update_table_and_fields, getTablesByDs, chooseTables, preview, updateTable, updateField, get_ds, fieldEnum, \
     check_status_by_id, sync_single_fields
@@ -39,6 +39,25 @@
 path = settings.EXCEL_PATH
 
 
+def _excel_upload_path(filename: str, trans) -> str:
+    """Resolve a stored Excel file inside EXCEL_PATH (CWE-22, issue #1448).
+
+    `filename` must already be a single path component; this is the second
+    barrier that keeps the resolved path inside the upload directory even when
+    the host treats another character as a separator.
+    """
+    base_dir = os.path.realpath(path)
+    try:
+        save_path = os.path.realpath(os.path.join(base_dir, filename))
+        inside = os.path.commonpath([save_path, base_dir]) == base_dir
+    except ValueError:  # embedded NUL, different drives or malformed path
+        inside = False
+        save_path = ""
+    if not inside or save_path == base_dir:
+        raise HTTPException(400, trans('i18n_ds_invalid_file_path'))
+    return save_path
+
+
 @router.get("/ws/{oid}", include_in_schema=False)
 @require_permissions(permission=SqlbotPermission(role=['ws_admin']))
 async def query_by_oid(session: SessionDep, user: CurrentUser, oid: int) -> List[CoreDatasource]:
@@ -326,13 +345,18 @@ def inner():
 @require_permissions(permission=SqlbotPermission(role=['ws_admin']))
 async def upload_excel(trans: Trans, session: SessionDep,
                        file: UploadFile = File(..., description=f"{PLACEHOLDER_PREFIX}ds_excel")):
-    ALLOWED_EXTENSIONS = {"xlsx", "xls", "csv"}
+    ALLOWED_EXTENSIONS = {".xlsx", ".xls", ".csv"}
     if not file.filename.lower().endswith(tuple(ALLOWED_EXTENSIONS)):
         raise HTTPException(400, trans('i18n_ds_unsupported_file_format'))
 
+    file_name = safe_upload_name(file.filename)
+    if not file_name:
+        raise HTTPException(400, trans('i18n_ds_invalid_file_path'))
+
     os.makedirs(path, exist_ok=True)
-    filename = f"{file.filename.split('.')[0]}_{hashlib.sha256(uuid.uuid4().bytes).hexdigest()[:10]}.{file.filename.split('.')[1]}"
-    save_path = os.path.join(path, filename)
+    name_root, name_ext = os.path.splitext(file_name)
+    filename = f"{name_root}_{hashlib.sha256(uuid.uuid4().bytes).hexdigest()[:10]}{name_ext}"
+    save_path = _excel_upload_path(filename, trans)
     with open(save_path, "wb") as f:
         f.write(await file.read())
 
@@ -561,9 +585,14 @@ async def parse_excel(trans: Trans, file: UploadFile = File(..., description=f"{
     if not file.filename.lower().endswith(tuple(ALLOWED_EXTENSIONS)):
         raise HTTPException(400, trans('i18n_ds_unsupported_file_format'))
 
+    file_name = safe_upload_name(file.filename)
+    if not file_name:
+        raise HTTPException(400, trans('i18n_ds_invalid_file_path'))
+
     os.makedirs(path, exist_ok=True)
-    filename = f"{file.filename.split('.')[0].split('/')[-1]}_{hashlib.sha256(uuid.uuid4().bytes).hexdigest()[:10]}.{file.filename.split('.')[-1]}"
-    save_path = os.path.join(path, filename)
+    name_root, name_ext = os.path.splitext(file_name)
+    filename = f"{name_root}_{hashlib.sha256(uuid.uuid4().bytes).hexdigest()[:10]}{name_ext}"
+    save_path = _excel_upload_path(filename, trans)
     with open(save_path, "wb") as f:
         f.write(await file.read())
 

backend/common/utils/utils.py

@@ -312,3 +312,22 @@ def equals_ignore_case(str1: str, *args: str) -> bool:
         if str1.casefold() == arg.casefold():
             return True
     return False
+
+
+def safe_upload_name(file_name: str | None) -> str | None:
+    """Reduce a client-supplied upload name to a single safe path component (issue #1448).
+
+    The multipart file name is attacker-controlled and the same request may be
+    served by a POSIX or a Windows host, so both separators count as path
+    separators: `..\\..\\evil.xlsx` and a full client path such as
+    `C:\\Users\\me\\orders.xlsx` must not keep any directory part. A name that
+    still addresses another location after that reduction (empty, `.`, `..`, a
+    drive-relative or alternate-stream `:`, an embedded NUL) is rejected with
+    `None` so the caller answers 400 instead of writing anywhere.
+    """
+    if not file_name:
+        return None
+    name = file_name.replace("\\", "/").rsplit("/", 1)[-1].strip()
+    if not name or name in {".", ".."} or ":" in name or "\x00" in name:
+        return None
+    return name

backend/tests/test_datasource_upload_path_traversal.py

@@ -0,0 +1,124 @@
+"""Offline HTTP regressions for the Excel upload routes (CWE-22, issue #1448).
+
+These tests post real multipart bodies to `/datasource/parseExcel` and to the
+deprecated `/datasource/uploadExcel`, and reuse the offline harness from
+`test_datasource_import_path_traversal.py`: identity, audit logging, the database
+session and the Excel reader are test adapters, so no application startup,
+external database or deployed service is used.
+
+Before the fix the stored name kept whatever the client sent apart from POSIX
+directory parts (`..\\..\\evil.xlsx` -> `..\\evil_<hash>.xlsx`), which resolves
+outside the upload directory on a Windows host, and `uploadExcel` matched bare
+suffixes and indexed the name with `split('.')[1]`. The assertions below are the
+platform-independent view of that contract: whatever the client sends, the stored
+name is a single component inside the upload directory, or the request is refused.
+"""
+
+from unittest.mock import Mock
+
+import pytest
+from test_datasource_import_path_traversal import (
+    api,  # noqa: F401  # shared offline harness, resolved by fixture name
+)
+
+from apps.datasource.api import datasource
+
+PARSE_URL = "/datasource/parseExcel"
+UPLOAD_URL = "/datasource/uploadExcel"
+XLSX_TYPE = "application/vnd.openxmlformats-officedocument.spreadsheetml.sheet"
+INVALID_FILE_PATH = "i18n_ds_invalid_file_path"
+
+# Names whose directory part must be dropped, not written.
+STRIPPED_NAMES = [
+    "..\\..\\evil.xlsx",
+    "..\\evil.xlsx",
+    "nested\\..\\..\\evil.xlsx",
+    "../../evil.xlsx",
+    "/tmp/orders.xlsx",
+    "C:\\Users\\me\\orders.xlsx",
+]
+
+# Names that would still address another location after dropping the separators.
+REJECTED_NAMES = ["C:evil.xlsx", "C:.xlsx", "orders.xlsx:extra"]
+
+
+def stored_files(root):
+    return {path for path in root.rglob("*") if path.is_file()}
+
+
+def upload(client, url, file_name):
+    return client.post(url, files={"file": (file_name, b"a\n1\n", XLSX_TYPE)})
+
+
+@pytest.fixture
+def harness(request):
+    """The offline API harness defined by the import-traversal guard."""
+    return request.getfixturevalue("api")
+
+
+@pytest.fixture
+def parse_harness(harness):
+    # The preview reader is not part of this contract and would need a real workbook.
+    harness.monkeypatch.setattr(
+        datasource, "parse_excel_preview", Mock(return_value=[])
+    )
+    return harness
+
+
+@pytest.mark.parametrize("file_name", STRIPPED_NAMES)
+def test_parse_excel_stores_one_component_inside_the_upload_dir(
+    parse_harness, file_name
+):
+    before = stored_files(parse_harness.tmp_path)
+    response = upload(parse_harness.client, PARSE_URL, file_name)
+    assert response.status_code == 200, response.text
+    stored = response.json()["filePath"]
+    assert "/" not in stored and "\\" not in stored and ":" not in stored, stored
+    assert (parse_harness.upload_dir / stored).is_file()
+    created = stored_files(parse_harness.tmp_path) - before
+    assert created, f"nothing was stored for {file_name!r}"
+    assert all(parse_harness.upload_dir in path.parents for path in created), created
+
+
+@pytest.mark.parametrize("file_name", STRIPPED_NAMES)
+def test_upload_excel_stores_one_component_inside_the_upload_dir(harness, file_name):
+    before = stored_files(harness.tmp_path)
+    response = upload(harness.client, UPLOAD_URL, file_name)
+    # The deprecated route continues into the import engine, which the harness
+    # stubs out; only the stored name is under test here.
+    assert response.status_code in (200, 500), response.text
+    created = stored_files(harness.tmp_path) - before
+    assert created, f"nothing was stored for {file_name!r}"
+    assert all(harness.upload_dir in path.parents for path in created), created
+    assert all(path.name == path.name.strip("/\\") for path in created)
+    assert all(":" not in path.name for path in created)
+
+
+@pytest.mark.parametrize("url", [PARSE_URL, UPLOAD_URL])
+@pytest.mark.parametrize("file_name", REJECTED_NAMES)
+def test_names_that_keep_a_drive_relative_location_are_refused(harness, url, file_name):
+    before = stored_files(harness.tmp_path)
+    response = upload(harness.client, url, file_name)
+    assert response.status_code == 400, response.text
+    assert response.json() == INVALID_FILE_PATH
+    assert stored_files(harness.tmp_path) == before
+
+
+@pytest.mark.parametrize("file_name", ["evil.txlsx", "evilxlsx"])
+def test_upload_excel_refuses_names_that_only_end_with_the_bare_suffix(
+    harness, file_name
+):
+    # The whitelist used to be {"xlsx", "xls", "csv"}: "evil.txlsx" was accepted
+    # and the dot-less "evilxlsx" made `split('.')[1]` raise IndexError.
+    before = stored_files(harness.tmp_path)
+    response = upload(harness.client, UPLOAD_URL, file_name)
+    assert response.status_code == 400, response.text
+    assert stored_files(harness.tmp_path) == before
+
+
+def test_normal_names_keep_their_own_stem_and_extension(parse_harness):
+    response = upload(parse_harness.client, PARSE_URL, "a.b.xlsx")
+    assert response.status_code == 200, response.text
+    stored = response.json()["filePath"]
+    assert stored.startswith("a.b_") and stored.endswith(".xlsx"), stored
+    assert (parse_harness.upload_dir / stored).is_file()

backend/tests/test_upload_file_name.py

@@ -0,0 +1,84 @@
+"""Regression coverage for client-supplied upload file names (issue #1448).
+
+`/datasource/parseExcel` and the deprecated `/datasource/uploadExcel` derive the
+stored file name from the multipart `file.filename`. Splitting that name on "/"
+only keeps Windows separators and a full client path intact, so on a Windows host
+the write could leave `settings.EXCEL_PATH`.
+
+`safe_upload_name` is now the single place that reduces the client name to one path
+component, so these tests pin its contract for both separator styles and for the
+names that must be rejected outright instead of being stored somewhere else.
+"""
+
+import os
+
+import pytest
+
+from common.utils.utils import safe_upload_name
+
+
+@pytest.mark.parametrize(
+    "file_name, expected",
+    [
+        ("orders.xlsx", "orders.xlsx"),
+        ("quarter-1.xls", "quarter-1.xls"),
+        ("2026-报表.csv", "2026-报表.csv"),
+        ("a.b.c.csv", "a.b.c.csv"),
+        ("  spaced .xlsx  ", "spaced .xlsx"),
+        ("/tmp/orders.xlsx", "orders.xlsx"),
+        ("./nested/orders.xlsx", "orders.xlsx"),
+        ("../../orders.xlsx", "orders.xlsx"),
+        ("nested/../../orders.xlsx", "orders.xlsx"),
+        ("..\\..\\evil.xlsx", "evil.xlsx"),
+        ("nested\\..\\..\\evil.xlsx", "evil.xlsx"),
+        ("C:\\Users\\me\\orders.xlsx", "orders.xlsx"),
+        ("\\\\server\\share\\orders.xlsx", "orders.xlsx"),
+    ],
+)
+def test_only_the_last_path_component_survives(file_name, expected):
+    assert safe_upload_name(file_name) == expected
+
+
+@pytest.mark.parametrize(
+    "file_name",
+    [
+        None,
+        "",
+        "   ",
+        ".",
+        "..",
+        "../",
+        "..\\",
+        "C:orders.xlsx",
+        "C:.xlsx",
+        "orders.xlsx:extra",
+        "orders.xlsx\x00",
+    ],
+)
+def test_names_that_still_address_another_location_are_rejected(file_name):
+    assert safe_upload_name(file_name) is None
+
+
+@pytest.mark.parametrize(
+    "file_name", ["dir/orders.xlsx", "dir\\orders.xlsx", "dir/dir\\orders.xlsx"]
+)
+def test_both_separator_styles_are_stripped(file_name):
+    # The same request body can be served by a POSIX or a Windows host, so
+    # neither "/" nor "\" may survive into the stored name.
+    assert safe_upload_name(file_name) == "orders.xlsx"
+
+
+@pytest.mark.parametrize(
+    "file_name",
+    [
+        "..\\..\\evil.xlsx",
+        "C:\\Users\\me\\orders.xlsx",
+        "/tmp/orders.xlsx",
+        "orders.xlsx",
+    ],
+)
+def test_accepted_names_never_carry_a_separator_or_drive(file_name):
+    name = safe_upload_name(file_name)
+    assert name
+    assert "/" not in name and "\\" not in name and ":" not in name
+    assert os.path.basename(name) == name