Срез 0 [4.2] T7.4: общие проверки имени и job_id в hotfolder.py, запрет символов Windows/SMB
validate_name/validate_job_id вынесены из LocalHotFolder для повторного использования в SmbHotFolder. Дополнительно отвергаются `: < > " | ? *` и управляющие символы 0x00-0x1F (в put, answer_path и inbox_name); пробелы и кириллица в именах допустимы. Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>
This commit is contained in:
parent
6d1f42fe1b
commit
ad5cb7f0f0
@ -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:
|
||||
"""Имя входящего файла `<job_id>_<original_name>`.
|
||||
|
||||
`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:
|
||||
"""Путь ответа ИС: `<output_dir>/<job_id>/<job_id>.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:
|
||||
|
||||
@ -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<b>.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", "a<b.pdf", "a>b.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)
|
||||
|
||||
|
||||
Loading…
x
Reference in New Issue
Block a user