diff --git a/backend/apps/settings/api/base.py b/backend/apps/settings/api/base.py index c9c59548e..2d790a2bc 100644 --- a/backend/apps/settings/api/base.py +++ b/backend/apps/settings/api/base.py @@ -1,4 +1,5 @@ import os +import unicodedata from fastapi import APIRouter, HTTPException from fastapi.responses import FileResponse @@ -6,6 +7,7 @@ from apps.swagger.i18n import PLACEHOLDER_PREFIX from common.core.config import settings from common.core.file import FileRequest +from common.utils.utils import safe_upload_name router = APIRouter(tags=["System"], prefix="/system") @@ -17,23 +19,35 @@ async def download_excel(req: FileRequest): """ 根据文件路径下载 Excel 文件 """ - filename = req.file - file_path = os.path.join(path, filename) + # 该文件名由导出接口生成，是单个文件名：带目录分量、盘符或控制字符的取值 + # 一律拒绝，不做静默改写；校验先于文件系统访问，避免用拒绝结果反推任意路径 + # 是否存在。 + # 必须在 strip 或文件名清理前拒绝控制字符，避免将非法输入改写为合法文件名。 + if any(unicodedata.category(char) == "Cc" for char in req.file): + raise HTTPException(400, "Only support _error.xlsx") + raw_name = (req.file or "").strip() + file_name = safe_upload_name(req.file) + if not file_name or file_name != raw_name or not file_name.endswith("_error.xlsx"): + raise HTTPException(400, "Only support _error.xlsx") - # 检查文件是否存在 - if not os.path.exists(file_path): - raise HTTPException(404, "File Not Exists") + base_dir = os.path.realpath(path) + try: + file_path = os.path.realpath(os.path.join(base_dir, file_name)) + inside = os.path.commonpath([file_path, base_dir]) == base_dir + except ValueError: # 含 NUL、跨盘符或非法路径 + inside = False + file_path = "" - # 检查文件是否是 Excel 文件 - if not filename.endswith('_error.xlsx'): - raise HTTPException(400, "Only support _error.xlsx") + # 检查文件是否位于上传目录内且确实是文件（符号链接由 realpath 一并解析） + if not inside or file_path == base_dir or not os.path.isfile(file_path): + raise HTTPException(404, "File Not Exists") # 获取文件名 - filename = os.path.basename(file_path) + file_name = os.path.basename(file_path) # 返回文件 return FileResponse( path=file_path, - filename=filename, - media_type='application/vnd.openxmlformats-officedocument.spreadsheetml.sheet' + filename=file_name, + media_type="application/vnd.openxmlformats-officedocument.spreadsheetml.sheet", ) diff --git a/backend/tests/test_download_fail_info_path.py b/backend/tests/test_download_fail_info_path.py new file mode 100644 index 000000000..07889f322 --- /dev/null +++ b/backend/tests/test_download_fail_info_path.py @@ -0,0 +1,154 @@ +"""Regression coverage for the fail-info download route (path handling). + +`/system/download-fail-info` joined the client-supplied `file` field onto +`settings.EXCEL_PATH` and only afterwards checked the suffix, so the route could +read a file outside the upload directory when the name ended with +`_error.xlsx`, and the order of the checks disclosed whether an arbitrary path +exists (`404` when missing, `400` when it exists with the wrong suffix). + +The route is exercised over HTTP here, with its router mounted on a bare app and +the upload directory redirected to a temporary path; no database, permission +layer or xpack module is involved. +""" + +import os +from types import SimpleNamespace +from unittest.mock import Mock + +import pytest +from fastapi import FastAPI +from fastapi.testclient import TestClient + +from apps.settings.api import base as settings_api + +URL = "/system/download-fail-info" +REPORT = b"error report body\n" + + +@pytest.fixture +def api(monkeypatch, tmp_path): + upload_dir = tmp_path / "excel" + upload_dir.mkdir() + outside = tmp_path / "outside" + outside.mkdir() + secret = outside / "secret_error.xlsx" + secret.write_bytes(REPORT) + # Sibling directory whose name starts with the upload directory name. + sibling = tmp_path / "excel_backup" + sibling.mkdir() + (sibling / "secret_error.xlsx").write_bytes(REPORT) + + (upload_dir / "orders_error.xlsx").write_bytes(REPORT) + (upload_dir / "orders.xlsx").write_bytes(REPORT) # exists, wrong suffix + (upload_dir / "nested").mkdir() + (upload_dir / "nested" / "orders_error.xlsx").write_bytes(REPORT) + (upload_dir / "sub_error.xlsx").mkdir() # a directory, not a file + + app = FastAPI() + app.include_router(settings_api.router) + monkeypatch.setattr(settings_api, "path", str(upload_dir)) + with TestClient(app) as client: + yield SimpleNamespace( + client=client, + upload_dir=upload_dir, + outside=outside, + secret=secret, + tmp_path=tmp_path, + monkeypatch=monkeypatch, + ) + + +def request(api, file_name): + return api.client.post(URL, json={"file": file_name}) + + +def test_report_in_the_upload_directory_is_served(api): + response = request(api, "orders_error.xlsx") + assert response.status_code == 200, response.text + assert response.content == REPORT + assert "orders_error.xlsx" in response.headers.get("content-disposition", "") + + +@pytest.mark.parametrize( + "control", ["\x00", "\x01", "\t", "\n", "\r", "\x1f", "\x7f", "\x85", "\x9f"] +) +@pytest.mark.parametrize( + "pattern", + [ + "{control}{name}_error.xlsx", + "{name}{control}_error.xlsx", + "{name}_error.xlsx{control}", + ], +) +def test_control_characters_are_rejected_before_filesystem_access( + api, control, pattern +): + # Patch only the route's os reference, preserving TestClient's filesystem use. + path = Mock(wraps=os.path) + api.monkeypatch.setattr(settings_api, "os", SimpleNamespace(path=path)) + for name in ("orders", "missing"): + response = request(api, pattern.format(control=control, name=name)) + assert response.status_code == 400, response.text + assert response.json() == {"detail": "Only support _error.xlsx"} + path.realpath.assert_not_called() + path.isfile.assert_not_called() + + +def test_name_with_a_directory_part_is_not_resolved_as_a_path(api): + # The export endpoints answer with a bare file name, so a name that carries a + # directory must not be used to walk into a subdirectory of the upload root. + response = request(api, "nested/orders_error.xlsx") + assert response.status_code in (400, 404), response.text + assert REPORT not in response.content + + +@pytest.mark.parametrize( + "file_name", + [ + "../outside/secret_error.xlsx", + "../../outside/secret_error.xlsx", + "./../outside/secret_error.xlsx", + "/etc/secret_error.xlsx", + "..\\..\\outside\\secret_error.xlsx", + "C:secret_error.xlsx", + "C:\\outside\\secret_error.xlsx", + ], +) +def test_names_that_leave_the_upload_directory_are_refused(api, file_name): + response = request(api, file_name) + assert response.status_code in (400, 404), response.text + assert REPORT not in response.content + assert api.secret.read_bytes() == REPORT # untouched + + +def test_a_sibling_directory_is_not_reachable(api): + response = request(api, "../excel_backup/secret_error.xlsx") + assert response.status_code in (400, 404), response.text + assert REPORT not in response.content + + +def test_existing_and_missing_names_are_indistinguishable(api): + # "orders.xlsx" exists inside the upload directory but carries the wrong + # suffix; "missing.xlsx" does not exist at all. Both must be refused the same + # way, so the route cannot be used to probe arbitrary paths. + existing = request(api, "orders.xlsx") + missing = request(api, "missing.xlsx") + assert existing.status_code == 400, existing.text + assert missing.status_code == 400, missing.text + assert existing.json() == missing.json() + + +def test_a_directory_is_never_downloaded(api): + response = request(api, "sub_error.xlsx") + assert response.status_code == 404, response.text + + +def test_a_symlink_that_leaves_the_upload_directory_is_refused(api): + link = api.upload_dir / "link_error.xlsx" + try: + os.symlink(api.secret, link) + except (OSError, NotImplementedError): + pytest.skip("symlinks are not available in this environment") + response = request(api, "link_error.xlsx") + assert response.status_code == 404, response.text + assert REPORT not in response.content