executablescheck.py - master - askarinka/executablescheck

executablescheck.py
1 846 строк79 KB
askarinka
Переработан подход к рекурсии
08 июн 2026, 19:48
08 июн 2026, 19:48543eecc
#!/usr/bin/env python3
"""
Поиск исполняемых файлов (PE, ELF, Mach-O) и Java-архивов в папках и архивах.
Рекурсивный обход вложенных архивов (ZIP, 7z, RPM, newc cpio, ar/.a) с определением типа по magic-байтам.
Симлинки в отчёте помечаются как " [symlink → <цель>]".
"""

import io
import os
import tarfile
import tempfile
from argparse import ArgumentParser, RawDescriptionHelpFormatter
from concurrent.futures import ThreadPoolExecutor, as_completed
from typing import Callable, Dict, Generator, Iterable, List, Optional, Set, Tuple
import lzma
import zipfile

try:
    import py7zr
    from py7zr import WriterFactory
    from py7zr.io import Py7zIO
    HAS_PY7ZR = True
except ImportError:
    HAS_PY7ZR = False
    WriterFactory = None  # type: ignore[misc, assignment]
    Py7zIO = None  # type: ignore[misc, assignment]

try:
    import rpmfile
    HAS_RPMFILE = True
except ImportError:
    HAS_RPMFILE = False
    rpmfile = None  # type: ignore[misc, assignment]

# Константы
STREAM_CHUNK_SIZE = 1024 * 1024  # 1 МБ, меньше syscall при стриминге
ARCHIVE_HEADER_LEN = 8  # байт для различения ZIP, 7z и OLE (DOC/XLS/PPT)
NESTED_IN_MEMORY_MB = 2  # вложенные архивы до N МБ читать в память вместо temp-файла
ZIP_LOAD_INTO_MEMORY_MB = 10  # ZIP до N МБ загружать в память целиком (быстрее на SSD/HDD)
FILE_READ_BUFFER = 2 * 1024 * 1024  # 2 МБ буфер при чтении архивов с диска

HEADER_WIN = b"MZ"
HEADER_ELF = b"\x7fELF"
HEADER_MACHO_32 = b"\xfe\xed\xfa\xce"
HEADER_MACHO_64 = b"\xfe\xed\xfa\xcf"
HEADER_MACHO_FAT = b"\xca\xfe\xba\xbe"
HEADER_ZIP = b"PK"
HEADER_7Z = b"7z\xbc\xaf\x27\x1c"
HEADER_OLE = b"\xd0\xcf\x11\xe0\xa1\xb1\x1a\xe1"  # DOC, XLS, PPT, MSI (OLE Compound Document)
HEADER_RPM = b"\xed\xab\xee\xdb"
# newc cpio (payload RPM и отдельные .cpio-файлы)
HEADER_CPIO_NEWC = b"070701"  # RPM lead (payload: cpio)
HEADER_RAR = b"Rar!\x1a\x07"
HEADER_GZIP = b"\x1f\x8b"
HEADER_BZ2 = b"BZ"
HEADER_XZ = b"\xfd7zXZ\x00"
TAR_USTAR_OFFSET = 257
TAR_USTAR = b"ustar"
ARCHIVE_MAGIC_READ_LEN = 262
# Unix static library / ar(1) archive (не thin)
HEADER_AR = b"!<arch>\n"


_SYMLINK_TAG = " [symlink → "


def list_folder(folder_path: str) -> Generator[Tuple[int, str, str], None, None]:
    """Рекурсивно обходит папку: (индекс, путь_для_чтения, путь_для_отчёта).

    Для симлинка путь_для_чтения это сама ссылка (open следует к цели); в отчёте добавляется
    " [symlink → <разрешённый_путь>]". Остальные файлы: оба пути совпадают (как раньше, realpath).
    """
    file_index = 0
    for root, _dirs, filenames in os.walk(folder_path):
        for filename in filenames:
            filepath = os.path.join(root, filename)
            try:
                abs_entry = os.path.abspath(filepath)
                if os.path.islink(abs_entry):
                    if "hardlink" in abs_entry or "fifo" in abs_entry:
                        continue
                    path_io = abs_entry
                    try:
                        resolved = os.path.realpath(abs_entry)
                    except OSError:
                        try:
                            resolved = os.readlink(abs_entry)
                        except OSError:
                            resolved = "?"
                    path_report = f"{abs_entry}{_SYMLINK_TAG}{resolved}]"
                else:
                    path_io = os.path.abspath(os.path.realpath(filepath))
                    if "hardlink" in path_io or "fifo" in path_io:
                        continue
                    path_report = path_io
            except OSError:
                print(f"Исключение при обработке, пропускаем файл {filepath!r}")
                continue
            file_index += 1
            yield file_index, path_io, path_report


# Классы для 7z (только при наличии py7zr)
def _make_7z_factories():
    """Фабрики для извлечения из 7z: только заголовок или во временный файл."""
    if not (HAS_PY7ZR and WriterFactory is not None and Py7zIO is not None):
        return None, None

    class HeadOnlyIO(Py7zIO):
        def __init__(self, limit: int) -> None:
            self._limit = limit
            self._buffer = b""

        def write(self, data: bytes | bytearray) -> None:
            if len(self._buffer) < self._limit:
                self._buffer += bytes(data)[: self._limit - len(self._buffer)]

        def read(self, size: int = -1) -> bytes:
            return self._buffer[:size] if size > 0 else self._buffer

        def seek(self, offset: int, whence: int = 0) -> int:
            return 0

        def flush(self) -> None:
            pass

        def size(self) -> int:
            return len(self._buffer)

    class HeadOnlyFactory(WriterFactory):
        def __init__(self, head_size: int) -> None:
            self.head_size = head_size
            self.products: Dict[str, HeadOnlyIO] = {}

        def create(self, filename: str) -> HeadOnlyIO:
            p = HeadOnlyIO(self.head_size)
            self.products[filename] = p
            return p

    class TempFileIO(Py7zIO):
        """Пишет во временный файл. limit=0 означает без лимита."""

        def __init__(self, limit: int) -> None:
            self._limit = limit
            self._written = 0
            self._fd, self._path = tempfile.mkstemp(suffix=".nested")
            self._f = os.fdopen(self._fd, "wb")

        def write(self, data: bytes | bytearray) -> None:
            if self._limit > 0 and self._written >= self._limit:
                return
            chunk = bytes(data) if self._limit <= 0 else bytes(data)[: self._limit - self._written]
            self._f.write(chunk)
            self._written += len(chunk)

        def close(self) -> None:
            self._f.close()

        def read(self, size: int = -1) -> bytes:
            return b""

        def seek(self, offset: int, whence: int = 0) -> int:
            return 0

        def flush(self) -> None:
            self._f.flush()

        def size(self) -> int:
            return self._written

    class TempFileFactory(WriterFactory):
        def __init__(self, limit: int) -> None:
            self.limit = limit
            self.products: Dict[str, TempFileIO] = {}

        def create(self, filename: str) -> TempFileIO:
            p = TempFileIO(self.limit)
            self.products[filename] = p
            return p

    return HeadOnlyFactory, TempFileFactory


_HeadOnlyFactory, _TempFileFactory = _make_7z_factories() if HAS_PY7ZR else (None, None)


# Проверки по заголовку
def _header_is_native_exec(header: bytes, name: str = "") -> bool:
    if len(header) < 4:
        return False
    if header.startswith(HEADER_WIN) or header == HEADER_ELF:
        return True
    if header in (HEADER_MACHO_32, HEADER_MACHO_64):
        return True
    if header == HEADER_MACHO_FAT and not name.lower().endswith(".class"):
        return True
    return False


def _header_is_zip(header: bytes) -> bool:
    return len(header) >= 2 and header.startswith(HEADER_ZIP)


def _header_is_7z(header: bytes) -> bool:
    return len(header) >= len(HEADER_7Z) and header[: len(HEADER_7Z)] == HEADER_7Z


def _header_is_rpm(header: bytes) -> bool:
    return len(header) >= 4 and header[:4] == HEADER_RPM


def _header_is_cpio_newc(header: bytes) -> bool:
    return bool(HAS_RPMFILE and len(header) >= 6 and header[:6] == HEADER_CPIO_NEWC)


def _header_is_ar(header: bytes) -> bool:
    return len(header) >= 8 and header[:8] == HEADER_AR


def _header_is_tar(header: bytes) -> bool:
    """Сжатый TAR (gzip/bz2/xz) или голый TAR с ustar-меткой."""
    if len(header) >= 2 and header[:2] == HEADER_GZIP:
        return True
    if len(header) >= 2 and header[:2] == HEADER_BZ2:
        return True
    if len(header) >= len(HEADER_XZ) and header[:len(HEADER_XZ)] == HEADER_XZ:
        return True
    if len(header) >= TAR_USTAR_OFFSET + len(TAR_USTAR):
        if header[TAR_USTAR_OFFSET : TAR_USTAR_OFFSET + len(TAR_USTAR)] == TAR_USTAR:
            return True
    return False


def _header_is_nested_archive(header: bytes) -> bool:
    return (
        _header_is_zip(header)
        or _header_is_7z(header)
        or (_header_is_rpm(header) and HAS_RPMFILE)
        or _header_is_cpio_newc(header)
        or _header_is_ar(header)
        or _header_is_tar(header)
    )


def _header_is_ole(header: bytes) -> bool:
    """OLE Compound Document (DOC, XLS, PPT, MSI)."""
    return len(header) >= 8 and header[:8] == HEADER_OLE


def _is_office_document(header: bytes, name: str) -> bool:
    """Исключить документы Office по сигнатуре и расширению."""
    if _header_is_ole(header):
        return True
    # DOCX, XLSX, PPTX это ZIP, различаем по расширению
    if _header_is_zip(header):
        ext = os.path.splitext(name.lower())[1]
        return ext in (".doc", ".docx", ".xls", ".xlsx", ".ppt", ".pptx", ".odt", ".ods", ".odp", ".docm", ".xlsm", ".pptm")
    return False


def _header_is_archive(header: bytes) -> bool:
    """Архив по magic-байтам (ZIP, 7z, RAR, GZIP, BZ2, XZ, TAR)."""
    if len(header) < 2:
        return False
    magic_checks = [
        (HEADER_ZIP, 2),
        (HEADER_GZIP, 2),
        (HEADER_BZ2, 2),
        (HEADER_7Z, len(HEADER_7Z)),
        (HEADER_RAR, len(HEADER_RAR)),
        (HEADER_XZ, len(HEADER_XZ)),
    ]
    if HAS_RPMFILE and len(header) >= 4 and header[:4] == HEADER_RPM:
        return True
    if _header_is_cpio_newc(header):
        return True
    if _header_is_ar(header):
        return True
    for magic_bytes, required_len in magic_checks:
        if len(header) >= required_len and header[:required_len] == magic_bytes:
            return True
    if len(header) >= TAR_USTAR_OFFSET + len(TAR_USTAR):
        if header[TAR_USTAR_OFFSET : TAR_USTAR_OFFSET + len(TAR_USTAR)] == TAR_USTAR:
            return True
    return False


def _ar_parse_member_header(block60: bytes) -> Tuple[bytes, int]:
    """Заголовок члена ar (60 байт): имя (16), ..., размер (10), fmag '`\\n'."""
    if len(block60) != 60 or block60[58:60] != b"`\n":
        raise ValueError("ar hdr")
    name_raw = block60[0:16]
    size_s = block60[48:58].strip()
    size = int(size_s or b"0")
    return name_raw, size


def _ar_member_display_name(name_raw: bytes, strtab: bytes, member_data: bytes) -> str:
    """Имя для отчёта: BSD/GNU #1/, индекс в таблице имён /N, иначе поле name."""
    n = name_raw.rstrip(b" ")
    if n.startswith(b"#1/"):
        try:
            ln = int(n[3:].strip() or b"0")
        except ValueError:
            return n.decode("latin1", "replace")
        return member_data[:ln].decode("latin1", "replace").strip("\x00").strip()
    if len(n) > 1 and n.startswith(b"/") and n[1:].isdigit() and strtab:
        off = int(n[1:])
        chunk = strtab[off:]
        end = chunk.find(b"/")
        if end >= 0:
            return chunk[:end].decode("latin1", "replace")
    return n.rstrip(b"/").decode("latin1", "replace")


def _iter_ar_members(fileobj: io.BufferedIOBase) -> Generator[Tuple[str, bytes], None, None]:
    """Члены ar после сигнатуры !<arch>\\n; таблица имён // пропускается как служебная."""
    sig = fileobj.read(8)
    if sig != HEADER_AR:
        return
    strtab = b""
    while True:
        hdr = fileobj.read(60)
        if len(hdr) == 0:
            break
        if len(hdr) < 60:
            break
        try:
            name_raw, size = _ar_parse_member_header(hdr)
        except ValueError:
            break
        data = fileobj.read(size)
        if len(data) < size:
            break
        if fileobj.tell() % 2 == 1:
            fileobj.read(1)

        nstrip = name_raw.strip()
        if nstrip == b"//":
            strtab = data
            continue

        if nstrip.startswith(b"#1/"):
            try:
                ln = int(nstrip[3:].strip() or b"0")
            except ValueError:
                ln = 0
            display = _ar_member_display_name(name_raw, strtab, data)
            payload = data[ln:] if ln > 0 else data
        else:
            display = _ar_member_display_name(name_raw, strtab, data)
            payload = data

        if payload:
            yield display, payload


def _read_file_header(filepath: str) -> Optional[bytes]:
    """Читает заголовок файла один раз. Возвращает None при ошибке."""
    try:
        with open(filepath, "rb") as f:
            first = f.read(4)
            if len(first) < 4:
                return first
            if _header_is_native_exec(first, filepath):
                return first  # исполняемый, не читаем остальное
            rest = f.read(ARCHIVE_MAGIC_READ_LEN - 4)
            return first + rest
    except OSError:
        return None


# Вспомогательные функции
def _safe_unlink(path: Optional[str]) -> None:
    if path and os.path.lexists(path):
        try:
            os.unlink(path)
        except OSError:
            pass


def _read_zip_member_to_bytes(
    zf: zipfile.ZipFile,
    info: zipfile.ZipInfo,
    header: bytes,
    max_bytes: int,
) -> Optional[bytes]:
    """Читает элемент ZIP в память. Для небольших вложенных архивов это быстрее temp-файла."""
    try:
        with zf.open(info) as member:
            data = header + member.read(max_bytes - len(header) if max_bytes > 0 else -1)
        return data if len(data) >= 4 else None
    except (OSError, RuntimeError, zipfile.BadZipFile):
        return None


def _stream_zip_member_to_temp(
    zf: zipfile.ZipFile,
    info: zipfile.ZipInfo,
    header: bytes,
    max_bytes: int,
) -> Optional[str]:
    """Стримит элемент ZIP во временный файл. Возвращает путь или None."""
    temp_fd, temp_path = tempfile.mkstemp(suffix=".nested")
    bytes_written = 0
    try:
        with os.fdopen(temp_fd, "wb") as temp_file:
            temp_file.write(header)
            bytes_written = len(header)
            with zf.open(info) as member:
                member.read(ARCHIVE_HEADER_LEN)
                while True:
                    if max_bytes > 0 and bytes_written >= max_bytes:
                        break
                    chunk_size = STREAM_CHUNK_SIZE if max_bytes <= 0 else min(STREAM_CHUNK_SIZE, max_bytes - bytes_written)
                    chunk = member.read(chunk_size)
                    if not chunk:
                        break
                    temp_file.write(chunk)
                    bytes_written += len(chunk)
        if bytes_written < 4:
            _safe_unlink(temp_path)
            return None
        return temp_path
    except (OSError, RuntimeError):
        _safe_unlink(temp_path)
        return None


def _read_rpm_member_to_bytes(rpm: object, member: object, header: bytes, max_bytes: int) -> Optional[bytes]:
    """Читает член RPM через extractfile (rpm: открытый RPMFile)."""
    try:
        mf = rpm.extractfile(member)
        with mf:
            data = header + mf.read(max_bytes - len(header) if max_bytes > 0 else -1)
        return data if len(data) >= 4 else None
    except (OSError, RuntimeError, TypeError, AttributeError, Exception):
        return None


def _stream_rpm_member_to_temp(
    rpm: object,
    member: object,
    header: bytes,
    max_bytes: int,
) -> Optional[str]:
    """Стримит файл из RPM во временный файл."""
    temp_fd, temp_path = tempfile.mkstemp(suffix=".nested")
    bytes_written = 0
    try:
        mf = rpm.extractfile(member)
        with os.fdopen(temp_fd, "wb") as temp_file:
            temp_file.write(header)
            bytes_written = len(header)
            mf.read(ARCHIVE_HEADER_LEN)
            while True:
                if max_bytes > 0 and bytes_written >= max_bytes:
                    break
                chunk_size = STREAM_CHUNK_SIZE if max_bytes <= 0 else min(STREAM_CHUNK_SIZE, max_bytes - bytes_written)
                chunk = mf.read(chunk_size)
                if not chunk:
                    break
                temp_file.write(chunk)
                bytes_written += len(chunk)
        try:
            mf.close()
        except Exception:
            pass
        if bytes_written < 4:
            _safe_unlink(temp_path)
            return None
        return temp_path
    except (OSError, RuntimeError, TypeError, AttributeError, Exception):
        _safe_unlink(temp_path)
        return None


def _iter_cpio_newc_members(fileobj: io.BufferedIOBase):
    """Члены newc cpio (070701), тот же формат, что у распакованного payload RPM."""
    if not HAS_RPMFILE or rpmfile is None:
        return
    RPMInfo = rpmfile.RPMInfo
    while True:
        magic2 = fileobj.read(2)
        if len(magic2) < 2:
            return
        if magic2 != b"07":
            return
        magic = magic2 + fileobj.read(4)
        if len(magic) < 6 or magic != HEADER_CPIO_NEWC:
            return
        try:
            member = RPMInfo._read(magic, fileobj)
        except Exception:
            return
        if member.name == "TRAILER!!!":
            return
        if not member.isdir:
            yield member


def _read_cpio_member_to_bytes(
    cpio_file: io.BufferedIOBase,
    member: object,
    header: bytes,
    max_bytes: int,
) -> Optional[bytes]:
    if not HAS_RPMFILE or rpmfile is None:
        return None
    try:
        from rpmfile.io_extra import _SubFile

        mf = _SubFile(cpio_file, member.file_start, member.size)
        with mf:
            data = header + mf.read(max_bytes - len(header) if max_bytes > 0 else -1)
        return data if len(data) >= 4 else None
    except (OSError, RuntimeError, TypeError, AttributeError, Exception):
        return None


def _stream_cpio_member_to_temp(
    cpio_file: io.BufferedIOBase,
    member: object,
    header: bytes,
    max_bytes: int,
) -> Optional[str]:
    if not HAS_RPMFILE or rpmfile is None:
        return None
    from rpmfile.io_extra import _SubFile

    temp_fd, temp_path = tempfile.mkstemp(suffix=".nested")
    bytes_written = 0
    try:
        mf = _SubFile(cpio_file, member.file_start, member.size)
        with os.fdopen(temp_fd, "wb") as temp_file:
            temp_file.write(header)
            bytes_written = len(header)
            mf.read(ARCHIVE_HEADER_LEN)
            while True:
                if max_bytes > 0 and bytes_written >= max_bytes:
                    break
                chunk_size = STREAM_CHUNK_SIZE if max_bytes <= 0 else min(STREAM_CHUNK_SIZE, max_bytes - bytes_written)
                chunk = mf.read(chunk_size)
                if not chunk:
                    break
                temp_file.write(chunk)
                bytes_written += len(chunk)
        try:
            mf.close()
        except Exception:
            pass
        if bytes_written < 4:
            _safe_unlink(temp_path)
            return None
        return temp_path
    except (OSError, RuntimeError, TypeError, AttributeError, Exception):
        _safe_unlink(temp_path)
        return None


def _recurse_into_nested(
    extracted_path: str,
    magic_bytes: bytes,
    path_prefix: str,
    scan_zip_func: Callable[..., Tuple[List[str], List[str], bool]],
    scan_7z_func: Callable[..., Tuple[List[str], List[str], bool]],
    max_nested_size: int,
    scan_rpm_func: Optional[Callable[..., Tuple[List[str], List[str], bool]]] = None,
    scan_cpio_func: Optional[Callable[..., Tuple[List[str], List[str], bool]]] = None,
    scan_ar_func: Optional[Callable[..., Tuple[List[str], List[str], bool]]] = None,
    scan_tar_func: Optional[Callable[..., Tuple[List[str], List[str], bool]]] = None,
) -> Tuple[List[str], List[str], bool]:
    """Рекурсивный вызов сканера по пути к распакованному архиву и magic."""
    if _header_is_zip(magic_bytes):
        return scan_zip_func(archive_path=extracted_path, prefix=path_prefix, max_nested_archive_size=max_nested_size)
    if _header_is_7z(magic_bytes) and HAS_PY7ZR:
        return scan_7z_func(archive_path=extracted_path, prefix=path_prefix, max_nested_archive_size=max_nested_size)
    if _header_is_rpm(magic_bytes) and scan_rpm_func:
        return scan_rpm_func(archive_path=extracted_path, prefix=path_prefix, max_nested_archive_size=max_nested_size)
    if _header_is_cpio_newc(magic_bytes) and scan_cpio_func:
        return scan_cpio_func(archive_path=extracted_path, prefix=path_prefix, max_nested_archive_size=max_nested_size)
    if _header_is_ar(magic_bytes) and scan_ar_func:
        return scan_ar_func(archive_path=extracted_path, prefix=path_prefix, max_nested_archive_size=max_nested_size)
    if _header_is_tar(magic_bytes) and scan_tar_func:
        return scan_tar_func(archive_path=extracted_path, prefix=path_prefix, max_nested_archive_size=max_nested_size)
    return [], [], False


def _write_bytes_to_nested_temp(data: bytes, max_bytes: int) -> Optional[str]:
    """Временный файл для вложенного архива из буфера (до max_bytes; 0 = весь data)."""
    chunk = data if max_bytes <= 0 else data[:max_bytes]
    if len(chunk) < 4:
        return None
    temp_fd, temp_path = tempfile.mkstemp(suffix=".nested")
    try:
        with os.fdopen(temp_fd, "wb") as tf:
            tf.write(chunk)
        return temp_path
    except OSError:
        _safe_unlink(temp_path)
        return None


# Определение типа файлов (используют уже прочитанный header)
def is_java_archive(filepath: str) -> bool:
    try:
        with open(filepath, "rb") as f:
            header = f.read(4)
        if not (header.startswith(HEADER_ZIP) and zipfile.is_zipfile(filepath)):
            return False
        with zipfile.ZipFile(filepath) as zip_archive:
            member_names = zip_archive.namelist()
            return (
                any(name.upper() == "META-INF/MANIFEST.MF" for name in member_names)
                or any(name.lower().endswith(".class") for name in member_names)
            )
    except (OSError, zipfile.BadZipFile):
        return False


def list_files_extension(file_paths: Iterable[str]) -> Set[str]:
    """Расширение по имени файла; суффикс ' [symlink → ...]' в отчёте отрезается."""
    exts: Set[str] = set()
    for path in file_paths:
        base = path.split(_SYMLINK_TAG, 1)[0]
        exts.add(os.path.splitext(base)[1])
    return exts


# Сканирование ZIP
def scan_zip_archive_for_binaries_and_java(
    archive_path: Optional[str] = None,
    archive_bytes: Optional[bytes] = None,
    prefix: str = "",
    max_nested_archive_size: int = 0,
    *,
    _scan_7z: Optional[Callable[..., Tuple[List[str], List[str], bool]]] = None,
    _scan_zip_self: Optional[Callable[..., Tuple[List[str], List[str], bool]]] = None,
    _scan_rpm: Optional[Callable[..., Tuple[List[str], List[str], bool]]] = None,
    _scan_cpio: Optional[Callable[..., Tuple[List[str], List[str], bool]]] = None,
    _scan_ar: Optional[Callable[..., Tuple[List[str], List[str], bool]]] = None,
    _scan_tar: Optional[Callable[..., Tuple[List[str], List[str], bool]]] = None,
) -> Tuple[List[str], List[str], bool]:
    """Сканирует ZIP (по пути или байтам). Обход в глубину, тип по magic.
    Возвращает (исполняемые, java-архивы, является_ли_сам_архив_JAR).
    """
    executables_found: List[str] = []
    java_archives_found: List[str] = []
    is_self_jar = False

    if (archive_path is None) == (archive_bytes is None):
        return [], [], False

    if archive_bytes is None and archive_path:
        try:
            size = os.path.getsize(archive_path)
            if size > 0 and size <= ZIP_LOAD_INTO_MEMORY_MB * 1024 * 1024:
                with open(archive_path, "rb", buffering=FILE_READ_BUFFER) as f:
                    archive_bytes = f.read()
                archive_path = None
        except OSError:
            pass

    try:
        if archive_bytes is not None:
            archive_source = io.BytesIO(archive_bytes)
        else:
            archive_source = open(archive_path, "rb", buffering=FILE_READ_BUFFER)
        zip_archive = zipfile.ZipFile(archive_source, "r")  # type: ignore[arg-type]
    except (OSError, zipfile.BadZipFile):
        return [], [], False

    with zip_archive:
        is_self_jar = False
        for entry in zip_archive.infolist():
            if entry.is_dir():
                continue
            member_name = entry.filename
            if not is_self_jar:
                n = member_name.upper()
                is_self_jar = n == "META-INF/MANIFEST.MF" or member_name.lower().endswith(".class")
            uncompressed_size = getattr(entry, "file_size", 0) or 0

            try:
                with zip_archive.open(entry) as member:
                    member_header = member.read(ARCHIVE_HEADER_LEN)
            except (OSError, RuntimeError, zipfile.BadZipFile):
                continue

            if _is_office_document(member_header, member_name):
                pass  # пропускаем DOC, DOCX, XLS, XLSX и т.д.
            else:
                if _header_is_native_exec(member_header[:4], member_name):
                    executables_found.append(prefix + member_name)

            if uncompressed_size <= 0 or not _header_is_nested_archive(member_header):
                continue

            limit = max_nested_archive_size if max_nested_archive_size > 0 else 0
            in_memory_limit = NESTED_IN_MEMORY_MB * 1024 * 1024
            use_memory = (
                uncompressed_size > 0
                and uncompressed_size <= in_memory_limit
                and (limit <= 0 or uncompressed_size <= limit)
            )

            nested_result: Tuple[List[str], List[str], bool] = ([], [], False)
            if use_memory:
                archive_bytes = _read_zip_member_to_bytes(
                    zip_archive, entry, member_header, limit if limit > 0 else uncompressed_size
                )
                if archive_bytes:
                    try:
                        if _header_is_zip(member_header):
                            nested_result = (_scan_zip_self or scan_zip_archive_for_binaries_and_java)(
                                archive_bytes=archive_bytes,
                                prefix=prefix + member_name + "/",
                                max_nested_archive_size=max_nested_archive_size,
                            )
                        elif _header_is_7z(member_header) and HAS_PY7ZR:
                            nested_result = (_scan_7z or (lambda **kw: ([], [], False)))(
                                archive_bytes=archive_bytes,
                                prefix=prefix + member_name + "/",
                                max_nested_archive_size=max_nested_archive_size,
                            )
                        elif _header_is_rpm(member_header):
                            nested_result = (_scan_rpm or (lambda **kw: ([], [], False)))(
                                archive_bytes=archive_bytes,
                                prefix=prefix + member_name + "/",
                                max_nested_archive_size=max_nested_archive_size,
                            )
                        elif _header_is_cpio_newc(member_header):
                            nested_result = (_scan_cpio or (lambda **kw: ([], [], False)))(
                                archive_bytes=archive_bytes,
                                prefix=prefix + member_name + "/",
                                max_nested_archive_size=max_nested_archive_size,
                            )
                        elif _header_is_ar(member_header):
                            nested_result = (_scan_ar or (lambda **kw: ([], [], False)))(
                                archive_bytes=archive_bytes,
                                prefix=prefix + member_name + "/",
                                max_nested_archive_size=max_nested_archive_size,
                            )
                        elif _header_is_tar(member_header):
                            nested_result = (_scan_tar or (lambda **kw: ([], [], False)))(
                                archive_bytes=archive_bytes,
                                prefix=prefix + member_name + "/",
                                max_nested_archive_size=max_nested_archive_size,
                            )
                    except (OSError, zipfile.BadZipFile, Exception):
                        pass
            else:
                extracted_path = _stream_zip_member_to_temp(
                    zip_archive, entry, member_header, max_nested_archive_size
                )
                if extracted_path:
                    try:
                        nested_result = _recurse_into_nested(
                            extracted_path,
                            member_header,
                            prefix + member_name + "/",
                            _scan_zip_self or scan_zip_archive_for_binaries_and_java,
                            _scan_7z or (lambda **kw: ([], [], False)),
                            max_nested_archive_size,
                            _scan_rpm,
                            _scan_cpio,
                            _scan_ar,
                            _scan_tar,
                        )
                    except (OSError, zipfile.BadZipFile):
                        pass
                    finally:
                        _safe_unlink(extracted_path)

            executables_found.extend(nested_result[0])
            java_archives_found.extend(nested_result[1])
            if len(nested_result) > 2 and nested_result[2]:
                java_archives_found.append(prefix + member_name)

    return executables_found, java_archives_found, is_self_jar


# Сканирование 7z
def scan_7z_archive_for_binaries_and_java(
    archive_path: Optional[str] = None,
    archive_bytes: Optional[bytes] = None,
    prefix: str = "",
    max_nested_archive_size: int = 0,
    *,
    _scan_zip: Optional[Callable[..., Tuple[List[str], List[str], bool]]] = None,
    _scan_rpm: Optional[Callable[..., Tuple[List[str], List[str], bool]]] = None,
    _scan_cpio: Optional[Callable[..., Tuple[List[str], List[str], bool]]] = None,
    _scan_ar: Optional[Callable[..., Tuple[List[str], List[str], bool]]] = None,
    _scan_tar: Optional[Callable[..., Tuple[List[str], List[str], bool]]] = None,
) -> Tuple[List[str], List[str], bool]:
    """Сканирует 7z (по пути или байтам). Обход в глубину, тип по magic."""
    if not HAS_PY7ZR or _HeadOnlyFactory is None or _TempFileFactory is None:
        return [], [], False
    if (archive_path is None) == (archive_bytes is None):
        return [], [], False

    executables_found: List[str] = []
    java_archives_found: List[str] = []

    if archive_bytes is None and archive_path:
        try:
            size = os.path.getsize(archive_path)
            if size > 0 and size <= ZIP_LOAD_INTO_MEMORY_MB * 1024 * 1024:
                with open(archive_path, "rb", buffering=FILE_READ_BUFFER) as f:
                    archive_bytes = f.read()
                archive_path = None
        except OSError:
            pass

    try:
        if archive_bytes is not None:
            archive_source = io.BytesIO(archive_bytes)
        else:
            archive_source = open(archive_path, "rb", buffering=FILE_READ_BUFFER)
        archive_7z = py7zr.SevenZipFile(archive_source, mode="r")  # type: ignore[arg-type]
    except (OSError, py7zr.Bad7zFile):
        return [], [], False
    except Exception:
        return [], [], False

    try:
        with archive_7z:
            file_entries = archive_7z.list()
            file_names = [
                getattr(entry, "filename", "")
                for entry in file_entries
                if not getattr(entry, "is_directory", False)
                and getattr(entry, "filename", "")
                and not getattr(entry, "filename", "").endswith("/")
            ]
            if not file_names:
                return [], [], False

            header_factory = _HeadOnlyFactory(ARCHIVE_HEADER_LEN)
            archive_7z.extract(targets=file_names, factory=header_factory)

            uncompressed_sizes = {
                getattr(entry, "filename", ""): getattr(entry, "uncompressed", 0) or 0
                for entry in file_entries
            }
            nested_archives: List[Tuple[str, int]] = []

            for member_name, header_buffer in header_factory.products.items():
                member_header = header_buffer.read(ARCHIVE_HEADER_LEN)
                if len(member_header) < 4:
                    continue
                if _is_office_document(member_header, member_name):
                    continue
                if _header_is_native_exec(member_header[:4], member_name):
                    executables_found.append(prefix + member_name)
                if _header_is_nested_archive(member_header):
                    member_size = uncompressed_sizes.get(member_name, 0)
                    if member_size >= 0:
                        nested_archives.append((member_name, member_size))

            scan_zip_func = _scan_zip or scan_zip_archive_for_binaries_and_java

            for member_name, uncompressed_size in nested_archives:
                extracted_path = None
                try:
                    if hasattr(archive_7z, "reset"):
                        archive_7z.reset()
                    extract_limit = (
                        uncompressed_size
                        if max_nested_archive_size <= 0
                        else min(uncompressed_size, max_nested_archive_size)
                    )
                    temp_factory = _TempFileFactory(extract_limit)
                    archive_7z.extract(targets=[member_name], factory=temp_factory)
                    if member_name not in temp_factory.products:
                        continue
                    temp_io = temp_factory.products[member_name]
                    temp_io.close()
                    extracted_path = temp_io._path
                    if temp_io.size() < 4:
                        continue
                    with open(extracted_path, "rb") as f:
                        magic_bytes = f.read(ARCHIVE_HEADER_LEN)
                    nested_result = _recurse_into_nested(
                        extracted_path,
                        magic_bytes,
                        prefix + member_name + "/",
                        scan_zip_func,
                        lambda **kw: scan_7z_archive_for_binaries_and_java(
                            _scan_zip=scan_zip_func,
                            _scan_rpm=_scan_rpm,
                            _scan_cpio=_scan_cpio,
                            _scan_ar=_scan_ar,
                            _scan_tar=_scan_tar,
                            **kw,
                        ),
                        max_nested_archive_size,
                        _scan_rpm,
                        _scan_cpio,
                        _scan_ar,
                        _scan_tar,
                    )
                    executables_found.extend(nested_result[0])
                    java_archives_found.extend(nested_result[1])
                    if len(nested_result) > 2 and nested_result[2]:
                        java_archives_found.append(prefix + member_name)
                except (OSError, py7zr.Bad7zFile, zipfile.BadZipFile, TypeError, KeyError):
                    pass
                finally:
                    _safe_unlink(extracted_path)
    except Exception:
        pass

    return executables_found, java_archives_found, False


def _rpmfile_prepare_payload_stream(rpm: object) -> None:
    """rpmfile для archive_compression=lzma открывает payload как gzip и ломается (BadGzipFile).

    У пакетов ALT/Sisyphus и др. встречается тег lzma (фактически поток, который принимает
    lzma.LZMAFile с FORMAT_AUTO). Патчим до первого чтения data_file.
    """
    if not HAS_RPMFILE or rpmfile is None:
        return
    try:
        comp = rpm.headers.get("archive_compression")
    except (AttributeError, KeyError, TypeError):
        return
    if comp != b"lzma":
        return
    if getattr(rpm, "_data_file", None) is not None:
        return
    try:
        from rpmfile.io_extra import _SubFile
    except ImportError:
        return
    try:
        fileobj = _SubFile(rpm._fileobj, rpm.data_offset)
        rpm._data_file = lzma.LZMAFile(fileobj, format=lzma.FORMAT_AUTO)
    except (OSError, TypeError, AttributeError, lzma.LZMAError, ValueError):
        pass


# Сканирование RPM (cpio внутри)
def scan_rpm_archive_for_binaries_and_java(
    archive_path: Optional[str] = None,
    archive_bytes: Optional[bytes] = None,
    prefix: str = "",
    max_nested_archive_size: int = 0,
    *,
    _scan_zip: Optional[Callable[..., Tuple[List[str], List[str], bool]]] = None,
    _scan_7z: Optional[Callable[..., Tuple[List[str], List[str], bool]]] = None,
    _scan_rpm_self: Optional[Callable[..., Tuple[List[str], List[str], bool]]] = None,
    _scan_cpio: Optional[Callable[..., Tuple[List[str], List[str], bool]]] = None,
    _scan_ar: Optional[Callable[..., Tuple[List[str], List[str], bool]]] = None,
    _scan_tar: Optional[Callable[..., Tuple[List[str], List[str], bool]]] = None,
) -> Tuple[List[str], List[str], bool]:
    """Сканирует RPM (payload: cpio). Возвращает (исполняемые, java, False)."""
    if not HAS_RPMFILE or rpmfile is None:
        return [], [], False
    if (archive_path is None) == (archive_bytes is None):
        return [], [], False

    executables_found: List[str] = []
    java_archives_found: List[str] = []

    try:
        if archive_bytes is not None:
            rpm_ctx = rpmfile.open(fileobj=io.BytesIO(archive_bytes))
        else:
            rpm_ctx = rpmfile.open(archive_path)
    except Exception:
        return [], [], False

    try:
        with rpm_ctx as rpm:
            _rpmfile_prepare_payload_stream(rpm)
            for member in rpm.getmembers():
                if member.isdir:
                    continue
                member_name = member.name
                try:
                    mf = rpm.extractfile(member)
                    member_header = mf.read(ARCHIVE_HEADER_LEN)
                    mf.close()
                except Exception:
                    continue

                if _is_office_document(member_header, member_name):
                    pass
                else:
                    if _header_is_native_exec(member_header[:4], member_name):
                        executables_found.append(prefix + member_name)

                uncompressed_size = int(getattr(member, "size", 0) or 0)
                if uncompressed_size <= 0 or not _header_is_nested_archive(member_header):
                    continue

                limit = max_nested_archive_size if max_nested_archive_size > 0 else 0
                in_memory_limit = NESTED_IN_MEMORY_MB * 1024 * 1024
                use_memory = (
                    uncompressed_size > 0
                    and uncompressed_size <= in_memory_limit
                    and (limit <= 0 or uncompressed_size <= limit)
                )

                nested_result: Tuple[List[str], List[str], bool] = ([], [], False)
                scan_zip_fn = _scan_zip or scan_zip_archive_for_binaries_and_java
                scan_7z_fn = _scan_7z or (lambda **kw: ([], [], False))
                scan_rpm_fn = _scan_rpm_self or scan_rpm_archive_for_binaries_and_java
                scan_cpio_fn = _scan_cpio or (lambda **kw: ([], [], False))
                scan_ar_fn = _scan_ar or (lambda **kw: ([], [], False))
                scan_tar_fn = _scan_tar or (lambda **kw: ([], [], False))

                if use_memory:
                    nested_bytes = _read_rpm_member_to_bytes(
                        rpm, member, member_header, limit if limit > 0 else uncompressed_size
                    )
                    if nested_bytes:
                        try:
                            if _header_is_zip(member_header):
                                nested_result = scan_zip_fn(
                                    archive_bytes=nested_bytes,
                                    prefix=prefix + member_name + "/",
                                    max_nested_archive_size=max_nested_archive_size,
                                )
                            elif _header_is_7z(member_header) and HAS_PY7ZR:
                                nested_result = scan_7z_fn(
                                    archive_bytes=nested_bytes,
                                    prefix=prefix + member_name + "/",
                                    max_nested_archive_size=max_nested_archive_size,
                                )
                            elif _header_is_rpm(member_header):
                                nested_result = scan_rpm_fn(
                                    archive_bytes=nested_bytes,
                                    prefix=prefix + member_name + "/",
                                    max_nested_archive_size=max_nested_archive_size,
                                )
                            elif _header_is_cpio_newc(member_header):
                                nested_result = scan_cpio_fn(
                                    archive_bytes=nested_bytes,
                                    prefix=prefix + member_name + "/",
                                    max_nested_archive_size=max_nested_archive_size,
                                )
                            elif _header_is_ar(member_header):
                                nested_result = scan_ar_fn(
                                    archive_bytes=nested_bytes,
                                    prefix=prefix + member_name + "/",
                                    max_nested_archive_size=max_nested_archive_size,
                                )
                            elif _header_is_tar(member_header):
                                nested_result = scan_tar_fn(
                                    archive_bytes=nested_bytes,
                                    prefix=prefix + member_name + "/",
                                    max_nested_archive_size=max_nested_archive_size,
                                )
                        except Exception:
                            pass
                else:
                    extracted_path = _stream_rpm_member_to_temp(
                        rpm, member, member_header, max_nested_archive_size
                    )
                    if extracted_path:
                        try:
                            nested_result = _recurse_into_nested(
                                extracted_path,
                                member_header,
                                prefix + member_name + "/",
                                scan_zip_fn,
                                scan_7z_fn,
                                max_nested_archive_size,
                                scan_rpm_fn,
                                scan_cpio_fn,
                                scan_ar_fn,
                                scan_tar_fn,
                            )
                        except (OSError, zipfile.BadZipFile):
                            pass
                        finally:
                            _safe_unlink(extracted_path)

                executables_found.extend(nested_result[0])
                java_archives_found.extend(nested_result[1])
                if len(nested_result) > 2 and nested_result[2]:
                    java_archives_found.append(prefix + member_name)
    except Exception:
        return [], [], False

    return executables_found, java_archives_found, False


# Сканирование сырого newc cpio (070701), в т.ч. извлечённый payload RPM
def scan_cpio_archive_for_binaries_and_java(
    archive_path: Optional[str] = None,
    archive_bytes: Optional[bytes] = None,
    prefix: str = "",
    max_nested_archive_size: int = 0,
    *,
    _scan_zip: Optional[Callable[..., Tuple[List[str], List[str], bool]]] = None,
    _scan_7z: Optional[Callable[..., Tuple[List[str], List[str], bool]]] = None,
    _scan_rpm: Optional[Callable[..., Tuple[List[str], List[str], bool]]] = None,
    _scan_cpio_self: Optional[Callable[..., Tuple[List[str], List[str], bool]]] = None,
    _scan_ar: Optional[Callable[..., Tuple[List[str], List[str], bool]]] = None,
    _scan_tar: Optional[Callable[..., Tuple[List[str], List[str], bool]]] = None,
) -> Tuple[List[str], List[str], bool]:
    """Сканирует newc cpio без оболочки RPM. Возвращает (исполняемые, java, False)."""
    if not HAS_RPMFILE or rpmfile is None:
        return [], [], False
    if (archive_path is None) == (archive_bytes is None):
        return [], [], False

    executables_found: List[str] = []
    java_archives_found: List[str] = []
    own_stream = False
    stream: Optional[io.BufferedIOBase] = None
    try:
        if archive_bytes is not None:
            stream = io.BytesIO(archive_bytes)
        else:
            stream = open(archive_path, "rb", buffering=FILE_READ_BUFFER)
            own_stream = True

        scan_zip_fn = _scan_zip or scan_zip_archive_for_binaries_and_java
        scan_7z_fn = _scan_7z or (lambda **kw: ([], [], False))
        scan_rpm_fn = _scan_rpm or (lambda **kw: ([], [], False))
        scan_cpio_fn = _scan_cpio_self or scan_cpio_archive_for_binaries_and_java
        scan_ar_fn = _scan_ar or (lambda **kw: ([], [], False))
        scan_tar_fn = _scan_tar or (lambda **kw: ([], [], False))

        from rpmfile.io_extra import _SubFile

        for member in _iter_cpio_newc_members(stream):
            member_name = member.name
            try:
                mf = _SubFile(stream, member.file_start, member.size)
                member_header = mf.read(ARCHIVE_HEADER_LEN)
                try:
                    mf.close()
                except Exception:
                    pass
            except Exception:
                continue

            if _is_office_document(member_header, member_name):
                pass
            else:
                if _header_is_native_exec(member_header[:4], member_name):
                    executables_found.append(prefix + member_name)

            uncompressed_size = int(getattr(member, "size", 0) or 0)
            if uncompressed_size <= 0 or not _header_is_nested_archive(member_header):
                continue

            limit = max_nested_archive_size if max_nested_archive_size > 0 else 0
            in_memory_limit = NESTED_IN_MEMORY_MB * 1024 * 1024
            use_memory = (
                uncompressed_size > 0
                and uncompressed_size <= in_memory_limit
                and (limit <= 0 or uncompressed_size <= limit)
            )

            nested_result: Tuple[List[str], List[str], bool] = ([], [], False)

            if use_memory:
                nested_bytes = _read_cpio_member_to_bytes(
                    stream, member, member_header, limit if limit > 0 else uncompressed_size
                )
                if nested_bytes:
                    try:
                        if _header_is_zip(member_header):
                            nested_result = scan_zip_fn(
                                archive_bytes=nested_bytes,
                                prefix=prefix + member_name + "/",
                                max_nested_archive_size=max_nested_archive_size,
                            )
                        elif _header_is_7z(member_header) and HAS_PY7ZR:
                            nested_result = scan_7z_fn(
                                archive_bytes=nested_bytes,
                                prefix=prefix + member_name + "/",
                                max_nested_archive_size=max_nested_archive_size,
                            )
                        elif _header_is_rpm(member_header):
                            nested_result = scan_rpm_fn(
                                archive_bytes=nested_bytes,
                                prefix=prefix + member_name + "/",
                                max_nested_archive_size=max_nested_archive_size,
                            )
                        elif _header_is_cpio_newc(member_header):
                            nested_result = scan_cpio_fn(
                                archive_bytes=nested_bytes,
                                prefix=prefix + member_name + "/",
                                max_nested_archive_size=max_nested_archive_size,
                            )
                        elif _header_is_ar(member_header):
                            nested_result = scan_ar_fn(
                                archive_bytes=nested_bytes,
                                prefix=prefix + member_name + "/",
                                max_nested_archive_size=max_nested_archive_size,
                            )
                        elif _header_is_tar(member_header):
                            nested_result = scan_tar_fn(
                                archive_bytes=nested_bytes,
                                prefix=prefix + member_name + "/",
                                max_nested_archive_size=max_nested_archive_size,
                            )
                    except Exception:
                        pass
            else:
                extracted_path = _stream_cpio_member_to_temp(
                    stream, member, member_header, max_nested_archive_size
                )
                if extracted_path:
                    try:
                        nested_result = _recurse_into_nested(
                            extracted_path,
                            member_header,
                            prefix + member_name + "/",
                            scan_zip_fn,
                            scan_7z_fn,
                            max_nested_archive_size,
                            scan_rpm_fn,
                            scan_cpio_fn,
                            scan_ar_fn,
                            scan_tar_fn,
                        )
                    except (OSError, zipfile.BadZipFile):
                        pass
                    finally:
                        _safe_unlink(extracted_path)

            executables_found.extend(nested_result[0])
            java_archives_found.extend(nested_result[1])
            if len(nested_result) > 2 and nested_result[2]:
                java_archives_found.append(prefix + member_name)
    except Exception:
        return [], [], False
    finally:
        if own_stream and stream is not None:
            try:
                stream.close()
            except OSError:
                pass

    return executables_found, java_archives_found, False


# Сканирование ar (статические библиотеки .a): ELF/Mach-O/PE в .o внутри
def scan_ar_archive_for_binaries_and_java(
    archive_path: Optional[str] = None,
    archive_bytes: Optional[bytes] = None,
    prefix: str = "",
    max_nested_archive_size: int = 0,
    *,
    _scan_zip: Optional[Callable[..., Tuple[List[str], List[str], bool]]] = None,
    _scan_7z: Optional[Callable[..., Tuple[List[str], List[str], bool]]] = None,
    _scan_rpm: Optional[Callable[..., Tuple[List[str], List[str], bool]]] = None,
    _scan_cpio: Optional[Callable[..., Tuple[List[str], List[str], bool]]] = None,
    _scan_ar_self: Optional[Callable[..., Tuple[List[str], List[str], bool]]] = None,
    _scan_tar: Optional[Callable[..., Tuple[List[str], List[str], bool]]] = None,
) -> Tuple[List[str], List[str], bool]:
    """Сканирует ar-архив (!<arch>\\n). Учитывает объектные .o (ELF ET_REL и т.д.) и вложенные архивы."""
    if (archive_path is None) == (archive_bytes is None):
        return [], [], False

    executables_found: List[str] = []
    java_archives_found: List[str] = []

    if archive_bytes is None and archive_path:
        try:
            size = os.path.getsize(archive_path)
            if size > 0 and size <= ZIP_LOAD_INTO_MEMORY_MB * 1024 * 1024:
                with open(archive_path, "rb", buffering=FILE_READ_BUFFER) as f:
                    archive_bytes = f.read()
                archive_path = None
        except OSError:
            pass

    own_stream = False
    stream: Optional[io.BufferedIOBase] = None
    try:
        if archive_bytes is not None:
            stream = io.BytesIO(archive_bytes)
        else:
            stream = open(archive_path, "rb", buffering=FILE_READ_BUFFER)
            own_stream = True

        scan_zip_fn = _scan_zip or scan_zip_archive_for_binaries_and_java
        scan_7z_fn = _scan_7z or (lambda **kw: ([], [], False))
        scan_rpm_fn = _scan_rpm or (lambda **kw: ([], [], False))
        scan_cpio_fn = _scan_cpio or (lambda **kw: ([], [], False))
        scan_ar_fn = _scan_ar_self or scan_ar_archive_for_binaries_and_java
        scan_tar_fn = _scan_tar or (lambda **kw: ([], [], False))

        for member_name, payload in _iter_ar_members(stream):
            member_header = payload[:ARCHIVE_HEADER_LEN]
            if len(member_header) < 4:
                continue

            if _is_office_document(member_header, member_name):
                pass
            else:
                if _header_is_native_exec(member_header[:4], member_name):
                    executables_found.append(prefix + member_name)

            plen = len(payload)
            if plen <= 0 or not _header_is_nested_archive(member_header):
                continue

            limit = max_nested_archive_size if max_nested_archive_size > 0 else 0
            in_memory_limit = NESTED_IN_MEMORY_MB * 1024 * 1024
            use_memory = plen > 0 and plen <= in_memory_limit and (limit <= 0 or plen <= limit)

            nested_result: Tuple[List[str], List[str], bool] = ([], [], False)

            if use_memory:
                nested_bytes = payload if limit <= 0 else payload[:limit]
                if len(nested_bytes) >= 4:
                    try:
                        if _header_is_zip(member_header):
                            nested_result = scan_zip_fn(
                                archive_bytes=nested_bytes,
                                prefix=prefix + member_name + "/",
                                max_nested_archive_size=max_nested_archive_size,
                            )
                        elif _header_is_7z(member_header) and HAS_PY7ZR:
                            nested_result = scan_7z_fn(
                                archive_bytes=nested_bytes,
                                prefix=prefix + member_name + "/",
                                max_nested_archive_size=max_nested_archive_size,
                            )
                        elif _header_is_rpm(member_header):
                            nested_result = scan_rpm_fn(
                                archive_bytes=nested_bytes,
                                prefix=prefix + member_name + "/",
                                max_nested_archive_size=max_nested_archive_size,
                            )
                        elif _header_is_cpio_newc(member_header):
                            nested_result = scan_cpio_fn(
                                archive_bytes=nested_bytes,
                                prefix=prefix + member_name + "/",
                                max_nested_archive_size=max_nested_archive_size,
                            )
                        elif _header_is_ar(member_header):
                            nested_result = scan_ar_fn(
                                archive_bytes=nested_bytes,
                                prefix=prefix + member_name + "/",
                                max_nested_archive_size=max_nested_archive_size,
                            )
                        elif _header_is_tar(member_header):
                            nested_result = scan_tar_fn(
                                archive_bytes=nested_bytes,
                                prefix=prefix + member_name + "/",
                                max_nested_archive_size=max_nested_archive_size,
                            )
                    except Exception:
                        pass
            else:
                extracted_path = _write_bytes_to_nested_temp(payload, limit if limit > 0 else 0)
                if extracted_path:
                    try:
                        with open(extracted_path, "rb") as ef:
                            magic_bytes = ef.read(ARCHIVE_HEADER_LEN)
                        nested_result = _recurse_into_nested(
                            extracted_path,
                            magic_bytes,
                            prefix + member_name + "/",
                            scan_zip_fn,
                            scan_7z_fn,
                            max_nested_archive_size,
                            scan_rpm_fn,
                            scan_cpio_fn,
                            scan_ar_fn,
                            scan_tar_fn,
                        )
                    except (OSError, zipfile.BadZipFile):
                        pass
                    finally:
                        _safe_unlink(extracted_path)

            executables_found.extend(nested_result[0])
            java_archives_found.extend(nested_result[1])
            if len(nested_result) > 2 and nested_result[2]:
                java_archives_found.append(prefix + member_name)
    except Exception:
        return [], [], False
    finally:
        if own_stream and stream is not None:
            try:
                stream.close()
            except OSError:
                pass

    return executables_found, java_archives_found, False


# Сканирование TAR (plain, gzip, bz2, xz)
def scan_tar_archive_for_binaries_and_java(
    archive_path: Optional[str] = None,
    archive_bytes: Optional[bytes] = None,
    prefix: str = "",
    max_nested_archive_size: int = 0,
    *,
    _scan_zip: Optional[Callable[..., Tuple[List[str], List[str], bool]]] = None,
    _scan_7z: Optional[Callable[..., Tuple[List[str], List[str], bool]]] = None,
    _scan_rpm: Optional[Callable[..., Tuple[List[str], List[str], bool]]] = None,
    _scan_cpio: Optional[Callable[..., Tuple[List[str], List[str], bool]]] = None,
    _scan_ar: Optional[Callable[..., Tuple[List[str], List[str], bool]]] = None,
    _scan_tar_self: Optional[Callable[..., Tuple[List[str], List[str], bool]]] = None,
) -> Tuple[List[str], List[str], bool]:
    """Сканирует TAR (plain/.tar.gz/.tar.bz2/.tar.xz). Возвращает (исполняемые, java, False)."""
    if (archive_path is None) == (archive_bytes is None):
        return [], [], False

    executables_found: List[str] = []
    java_archives_found: List[str] = []

    try:
        if archive_bytes is not None:
            tf = tarfile.open(fileobj=io.BytesIO(archive_bytes), mode="r:*")
        else:
            tf = tarfile.open(archive_path, mode="r:*")
    except (tarfile.TarError, OSError, EOFError, Exception):
        return [], [], False

    scan_zip_fn = _scan_zip or scan_zip_archive_for_binaries_and_java
    scan_7z_fn = _scan_7z or (lambda **kw: ([], [], False))
    scan_rpm_fn = _scan_rpm or (lambda **kw: ([], [], False))
    scan_cpio_fn = _scan_cpio or (lambda **kw: ([], [], False))
    scan_ar_fn = _scan_ar or (lambda **kw: ([], [], False))
    scan_tar_fn = _scan_tar_self or scan_tar_archive_for_binaries_and_java

    try:
        with tf:
            for member in tf:
                if not member.isfile():
                    continue
                member_name = member.name
                uncompressed_size = member.size or 0

                mf = tf.extractfile(member)
                if mf is None:
                    continue

                limit = max_nested_archive_size if max_nested_archive_size > 0 else 0

                try:
                    member_header = mf.read(ARCHIVE_HEADER_LEN)
                except (tarfile.TarError, OSError):
                    try:
                        mf.close()
                    except Exception:
                        pass
                    continue

                if len(member_header) < 4:
                    try:
                        mf.close()
                    except Exception:
                        pass
                    continue

                if not _is_office_document(member_header, member_name):
                    if _header_is_native_exec(member_header[:4], member_name):
                        executables_found.append(prefix + member_name)

                is_nested = _header_is_nested_archive(member_header)
                if not is_nested or uncompressed_size <= 0:
                    try:
                        mf.close()
                    except Exception:
                        pass
                    continue

                # Записываем member в temp-файл (уже прочли header, дочитываем остаток)
                extracted_path = None
                try:
                    temp_fd, extracted_path = tempfile.mkstemp(suffix=".nested")
                    bytes_written = len(member_header)
                    with os.fdopen(temp_fd, "wb") as out:
                        out.write(member_header)
                        while True:
                            if limit > 0 and bytes_written >= limit:
                                break
                            chunk_size = STREAM_CHUNK_SIZE if limit <= 0 else min(STREAM_CHUNK_SIZE, limit - bytes_written)
                            chunk = mf.read(chunk_size)
                            if not chunk:
                                break
                            out.write(chunk)
                            bytes_written += len(chunk)

                    if bytes_written < 4:
                        _safe_unlink(extracted_path)
                        extracted_path = None
                        continue

                    nested_result: Tuple[List[str], List[str], bool] = ([], [], False)
                    if _header_is_zip(member_header):
                        nested_result = scan_zip_fn(archive_path=extracted_path, prefix=prefix + member_name + "/", max_nested_archive_size=max_nested_archive_size)
                    elif _header_is_7z(member_header) and HAS_PY7ZR:
                        nested_result = scan_7z_fn(archive_path=extracted_path, prefix=prefix + member_name + "/", max_nested_archive_size=max_nested_archive_size)
                    elif _header_is_rpm(member_header) and HAS_RPMFILE:
                        nested_result = scan_rpm_fn(archive_path=extracted_path, prefix=prefix + member_name + "/", max_nested_archive_size=max_nested_archive_size)
                    elif _header_is_cpio_newc(member_header):
                        nested_result = scan_cpio_fn(archive_path=extracted_path, prefix=prefix + member_name + "/", max_nested_archive_size=max_nested_archive_size)
                    elif _header_is_ar(member_header):
                        nested_result = scan_ar_fn(archive_path=extracted_path, prefix=prefix + member_name + "/", max_nested_archive_size=max_nested_archive_size)
                    elif _header_is_tar(member_header):
                        nested_result = scan_tar_fn(archive_path=extracted_path, prefix=prefix + member_name + "/", max_nested_archive_size=max_nested_archive_size)

                    executables_found.extend(nested_result[0])
                    java_archives_found.extend(nested_result[1])
                    if len(nested_result) > 2 and nested_result[2]:
                        java_archives_found.append(prefix + member_name)
                except (OSError, tarfile.TarError, zipfile.BadZipFile, Exception):
                    pass
                finally:
                    try:
                        mf.close()
                    except Exception:
                        pass
                    _safe_unlink(extracted_path)
    except (tarfile.TarError, OSError, EOFError, Exception):
        pass

    return executables_found, java_archives_found, False


# Отчёты
def make_report_file(
    report_path: str,
    file_paths: List[str],
    extensions: Iterable[str],
) -> None:
    extensions_sorted = sorted(extensions)
    with open(report_path, "w", encoding="utf-8") as report_file:
        if extensions_sorted:
            for ext in extensions_sorted:
                report_file.write(f"{ext}\n")
            report_file.write("-----\n")
        for filepath in file_paths:
            report_file.write(f"{filepath}\n")


def make_archives_report_file(
    report_path: str,
    archives_with_executables: Dict[str, List[str]],
    archives_to_exclude: Iterable[str] = (),
) -> None:
    exclude_set = set(archives_to_exclude)
    with open(report_path, "w", encoding="utf-8") as report_file:
        for archive_path in sorted(archives_with_executables):
            if archive_path in exclude_set:
                continue
            report_file.write(f"{archive_path}\n")
            for member_path in sorted(archives_with_executables[archive_path]):
                report_file.write(f"  {member_path}\n")
            report_file.write("-----\n")


def main() -> None:
    epilog = """
Примеры:
  %(prog)s
    Поиск в текущей папке, отчёты в executables_report.txt и archives_executables_report.txt

  %(prog)s C:\\Downloads report.txt
    Поиск в C:\\Downloads, основной отчёт в report.txt

  %(prog)s ./out -e --max-nested-mb 100
    Список расширений в отчёте, лимит вложенных архивов 100 МБ

  %(prog)s --no-scan-archives
    Только файлы в папке, без сканирования архивов

  %(prog)s -j 4
    Сканирование архивов в 4 потока (без -j: ядра - 1)

  %(prog)s --only-archives-with-exec
    В отчёте по архивам только архивы с исполняемыми внутри

  %(prog)s --help
    Показать эту справку
"""
    parser = ArgumentParser(
        description="Поиск исполняемых файлов (PE, ELF, Mach-O) и Java-архивов в папках и архивах. "
        "Рекурсивный обход ZIP, 7z, RPM и сырого newc cpio (070701). Тип определяется по magic-байтам.",
        epilog=epilog,
        formatter_class=RawDescriptionHelpFormatter,
    )
    parser.add_argument(
        "path_to_folder",
        nargs="?",
        default="./",
        help="путь к папке для поиска (по умолчанию: текущая)",
    )
    parser.add_argument(
        "path_to_report",
        nargs="?",
        default="./executables_report.txt",
        help="файл отчёта: список исполняемых и Java-архивов (по умолчанию: executables_report.txt)",
    )
    parser.add_argument(
        "--archives-report",
        default="./archives_executables_report.txt",
        metavar="ФАЙЛ",
        help="файл отчёта по архивам с исполняемыми внутри (по умолчанию: archives_executables_report.txt)",
    )
    parser.add_argument(
        "-e",
        "--extensions",
        action="store_true",
        help="добавить в начало отчёта список найденных расширений",
    )
    parser.add_argument(
        "--max-nested-mb",
        type=int,
        default=0,
        metavar="МБ",
        help="макс. размер вложенного архива для рекурсии, МБ; 0 = без лимита (по умолчанию)",
    )
    parser.add_argument(
        "--no-scan-archives",
        action="store_true",
        help="не сканировать содержимое архивов (только файлы в папке)",
    )
    parser.add_argument(
        "-j",
        "--jobs",
        type=int,
        default=max(1, (os.cpu_count() or 1) - 1),
        metavar="N",
        help="число параллельных потоков для сканирования архивов (по умолчанию: ядра - 1)",
    )
    parser.add_argument(
        "--only-archives-with-exec",
        action="store_true",
        help="в отчёте по архивам оставлять только архивы, содержащие исполняемые файлы",
    )
    args = parser.parse_args()

    max_nested_bytes = args.max_nested_mb * 1024 * 1024
    search_folder = os.path.abspath(os.path.realpath(args.path_to_folder))
    main_report_path = os.path.abspath(os.path.realpath(args.path_to_report))
    archives_report_path = os.path.abspath(os.path.realpath(args.archives_report))

    if not os.path.isdir(search_folder):
        print(f"Папка не найдена: {search_folder}")
        raise SystemExit(1)

    found_executables: List[str] = []
    found_java_archives: List[str] = []
    found_archives: List[str] = []
    archives_containing_executables: Dict[str, List[str]] = {}
    archives_containing_java: Dict[str, List[str]] = {}

    # Обёртки для взаимной рекурсии ZIP <-> 7z <-> RPM <-> cpio <-> ar <-> tar
    def scan_zip(**kwargs) -> Tuple[List[str], List[str], bool]:
        return scan_zip_archive_for_binaries_and_java(
            _scan_7z=scan_7z,
            _scan_zip_self=scan_zip,
            _scan_rpm=scan_rpm,
            _scan_cpio=scan_cpio,
            _scan_ar=scan_ar,
            _scan_tar=scan_tar,
            **kwargs,
        )

    def scan_7z(**kwargs) -> Tuple[List[str], List[str], bool]:
        return scan_7z_archive_for_binaries_and_java(
            _scan_zip=scan_zip,
            _scan_rpm=scan_rpm,
            _scan_cpio=scan_cpio,
            _scan_ar=scan_ar,
            _scan_tar=scan_tar,
            **kwargs,
        )

    def scan_rpm(**kwargs) -> Tuple[List[str], List[str], bool]:
        return scan_rpm_archive_for_binaries_and_java(
            _scan_zip=scan_zip,
            _scan_7z=scan_7z,
            _scan_rpm_self=scan_rpm,
            _scan_cpio=scan_cpio,
            _scan_ar=scan_ar,
            _scan_tar=scan_tar,
            **kwargs,
        )

    def scan_cpio(**kwargs) -> Tuple[List[str], List[str], bool]:
        return scan_cpio_archive_for_binaries_and_java(
            _scan_zip=scan_zip,
            _scan_7z=scan_7z,
            _scan_rpm=scan_rpm,
            _scan_cpio_self=scan_cpio,
            _scan_ar=scan_ar,
            _scan_tar=scan_tar,
            **kwargs,
        )

    def scan_ar(**kwargs) -> Tuple[List[str], List[str], bool]:
        return scan_ar_archive_for_binaries_and_java(
            _scan_zip=scan_zip,
            _scan_7z=scan_7z,
            _scan_rpm=scan_rpm,
            _scan_cpio=scan_cpio,
            _scan_ar_self=scan_ar,
            _scan_tar=scan_tar,
            **kwargs,
        )

    def scan_tar(**kwargs) -> Tuple[List[str], List[str], bool]:
        return scan_tar_archive_for_binaries_and_java(
            _scan_zip=scan_zip,
            _scan_7z=scan_7z,
            _scan_rpm=scan_rpm,
            _scan_cpio=scan_cpio,
            _scan_ar=scan_ar,
            _scan_tar_self=scan_tar,
            **kwargs,
        )

    def _scan_one_archive(archive_path: str, kind: str) -> Tuple[str, List[str], List[str], bool]:
        """Сканирует один архив. kind: zip | 7z | rpm | cpio | ar | tar. Возвращает (путь, исполняемые, java, is_jar)."""
        try:
            if kind == "zip":
                exec_in, java_in, is_self_jar = scan_zip(
                    archive_path=archive_path, max_nested_archive_size=max_nested_bytes
                )
                return (archive_path, exec_in, java_in, is_self_jar)
            if kind == "7z":
                exec_in, java_in, _ = scan_7z(
                    archive_path=archive_path, max_nested_archive_size=max_nested_bytes
                )
                return (archive_path, exec_in, java_in, False)
            if kind == "rpm":
                exec_in, java_in, _ = scan_rpm(
                    archive_path=archive_path, max_nested_archive_size=max_nested_bytes
                )
                return (archive_path, exec_in, java_in, False)
            if kind == "cpio":
                exec_in, java_in, _ = scan_cpio(
                    archive_path=archive_path, max_nested_archive_size=max_nested_bytes
                )
                return (archive_path, exec_in, java_in, False)
            if kind == "ar":
                exec_in, java_in, _ = scan_ar(
                    archive_path=archive_path, max_nested_archive_size=max_nested_bytes
                )
                return (archive_path, exec_in, java_in, False)
            if kind == "tar":
                exec_in, java_in, _ = scan_tar(
                    archive_path=archive_path, max_nested_archive_size=max_nested_bytes
                )
                return (archive_path, exec_in, java_in, False)
        except Exception:
            pass
        return (archive_path, [], [], False)

    archives_to_scan: List[Tuple[str, str, str]] = []
    files_scanned = 0
    for files_scanned, path_io, path_report in list_folder(search_folder):
        header = _read_file_header(path_io)
        if not header:
            continue

        if _header_is_native_exec(header[:4], path_io):
            found_executables.append(path_report)

        if _header_is_archive(header) and not _is_office_document(header, path_io):
            found_archives.append(path_report)
            if not args.no_scan_archives:
                if _header_is_rpm(header) and HAS_RPMFILE:
                    archives_to_scan.append((path_io, "rpm", path_report))
                elif _header_is_cpio_newc(header):
                    archives_to_scan.append((path_io, "cpio", path_report))
                elif _header_is_zip(header):
                    archives_to_scan.append((path_io, "zip", path_report))
                elif _header_is_7z(header) and HAS_PY7ZR:
                    archives_to_scan.append((path_io, "7z", path_report))
                elif _header_is_ar(header):
                    archives_to_scan.append((path_io, "ar", path_report))
                elif _header_is_tar(header):
                    archives_to_scan.append((path_io, "tar", path_report))
            elif _header_is_zip(header) and is_java_archive(path_io):
                found_java_archives.append(path_report)

    # Сканирование архивов (последовательно или параллельно)
    jobs = max(1, args.jobs)
    if jobs > 1 and archives_to_scan:
        with ThreadPoolExecutor(max_workers=jobs) as executor:
            futures = {
                executor.submit(_scan_one_archive, io_p, kind): path_report
                for io_p, kind, path_report in archives_to_scan
            }
            for future in as_completed(futures):
                path_report = futures[future]
                try:
                    _, exec_in, java_in, is_self_jar = future.result()
                    if is_self_jar:
                        found_java_archives.append(path_report)
                    archives_containing_executables[path_report] = exec_in
                    if java_in:
                        archives_containing_java[path_report] = java_in
                except Exception:
                    pass
    else:
        for io_p, kind, path_report in archives_to_scan:
            _, exec_in, java_in, is_self_jar = _scan_one_archive(io_p, kind)
            if is_self_jar:
                found_java_archives.append(path_report)
            archives_containing_executables[path_report] = exec_in
            if java_in:
                archives_containing_java[path_report] = java_in

    # Добавить архивы, которые не сканировались (RAR, GZIP и т.д. или --no-scan-archives)
    for archive_path in found_archives:
        if archive_path not in archives_containing_executables:
            archives_containing_executables[archive_path] = []

    total_exec_in_archives = sum(len(v) for v in archives_containing_executables.values())
    total_java_in_archives = sum(len(v) for v in archives_containing_java.values())

    print(f"Просмотрено файлов: {files_scanned}")
    print(f"Найдено исполняемых файлов: {len(found_executables)}")
    print(f"Найдено архивов: {len(found_archives)}")
    print(f"Исполняемых файлов внутри просканированных архивов: {total_exec_in_archives}")
    print(f"Java-архивов внутри просканированных архивов: {total_java_in_archives}")

    executables_in_archives = [
        f"{archive_path}/{member_path}"
        for archive_path, members in archives_containing_executables.items()
        for member_path in members
    ]
    all_executables = found_executables + sorted(executables_in_archives)
    extensions = list_files_extension(all_executables + found_java_archives) if args.extensions else []
    make_report_file(main_report_path, all_executables + found_java_archives, extensions)
    archives_for_report = archives_containing_executables
    if args.only_archives_with_exec:
        archives_for_report = {k: v for k, v in archives_containing_executables.items() if v}
    make_archives_report_file(
        archives_report_path, archives_for_report, archives_to_exclude=found_java_archives
    )


if __name__ == "__main__":
    main()