"""Лимиты фазы чтения (ruling R-23, находка I-1 финального ревью): распакованный размер zip-архива и суммарная площадь объединённых ячеек проверяются до библиотеки чтения — бомба из нескольких килобайт не доходит до openpyxl. Книги собираются в памяти: в fixtures ничего не пишется. """ import io import time import zipfile from collections.abc import Callable, Iterator from typing import Any import openpyxl import pytest from arm_statements import default_rules from src.contract import ErrorCode, ResultStatus from src.parser.book import BookParser from src.parser.errors import UnreadableError, UnsupportedFormatError from src.parser.grid import MAX_INDEXED_AREA, MergedRange, Sheet from src.parser.osv import OsvParser from src.parser.readers import _READERS, read_book from src.parser.readers._limits import ( LARGE_MERGED_COUNT_LIMIT, MERGED_AREA_LIMIT, ZIP_UNCOMPRESSED_LIMIT_FACTOR, check_merged_area, check_xlsx_merged_area, merged_area_of_xml, ) from src.parser.rules import Rules, load_rules from src.parser.statements.pipeline import StatementPipeline 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 байт MAIN_NS = "http://schemas.openxmlformats.org/spreadsheetml/2006/main" _PACKAGE_RELS_NS = "http://schemas.openxmlformats.org/package/2006/relationships" _OFFICE_RELS_NS = "http://schemas.openxmlformats.org/officeDocument/2006/relationships" _STYLES = ( '' '' '' '' ) @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) -> Iterator[None]: """openpyxl не должен вызываться: бомба отбрасывается до него (и до выделения памяти под ячейки). Вызов не только падает, но и запоминается: `read_book` превращает любое исключение читателя в `UnreadableError`, и тест на UNREADABLE без проверки после теста прошёл бы и при вызове openpyxl. """ calls: list[str] = [] def forbidden(*_args: Any, **_kwargs: Any) -> None: calls.append("load_workbook") raise AssertionError("openpyxl.load_workbook вызван: лимит должен сработать до чтения книги") monkeypatch.setattr(openpyxl, "load_workbook", forbidden) yield assert calls == [], "openpyxl.load_workbook вызван: лимит должен сработать до чтения книги" # --- Площадь объединённых ячеек (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 def _book(rules: Rules, max_bytes: int) -> BookParser: statement_rules = default_rules() return BookParser( rules, osv=OsvParser(rules), statements=StatementPipeline(statement_rules), statement_rules=statement_rules, max_bytes=max_bytes, timeout_seconds=120, ) @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_huge_merged_range_gives_unsupported_format_from_parser(rules: Rules) -> None: outcome = _book(rules, 50 * 1024 * 1024).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'' ) % (MAIN_NS.encode(), MAIN_NS.encode()) 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) # --- Обходы скана (повторное ревью I-1б): книги собираются вручную, `zipfile`, без openpyxl --- BOMB_REF = "A1:Z200000" # 5,2 млн ячеек SMALL_REF = "A1:B2" SMALL_SPAN = MergedRange(1, 1, 2, 2) XML_DECLARATION = '' def _hand_xlsx( merge_cells: str, *, sheet_part: str = SHEET_PART, extra: dict[str, bytes] | None = None, sheet_head: str = XML_DECLARATION, sheet_encoding: str = "utf-8", ) -> bytes: """Минимальная книга xlsx из одного листа: `merge_cells` вставляется после `sheetData`, лист лежит в `sheet_part` (цель связи в `xl/_rels/workbook.xml.rels` — openpyxl идёт по ней, имя части не проверяет), начинается с `sheet_head` (по умолчанию XML-декларация) и записан в кодировке `sheet_encoding`.""" target = sheet_part.removeprefix("xl/") parts = { "[Content_Types].xml": ( '' '' '' '' '' ), "_rels/.rels": ( f'' f'' "" ), "xl/workbook.xml": ( f'' '' ), "xl/_rels/workbook.xml.rels": ( f'' f'' f'' ), "xl/styles.xml": f'{_STYLES}', } sheet = ( f'{sheet_head}' 'Оборотно-сальдовая ведомость по счету 60' f"{merge_cells}" ) out = io.BytesIO() with zipfile.ZipFile(out, "w", zipfile.ZIP_DEFLATED) as archive: for name, content in parts.items(): archive.writestr(name, content) archive.writestr(sheet_part, sheet.encode(sheet_encoding)) for name, raw in (extra or {}).items(): archive.writestr(name, raw) return out.getvalue() def _merges(element: str) -> str: return f'{element}' # Вариант → (элемент mergeCell с диапазоном SMALL_REF, тот же с BOMB_REF, часть листа) _BYPASSES: dict[str, tuple[str, str, str]] = { "исходная бомба": (f'', f'', SHEET_PART), "лист вне xl/worksheets": (f'', f'', "xl/data/s.xml"), "лист без расширения .xml": (f'', f'', "xl/data/s.dat"), "десятичная ссылка на символ": ('', '', SHEET_PART), "шестнадцатеричная ссылка на символ": ( '', '', SHEET_PART, ), "атрибут a:ref перед ref": ( f'', f'', SHEET_PART, ), "«>» в значении атрибута": ( f'', f'', SHEET_PART, ), "элемент с префиксом основного пространства": ( f'', f'', SHEET_PART, ), "элемент чужого пространства имён": ( f'', f'', SHEET_PART, ), } @pytest.mark.parametrize("variant", list(_BYPASSES)) def test_bypass_variant_is_honoured_by_openpyxl(variant: str) -> None: """Каждый вариант — не выдумка: openpyxl на малом диапазоне действительно читает объединение (значит, на большом разложил бы его по ячейкам), а скан пропускает книгу в пределах лимита и видит площадь диапазона.""" small, _, part = _BYPASSES[variant] data = _hand_xlsx(_merges(small), sheet_part=part) sheets, _ = read_book(data, "XLSX") assert sheets[0].merged == (SMALL_SPAN,) with zipfile.ZipFile(io.BytesIO(data)) as archive, pytest.raises(UnsupportedFormatError): check_xlsx_merged_area(archive, limit=SMALL_SPAN.area - 1) @pytest.mark.usefixtures("openpyxl_forbidden") @pytest.mark.parametrize("variant", list(_BYPASSES)) def test_bypass_variant_bomb_is_rejected_before_openpyxl(variant: str) -> None: _, bomb, part = _BYPASSES[variant] data = _hand_xlsx(_merges(bomb), sheet_part=part) 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 @pytest.mark.usefixtures("openpyxl_forbidden") def test_whole_sheet_merge_costs_one_step() -> None: """Площадь считается по строке ref, ячейки не перебираются: диапазон на весь лист отвергается сразу.""" started = time.monotonic() with pytest.raises(UnsupportedFormatError, match="лимит объединённых ячеек"): read_book(_hand_xlsx(_merges('')), "XLSX") assert time.monotonic() - started < FAST_SECONDS def test_unprefixed_ref_only_is_counted() -> None: """Считается только атрибут `ref` без префикса (его читает openpyxl); `a:ref` с бомбой — чужой атрибут.""" xml = ( b'' b"" ) % MAIN_NS.encode() assert merged_area_of_xml(io.BytesIO(xml), limit=MERGED_AREA_LIMIT) == SMALL_SPAN.area def test_scan_stops_reading_as_soon_as_limit_is_exceeded() -> None: """Ранний отказ: поток после бомбы не дочитывается (и даже битый XML за ней не важен).""" head = b'' stream = io.BytesIO(head + b"x" * 1_000_000 + b"<<<") with pytest.raises(UnsupportedFormatError, match="лимит объединённых ячеек"): merged_area_of_xml(stream, limit=MERGED_AREA_LIMIT, chunk_size=len(head)) assert stream.tell() < 2 * len(head) @pytest.mark.usefixtures("openpyxl_forbidden") def test_malformed_xml_part_is_unreadable() -> None: data = _hand_xlsx(_merges('')) # элемент не закрыт with pytest.raises(UnreadableError) as caught: read_book(data, "XLSX") assert caught.value.code is ErrorCode.UNREADABLE @pytest.mark.usefixtures("openpyxl_forbidden") @pytest.mark.parametrize("element", ['', '', ""]) def test_unparseable_merge_ref_is_unreadable(element: str) -> None: """ref, который openpyxl не разберёт (или его нет), — книга не читается, как и у openpyxl.""" with pytest.raises(UnreadableError, match="mergeCell"): read_book(_hand_xlsx(_merges(element)), "XLSX") def test_non_xml_parts_that_are_not_xml_do_not_break_reading() -> None: """Части без расширения .xml сканируются все, но мягко: бинарные (картинки, printerSettings), текст и не вполне корректный VML книгу не ломают — ошибка разбора лишь заканчивает скан части (openpyxl их не разбирает).""" extra = { "xl/media/image1.png": b"\x89PNG\r\n\x1a\n" + bytes(64), "xl/printerSettings/printerSettings1.bin": "\\\\принтер".encode("utf-16-le") + bytes(64), "xl/drawings/vmlDrawing1.vml": b"
", "xl/notes.txt": b" " * 100 + b"plain text, not XML", "xl/media/empty.bin": b"", } sheets, _ = read_book(_hand_xlsx(_merges(f''), extra=extra), "XLSX") assert sheets[0].merged == (SMALL_SPAN,) @pytest.mark.usefixtures("openpyxl_forbidden") def test_utf16_part_is_scanned() -> None: """Кодировка части не прячет диапазон: expat сам определяет UTF-16 по BOM.""" bomb = _merges(f'') sheet = f'{bomb}' data = _hand_xlsx("", extra={"xl/data/u.bin": sheet.encode("utf-16")}) with pytest.raises(UnsupportedFormatError, match="лимит объединённых ячеек"): read_book(data, "XLSX") # Обход «sniff» (повторное ревью раунда 2, I-1б): лист в части без `.xml`, ≥ 64 байт ведущих пробелов, без # XML-декларации — первые 64 байта не начинаются с «<», и проверка «похоже на XML» часть пропускала (27 с / 1,4 ГБ) CRAFTED_PART = "xl/worksheets/sheet1.dat" CRAFTED_PADDING = " " * 64 SLOW_READ_SECONDS = 2.0 # Вариант → (начало части листа, кодировка) _SNIFF_BYPASSES: dict[str, tuple[str, str]] = { "64 пробела без декларации": (CRAFTED_PADDING, "utf-8"), "BOM UTF-16 и 64 пробела без декларации": (CRAFTED_PADDING, "utf-16"), "4096 пробелов без декларации": (" " * 4096, "utf-8"), } @pytest.mark.parametrize("variant", list(_SNIFF_BYPASSES)) def test_padded_non_xml_sheet_part_is_honoured_by_openpyxl(variant: str) -> None: """Вариант не выдумка: openpyxl читает такой лист и раскладывает его объединения.""" head, encoding = _SNIFF_BYPASSES[variant] data = _hand_xlsx( _merges(f''), sheet_part=CRAFTED_PART, sheet_head=head, sheet_encoding=encoding ) sheets, _ = read_book(data, "XLSX") assert sheets[0].merged == (SMALL_SPAN,) with zipfile.ZipFile(io.BytesIO(data)) as archive, pytest.raises(UnsupportedFormatError): check_xlsx_merged_area(archive, limit=SMALL_SPAN.area - 1) @pytest.mark.usefixtures("openpyxl_forbidden") @pytest.mark.parametrize("variant", list(_SNIFF_BYPASSES)) def test_padded_non_xml_sheet_part_bomb_is_rejected_before_openpyxl(variant: str) -> None: head, encoding = _SNIFF_BYPASSES[variant] data = _hand_xlsx( _merges(f''), sheet_part=CRAFTED_PART, sheet_head=head, sheet_encoding=encoding ) assert len(data) < BOMB_MAX_FILE_BYTES started = time.monotonic() with pytest.raises(UnsupportedFormatError) as caught: read_book(data, "XLSX") assert time.monotonic() - started < SLOW_READ_SECONDS assert caught.value.code is ErrorCode.UNSUPPORTED_FORMAT assert "лимит объединённых ячеек" in caught.value.message # --- Площадь объединённых ячеек после чтения (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 _indexable_spans(count: int) -> tuple[MergedRange, ...]: """`count` диапазонов площадью ровно `MAX_INDEXED_AREA` (столбец за столбцом): каждый индекс раскладывает.""" return tuple(MergedRange(1, col, MAX_INDEXED_AREA, col) for col in range(1, count + 1)) def test_merged_area_of_read_sheets_is_limited() -> None: """xlrd хранит merged списком (8 байт на диапазон), но индекс `Sheet` раскладывает диапазоны по ячейкам: тысячи средних диапазонов в xls дали бы десятки миллионов записей. Та же сумма площадей — после чтения. """ half = _indexable_spans(MERGED_AREA_LIMIT // MAX_INDEXED_AREA // 2) 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, 99, 1, 99))], limit=MERGED_AREA_LIMIT) def test_spans_not_indexed_by_sheet_are_not_counted_after_read() -> None: """Диапазон больше `MAX_INDEXED_AREA` (merged на весь столбец, на весь лист) `Sheet` по ячейкам не раскладывает — риска нет, и законная книга xls с таким диапазоном не отвергается (находка N1).""" whole_column = MergedRange(1, 1, 1_048_576, 1) whole_sheet = MergedRange(1, 1, 1_048_576, 16_384) just_above = MergedRange(1, 1, MAX_INDEXED_AREA + 1, 1) check_merged_area([_sheet_with(whole_column, whole_sheet, just_above)], limit=MERGED_AREA_LIMIT) with pytest.raises(UnsupportedFormatError, match="лимит объединённых ячеек"): check_merged_area([_sheet_with(whole_column, *_indexable_spans(MERGED_AREA_LIMIT // MAX_INDEXED_AREA + 1))]) def test_xls_reader_result_is_checked(monkeypatch: pytest.MonkeyPatch) -> None: many = _sheet_with(*_indexable_spans(MERGED_AREA_LIMIT // MAX_INDEXED_AREA + 1)) monkeypatch.setitem(_READERS, "XLS", ("BIFF8", lambda _data: (many,))) with pytest.raises(UnsupportedFormatError, match="лимит объединённых ячеек"): read_book(bytes.fromhex("D0CF11E0A1B11AE1") + b"\x00" * 64, "XLS") def test_xls_with_whole_column_merge_is_read(monkeypatch: pytest.MonkeyPatch) -> None: sheet = _sheet_with(MergedRange(1, 1, 1_048_576, 1)) monkeypatch.setitem(_READERS, "XLS", ("BIFF8", lambda _data: (sheet,))) sheets, _ = read_book(bytes.fromhex("D0CF11E0A1B11AE1") + b"\x00" * 64, "XLS") assert sheets == (sheet,) def _large_spans(count: int) -> tuple[MergedRange, ...]: """`count` диапазонов площадью `MAX_INDEXED_AREA + 1` (столбец за столбцом): индекс не раскладывает ни один.""" return tuple(MergedRange(1, col, MAX_INDEXED_AREA + 1, col) for col in range(1, count + 1)) def test_large_spans_count_is_limited_per_sheet() -> None: """Крупные диапазоны (площадь больше `MAX_INDEXED_AREA`) `Sheet.merged_span` перебирает линейно на каждое обращение: в xls их 100 000 — ≈ 4,4 мс на обращение (повторное ревью раунда 2, минор #2). Потолок — на лист.""" at_limit = _large_spans(LARGE_MERGED_COUNT_LIMIT) check_merged_area([_sheet_with(*at_limit), _sheet_with(*at_limit)]) with pytest.raises(UnsupportedFormatError, match="лимит крупных объединённых диапазонов") as caught: check_merged_area([_sheet_with(*_large_spans(LARGE_MERGED_COUNT_LIMIT + 1))]) assert str(LARGE_MERGED_COUNT_LIMIT) in caught.value.message def test_xls_with_too_many_large_spans_is_unsupported(monkeypatch: pytest.MonkeyPatch) -> None: many = _sheet_with(*_large_spans(100_000)) monkeypatch.setitem(_READERS, "XLS", ("BIFF8", lambda _data: (many,))) started = time.monotonic() with pytest.raises(UnsupportedFormatError) as caught: read_book(bytes.fromhex("D0CF11E0A1B11AE1") + b"\x00" * 64, "XLS") assert time.monotonic() - started < FAST_SECONDS assert caught.value.code is ErrorCode.UNSUPPORTED_FORMAT assert "лимит крупных объединённых диапазонов" in caught.value.message # --- Распакованный размер 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 = _book(rules, SMALL_MAX_BYTES).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] == ["ОСВ"]