diff --git a/docs/adr/0005-leitor-de-planilhas.md b/docs/adr/0005-leitor-de-planilhas.md index bcc2f25..d89cfb4 100644 --- a/docs/adr/0005-leitor-de-planilhas.md +++ b/docs/adr/0005-leitor-de-planilhas.md @@ -25,7 +25,7 @@ Opções avaliadas: ## Decisão -- O módulo `namescommons.spreadsheet` lê XLSX com a biblioteca padrão (`zipfile` e `iterparse`), em fluxo. +- O módulo `namescommons.spreadsheet` lê XLSX com a biblioteca padrão (`zipfile` e `xml.parsers.expat` com `TreeBuilder`), em fluxo. - XLS é lido com `xlrd`, fixado em `==2.0.2` em `pyproject.toml`. É a única dependência de runtime acrescentada. - O formato é reconhecido pelos primeiros bytes do arquivo, e os dois caminhos devolvem valores do mesmo tipo (texto, inteiro, fração, booleano, vazio). - Adaptadores não importam `xlrd` nem leem XML de planilha: usam só `namescommons.spreadsheet`. @@ -33,7 +33,7 @@ Opções avaliadas: ## Consequências - XLSX não depende de terceiros, e o pico de memória é o de uma linha mais os textos compartilhados. -- O `ElementTree` da biblioteca padrão não busca entidades externas, e o expat do Python 3.12 limita a expansão de entidades; por isso `defusedxml` não entra. Como defesa em profundidade, o leitor recusa com `FormatError` qualquer membro XML cujo prólogo contenha ` list[str]` devolve as abas na ordem do arquivo. `rows(path, sheet) -> Iterator[tuple[int, list]]` devolve `(número da linha, valores)` de cada linha não vazia, com o número da linha como o Excel mostra (a partir de 1) e os valores na posição da coluna (A = 0). Uma aba fora de `sheet_names` levanta `FormatError` com o nome pedido e a lista das abas. - **R1.2 Formato pelo conteúdo.** O formato é reconhecido pelos primeiros bytes, não pela extensão: `PK\x03\x04` é XLSX (ZIP do Office Open XML) e `D0 CF 11 E0` é XLS (documento composto do Excel, da versão 5 em diante). Outro conteúdo, inclusive uma página HTML de erro gravada no lugar do arquivo, levanta `FormatError`. - **R1.3 Valores iguais nos dois formatos.** Texto vira `str` sem alteração, número inteiro vira `int`, número com fração vira `float`, booleano vira `bool` e célula vazia vira `None`. Valores `None` no fim da linha são cortados, e linhas sem nenhum valor são puladas. Datas chegam como o número de série do Excel (nenhuma fonte de nomes usa datas em célula); a célula XLSX de tipo `d` (data ISO 8601, rara) não é aceita e levanta `FormatError` com o tipo. Em células mescladas, o valor fica só na célula do canto superior esquerdo, e as demais chegam vazias (`None`), como estão gravadas no arquivo. Célula com erro de fórmula (`#N/A`, `#REF!`) levanta `FormatError` com a aba e a referência da célula. -- **R1.4 XLSX com a biblioteca padrão.** O XLSX é lido com `zipfile` e `xml.etree.ElementTree.iterparse`, linha a linha, sem carregar a aba inteira na memória. São aceitos os tipos de célula `s` (texto compartilhado, inclusive com trechos de formatação; o texto fonético `rPh` é ignorado), `inlineStr`, `str`, `n` (ou sem tipo), `b` e `e`. A aba é localizada pelo nome em `xl/workbook.xml` e pelo alvo em `xl/_rels/workbook.xml.rels`. Cada linha lida sai da árvore XML, e a memória fica plana na aba inteira. Viram `FormatError`, com o arquivo e a aba ou a célula: membro XML com DTD no prólogo (` int | None` converte uma célula numa contagem pelo mesmo critério de `parse_count` (spec 003, R2.2): `int` não negativo passa, `float` com fração, negativo ou `bool` é `FormatError`, texto passa por `parse_count` com os marcadores de `missing`, e célula vazia só é aceita se `""` estiver em `missing`. diff --git a/src/namescommons/spreadsheet.py b/src/namescommons/spreadsheet.py index 3a02c9c..f723d03 100644 --- a/src/namescommons/spreadsheet.py +++ b/src/namescommons/spreadsheet.py @@ -4,11 +4,13 @@ XLS, com xlrd. Os dois devolvem valores do mesmo tipo: str, int, float, bool ou None. """ +import codecs import io import posixpath import re import zipfile import xml.etree.ElementTree as ET +import xml.parsers.expat as expat from collections.abc import Iterable, Iterator from pathlib import Path @@ -35,6 +37,10 @@ # Defesa em profundidade (ADR 0005): membro XML com DTD é recusado antes de ir ao parser. _MAX_PROLOGO = 1 << 16 _ESPACO = re.compile(r"[\s\ufeff]*") +_DECLARACAO = re.compile(r"<\?xml(?=\s)(.*?)\?>", re.DOTALL) +_ENCODING = re.compile(r"""\sencoding\s*=\s*(?:"([^"]*)"|'([^']*)')""") +_ENCODING_OK = re.compile(r"utf-?(?:8|16)", re.IGNORECASE) +_TAMANHO_BLOCO = 1 << 16 def _formato(caminho: Path) -> str: @@ -135,7 +141,11 @@ def __init__(self, fh, nome: str, caminho: Path): while True: parte = fh.read(4096) prefixo += parte - decisao = _prologo(prefixo) + try: + decisao = _prologo(prefixo, f"{caminho.name}: membro {nome}") + except FormatError: + fh.close() + raise if decisao == "dtd": fh.close() raise FormatError(f"{caminho.name}: membro {nome} tem DTD (DOCTYPE ou ENTITY); recusado") @@ -162,15 +172,47 @@ def close(self) -> None: super().close() -def _prologo(dados: bytes) -> str: - """'dtd' se há DOCTYPE/ENTITY antes do elemento raiz, 'raiz' se a raiz já começou, senão 'mais'.""" +def _texto_do_prefixo(dados: bytes, onde: str) -> str: + """Decodifica o prefixo como UTF-8 ou UTF-16 (por BOM ou NUL); outra coisa é FormatError. + + O expat aceita qualquer codificação declarada (EBCDIC, UTF-32, ISO-8859-x...) e um DOCTYPE + escrito assim passaria despercebido por uma leitura só em UTF-8; por isso o resto é recusado. + """ + if dados[:4] in (b"\xff\xfe\x00\x00", b"\x00\x00\xfe\xff") or dados[:4] in (b"<\x00\x00\x00", b"\x00\x00\x00<"): + raise FormatError(f"{onde} está em UTF-32; só UTF-8 e UTF-16 são aceitos") if dados[:2] in (b"\xff\xfe", b"\xfe\xff"): - texto = dados.decode("utf-16", errors="replace") + codec = "utf-16" + elif dados[:4] == b"<\x00?\x00": + codec = "utf-16-le" + elif dados[:4] == b"\x00<\x00?": + codec = "utf-16-be" elif b"\x00" in dados[:4]: - texto = dados.decode("utf-16-le" if dados[1:2] == b"\x00" else "utf-16-be", errors="replace") + raise FormatError(f"{onde} começa com bytes nulos que não são UTF-16; só UTF-8 e UTF-16 são aceitos") else: - # UTF-8 (BOM opcional). UTF-32 e EBCDIC não são inspecionados e falham no expat. - texto = dados.removeprefix(b"\xef\xbb\xbf").decode("utf-8", errors="replace") + codec = "utf-8-sig" + try: + # final=False: o prefixo pode terminar no meio de um caractere de vários bytes. + return codecs.getincrementaldecoder(codec)("strict").decode(dados, final=False) + except UnicodeDecodeError: + raise FormatError(f"{onde} não é UTF-8 nem UTF-16 válido; só essas codificações são aceitas") from None + + +def _confere_declaracao(texto: str, onde: str) -> None: + """A declaração `` só pode dizer UTF-8 ou UTF-16 (ou não dizer nada).""" + m = _DECLARACAO.match(texto) + if m and (e := _ENCODING.search(m.group(0))): + nome = e.group(1) if e.group(1) is not None else e.group(2) + if not _ENCODING_OK.fullmatch(nome): + raise FormatError(f"{onde} declara a codificação {nome!r}; só UTF-8 e UTF-16 são aceitos") + + +def _prologo(dados: bytes, onde: str = "membro XML") -> str: + """'dtd' se há DOCTYPE/ENTITY antes do elemento raiz, 'raiz' se a raiz já começou, senão 'mais'. + + FormatError se o prefixo não é UTF-8 nem UTF-16 ou se a declaração XML pede outra codificação. + """ + texto = _texto_do_prefixo(dados, onde) + _confere_declaracao(texto.lstrip("\ufeff"), onde) i = 0 while True: i = _ESPACO.match(texto, i).end() @@ -191,13 +233,70 @@ def _prologo(dados: bytes) -> str: i = fim + salto +def _eventos(fh, nome: str, caminho: Path, events=("end",)) -> Iterator[tuple[str, ET.Element]]: + """Eventos ("start"/"end", elemento) do XML de um membro, em fluxo, num expat que recusa DTD. + + Defesa em profundidade, independente da pré-checagem do prólogo (ADR 0005): o próprio expat + levanta FormatError ao ver DOCTYPE ou ENTITY, em qualquer codificação que ele aceite. O + `ET.XMLParser` em C não expõe o expat; por isso o expat é usado direto, com `ET.TreeBuilder`. + Erro de sintaxe vira `ET.ParseError`, como no `iterparse`. + """ + def recusa(*_args): + raise FormatError(f"{caminho.name}: membro {nome} tem DTD (DOCTYPE ou ENTITY); recusado") + + construtor = ET.TreeBuilder() + pendentes: list[tuple[str, ET.Element]] = [] + quer_start, quer_end = "start" in events, "end" in events + + def inicio(tag, atributos): + el = construtor.start(tag, atributos) + if quer_start: + pendentes.append(("start", el)) + + def fim(tag): + el = construtor.end(tag) + if quer_end: + pendentes.append(("end", el)) + + parser = expat.ParserCreate(namespace_separator="}") + parser.buffer_text = True + parser.StartDoctypeDeclHandler = recusa + parser.EntityDeclHandler = recusa + parser.StartElementHandler = lambda tag, atr: inicio(_tag(tag), {_tag(k): v for k, v in atr.items()}) + parser.EndElementHandler = lambda tag: fim(_tag(tag)) + parser.CharacterDataHandler = construtor.data + try: + while True: + bloco = fh.read(_TAMANHO_BLOCO) + parser.Parse(bloco, not bloco) + yield from pendentes + pendentes.clear() + if not bloco: + break + except expat.ExpatError as e: + raise ET.ParseError(str(e)) from None + + +def _tag(nome: str) -> str: + """'uri}local' do expat vira '{uri}local' (nome qualificado do ElementTree).""" + return "{" + nome if "}" in nome else nome + + +def _raiz(fh, nome: str, caminho: Path) -> ET.Element: + """Elemento raiz do XML de um membro (o `ET.parse(...).getroot()` com DTD recusado).""" + raiz = None + for _, el in _eventos(fh, nome, caminho): + raiz = el + return raiz + + def _xlsx_abas(z: zipfile.ZipFile, caminho: Path) -> list[tuple[str, str]]: """(nome, membro do ZIP) de cada aba, na ordem de xl/workbook.xml.""" try: with _membro(z, "xl/workbook.xml", caminho) as fh: - livro = ET.parse(fh).getroot() + livro = _raiz(fh, "xl/workbook.xml", caminho) with _membro(z, "xl/_rels/workbook.xml.rels", caminho) as fh: - rels = ET.parse(fh).getroot() + rels = _raiz(fh, "xl/_rels/workbook.xml.rels", caminho) except ET.ParseError as e: raise FormatError(f"{caminho.name}: workbook do XLSX ilegível: {e}") from None alvos = {r.get("Id"): r.get("Target", "") for r in rels.iter(_REL_PKG + "Relationship")} @@ -228,7 +327,7 @@ def _textos_compartilhados(z: zipfile.ZipFile, caminho: Path) -> list[str]: textos = [] with _membro(z, "xl/sharedStrings.xml", caminho) as fh: try: - for _, el in ET.iterparse(fh): + for _, el in _eventos(fh, "xl/sharedStrings.xml", caminho): if el.tag == _SI: textos.append(_texto(el)) el.clear() @@ -301,7 +400,7 @@ def _xlsx_rows(caminho: Path, sheet: str) -> Iterator[tuple[int, list]]: with _membro(z, membro, caminho) as fh: numero, pai = 0, None try: - for evento, el in ET.iterparse(fh, events=("start", "end")): + for evento, el in _eventos(fh, membro, caminho, ("start", "end")): if evento == "start": if el.tag == _SHEET_DATA: pai = el diff --git a/tests/test_spreadsheet.py b/tests/test_spreadsheet.py index 4aad4c7..c9c9399 100644 --- a/tests/test_spreadsheet.py +++ b/tests/test_spreadsheet.py @@ -312,17 +312,16 @@ def test_linhas_lidas_saem_da_arvore(tmp_path): dados = "".join(f'{i}' for i in range(1, 2001)) arq = _xlsx(tmp_path / "a.xlsx", [("X", dados)]) vistos = [] - original = planilha.ET.iterparse + original = planilha._eventos def espia(*args, **kwargs): - it = original(*args, **kwargs) - for evento, el in it: + for evento, el in original(*args, **kwargs): yield evento, el if evento == "end" and el.tag.endswith("sheetData"): vistos.append(len(el)) monkeypatch = pytest.MonkeyPatch() - monkeypatch.setattr(planilha.ET, "iterparse", espia) + monkeypatch.setattr(planilha, "_eventos", espia) try: assert sum(1 for _ in rows(arq, "X")) == 2000 finally: @@ -438,3 +437,94 @@ def test_r1_4_dtd_depois_de_bom_utf8_e_recusado(tmp_path, membro, operacao, prol arq = _regravar(base, tmp_path / "a.xlsx", membro, f'{prolog}{raiz}') with pytest.raises(FormatError, match=rf"a\.xlsx: membro {membro} tem DTD"): operacao(arq) + + +# DTD em codificações que fogem da pré-checagem em UTF-8 (parser differential) ---------------------- + +_MEMBROS = [ + ("xl/workbook.xml", lambda a: sheet_names(a)), + ("xl/_rels/workbook.xml.rels", lambda a: sheet_names(a)), + ("xl/sharedStrings.xml", lambda a: list(rows(a, "X"))), + ("xl/worksheets/sheet1.xml", lambda a: list(rows(a, "X"))), +] +_RAIZ = {"xl/workbook.xml": f"", "xl/_rels/workbook.xml.rels": "", + "xl/sharedStrings.xml": f"", "xl/worksheets/sheet1.xml": f""} +_DTD = ']>' + + +def _regravar_bytes(origem: Path, destino: Path, membro: str, dados: bytes) -> Path: + with zipfile.ZipFile(origem) as zin, zipfile.ZipFile(destino, "w", zipfile.ZIP_DEFLATED) as zout: + for info in zin.infolist(): + zout.writestr(info.filename, dados if info.filename == membro else zin.read(info.filename)) + return destino + + +@pytest.mark.parametrize("membro, operacao", _MEMBROS) +@pytest.mark.parametrize("declarada, codec", [ + ("cp037", "cp037"), + ("utf-32", "utf-32"), + ("UTF-32LE", "utf-32-le"), + ("UTF-32BE", "utf-32-be"), + ("latin-1", "latin-1"), + ("iso-8859-1", "ascii"), + ("windows-1252", "cp1252"), + ("us-ascii", "ascii"), +]) +def test_r1_4_dtd_em_codificacao_nao_utf_e_recusado(tmp_path, membro, operacao, declarada, codec): + base = _xlsx(tmp_path / "base.xlsx", [("X", '1')]) + texto = f'{_DTD}{_RAIZ[membro]}' + arq = _regravar_bytes(base, tmp_path / "a.xlsx", membro, texto.encode(codec)) + with pytest.raises(FormatError, match=rf"a\.xlsx: membro {membro} "): + operacao(arq) + + +@pytest.mark.parametrize("membro, operacao", _MEMBROS) +@pytest.mark.parametrize("dados", [ + '{dtd}{raiz}'.encode("cp037"), # sem declaração: EBCDIC não é UTF-8 + '{dtd}{raiz}'.encode("utf-32-le"), + '{dtd}{raiz}'.encode("utf-32-be"), + b"\xff\xfe\x00\x00<\x00\x00\x00", + b"\x00\x00\xfe\xff\x00\x00\x00<", + b'\xe9{dtd}{raiz}', # UTF-8 inválido +], ids=["ebcdic", "utf32le", "utf32be", "bom-utf32le", "bom-utf32be", "utf8-invalido"]) +def test_r1_4_prefixo_que_nao_e_utf8_nem_utf16_e_recusado(tmp_path, membro, operacao, dados): + base = _xlsx(tmp_path / "base.xlsx", [("X", '1')]) + arq = _regravar_bytes(base, tmp_path / "a.xlsx", membro, dados) + with pytest.raises(FormatError, match=rf"a\.xlsx: membro {membro} "): + operacao(arq) + + +@pytest.mark.parametrize("membro, operacao", _MEMBROS) +@pytest.mark.parametrize("codec, declarada", [ + ("utf-8", "UTF-8"), ("utf-8", "utf8"), ("utf-8", None), ("utf-16", "UTF-16"), ("utf-16", "utf16"), +]) +def test_r1_4_dtd_em_utf8_e_utf16_com_declaracao_valida_continua_recusado(tmp_path, membro, operacao, codec, declarada): + base = _xlsx(tmp_path / "base.xlsx", [("X", '1')]) + enc = f' encoding="{declarada}"' if declarada else "" + texto = f'{_DTD}{_RAIZ[membro]}' + arq = _regravar_bytes(base, tmp_path / "a.xlsx", membro, texto.encode(codec)) + with pytest.raises(FormatError, match=rf"a\.xlsx: membro {membro} tem DTD"): + operacao(arq) + + +@pytest.mark.parametrize("codec, declarada", [ + ("utf-8", "UTF-8"), ("utf-8", "utf-8"), ("utf-16", "UTF-16"), ("utf-16-le", None)]) +def test_r1_4_xlsx_normal_em_utf8_ou_utf16_e_lido(tmp_path, codec, declarada): + base = _xlsx(tmp_path / "base.xlsx", [("X", "")]) + enc = f' encoding="{declarada}"' if declarada else "" + texto = (f'' + 'IÑIGO') + arq = _regravar_bytes(base, tmp_path / "a.xlsx", "xl/worksheets/sheet1.xml", texto.encode(codec)) + assert list(rows(arq, "X")) == [(1, ["IÑIGO"])] + + +@pytest.mark.parametrize("membro, operacao", _MEMBROS) +def test_r1_4_o_parser_recusa_dtd_mesmo_sem_a_pre_checagem(tmp_path, monkeypatch, membro, operacao): + """Defesa em profundidade: com a pré-checagem desligada, os handlers do expat ainda recusam DTD.""" + import namescommons.spreadsheet as planilha + monkeypatch.setattr(planilha, "_prologo", lambda dados, onde="": "raiz") + base = _xlsx(tmp_path / "base.xlsx", [("X", '1')]) + for dtd in (_DTD, BILLION_LAUGHS.split("')): + arq = _regravar(base, tmp_path / "a.xlsx", membro, f'{dtd}{_RAIZ[membro]}') + with pytest.raises(FormatError, match=rf"a\.xlsx: membro {membro} tem DTD"): + operacao(arq)