"""Лимиты фазы чтения (ruling R-23, находка I-1 финального ревью): распакованный размер zip-архива и суммарная площадь объединённых ячеек проверяются до библиотеки чтения — бомба из нескольких килобайт не доходит до openpyxl. Книги собираются в памяти: в fixtures ничего не пишется. """ import io import time import zipfile from collections.abc import Callable from typing import Any import openpyxl import pytest from src.contract import ErrorCode, ResultStatus from src.parser.errors import UnsupportedFormatError from src.parser.grid import MergedRange, Sheet from src.parser.osv import OsvParser from src.parser.readers import _READERS, read_book from src.parser.readers._limits import ( MERGED_AREA_LIMIT, ZIP_UNCOMPRESSED_LIMIT_FACTOR, check_merged_area, merged_area_of_xml, ) from src.parser.rules import Rules, load_rules FAST_SECONDS = 1.0 SHEET_PART = "xl/worksheets/sheet1.xml" ZEROS_BYTES = 10 * 1024 * 1024 # «десятки МБ» нулей сжимаются в ~10 КБ BOMB_MAX_FILE_BYTES = 10_000 # бомба — несколько килобайт файла SMALL_MAX_BYTES = 100_000 # лимит размера файла в тесте: распакованный лимит = 10 × 100 000 байт @pytest.fixture(scope="module") def rules() -> Rules: return load_rules() def _plain_xlsx() -> bytes: book = openpyxl.Workbook() sheet = book.active assert sheet is not None sheet.title = "ОСВ" sheet["A1"] = "Оборотно-сальдовая ведомость по счету 60" buffer = io.BytesIO() book.save(buffer) return buffer.getvalue() def _rewrite(data: bytes, patch: Callable[[str, bytes], bytes], extra: dict[str, bytes] | None = None) -> bytes: """Копия zip-архива книги: содержимое частей через `patch(имя, байты)` и дополнительные части `extra`.""" source = zipfile.ZipFile(io.BytesIO(data)) out = io.BytesIO() with zipfile.ZipFile(out, "w", zipfile.ZIP_DEFLATED) as target: for info in source.infolist(): target.writestr(info.filename, patch(info.filename, source.read(info.filename))) for name, content in (extra or {}).items(): target.writestr(name, content) return out.getvalue() def _xlsx_with_merges(*refs: str, tag: str = "mergeCell") -> bytes: cells = "".join(f'<{tag} ref="{ref}"/>' for ref in refs).encode() block = b'' % len(refs) + cells + b"" def patch(name: str, content: bytes) -> bytes: return content.replace(b"", b"" + block) if name == SHEET_PART else content return _rewrite(_plain_xlsx(), patch) def _xlsx_with_zeros(size: int) -> bytes: return _rewrite(_plain_xlsx(), lambda _name, content: content, {"xl/media/zeros.bin": bytes(size)}) @pytest.fixture def openpyxl_forbidden(monkeypatch: pytest.MonkeyPatch) -> None: """openpyxl не должен вызываться: бомба отбрасывается до него (и до выделения памяти под ячейки).""" def forbidden(*_args: Any, **_kwargs: Any) -> None: raise AssertionError("openpyxl.load_workbook вызван: лимит должен сработать до чтения книги") monkeypatch.setattr(openpyxl, "load_workbook", forbidden) # --- Площадь объединённых ячеек (xlsx) --- @pytest.mark.usefixtures("openpyxl_forbidden") def test_huge_merged_range_is_rejected_before_openpyxl() -> None: data = _xlsx_with_merges("A1:Z200000") # 5,2 млн ячеек в нескольких килобайтах файла assert len(data) < BOMB_MAX_FILE_BYTES started = time.monotonic() with pytest.raises(UnsupportedFormatError) as caught: read_book(data, "XLSX") assert time.monotonic() - started < FAST_SECONDS assert caught.value.code is ErrorCode.UNSUPPORTED_FORMAT assert "лимит объединённых ячеек" in caught.value.message assert str(MERGED_AREA_LIMIT) in caught.value.message @pytest.mark.usefixtures("openpyxl_forbidden") def test_many_small_merged_ranges_are_summed() -> None: per_range = 1000 # A1:A1000 и т. д.: каждый диапазон по отдельности невелик count = MERGED_AREA_LIMIT // per_range + 1 refs = [f"B{start}:B{start + per_range - 1}" for start in range(1, count * per_range, per_range)] with pytest.raises(UnsupportedFormatError, match="лимит объединённых ячеек"): read_book(_xlsx_with_merges(*refs), "XLSX") @pytest.mark.usefixtures("openpyxl_forbidden") def test_prefixed_merge_element_is_counted() -> None: """Пространство имён с префиксом (``) — тот же элемент; openpyxl его прочитал бы.""" with pytest.raises(UnsupportedFormatError, match="лимит объединённых ячеек"): read_book(_xlsx_with_merges("A1:Z200000", tag="x:mergeCell"), "XLSX") @pytest.mark.usefixtures("openpyxl_forbidden") def test_huge_merged_range_gives_unsupported_format_from_parser(rules: Rules) -> None: outcome = OsvParser(rules, max_bytes=50 * 1024 * 1024, timeout_seconds=120).parse( _xlsx_with_merges("A1:Z200000"), "XLSX" ) assert outcome.status is ResultStatus.ERROR assert outcome.error is not None assert outcome.error.code is ErrorCode.UNSUPPORTED_FORMAT def test_normal_merges_are_read() -> None: sheets, _ = read_book(_xlsx_with_merges("A3:A4", "B3:C3"), "XLSX") assert set(sheets[0].merged) == {MergedRange(3, 1, 4, 1), MergedRange(3, 2, 3, 3)} @pytest.mark.parametrize("chunk_size", [1, 7, 64, 65_536]) def test_scan_is_independent_of_chunk_boundaries(chunk_size: int) -> None: xml = ( b'1' b'' b'' ) expected = 4 + 1 + 10 + 4 assert merged_area_of_xml(io.BytesIO(xml), limit=MERGED_AREA_LIMIT, chunk_size=chunk_size) == expected def test_scan_limit_is_inclusive() -> None: xml = b'' area = 100 assert merged_area_of_xml(io.BytesIO(xml), limit=area) == area with pytest.raises(UnsupportedFormatError, match="лимит объединённых ячеек"): merged_area_of_xml(io.BytesIO(xml), limit=area - 1) def test_scan_ignores_merge_cells_container_and_other_elements() -> None: xml = b'' assert merged_area_of_xml(io.BytesIO(xml), limit=MERGED_AREA_LIMIT, chunk_size=5) == 0 def test_overlong_merge_element_is_rejected() -> None: """Раздутый атрибут не прячет диапазон от сканера: слишком длинный элемент mergeCell — отказ, а не пропуск.""" xml = b'' with pytest.raises(UnsupportedFormatError, match="лимит объединённых ячеек"): merged_area_of_xml(io.BytesIO(xml), limit=MERGED_AREA_LIMIT, chunk_size=4096) # --- Площадь объединённых ячеек после чтения (xls/xlsb: индекс merged в `Sheet`) --- def _sheet_with(*spans: MergedRange) -> Sheet: return Sheet(name="Лист", cells={}, merged=spans, row_outline={}, n_rows=1, n_cols=1) def test_merged_area_of_read_sheets_is_limited() -> None: """xlrd хранит merged списком (8 байт на диапазон), но индекс `Sheet` раскладывает диапазоны по ячейкам: тысячи средних диапазонов в xls дали бы десятки миллионов записей. Та же сумма площадей — после чтения. """ half = MergedRange(1, 1, MERGED_AREA_LIMIT // 2, 1) check_merged_area([_sheet_with(half), _sheet_with(half)], limit=MERGED_AREA_LIMIT) with pytest.raises(UnsupportedFormatError, match="лимит объединённых ячеек"): check_merged_area([_sheet_with(half), _sheet_with(half, MergedRange(1, 2, 1, 2))], limit=MERGED_AREA_LIMIT) def test_xls_reader_result_is_checked(monkeypatch: pytest.MonkeyPatch) -> None: huge = _sheet_with(MergedRange(1, 1, MERGED_AREA_LIMIT + 1, 1)) monkeypatch.setitem(_READERS, "XLS", ("BIFF8", lambda _data: (huge,))) with pytest.raises(UnsupportedFormatError, match="лимит объединённых ячеек"): read_book(bytes.fromhex("D0CF11E0A1B11AE1") + b"\x00" * 64, "XLS") # --- Распакованный размер zip --- @pytest.mark.usefixtures("openpyxl_forbidden") def test_zip_expanding_beyond_limit_is_rejected() -> None: data = _xlsx_with_zeros(ZEROS_BYTES) assert len(data) < SMALL_MAX_BYTES with pytest.raises(UnsupportedFormatError) as caught: read_book(data, "XLSX", max_uncompressed_bytes=ZIP_UNCOMPRESSED_LIMIT_FACTOR * SMALL_MAX_BYTES) assert caught.value.code is ErrorCode.UNSUPPORTED_FORMAT assert "распакованный размер" in caught.value.message assert str(ZIP_UNCOMPRESSED_LIMIT_FACTOR * SMALL_MAX_BYTES) in caught.value.message @pytest.mark.usefixtures("openpyxl_forbidden") def test_parser_passes_ten_times_max_bytes_as_zip_limit(rules: Rules) -> None: outcome = OsvParser(rules, max_bytes=SMALL_MAX_BYTES, timeout_seconds=120).parse( _xlsx_with_zeros(ZEROS_BYTES), "XLSX" ) assert outcome.error is not None assert outcome.error.code is ErrorCode.UNSUPPORTED_FORMAT assert "распакованный размер" in outcome.error.message def test_zip_limit_applies_to_xlsb_too() -> None: buffer = io.BytesIO() with zipfile.ZipFile(buffer, "w", zipfile.ZIP_DEFLATED) as archive: archive.writestr("xl/workbook.bin", bytes(ZEROS_BYTES)) with pytest.raises(UnsupportedFormatError, match="распакованный размер"): read_book(buffer.getvalue(), "XLSB", max_uncompressed_bytes=SMALL_MAX_BYTES) def test_zip_within_limit_is_read() -> None: data = _xlsx_with_zeros(1000) sheets, _ = read_book(data, "XLSX", max_uncompressed_bytes=ZIP_UNCOMPRESSED_LIMIT_FACTOR * SMALL_MAX_BYTES) assert [sheet.name for sheet in sheets] == ["ОСВ"]