diff --git a/backend/apps/datasource/api/datasource.py b/backend/apps/datasource/api/datasource.py index 4949b3730..db03b2fae 100644 --- a/backend/apps/datasource/api/datasource.py +++ b/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()) diff --git a/backend/common/utils/utils.py b/backend/common/utils/utils.py index d2836f438..3b66c3ff3 100644 --- a/backend/common/utils/utils.py +++ b/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 diff --git a/backend/tests/test_datasource_upload_path_traversal.py b/backend/tests/test_datasource_upload_path_traversal.py new file mode 100644 index 000000000..3505bc782 --- /dev/null +++ b/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_.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() diff --git a/backend/tests/test_upload_file_name.py b/backend/tests/test_upload_file_name.py new file mode 100644 index 000000000..02130a91a --- /dev/null +++ b/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