diff --git a/api/src/domain/directory_import.py b/api/src/domain/directory_import.py
index 5141fde..58ae3c9 100644
--- a/api/src/domain/directory_import.py
+++ b/api/src/domain/directory_import.py
@@ -4,6 +4,11 @@
снимку справочника (`DirectorySnapshot`, строит сервис) решает судьбу каждой строки и возвращает операции (только
данные — применяет сервис) и отчёт с кодами §13.2. Идентификаторы новых объектов генерируются здесь (`uuid4`), чтобы
операции плана ссылались друг на друга до записи в БД.
+
+Технические пределы чтения (защита от повреждённых и «раздутых» файлов, не правила предметной области):
+`MAX_IMPORT_ROWS` — число строк листа/записей csv, включая пустые (реальная выгрузка ЕЦП — 8400 строк);
+`MAX_UNCOMPRESSED_BYTES` — суммарный распакованный размер частей xlsx (zip), проверяется до разбора. Превышение —
+`UNSUPPORTED_FILE`.
"""
import csv
@@ -12,7 +17,8 @@ import io
import re
import uuid
import zipfile
-from collections.abc import Iterable, Sequence
+import zlib
+from collections.abc import Iterable, Iterator, Sequence
from dataclasses import asdict, dataclass, field
from typing import Any
@@ -73,6 +79,22 @@ FLAG_GSZ = "входитвгсз"
MISSING_COLUMNS = "MISSING_COLUMNS"
UNSUPPORTED_FILE = "UNSUPPORTED_FILE"
+MAX_IMPORT_ROWS = 100_000
+MAX_UNCOMPRESSED_BYTES = 200 * 1024 * 1024
+# Сбои разбора повреждённого xlsx (zip, XML, ссылки ячеек, книга без листов); `SyntaxError` покрывает
+# `xml.etree.ElementTree.ParseError`. openpyxl в read-only читает лист лениво — ошибки возникают и при итерации строк
+_XLSX_ERRORS: tuple[type[BaseException], ...] = (
+ zipfile.BadZipFile,
+ InvalidFileException,
+ KeyError,
+ ValueError,
+ IndexError,
+ OSError,
+ EOFError,
+ SyntaxError,
+ zlib.error,
+)
+
_WHITESPACE_RE = re.compile(r"\s+")
@@ -134,35 +156,50 @@ def read_rows(filename: str, content: bytes) -> list[ImportRow]:
return _rows_from_table(raw_rows)
-def _xlsx_rows(content: bytes) -> list[tuple[int, list[str]]]:
+def _xlsx_rows(content: bytes) -> Iterator[tuple[int, list[str]]]:
+ """Строки первого листа лениво (пустые строки openpyxl отдаёт пустыми кортежами, номер = позиция в листе)."""
+ _check_uncompressed_size(content)
try:
workbook = load_workbook(io.BytesIO(content), read_only=True, data_only=True)
- except (zipfile.BadZipFile, InvalidFileException, KeyError, ValueError, OSError, EOFError) as error:
+ except _XLSX_ERRORS as error:
raise FileFormatError(UNSUPPORTED_FILE, "файл xlsx не читается") from error
try:
sheet = workbook.worksheets[0]
# Размер листа из файла бывает неверным (экспорт сторонних систем) — читаем все строки фактически
sheet.reset_dimensions()
- return [
- (number, [normalize_cell(value) for value in values])
- for number, values in enumerate(sheet.iter_rows(values_only=True), start=1)
- ]
+ for number, values in enumerate(sheet.iter_rows(values_only=True), start=1):
+ yield number, [normalize_cell(value) for value in values]
+ except _XLSX_ERRORS as error:
+ raise FileFormatError(UNSUPPORTED_FILE, "файл xlsx не читается") from error
finally:
workbook.close()
-def _csv_rows(content: bytes) -> list[tuple[int, list[str]]]:
+def _check_uncompressed_size(content: bytes) -> None:
+ try:
+ with zipfile.ZipFile(io.BytesIO(content)) as archive:
+ total = sum(info.file_size for info in archive.infolist())
+ except _XLSX_ERRORS as error:
+ raise FileFormatError(UNSUPPORTED_FILE, "файл xlsx не читается") from error
+ if total > MAX_UNCOMPRESSED_BYTES:
+ raise FileFormatError(UNSUPPORTED_FILE, "файл xlsx слишком велик после распаковки")
+
+
+def _csv_rows(content: bytes) -> Iterator[tuple[int, list[str]]]:
+ """Записи csv лениво; номер — физическая строка файла, с которой начинается запись (поля в кавычках бывают
+ многострочными)."""
try:
text = content.decode("utf-8-sig")
except UnicodeDecodeError as error:
raise FileFormatError(UNSUPPORTED_FILE, "кодировка csv не распознана: ожидается UTF-8") from error
header_line = next((line for line in text.splitlines() if line.strip()), "")
delimiter = ";" if header_line.count(";") > header_line.count(",") else ","
+ reader = csv.reader(io.StringIO(text, newline=""), delimiter=delimiter)
+ start = 1
try:
- return [
- (number, [normalize_cell(value) for value in values])
- for number, values in enumerate(csv.reader(io.StringIO(text, newline=""), delimiter=delimiter), start=1)
- ]
+ for values in reader:
+ yield start, [normalize_cell(value) for value in values]
+ start = reader.line_num + 1
except csv.Error as error:
raise FileFormatError(UNSUPPORTED_FILE, "файл csv не читается") from error
@@ -170,7 +207,10 @@ def _csv_rows(content: bytes) -> list[tuple[int, list[str]]]:
def _rows_from_table(raw_rows: Iterable[tuple[int, list[str]]]) -> list[ImportRow]:
columns: dict[str, int] | None = None
result: list[ImportRow] = []
- for number, cells in raw_rows:
+ for count, (number, cells) in enumerate(raw_rows, start=1):
+ # Считаются и пустые строки: лист с одной ячейкой в строке 2 000 000 отвергается, не дочитываясь
+ if count > MAX_IMPORT_ROWS:
+ raise FileFormatError(UNSUPPORTED_FILE, f"слишком много строк: больше {MAX_IMPORT_ROWS}")
if not any(cells):
continue
if columns is None:
diff --git a/api/tests/test_domain_directory_import.py b/api/tests/test_domain_directory_import.py
index 0494d57..87a86fb 100644
--- a/api/tests/test_domain_directory_import.py
+++ b/api/tests/test_domain_directory_import.py
@@ -2,13 +2,17 @@
import csv
import io
+import re
import uuid
+import zipfile
+from collections.abc import Callable
from pathlib import Path
from typing import TypeVar
import openpyxl
import pytest
+from src.domain import directory_import
from src.domain.directory_import import (
AddMember,
CreateEntity,
@@ -216,6 +220,114 @@ def test_read_xlsx_header_in_second_row_numeric_inn_and_trailing_empty_rows() ->
assert rows[0].gsk_flag == "входит в ГСК"
+def test_read_csv_row_is_file_line_where_record_starts() -> None:
+ """Поле в кавычках с переводом строки не сдвигает номера: запись после него — со своей строки файла."""
+ content = _csv_bytes(
+ [HEADERS, _data_row("3528052718")[:4] + ["АО\nмногострочное"] + _data_row()[5:], _data_row("3528104476")]
+ )
+ rows = read_rows("export.csv", content)
+ assert [(r.row, r.inn) for r in rows] == [(2, "3528052718"), (4, "3528104476")]
+ assert rows[0].entity_name == "АО\nмногострочное"
+
+
+# ---------------------------------------------------------------- повреждённые и «раздутые» xlsx (T9.1, T9.2)
+
+
+def _valid_xlsx() -> bytes:
+ workbook = openpyxl.Workbook()
+ sheet = workbook.active
+ assert sheet is not None
+ sheet.append(HEADERS)
+ sheet.append(_data_row())
+ buffer = io.BytesIO()
+ workbook.save(buffer)
+ return buffer.getvalue()
+
+
+def _rewrite_xlsx(content: bytes, member: str, change: Callable[[bytes], bytes]) -> bytes:
+ """Тот же zip (валидный архив), в котором часть `member` заменена результатом `change`."""
+ source = zipfile.ZipFile(io.BytesIO(content))
+ buffer = io.BytesIO()
+ with zipfile.ZipFile(buffer, "w", zipfile.ZIP_DEFLATED) as target:
+ for info in source.infolist():
+ data = source.read(info.filename)
+ target.writestr(info.filename, change(data) if info.filename == member else data)
+ return buffer.getvalue()
+
+
+def _bad_cell_ref(data: bytes) -> bytes:
+ return data.replace(b'r="A2"', b'r="AAAAA2"')
+
+
+def _no_sheets(data: bytes) -> bytes:
+ return re.sub(rb".*", b"", data, flags=re.DOTALL)
+
+
+@pytest.mark.parametrize(
+ ("member", "change"),
+ [
+ pytest.param("xl/worksheets/sheet1.xml", lambda data: data[: len(data) // 2], id="лист-обрезан"),
+ pytest.param("xl/worksheets/sheet1.xml", lambda data: b"\x00garbage<<", id="лист-мусор"),
+ pytest.param("xl/workbook.xml", lambda data: b"garbage", id="книга-мусор"),
+ pytest.param("xl/styles.xml", lambda data: b" None:
+ content = _rewrite_xlsx(_valid_xlsx(), member, change)
+ with pytest.raises(FileFormatError) as error:
+ read_rows("export.xlsx", content)
+ assert error.value.code == "UNSUPPORTED_FILE"
+ assert error.value.message == "файл xlsx не читается"
+
+
+def test_rewritten_valid_xlsx_is_still_read() -> None:
+ """Контроль помощника: перепаковка без изменений даёт читаемый файл."""
+ content = _rewrite_xlsx(_valid_xlsx(), "xl/worksheets/sheet1.xml", lambda data: data)
+ assert [r.inn for r in read_rows("export.xlsx", content)] == ["3528052718"]
+
+
+def test_read_xlsx_with_far_row_index_is_rejected_by_row_limit() -> None:
+ """Одна строка данных с r="2000000": пустые строки до неё считаются, чтение останавливается на пределе."""
+
+ def far_row(data: bytes) -> bytes:
+ data = re.sub(rb'r="([A-Z]+)2"', rb'r="\g<1>2000000"', data)
+ return data.replace(b' None:
+ monkeypatch.setattr(directory_import, "MAX_IMPORT_ROWS", 3)
+ within = _csv_bytes([HEADERS, ["", "", "", "", "", "", "", ""], _data_row()])
+ assert len(read_rows("export.csv", within)) == 1
+ beyond = _csv_bytes([HEADERS, ["", "", "", "", "", "", "", ""], ["", "", "", "", "", "", "", ""], _data_row()])
+ with pytest.raises(FileFormatError) as error:
+ read_rows("export.csv", beyond)
+ assert error.value.code == "UNSUPPORTED_FILE"
+
+
+def test_xlsx_uncompressed_size_bomb_is_rejected(monkeypatch: pytest.MonkeyPatch) -> None:
+ """Сумма распакованных размеров частей zip выше предела → отказ до разбора (предел уменьшен для скорости)."""
+ limit = 1024 * 1024
+ monkeypatch.setattr(directory_import, "MAX_UNCOMPRESSED_BYTES", limit)
+ buffer = io.BytesIO(_valid_xlsx())
+ with zipfile.ZipFile(buffer, "a", zipfile.ZIP_DEFLATED) as archive:
+ archive.writestr("xl/padding.bin", b"\x00" * (limit + 1))
+ content = buffer.getvalue()
+ assert len(content) < limit // 10
+ with pytest.raises(FileFormatError) as error:
+ read_rows("export.xlsx", content)
+ assert error.value.code == "UNSUPPORTED_FILE"
+ assert error.value.message == "файл xlsx слишком велик после распаковки"
+
+
# ---------------------------------------------------------------- план импорта: помощники