Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
36 changes: 25 additions & 11 deletions backend/apps/settings/api/base.py
Original file line number Diff line number Diff line change
@@ -1,11 +1,13 @@
import os
import unicodedata

from fastapi import APIRouter, HTTPException
from fastapi.responses import FileResponse

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")

Expand All @@ -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",
)
154 changes: 154 additions & 0 deletions backend/tests/test_download_fail_info_path.py
Original file line number Diff line number Diff line change
@@ -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
Loading