"""Optional libarchive ZIPCrypto acceleration, without filesystem extraction. Only public libarchive read APIs are bound. Callers must preflight the complete ZIP directory with the ordinary archive validator. Missing libraries/symbols select the Python path before decoding; native failures never retry a decoder. Public ABI reference: https://github.com/libarchive/libarchive/blob/master/libarchive/archive.h """ from __future__ import annotations from collections.abc import Callable, Iterator, Mapping import ctypes from ctypes.util import find_library from functools import lru_cache import os from os import PathLike import stat from typing import Any import zlib from govoplan_files.backend.storage.common import FileStorageError _READ_CHUNK_SIZE = 1024 * 1024 _ARCHIVE_OK = 0 _ARCHIVE_EOF = 1 @lru_cache(maxsize=1) def native_zip_library() -> Any | None: name = find_library("archive") if not name: return None try: library = ctypes.CDLL(name) signatures = { "archive_read_new": ([], ctypes.c_void_p), "archive_read_support_filter_none": ([ctypes.c_void_p], ctypes.c_int), "archive_read_support_format_zip": ([ctypes.c_void_p], ctypes.c_int), "archive_read_set_format_option": ( [ctypes.c_void_p, ctypes.c_char_p, ctypes.c_char_p, ctypes.c_char_p], ctypes.c_int, ), "archive_read_add_passphrase": ( [ctypes.c_void_p, ctypes.c_char_p], ctypes.c_int, ), "archive_read_open_memory": ( [ctypes.c_void_p, ctypes.c_void_p, ctypes.c_size_t], ctypes.c_int, ), "archive_read_open_filename": ( [ctypes.c_void_p, ctypes.c_char_p, ctypes.c_size_t], ctypes.c_int, ), "archive_read_next_header": ( [ctypes.c_void_p, ctypes.POINTER(ctypes.c_void_p)], ctypes.c_int, ), "archive_read_data": ( [ctypes.c_void_p, ctypes.c_void_p, ctypes.c_size_t], ctypes.c_ssize_t, ), "archive_read_free": ([ctypes.c_void_p], ctypes.c_int), "archive_entry_pathname_utf8": ([ctypes.c_void_p], ctypes.c_char_p), "archive_entry_size": ([ctypes.c_void_p], ctypes.c_int64), "archive_entry_filetype": ([ctypes.c_void_p], ctypes.c_uint), } for symbol, (arguments, result) in signatures.items(): function = getattr(library, symbol) function.argtypes = arguments function.restype = result return library except (OSError, AttributeError): return None def _require_ok(status: int) -> None: if status != _ARCHIVE_OK: # Do not expose native error strings: they can contain source paths. raise FileStorageError( "Native ZIP decoding failed: incorrect password, corrupt or unsupported archive" ) def read_native_zip_members( library: Any, archive_data: bytes | str | PathLike[str], *, infos_by_path: Mapping[str, Any], selected_files: set[str], normalize_name: Callable[[str], str], password: str, max_file_bytes: int, max_total_bytes: int, progress: Callable[[str, int, int, int, int], None] | None = None, total_bytes: int = 0, ) -> Iterator[tuple[str, bytes]]: # Keep memory input, password and output buffers alive until reader free. source_buffer = None password_bytes = password.encode("utf-8") buffer = ctypes.create_string_buffer(_READ_CHUNK_SIZE) archive = library.archive_read_new() if not archive: raise FileStorageError("Native ZIP reader could not be allocated") total = 0 completed = 0 seen: set[str] = set() try: _require_ok(library.archive_read_support_filter_none(archive)) _require_ok(library.archive_read_support_format_zip(archive)) _require_ok( library.archive_read_set_format_option( archive, b"zip", b"hdrcharset", b"UTF-8" ) ) _require_ok(library.archive_read_add_passphrase(archive, password_bytes)) if isinstance(archive_data, bytes): source_buffer = ctypes.create_string_buffer(archive_data) _require_ok( library.archive_read_open_memory( archive, source_buffer, len(archive_data) ) ) else: _require_ok( library.archive_read_open_filename( archive, os.fsencode(archive_data), _READ_CHUNK_SIZE ) ) entry = ctypes.c_void_p() while True: status = library.archive_read_next_header(archive, ctypes.byref(entry)) if status == _ARCHIVE_EOF: break _require_ok(status) raw_path = library.archive_entry_pathname_utf8(entry) if not raw_path: raise FileStorageError("Native ZIP entry has no path") try: path = normalize_name(raw_path.decode("utf-8")) except UnicodeDecodeError as exc: raise FileStorageError( "Native ZIP path does not match the inspected directory" ) from exc info = infos_by_path.get(path) if info is None or path in seen: raise FileStorageError( "Native ZIP path does not match the inspected directory" ) seen.add(path) directory = info.is_dir() file_type = library.archive_entry_filetype(entry) if not (stat.S_ISDIR(file_type) if directory else stat.S_ISREG(file_type)): raise FileStorageError( "Native ZIP entry type does not match the inspected directory" ) if library.archive_entry_size(entry) != info.file_size: raise FileStorageError( "Native ZIP entry size does not match the inspected directory" ) if directory or path not in selected_files: continue parts = [] size = 0 checksum = 0 while True: read_size = min( _READ_CHUNK_SIZE, max_file_bytes + 1 - size, max_total_bytes + 1 - total, ) if read_size <= 0: raise FileStorageError("Archive exceeds its extraction limits") count = library.archive_read_data(archive, buffer, read_size) if count < 0: _require_ok(count) if count == 0: break if count > read_size: raise FileStorageError("Native ZIP reader returned an invalid byte count") size += count total += count if size > max_file_bytes: raise FileStorageError( f"Archive member {path!r} exceeds per-file limit" ) if total > max_total_bytes: raise FileStorageError("Archive is too large after extraction") chunk = ctypes.string_at(buffer, count) checksum = zlib.crc32(chunk, checksum) parts.append(chunk) if progress: progress( "extracting", completed, len(selected_files), total, total_bytes ) # Independent checks preserve the Python ZIP reader's CRC and size # guarantees even when native decoding behavior changes upstream. if size != info.file_size or checksum != info.CRC: raise FileStorageError( "Native ZIP member failed size or CRC verification" ) completed += 1 if progress: progress( "extracting", completed, len(selected_files), total, total_bytes ) yield path, b"".join(parts) parts.clear() if seen != set(infos_by_path) or completed != len(selected_files): raise FileStorageError( "Native ZIP directory does not match the inspected archive" ) finally: library.archive_read_free(archive)