From ad5cb7f0f099f4b86dca3f9a8ef617f362929f53 Mon Sep 17 00:00:00 2001 From: Andrey Epifancev Date: Sun, 4 Oct 2026 02:53:48 +0400 Subject: [PATCH] =?UTF-8?q?=D0=A1=D1=80=D0=B5=D0=B7=200=20[4.2]=20T7.4:=20?= =?UTF-8?q?=D0=BE=D0=B1=D1=89=D0=B8=D0=B5=20=D0=BF=D1=80=D0=BE=D0=B2=D0=B5?= =?UTF-8?q?=D1=80=D0=BA=D0=B8=20=D0=B8=D0=BC=D0=B5=D0=BD=D0=B8=20=D0=B8=20?= =?UTF-8?q?job=5Fid=20=D0=B2=20hotfolder.py,=20=D0=B7=D0=B0=D0=BF=D1=80?= =?UTF-8?q?=D0=B5=D1=82=20=D1=81=D0=B8=D0=BC=D0=B2=D0=BE=D0=BB=D0=BE=D0=B2?= =?UTF-8?q?=20Windows/SMB?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit validate_name/validate_job_id вынесены из LocalHotFolder для повторного использования в SmbHotFolder. Дополнительно отвергаются `: < > " | ? *` и управляющие символы 0x00-0x1F (в put, answer_path и inbox_name); пробелы и кириллица в именах допустимы. Co-Authored-By: Claude Fable 5.1 --- api/src/gateway/hotfolder.py | 32 +++++++++++++++++++++----- api/tests/unit/test_local_hotfolder.py | 32 +++++++++++++++++++++++++- 2 files changed, 57 insertions(+), 7 deletions(-) diff --git a/api/src/gateway/hotfolder.py b/api/src/gateway/hotfolder.py index 685c173..1122051 100644 --- a/api/src/gateway/hotfolder.py +++ b/api/src/gateway/hotfolder.py @@ -9,6 +9,27 @@ from pathlib import Path from typing import Protocol _PATH_SEPARATORS = ("/", "\\") +# Недопустимые в именах файлов Windows/SMB символы (кроме разделителей пути) и управляющие 0x00-0x1F (с NUL) +_ILLEGAL_CHARS = frozenset(':<>"|?*') | frozenset(chr(code) for code in range(0x20)) +_FORBIDDEN_CHARS = _ILLEGAL_CHARS | frozenset(_PATH_SEPARATORS) + + +def validate_name(name: str) -> None: + """Проверка имени входящего файла для `put` (общая для `LocalHotFolder` и `SmbHotFolder`). + + Имя — один компонент пути: не пустое, не начинается с `.` (dot-файлы — служебные `.part`, их игнорирует эмулятор), + без разделителей `/` и `\\`, без символов, недопустимых в Windows/SMB (`: < > " | ? *`), и без управляющих + символов. Пробелы и кириллица допустимы. Нарушение — `ValueError`. + """ + if name == "" or name.startswith(".") or any(char in _FORBIDDEN_CHARS for char in name): + raise ValueError(f"недопустимое имя файла: {name!r}") + + +def validate_job_id(job_id: str) -> None: + """Проверка `job_id` перед построением пути ответа: защита от обхода каталога. UUID проходит; пусто, `.`, `..`, + разделители, недопустимые в Windows/SMB и управляющие символы — `ValueError`.""" + if job_id in ("", ".", "..") or any(char in _FORBIDDEN_CHARS for char in job_id): + raise ValueError(f"недопустимый job_id: {job_id!r}") def _split_basename(name: str) -> str: @@ -20,10 +41,11 @@ def inbox_name(job_id: str, original_name: str) -> str: """Имя входящего файла `_`. `original_name` приходит из недоверенного запроса: берётся только последний компонент пути (без разделителей - и `..`). Если после этого имени не осталось (пусто, `.`, `..`, NUL-байт) — `ValueError`. + и `..`). Если после этого имени не осталось (пусто, `.`, `..`) или в нём есть символы, недопустимые в имени файла + Windows/SMB, либо управляющие (см. `validate_name`) — `ValueError`. Пробелы и кириллица сохраняются как есть. """ base = _split_basename(original_name) - if base in ("", ".", "..") or "\x00" in base: + if base in ("", ".", "..") or any(char in _ILLEGAL_CHARS for char in base): raise ValueError(f"недопустимое имя файла: {original_name!r}") return f"{job_id}_{base}" @@ -62,8 +84,7 @@ class LocalHotFolder: def answer_path(self, job_id: str) -> Path: """Путь ответа ИС: `//.json`.""" - if job_id in ("", ".", "..") or any(sep in job_id for sep in _PATH_SEPARATORS) or "\x00" in job_id: - raise ValueError(f"недопустимый job_id: {job_id!r}") + validate_job_id(job_id) return self.output_dir / job_id / f"{job_id}.json" def put(self, name: str, data: bytes) -> None: @@ -72,8 +93,7 @@ class LocalHotFolder: Корень не создаётся (нет корня — `FileNotFoundError`), входящий каталог создаётся без родителей. При сбое временный файл удаляется. """ - if name == "" or name.startswith(".") or "\x00" in name or any(sep in name for sep in _PATH_SEPARATORS): - raise ValueError(f"недопустимое имя файла: {name!r}") + validate_name(name) self.input_dir.mkdir(exist_ok=True) part = self.input_dir / f".{name}.part" try: diff --git a/api/tests/unit/test_local_hotfolder.py b/api/tests/unit/test_local_hotfolder.py index ed3e258..1bc5b18 100644 --- a/api/tests/unit/test_local_hotfolder.py +++ b/api/tests/unit/test_local_hotfolder.py @@ -45,6 +45,17 @@ class TestInboxName: with pytest.raises(ValueError, match="имя файла"): inbox_name("job", original) + @pytest.mark.parametrize( + "original", ["a:b.pdf", "отчёт?.pdf", "a*b.pdf", 'a"b.pdf', "a|b.pdf", "a.pdf", "a\x01b.pdf"] + ) + def test_rejects_names_illegal_on_windows(self, original: str) -> None: + with pytest.raises(ValueError, match="имя файла"): + inbox_name("job", original) + + def test_keeps_spaces_and_cyrillic(self) -> None: + job_id = _job_id() + assert inbox_name(job_id, "Отчёт за 2024 год (копия).PDF") == f"{job_id}_Отчёт за 2024 год (копия).PDF" + def test_result_is_accepted_by_put(self, tmp_path: Path) -> None: hf = LocalHotFolder(tmp_path) name = inbox_name(_job_id(), "../../scan.pdf") @@ -68,7 +79,7 @@ class TestPaths: assert hf.output_dir == tmp_path / "Исходящие" assert hf.answer_path("abc") == tmp_path / "Исходящие" / "abc" / "abc.json" - @pytest.mark.parametrize("job_id", ["", ".", "..", "../x", "a/b", "a\\b"]) + @pytest.mark.parametrize("job_id", ["", ".", "..", "../x", "a/b", "a\\b", "a:b", "a?b", "a*b", "a\x00b", "a\nb"]) def test_answer_path_rejects_unsafe_job_id(self, tmp_path: Path, job_id: str) -> None: with pytest.raises(ValueError, match="job_id"): LocalHotFolder(tmp_path).answer_path(job_id) @@ -134,6 +145,25 @@ class TestPut: assert not hf.input_dir.exists() assert not (tmp_path.parent / "escape.pdf").exists() + @pytest.mark.parametrize("name", ["a:b.pdf", "ab.pdf", 'a"b.pdf', "a|b.pdf", "a?b.pdf", "a*b.pdf"]) + def test_rejects_names_illegal_on_windows(self, tmp_path: Path, name: str) -> None: + hf = LocalHotFolder(tmp_path) + with pytest.raises(ValueError, match="имя файла"): + hf.put(name, b"x") + assert not hf.input_dir.exists() + + @pytest.mark.parametrize("name", ["a\x01b.pdf", "a\tb.pdf", "a\nb.pdf", "a\x1fb.pdf"]) + def test_rejects_control_chars(self, tmp_path: Path, name: str) -> None: + with pytest.raises(ValueError, match="имя файла"): + LocalHotFolder(tmp_path).put(name, b"x") + + def test_accepts_spaces_cyrillic_and_inbox_names(self, tmp_path: Path) -> None: + hf = LocalHotFolder(tmp_path) + names = [inbox_name(_job_id(), "Отчёт за 2024 год (копия).PDF"), inbox_name(_job_id(), "aep_8.pdf"), "a b.pdf"] + for name in names: + hf.put(name, b"x") + assert sorted(p.name for p in hf.input_dir.iterdir()) == sorted(names) + def test_cleans_part_file_on_failure(self, tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None: hf = LocalHotFolder(tmp_path)