MeshSate/app/engines/structure.py

"""L0 结构与完整性检测引擎。

职责：
- 真实文件类型识别（不信任扩展名）
- 扩展名伪装、双扩展名检测
- 危险可执行扩展名识别
- 压缩包结构安全预检（压缩比、成员数量、路径穿越）
"""

from __future__ import annotations

import os
import tarfile
import zipfile
from typing import Any, Dict, List

from app.engines.base import Engine, EngineResult, Finding, ScanTarget, Severity

# 高危可执行/脚本扩展名。这些类型即使内容"干净"，在内网被打开即可能造成危害。
# 包含跨平台 + Windows 专属 + Linux 专属三类（见总体方案 §4.8.3）
DANGEROUS_EXTS = {
    # === Windows 专属 ===
    ".exe", ".dll", ".scr", ".com", ".pif", ".cpl", ".msi", ".msp", ".msc",
    ".bat", ".cmd", ".com2", ".ps1", ".psm1", ".vbs", ".vbe", ".js", ".jse",
    ".wsf", ".wsh", ".hta", ".reg", ".chm", ".lnk", ".gadget", ".inf", ".scpt",
    # === Linux 专属 ===
    ".elf", ".so", ".desktop", ".appimage", ".flatpak", ".snap", ".deb", ".rpm",
    ".run", ".bin",
    # === 跨平台脚本 ===
    ".jar", ".apk", ".app", ".iso",
    ".sh", ".bash", ".zsh", ".py", ".pyc", ".pl", ".rb", ".php", ".jsp", ".asp", ".aspx",
}

# 常见文档扩展名，用于识别"文档里藏可执行文件"的场景
DOCUMENT_EXTS = {
    ".doc", ".docx", ".docm", ".dot", ".dotm", ".rtf", ".odt",
    ".xls", ".xlsx", ".xlsm", ".xlsb", ".xlt", ".xltm", ".ods",
    ".ppt", ".pptx", ".pptm", ".pot", ".potm", ".odp",
    ".pdf", ".wps", ".et", ".dps",
}

# 归档扩展名
ARCHIVE_EXTS = {
    ".zip", ".rar", ".7z", ".tar", ".gz", ".bz2", ".xz", ".tgz", ".cab",
    ".iso", ".jar", ".war", ".apk", ".apks", ".xpi",
}

# 双扩展名检查：前缀扩展名落在危险集合中即判定为伪装
SUSPICIOUS_DOUBLE_EXT = {
    ".pdf", ".doc", ".docx", ".xls", ".xlsx", ".ppt", ".pptx", ".txt", ".jpg",
    ".png", ".zip", ".rar", ".csv",
}

# 允许在归档中出现的无害扩展名（用于评估"归档内含可执行"的风险）
BENIGN_IN_ARCHIVE = {
    ".txt", ".md", ".csv", ".json", ".xml", ".html", ".htm", ".pdf", ".jpg",
    ".jpeg", ".png", ".gif", ".bmp", ".doc", ".docx", ".xls", ".xlsx", ".ppt",
    ".pptx", ".mp3", ".mp4", ".avi", ".wav",
}


class StructureEngine(Engine):
    """结构层检测引擎。"""

    name = "structure"
    layer = "L0"

    def scan(self, target: ScanTarget, ctx: Any) -> EngineResult:
        result = EngineResult(engine=self.name)
        findings: List[Finding] = []

        ext = (target.declared_ext or "").lower()
        real_type, real_ext, mime = identify_type(target.path)
        result.meta.update(
            {
                "real_type": real_type,
                "real_ext": real_ext,
                "mime": mime,
                "size": os.path.getsize(target.path),
                "declared_ext": ext,
            }
        )

        # 1) 扩展名与真实类型不一致
        if ext and real_ext and ext not in _compatible_exts(real_ext, real_type):
            findings.append(
                Finding(
                    engine=self.name,
                    layer=self.layer,
                    severity=Severity.HIGH.value,
                    rule="ExtensionMismatch",
                    detail=f"扩展名声明为 {ext}，实际内容为 {real_type}（{real_ext}）",
                    weight=30.0,
                )
            )

        # 2) 双扩展名伪装
        if _has_double_ext(target.filename):
            findings.append(
                Finding(
                    engine=self.name,
                    layer=self.layer,
                    severity=Severity.HIGH.value,
                    rule="DoubleExtension",
                    detail=f"文件名包含多个扩展名：{target.filename}",
                    weight=35.0,
                )
            )

        # 3) 危险可执行扩展名（顶层文件）
        if ext in DANGEROUS_EXTS and not target.is_extracted:
            severity = Severity.CRITICAL.value if ext in {".exe", ".scr", ".dll", ".com", ".pif"} else Severity.HIGH.value
            findings.append(
                Finding(
                    engine=self.name,
                    layer=self.layer,
                    severity=severity,
                    rule="DangerousExecutableExtension",
                    detail=f"文件类型为可执行/脚本格式：{ext}",
                    weight=45.0 if severity == Severity.CRITICAL.value else 30.0,
                )
            )

        # 4) 归档结构预检
        if ext in ARCHIVE_EXTS or real_ext in ARCHIVE_EXTS:
            findings.extend(self._inspect_archive(target, ctx))

        # 4b) 畸形文件启发式（P1）——结构完整性校验，仅降权不零容忍
        findings.extend(self._check_malformed(target, real_type))

        # 4c) 伪装扩展名 & 双扩展名
        findings.extend(self._check_extension_mismatch(target, real_type, real_ext))

        # 5) LNK 快捷方式深度解析（Windows 专属，Linux 纵深防御）
        if ext == ".lnk" or real_ext == ".lnk":
            findings.extend(self._inspect_lnk(target))

        # 6) ISO 挂载递归扫描（IsoWithExecutable 零容忍规则）
        if ext == ".iso" or real_ext == ".iso":
            findings.extend(self._inspect_iso(target))

        # 7) CHM 帮助文件结构校验（ChmWithScript 零容忍规则）
        if ext == ".chm" or real_ext == ".chm":
            findings.extend(self._inspect_chm(target))

        # 8) Linux 专属零容忍检测
        findings.extend(self._check_linux_threats(target, real_type, real_ext))

        result.findings = findings
        return result

    # ------------------------------------------------------------------ LNK 解析

    def _inspect_lnk(self, target: ScanTarget) -> List[Finding]:
        """深度解析 .lnk 快捷方式的目标路径、参数与远程引用。

        Windows 专属载体，在内网麒麟/UOS 终端上不生效，但保持检测——
        文件可能通过其他路径流向运维人员的 Windows 工作站（纵深防御）。
        """
        import struct

        findings: List[Finding] = []
        path = target.path

        try:
            with open(path, "rb") as fh:
                data = fh.read(8192)
        except OSError:
            return findings

        if len(data) < 76:
            return findings

        # 1) Shell Link Header 校验
        header_size = struct.unpack_from("<I", data, 0)[0]
        if header_size != 76:
            return findings  # 不是合法 Shell Link

        # CLSID: 758F0230-BEEF-11D0-A285-00AA00304F65（CLSID_ShellLink）
        expected_clsid = bytes([
            0x00, 0x02, 0x14, 0x01, 0x00, 0x00, 0x00, 0x00,
            0xC0, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x46,
        ])
        clsid = data[4:20]
        if clsid != expected_clsid:
            return findings  # CLSID 不匹配，不是 Shell Link

        link_flags = struct.unpack_from("<I", data, 20)[0]
        offset = header_size

        # 2) LinkTargetIDList（可变，有 size 前缀）
        if offset + 2 <= len(data):
            idlist_size = struct.unpack_from("<H", data, offset)[0]
            if idlist_size > 0:
                offset += 2 + idlist_size

        # 3) LinkInfo（有 size 前缀；HasLinkInfo flag = bit 1）
        has_link_info = bool(link_flags & 0x00000001)
        target_path = ""
        arguments = ""

        if has_link_info and offset + 4 <= len(data):
            linkinfo_size = struct.unpack_from("<I", data, offset)[0]
            if linkinfo_size >= 4 and offset + linkinfo_size <= len(data):
                linkinfo = data[offset: offset + linkinfo_size]
                strings = self._extract_utf16_strings(linkinfo)
                # 优先选 UNC 或盘符开头的路径；否则选第一个可疑路径
                for s in strings:
                    lowered = s.lower()
                    # UNC 远程路径
                    if s.startswith("\\\\"):
                        target_path = s
                        break
                    # 盘符开头的本地路径
                    if len(s) >= 2 and s[1] == ":" and s[0].isalpha():
                        target_path = s
                        break
                if not target_path:
                    # 备选：任何以危险后缀结尾的字符串都记为 target_path
                    dangerous_suffixes = (".exe", ".cmd", ".ps1", ".vbs", ".bat", ".com",
                                         ".scr", ".hta", ".js", ".jar", ".msi", ".run", ".bin", ".sh", ".bash")
                    for s in strings:
                        if s.lower().endswith(dangerous_suffixes):
                            target_path = s
                            break

                offset += linkinfo_size

        # 4) Name / RelativePath / WorkingDir / Arguments（每个：2字节 length + UTF-16LE 内容）
        # LinkFlags bit 5 = HasArguments → Arguments 存在
        flags_table = [
            ("HasName", 0x00000004),      # bit 2
            ("HasRelativePath", 0x00000008),  # bit 3
            ("HasWorkingDir", 0x00000010),    # bit 4
            ("HasArguments", 0x00000020),     # bit 5
        ]

        for name, flag in flags_table:
            if not (link_flags & flag):
                continue
            if offset + 2 > len(data):
                break
            str_len = struct.unpack_from("<H", data, offset)[0]
            offset += 2
            byte_len = str_len * 2
            if offset + byte_len > len(data):
                break
            try:
                text = data[offset: offset + byte_len].decode("utf-16-le", errors="replace").rstrip("\x00")
                offset += byte_len + 2  # null terminator
            except Exception:  # noqa: BLE001
                break

            if name == "HasArguments" and text:
                arguments = text
            elif name == "HasRelativePath" and text and not target_path:
                target_path = text

        # 5) 安全判定
        dangerous_exts = (".exe", ".cmd", ".ps1", ".vbs", ".bat", ".com",
                          ".scr", ".hta", ".js", ".jar", ".msi", ".lnk", ".bin", ".run",
                          ".sh", ".bash")

        # 远程 UNC 引用
        if target_path.startswith("\\\\"):
            findings.append(
                Finding(
                    engine=self.name,
                    layer=self.layer,
                    severity=Severity.CRITICAL.value,
                    rule="LnkWithAutoExec",
                    detail=f"LNK 指向远程 UNC 路径：{target_path[:120]}",
                    weight=85.0,
                )
            )

        # 指向可执行文件
        if target_path.lower().endswith(dangerous_exts):
            findings.append(
                Finding(
                    engine=self.name,
                    layer=self.layer,
                    severity=Severity.CRITICAL.value,
                    rule="LnkWithAutoExec",
                    detail=f"LNK 指向可执行文件：{target_path[:120]}",
                    weight=85.0,
                )
            )

        # Arguments 含危险命令
        if arguments:
            dangerous_args = (
                ("cmd /c", "调用 cmd 执行命令"),
                ("powershell", "调用 PowerShell"),
                ("mshta", "调用 mshta 执行 HTML 应用"),
                ("rundll32", "调用 rundll32"),
                ("regsvr32", "调用 regsvr32"),
                ("wscript", "调用 WScript"),
                ("cscript", "调用 CScript"),
                ("certutil", "certutil -decode 下载"),
                ("bitsadmin", "BITS 下载"),
                ("wget", "下载"),
                ("curl", "下载"),
                ("-enc ", "PowerShell 编码命令"),
                ("-EncodedCommand", "PowerShell 编码命令"),
                ("/bin/sh", "Linux Shell 调用"),
            )
            lowered_args = arguments.lower()
            for trigger, desc in dangerous_args:
                if trigger.lower() in lowered_args:
                    findings.append(
                        Finding(
                            engine=self.name,
                            layer=self.layer,
                            severity=Severity.CRITICAL.value,
                            rule="LnkWithAutoExec",
                            detail=f"LNK Arguments 含危险命令（{desc}）：{arguments[:120]}",
                            weight=85.0,
                        )
                    )
                    break  # 一条足够触发零容忍

        return findings

    @staticmethod
    def _extract_utf16_strings(data: bytes) -> List[str]:
        """从二进制数据中提取所有可打印 UTF-16LE 字符串。"""
        strings: List[str] = []
        try:
            decoded = data.decode("utf-16-le", errors="replace")
        except Exception:  # noqa: BLE001
            return strings

        # 按 null 分割，过滤短/非 ASCII
        for s in decoded.split("\x00"):
            s = s.strip("\x00")
            if len(s) >= 4 and any(c.isprintable() for c in s):
                strings.append(s)
        return strings

    # ------------------------------------------------------------------ ISO 扫描

    def _inspect_iso(self, target: ScanTarget) -> List[Finding]:
        """ISO 9660 镜像目录树扫描——识别内部可疑可执行文件（触发 IsoWithExecutable）。

        不依赖 mount 或外部库，直接解析 ISO 9660 的 PVD + 根目录 + 子目录。
        在 Linux 生产环境中，提取器层（SafeExtractor）会用 mount -o loop 递归扫描，
        此处提供结构级快速识别——ISO 包含可执行类型文件时即触发零容忍。
        """
        import struct

        findings: List[Finding] = []
        path = target.path

        try:
            with open(path, "rb") as fh:
                data = fh.read(256 * 1024)  # 读前 256KB（足够覆盖 PVD + 多个目录）
        except OSError:
            return findings

        # ISO 9660 PVD 位于 sector 16，每个 sector 2048 字节
        # PVD 标识字节 0x01，其后 "CD001" 标识
        pvd_offset = 16 * 2048
        if pvd_offset + 8 > len(data):
            return findings

        if data[pvd_offset] != 0x01 or data[pvd_offset + 1: pvd_offset + 6] != b"CD001":
            return findings  # 不是合法 ISO 9660

        # PVD 偏移 156-164: 根目录 record 的 LBA + size
        root_lba = struct.unpack_from("<I", data, pvd_offset + 156)[0]
        root_size = struct.unpack_from("<I", data, pvd_offset + 164)[0]

        # 解析根目录 + 一层子目录（保守，避免深递归）
        found_exec = self._walk_iso_dirs(data, root_lba, max_depth=2, current_depth=0)
        if found_exec:
            findings.append(
                Finding(
                    engine=self.name,
                    layer=self.layer,
                    severity=Severity.CRITICAL.value,
                    rule="IsoWithExecutable",
                    detail=f"ISO 镜像包含 {len(found_exec)} 个可执行/脚本文件：{found_exec[:5]}",
                    weight=85.0,
                )
            )
        return findings

    def _walk_iso_dirs(
        self, data: bytes, lba: int, *, max_depth: int, current_depth: int
    ) -> List[str]:
        """递归遍历 ISO 目录树，收集可执行文件路径。"""
        import struct

        if current_depth >= max_depth:
            return []

        dir_offset = lba * 2048
        if dir_offset + 4 > len(data):
            return []

        found: List[str] = []
        pos = dir_offset

        while pos < len(data):
            entry_len = data[pos]
            if entry_len == 0:
                pos += 1  # padding
                continue
            if entry_len == 0xFF:
                break  # end of directory entries

            if pos + entry_len > len(data):
                break

            entry = data[pos: pos + entry_len]

            # Directory Record 结构：
            #   0  entry_len (1)
            #   1  ext_len (1)
            #   2-9  LBA (8, both endian)
            #   10-13  size (4, both endian)
            #   14-20  日期时间 (7)
            #   21  flags (1)
            #   22-23  interleave (2)
            #   24-25  volume_seq (2)
            #   26  name_len (1)
            #   27+ name (variable)
            if len(entry) < 28:
                pos += entry_len
                continue

            flags = entry[21]
            name_len = entry[26]
            lba_child = struct.unpack_from("<I", entry, 2)[0]
            size_child = struct.unpack_from("<I", entry, 10)[0]
            name_bytes = entry[27: 27 + name_len]

            # 过滤 . 和 ..
            try:
                name = name_bytes.decode("ascii", errors="replace").rstrip(";1")
                if name in (".", "..", ""):
                    pos += entry_len
                    continue
            except Exception:  # noqa: BLE001
                pos += entry_len
                continue

            is_dir = bool(flags & 0x02)
            ext = os.path.splitext(name)[1].lower()

            if is_dir:
                # 递归进入
                sub_found = self._walk_iso_dirs(
                    data, lba_child, max_depth=max_depth, current_depth=current_depth + 1
                )
                found.extend(f"{name}/{s}" for s in sub_found)
            else:
                # 检查扩展名是否为可执行/脚本类型
                if ext in {
                    ".exe", ".dll", ".scr", ".com", ".pif", ".cpl", ".msi",
                    ".bat", ".cmd", ".ps1", ".vbs", ".vbe", ".js", ".jse", ".hta", ".wsf",
                    ".elf", ".so", ".desktop", ".appimage", ".run", ".bin", ".deb", ".rpm",
                    ".jar", ".apk", ".app", ".iso",
                    ".sh", ".bash", ".zsh", ".py", ".pl", ".rb", ".php",
                }:
                    found.append(name)

            pos += entry_len

        return found

    # ------------------------------------------------------------------ CHM 扫描

    def _inspect_chm(self, target: ScanTarget) -> List[Finding]:
        """HTML Help (.chm) 文件结构校验——识别脚本/可执行内容（触发 ChmWithScript）。

        CHM 本质是一个 ITSF 容器，内部的 #SYSTEM 流定义了整个帮助系统的结构。
        重点检查：1) ITSF 头合法性；2) #SYSTEM 中是否有 .exe/.cmd/.vbs/.js/.hta 引用；
        3) 是否包含 ActiveX Object / Shell Script Object 等可执行 HTML 组件。
        """
        import struct
        import zlib

        findings: List[Finding] = []
        path = target.path

        try:
            with open(path, "rb") as fh:
                data = fh.read(256 * 1024)  # 前 256KB
        except OSError:
            return findings

        # ITSF 头: signature "ITSF" (4) + version (4) + unknown (4) + section_offset (4)
        if len(data) < 16 or data[:4] != b"ITSF":
            return findings

        # 尝试在数据中搜索脚本/可执行特征
        text_hits: List[str] = []

        # 直接搜索整个文件（CHM 通常只有几 MB）
        lowered = data.lower()

        # 1) ITSF section header + streams 枚举 → 找 .exe/.cmd/.vbs/.js/.hta
        executable_refs = [
            b".exe", b".cmd", b".bat", b".ps1", b".vbs", b".vbe",
            b".hta", b".js", b".jse", b".wsf", b".scr",
        ]
        for ref in executable_refs:
            pos = 0
            while True:
                idx = lowered.find(ref, pos)
                if idx < 0:
                    break
                # 取前后各 20 字节作为上下文
                start = max(0, idx - 20)
                end = min(len(data), idx + len(ref) + 20)
                context = data[start:end].decode("ascii", errors="replace").strip("\x00")
                text_hits.append(context)
                pos = idx + 1
                if len(text_hits) >= 5:
                    break
            if len(text_hits) >= 5:
                break

        # 2) ActiveX / Shell Script Object 引用
        ax_refs = [
            b"object. Shell", b"wscript.shell", b"scriptlet",
            b"mshtmbrowser", b"html.application", b"objshell.application",
        ]
        for ref in ax_refs:
            if ref in lowered:
                text_hits.append(f"ActiveX/Script 引用: {ref.decode('ascii', errors='replace')}")

        if text_hits:
            findings.append(
                Finding(
                    engine=self.name,
                    layer=self.layer,
                    severity=Severity.CRITICAL.value,
                    rule="ChmWithScript",
                    detail=f"CHM 帮助文件包含脚本/可执行引用：{'; '.join(text_hits[:3])}",
                    weight=80.0,
                )
            )

        # 3) 快速扫描：HTML 内容中是否包含 <script 或 eval( 动态执行
        try:
            ascii_text = lowered.decode("ascii", errors="ignore")
            if "<script" in ascii_text.lower() or "eval(" in ascii_text:
                if not text_hits:  # 已经有 hits 的话不重复报
                    findings.append(
                        Finding(
                            engine=self.name,
                            layer=self.layer,
                            severity=Severity.HIGH.value,
                            rule="ChmWithScript",
                            detail="CHM 内嵌 HTML 包含 <script> 或 eval() 动态执行",
                            weight=55.0,
                        )
                    )
        except Exception:  # noqa: BLE001
            pass

        return findings

    # ------------------------------------------------------------------ 畸形文件启发式 + 伪装扩展名

    def _check_malformed(self, target: ScanTarget, real_type: str) -> List[Finding]:
        """结构完整性校验——仅当损坏导致无法安全解析时触发 MalformedStructure。"""
        import zipfile as zf

        findings: List[Finding] = []
        path = target.path
        real_lower = real_type.lower()
        ext = (target.declared_ext or "").lower()

        # ZIP/OOXML 结构损坏（CRC 校验或 central directory）
        if real_lower.startswith("zip") or ext in {".docx", ".xlsx", ".pptx", ".xlsm", ".docm", ".pptm", ".zip", ".jar", ".apk"}:
            try:
                with zf.ZipFile(path, "r") as zfh:
                    bad = zfh.testzip()
                    if bad is not None:
                        findings.append(
                            Finding(
                                engine=self.name, layer=self.layer,
                                severity=Severity.MEDIUM.value, rule="MalformedStructure",
                                detail=f"ZIP/OOXML 包 CRC 校验失败，首条损坏条目: {bad}",
                                weight=25.0,
                            )
                        )
            except zf.BadZipFile as exc:
                findings.append(
                    Finding(
                        engine=self.name, layer=self.layer,
                        severity=Severity.MEDIUM.value, rule="MalformedStructure",
                        detail=f"ZIP/OOXML 包结构损坏: {exc}",
                        weight=25.0,
                    )
                )
            except Exception:  # noqa: BLE001
                pass

        # OLE 复合文档头部校验
        if real_lower.startswith("ole compound") or real_lower.startswith("msoffice"):
            try:
                with open(path, "rb") as fh:
                    head = fh.read(4096)
                ole_magic = b"\xd0\xcf\x11\xe0\xa1\xb1\x1a\xe1"
                if len(head) < 512:
                    findings.append(Finding(engine=self.name, layer=self.layer,
                        severity=Severity.MEDIUM.value, rule="MalformedStructure",
                        detail="OLE 复合文档过小（< 512B），结构异常", weight=20.0))
                elif head[:8] != ole_magic:
                    findings.append(Finding(engine=self.name, layer=self.layer,
                        severity=Severity.MEDIUM.value, rule="MalformedStructure",
                        detail="OLE 复合文档头部魔数异常（可能被篡改）", weight=20.0))
            except OSError:
                pass

        # PDF 尾部标记检查
        if real_lower.startswith("pdf") or ext == ".pdf":
            try:
                with open(path, "rb") as fh:
                    fh.seek(0, 2)
                    size = fh.tell()
                    tail_size = min(64 * 1024, max(size, 4096))
                    fh.seek(max(0, size - tail_size))
                    tail = fh.read(tail_size)
                if b"%%EOF" not in tail[-4096:]:
                    findings.append(Finding(engine=self.name, layer=self.layer,
                        severity=Severity.MEDIUM.value, rule="MalformedStructure",
                        detail="PDF 文件缺少 %%EOF 标记，可能被截断或损坏", weight=20.0))
                elif b"xref" not in tail and b"startxref" not in tail:
                    findings.append(Finding(engine=self.name, layer=self.layer,
                        severity=Severity.MEDIUM.value, rule="MalformedStructure",
                        detail="PDF 尾部缺少交叉引用表或 startxref 标记", weight=20.0))
            except OSError:
                pass

        return findings

    def _check_extension_mismatch(self, target: ScanTarget, real_type: str, real_ext: str) -> List[Finding]:
        """双扩展名伪装 + 声明扩展名与真实类型不一致检测。"""
        findings: List[Finding] = []
        ext = (target.declared_ext or "").lower()
        filename = (target.filename or "").lower()

        # 双扩展名：安全扩展名 + 危险扩展名结尾（如 invoice.pdf.exe）
        # 用 DANGEROUS_EXTS 做单一事实源，避免手工维护 endswith 链漏覆盖
        _dangerous_tuple = tuple(sorted(DANGEROUS_EXTS, key=len, reverse=True))
        safe_then_dangerous = filename.endswith(_dangerous_tuple)
        safe_exts = _SAFE_PREFIX_EXTS

        if safe_then_dangerous:
            # 取最后两个扩展名
            parts = filename.rsplit(".", 2)
            if len(parts) >= 3:
                tail2 = "." + parts[-2] + "." + parts[-1]
                first_ext = "." + parts[-2]
                if first_ext in safe_exts:
                    findings.append(
                        Finding(
                            engine=self.name, layer=self.layer,
                            severity=Severity.CRITICAL.value, rule="DoubleExtension",
                            detail=f"双扩展名伪装：{target.filename}",
                            weight=90.0,
                        )
                    )

        # 声明扩展名 vs 真实类型不一致
        if real_ext and ext and ext != real_ext:
            # 允许：docx→zip, xlsx→zip, pptx→zip, xlsm→zip 等 OOXML 本质是 zip
            ooxml_as_zip = {(ext, ".zip") for ext in (".docx", ".xlsx", ".xlsm", ".pptx", ".pptm", ".docm")}
            if (ext, real_ext) in ooxml_as_zip:
                return findings

            # 真实类型危险可执行，声明安全类型 → 伪装
            # DANGEROUS_EXTS 是模块级常量，此处直接引用
            if real_ext in DANGEROUS_EXTS and ext not in DANGEROUS_EXTS and ext:
                findings.append(
                    Finding(
                        engine=self.name, layer=self.layer,
                        severity=Severity.CRITICAL.value, rule="ExtensionMismatch",
                        detail=f"伪装扩展名：声明 {ext}，真实 {real_ext}",
                        weight=85.0,
                    )
                )

        return findings

    def _check_linux_threats(self, target: ScanTarget, real_type: str, real_ext: str) -> List[Finding]:
        """Linux 专属威胁零容忍检测。"""
        findings: List[Finding] = []
        path = target.path
        ext = (target.declared_ext or "").lower()
        target_depth = 0 if target.depth == 0 else 1

        # 5a) ELF 原生二进制
        if real_ext in {".elf", ".so"} or real_type.startswith("ELF"):
            findings.append(
                Finding(
                    engine=self.name,
                    layer=self.layer,
                    severity=Severity.CRITICAL.value,
                    rule="LinuxExecutable",
                    detail=f"检测到 Linux ELF 原生可执行文件：{target.filename}",
                    weight=85.0,
                )
            )

        # 5b) .desktop 文件 Exec 字段危险命令检查
        if ext == ".desktop" or real_ext == ".desktop":
            desktop_findings = self._inspect_desktop(path, target.filename)
            findings.extend(desktop_findings)

        # 5c) Shell / Python / Perl / Ruby 脚本危险函数检查
        script_exts = {".sh", ".bash", ".zsh", ".py", ".pl", ".rb"}
        if ext in script_exts:
            script_findings = self._inspect_script(path, target.filename)
            findings.extend(script_findings)

        return findings

    def _inspect_desktop(self, path: str, filename: str) -> List[Finding]:
        """解析 .desktop 文件的 Exec 字段，检测危险命令。"""
        import re

        findings: List[Finding] = []
        try:
            content = _safe_read_text(path, max_bytes=8192)
        except Exception:  # noqa: BLE001
            return findings

        # 提取 Exec= 字段
        exec_pattern = re.compile(r"^Exec\s*=\s*(.+)$", re.MULTILINE)
        matches = exec_pattern.findall(content)
        if not matches:
            return findings

        dangerous_patterns = [
            (r"/bin/sh\s+-c\s+[\"']", "Shell 调用 /bin/sh -c 执行任意命令"),
            (r"/bin/bash\s+-c", "Shell 调用 /bin/bash -c 执行任意命令"),
            (r"wget\s+", "Shell 包含 wget 下载命令"),
            (r"curl\s+.*\|\s*(sh|bash|zsh)", "Shell 通过 curl 管道直接执行远程脚本"),
            (r"curl\s+-o\s+\S+\s+\S+", "Shell 用 curl 下载文件"),
            (r">?\s*/dev/tcp/", "Shell 包含 /dev/tcp 反弹 Shell"),
            (r"bash\s+-i\s+>&\s*/dev/tcp/", "Shell 包含 bash -i 反弹 Shell"),
            (r"nc\s+.*-e", "Shell 包含 nc -e 反弹 Shell"),
            (r"python[23]?\s+-c", "Shell 调用 python -c 执行任意代码"),
        ]

        for exec_cmd in matches:
            matched_types = set()
            for pattern, description in dangerous_patterns:
                if re.search(pattern, exec_cmd):
                    finding_type = "critical" if any(
                        kw in description for kw in ("反弹 Shell", "反弹 shell", "/dev/tcp", "反弹")
                    ) else "high"
                    # 去重：同一文件同一规则同一 Exec 只报一次最严重的
                    if description not in matched_types:
                        matched_types.add(description)
                        findings.append(
                            Finding(
                                engine=self.name,
                                layer=self.layer,
                                severity=Severity.CRITICAL.value,
                                rule="DesktopFileAutoExec",
                                detail=f".desktop 文件 Exec 字段包含危险命令（{description}）：{exec_cmd[:120]}",
                                weight=85.0,
                            )
                        )

        return findings

    def _inspect_script(self, path: str, filename: str) -> List[Finding]:
        """检查 Shell / Python 等脚本中的危险函数和命令。"""
        import re

        findings: List[Finding] = []
        try:
            content = _safe_read_text(path, max_bytes=65536)
        except Exception:  # noqa: BLE001
            return findings

        # 反弹 Shell 模式（CRITICAL）
        reverse_shell_patterns = [
            (r"bash\s+-i\s+>&\s*/dev/tcp/\S+/\d+\s+0>&1", "bash -i 反弹 Shell"),
            (r"/bin/bash\s+-i", "bash -i 反弹 Shell"),
            (r">?\s*/dev/tcp/\S+/\d+", "/dev/tcp 反弹 Shell"),
            (r"nc\s+\S+\s+\d+\s+-e", "netcat -e 反弹 Shell"),
            (r"python[23]?\s+-c\s+['\"].*socket.*connect", "Python socket.connect 反弹 Shell"),
        ]

        for pattern, description in reverse_shell_patterns:
            if re.search(pattern, content, re.IGNORECASE):
                findings.append(
                    Finding(
                        engine=self.name,
                        layer=self.layer,
                        severity=Severity.CRITICAL.value,
                        rule="ShellScriptDangerous",
                        detail=f"脚本包含反弹 Shell 特征（{description}）",
                        weight=85.0,
                    )
                )
                return findings  # 反弹 Shell 已足够触发零容忍

        # 高危系统调用模式（HIGH）
        system_call_patterns = [
            (r"\bsystem\s*\(\s*[\"']", "脚本包含 system() 执行系统命令"),
            (r"\bexec\s*\(\s*[\"']", "脚本包含 exec() 执行命令"),
            (r"\beval\s*\(\s*[\"']", "脚本包含 eval() 动态执行"),
            (r"subprocess\.call\s*\(\s*\[", "Python subprocess 执行外部程序"),
            (r"os\.system\s*\(", "Python os.system 执行系统命令"),
            (r"Shell\s+True", "Python subprocess shell=True 危险参数"),
            (r"wget\s+.*\|\s*(sh|bash)", "脚本 wget 下载并管道执行"),
            (r"curl\s+.*\|\s*(sh|bash)", "脚本 curl 下载并管道执行"),
            (r"curl\s+-o\s+\S+", "脚本 curl 下载文件到本地"),
            (r"wget\s+", "脚本包含 wget 下载"),
            (r"cron|crontab", "脚本包含 cron 持久化"),
            (r"systemctl\s+enable", "脚本尝试 systemctl enable 持久化"),
        ]

        for pattern, description in system_call_patterns:
            if re.search(pattern, content, re.IGNORECASE):
                findings.append(
                    Finding(
                        engine=self.name,
                        layer=self.layer,
                        severity=Severity.HIGH.value,
                        rule="ShellScriptDangerous",
                        detail=f"脚本包含危险函数（{description}）",
                        weight=65.0,
                    )
                )

        return findings

    def _inspect_archive(self, target: ScanTarget, ctx: Any) -> List[Finding]:
        """归档安全预检：压缩比炸弹、路径穿越、成员数量异常、内含可执行。"""
        findings: List[Finding] = []
        max_ratio = ctx.settings.max_extract_ratio
        max_total = ctx.settings.max_extract_total_bytes

        try:
            if zipfile.is_zipfile(target.path):
                findings.extend(self._inspect_zip(target, max_ratio, max_total))
            elif tarfile.is_tarfile(target.path):
                findings.extend(self._inspect_tar(target, max_ratio, max_total))
        except Exception:  # noqa: BLE001
            # 结构损坏的归档由解包器层报告，这里不重复告警
            return findings

        return findings

    def _inspect_zip(self, target: ScanTarget, max_ratio: int, max_total: int) -> List[Finding]:
        findings: List[Finding] = []
        total_compressed = 0
        total_uncompressed = 0
        executable_members: List[str] = []
        traversal_members: List[str] = []
        long_filename_members: List[str] = []
        member_count = 0
        encrypted_count = 0

        try:
            with zipfile.ZipFile(target.path) as zf:
                for info in zf.infolist():
                    member_count += 1
                    total_compressed += info.compress_size
                    total_uncompressed += info.file_size
                    name = info.filename
                    ext = os.path.splitext(name)[1].lower()
                    if ext in DANGEROUS_EXTS:
                        executable_members.append(name)
                    if _is_path_traversal(name):
                        traversal_members.append(name)
                    if info.flag_bits & 0x1:  # ZIP 加密标志（bit 0）
                        encrypted_count += 1
                    if len(os.path.basename(name)) > 120:
                        long_filename_members.append(name)
        except Exception as exc:  # noqa: BLE001
            return [
                Finding(
                    engine=self.name,
                    layer=self.layer,
                    severity=Severity.MEDIUM.value,
                    rule="ArchiveUnreadable",
                    detail=f"归档结构异常，无法解析：{exc}",
                    weight=10.0,
                )
            ]

        if member_count > 10000:
            findings.append(
                Finding(
                    engine=self.name,
                    layer=self.layer,
                    severity=Severity.MEDIUM.value,
                    rule="ArchiveTooManyMembers",
                    detail=f"归档成员数量异常：{member_count}",
                    weight=15.0,
                )
            )

        if total_uncompressed > max_total:
            findings.append(
                Finding(
                    engine=self.name,
                    layer=self.layer,
                    severity=Severity.HIGH.value,
                    rule="ArchiveTooLarge",
                    detail=f"解压后总大小 {total_uncompressed} 字节，超过上限 {max_total} 字节",
                    weight=30.0,
                )
            )
        elif total_compressed > 0 and total_uncompressed / total_compressed > max_ratio:
            ratio = total_uncompressed / total_compressed
            findings.append(
                Finding(
                    engine=self.name,
                    layer=self.layer,
                    severity=Severity.HIGH.value,
                    rule="ArchiveBombSuspect",
                    detail=f"压缩比 {ratio:.1f}:1 超过阈值 {max_ratio}:1，疑似压缩炸弹",
                    weight=35.0,
                )
            )

        if traversal_members:
            findings.append(
                Finding(
                    engine=self.name,
                    layer=self.layer,
                    severity=Severity.CRITICAL.value,
                    rule="PathTraversal",
                    detail=f"归档内含路径穿越条目：{', '.join(traversal_members[:5])}",
                    weight=50.0,
                )
            )

        if executable_members:
            findings.append(
                Finding(
                    engine=self.name,
                    layer=self.layer,
                    severity=Severity.HIGH.value,
                    rule="ArchiveContainsExecutable",
                    detail=f"归档内含可执行文件 {len(executable_members)} 个：{', '.join(executable_members[:5])}",
                    weight=30.0,
                )
            )

        if encrypted_count > 0:
            findings.append(
                Finding(
                    engine=self.name,
                    layer=self.layer,
                    severity=Severity.MEDIUM.value,
                    rule="EncryptedArchive",
                    detail=f"归档内含加密成员 {encrypted_count} 个（bit0 flag 置位），无法扫描内部内容",
                    weight=20.0,
                )
            )

        if long_filename_members:
            findings.append(
                Finding(
                    engine=self.name,
                    layer=self.layer,
                    severity=Severity.LOW.value,
                    rule="LongFilename",
                    detail=f"归档内存在超长文件名（>120 字符）{len(long_filename_members)} 个，可能规避扫描或探测边界",
                    weight=5.0,
                )
            )

        return findings

    def _inspect_tar(self, target: ScanTarget, max_ratio: int, max_total: int) -> List[Finding]:
        findings: List[Finding] = []
        total_uncompressed = 0
        executable_members: List[str] = []
        traversal_members: List[str] = []
        member_count = 0

        try:
            with tarfile.open(target.path) as tf:
                for member in tf.getmembers():
                    member_count += 1
                    total_uncompressed += member.size
                    ext = os.path.splitext(member.name)[1].lower()
                    if ext in DANGEROUS_EXTS:
                        executable_members.append(member.name)
                    if _is_path_traversal(member.name):
                        traversal_members.append(member.name)
                    if member.issym() or member.islnk():
                        traversal_members.append(member.name)
        except Exception as exc:  # noqa: BLE001
            return [
                Finding(
                    engine=self.name,
                    layer=self.layer,
                    severity=Severity.MEDIUM.value,
                    rule="ArchiveUnreadable",
                    detail=f"归档结构异常，无法解析：{exc}",
                    weight=10.0,
                )
            ]

        if total_uncompressed > max_total:
            findings.append(
                Finding(
                    engine=self.name,
                    layer=self.layer,
                    severity=Severity.HIGH.value,
                    rule="ArchiveTooLarge",
                    detail=f"解压后总大小 {total_uncompressed} 字节，超过上限",
                    weight=30.0,
                )
            )
        if traversal_members:
            findings.append(
                Finding(
                    engine=self.name,
                    layer=self.layer,
                    severity=Severity.CRITICAL.value,
                    rule="PathTraversal",
                    detail=f"归档内含路径穿越或链接条目：{', '.join(traversal_members[:5])}",
                    weight=50.0,
                )
            )
        if executable_members:
            findings.append(
                Finding(
                    engine=self.name,
                    layer=self.layer,
                    severity=Severity.HIGH.value,
                    rule="ArchiveContainsExecutable",
                    detail=f"归档内含可执行文件 {len(executable_members)} 个：{', '.join(executable_members[:5])}",
                    weight=30.0,
                )
            )
        return findings


def _is_path_traversal(name: str) -> bool:
    """判断归档成员名是否试图跳出解包目录。"""
    normalized = name.replace("\\", "/")
    if normalized.startswith("/") or (len(normalized) > 1 and normalized[1] == ":"):
        return True
    parts = normalized.split("/")
    return ".." in parts


# 安全/文档类扩展名集合——双扩展名伪装中作为"前缀"出现
_SAFE_PREFIX_EXTS = DOCUMENT_EXTS | {
    ".txt", ".log", ".csv", ".json", ".xml", ".html", ".htm",
    ".jpg", ".jpeg", ".png", ".gif", ".bmp", ".tiff", ".svg",
    ".zip", ".rar", ".7z", ".tar", ".gz", ".bz2",
    ".wps", ".et", ".dps", ".wpt", ".ett", ".dpt", ".rtf", ".odt",
}


def _has_double_ext(filename: str) -> bool:
    """判断文件名是否为双扩展名伪装。

    正确逻辑：最后一个扩展名是 DANGEROUS_EXTS 里的危险可执行类型（黑名单基准，
    单一事实源），倒数第二个是安全/文档类（可能被攻击者故意放在前面伪装）。
    例如 invoice.docx.js → .js 危险 + .docx 安全 → True。
    """
    name = os.path.basename(filename)
    parts = name.split(".")
    if len(parts) < 3:
        return False
    prefix_ext = "." + parts[-2].lower()
    suffix_ext = "." + parts[-1].lower()
    # 核心：以 DANGEROUS_EXTS 为单一事实源
    if suffix_ext not in DANGEROUS_EXTS:
        return False
    # 前缀必须是安全/文档类，才算"伪装"
    return prefix_ext in _SAFE_PREFIX_EXTS


def _compatible_exts(real_ext: str, real_type: str) -> set[str]:
    """返回与该真实类型相容的扩展名集合。

    特别处理：libmagic 把所有纯文本脚本（.py/.js/.sh/.php/.jsp/.aspx/.eml/.desktop/.vbs/.pl/.rb/.ps1 等）
    统一识别成 text/plain → real_ext='.txt'，但这些扩展名合法地属于纯文本格式，
    不能因为 real_ext='.txt' 就判"扩展名伪装"。所以当 base='.txt' 时，也把所有常见脚本组合并进来。
    """
    base = real_ext.lower()
    groups: List[set[str]] = [
        {".doc", ".docx", ".docm", ".dot", ".dotm", ".rtf", ".odt"},
        {".xls", ".xlsx", ".xlsm", ".xlsb", ".xlt", ".xltm", ".ods"},
        {".ppt", ".pptx", ".pptm", ".pot", ".potm", ".odp"},
        {".zip", ".jar", ".war", ".apk", ".odt", ".ods", ".odp", ".epub", ".docx", ".xlsx", ".pptx"},
        {".tar", ".gz", ".tgz", ".bz2", ".xz"},
        {".txt", ".log", ".csv", ".ini", ".conf", ".yaml", ".yml", ".json", ".xml", ".md", ".markdown"},
        {".htm", ".html"},
        {".jpg", ".jpeg", ".jpe"},
        {".exe", ".dll", ".sys", ".ocx", ".cpl", ".scr"},
        {".py", ".pyc", ".pyo"},
        {".js", ".mjs"},
        {".png", ".apng"},
        {".gif"},
        {".bmp"},
        {".mp3"},
        {".mp4", ".m4v", ".avi", ".mkv", ".mov"},
        {".pdf"},
        {".rtf"},
        {".ps1", ".psm1"},
        {".bat", ".cmd"},
        {".vbs", ".vbe"},
        {".sh", ".bash"},
        {".cab"},
        {".iso"},
        {".rar"},
        {".7z"},
    ]
    compatible: set[str] = {base}
    for group in groups:
        if base in group:
            compatible |= group
            # ⚠️ 关键：当 base='.txt'（纯文本）或 '.html'（Web 文本）时，
            # 额外合并所有脚本/标记/邮件组。因为 libmagic 对纯文本内容
            # 可能识别成 text/plain 或 text/html，但 .py/.js/.aspx/.jsp/.php
            # 等都是合法的纯文本/Web 脚本，不能因此判"扩展名伪装"。
            if base in (".txt", ".html"):
                for extra in [
                    {".htm", ".html"},
                    {".py", ".pyc", ".pyo"},
                    {".js", ".mjs"},
                    {".ps1", ".psm1"},
                    {".bat", ".cmd"},
                    {".vbs", ".vbe"},
                    {".sh", ".bash"},
                    {".rb"},
                    {".pl"},
                    {".php", ".php3", ".php4", ".php5", ".phtml"},
                    {".jsp"},
                    {".aspx", ".ashx", ".asmx"},
                    {".asp"},
                    {".cgi"},
                    {".bas", ".vba"},
                    {".desktop"},
                    {".eml", ".msg"},
                    {".mib"},
                    {".yaml", ".yml"},
                    {".xml"},
                    {".markdown", ".md"},
                ]:
                    compatible |= extra
            break
    return compatible


def identify_type(path: str) -> tuple[str, str, str]:
    """识别文件真实类型，返回 (描述, 扩展名, MIME)。

    优先使用 libmagic；不可用时退回到基于文件头魔数的内置识别。
    无论哪条路径，只要判定为 ZIP，都进一步检查内部结构来区分
    OOXML（docx/xlsx/pptx）、EPUB、JAR、APK 和普通 ZIP。

    ext 字段**绝对不能**用 os.path.splitext(desc) 从 libmagic 的文字描述里
    截取——描述里常带 "version 1.4"、"v2.0"、"last modified Oct 27" 等，
    splitext 会把 ".4"、".0"、".27" 误当成扩展名。正确做法是从 MIME 反推 ext。
    """
    # ---- 1) 先拿到一个初始判断 ----
    try:
        import magic

        mime = magic.from_file(path, mime=True)
        desc = magic.from_file(path)
        # ⚠️ 关键修复：从 MIME 反推 ext，绝不用 splitext(desc)
        ext = _mime_to_ext(mime)
    except Exception:  # noqa: BLE001
        desc, ext, mime = _fallback_identify(path)

    # ---- 2) 只要判定为 ZIP（或 MIME 是 zip/OOXML 相关），就进一步探测内部结构 ----
    #    libmagic 对所有 OOXML 都报 "Zip archive data"，必须手动区分
    if ext == ".zip" or "zip" in mime.lower() or "zip" in desc.lower():
        refined = _refine_zip_type(path)
        if refined is not None:
            return refined

    return desc, ext, mime


def _mime_to_ext(mime: str) -> str:
    """从 libmagic 返回的 MIME 反推文件扩展名（带点、小写）。

    libmagic 的 MIME 值是可靠的，但描述文本里常带版本号/日期，
    不能直接用 os.path.splitext(desc)。
    """
    if not mime:
        return ""
    # 精确匹配优先
    mapping = {
        "application/pdf": ".pdf",
        "application/zip": ".zip",
        "application/x-zip": ".zip",
        "application/x-7z-compressed": ".7z",
        "application/x-rar": ".rar",
        "application/x-tar": ".tar",
        "application/gzip": ".gz",
        "application/x-bzip2": ".bz2",
        "application/x-xz": ".xz",
        "application/x-iso9660-image": ".iso",
        "application/vnd.ms-cab-compressed": ".cab",
        "application/x-msdownload": ".exe",
        "application/x-dosexec": ".exe",
        "application/vnd.microsoft.portable-executable": ".exe",
        "application/x-sharedlib": ".so",
        "application/x-executable": ".elf",
        "application/x-elf": ".elf",
        "application/vnd.android.package-archive": ".apk",
        "application/java-archive": ".jar",
        "application/epub+zip": ".epub",
        "application/msword": ".doc",
        "application/x-ole-storage": ".doc",
        "application/vnd.ms-excel": ".xls",
        "application/vnd.ms-powerpoint": ".ppt",
        "application/encrypted": ".doc",
        "application/vnd.openxmlformats-officedocument.wordprocessingml.document": ".docx",
        "application/vnd.openxmlformats-officedocument.spreadsheetml.sheet": ".xlsx",
        "application/vnd.openxmlformats-officedocument.presentationml.presentation": ".pptx",
        "application/vnd.openxmlformats-officedocument.wordprocessingml.template": ".dotx",
        "application/vnd.openxmlformats-officedocument.presentationml.template": ".potx",
        "application/vnd.openxmlformats-officedocument.presentationml.slideshow": ".ppsx",
        "application/vnd.ms-office.chm": ".chm",
        "application/vnd.ms-htmlhelp": ".chm",
        "application/x-chm": ".chm",
        "application/vnd.oasis.opendocument.text": ".odt",
        "application/vnd.oasis.opendocument.spreadsheet": ".ods",
        "application/vnd.oasis.opendocument.presentation": ".odp",
        "application/x-sqlite3": ".sqlite",
        "application/x-lnk": ".lnk",
        "application/x-shockwave-flash": ".swf",
        "application/x-gzip": ".gz",
        "application/json": ".json",
        "application/xml": ".xml",
        "application/xhtml+xml": ".xhtml",
        "application/x-httpd-php": ".php",
        "text/html": ".html",
        "text/plain": ".txt",
        "text/csv": ".csv",
        "text/x-python": ".py",
        "text/x-shellscript": ".sh",
        "text/x-perl": ".pl",
        "image/jpeg": ".jpg",
        "image/png": ".png",
        "image/gif": ".gif",
        "image/bmp": ".bmp",
        "image/tiff": ".tiff",
        "image/svg+xml": ".svg",
        "image/webp": ".webp",
        "video/mp4": ".mp4",
        "video/quicktime": ".mov",
        "video/x-msvideo": ".avi",
        "video/x-matroska": ".mkv",
        "video/webm": ".webm",
        "audio/mpeg": ".mp3",
        "audio/wav": ".wav",
        "audio/flac": ".flac",
        "audio/ogg": ".ogg",
        "application/vnd.ms-outlook": ".msg",
        "message/rfc822": ".eml",
    }
    if mime in mapping:
        return mapping[mime]
    # 模糊匹配兜底
    if "pdf" in mime:
        return ".pdf"
    if "zip" in mime:
        return ".zip"
    if "rar" in mime:
        return ".rar"
    if "7z" in mime:
        return ".7z"
    if "elf" in mime or "executable" in mime:
        return ".elf"
    if "octet-stream" in mime:
        return ""  # 二进制流，未知类型
    return ""


def _refine_zip_type(path: str) -> Optional[tuple[str, str, str]]:
    """对被识别为 ZIP 的文件进一步检查内部结构，区分 OOXML/EPUB/JAR/APK。

    即使 zipfile 打不开（损坏的 ZIP），也返回短中文名 "ZIP 压缩包"。
    只有完全不是 ZIP 时才返回 None。
    """
    try:
        import zipfile

        with zipfile.ZipFile(path) as zf:
            names = set(zf.namelist())
    except Exception:  # noqa: BLE001
        # libmagic 说它是 ZIP，但 zipfile 打不开 → 损坏的 ZIP，还是按 ZIP 处理
        return ("ZIP 压缩包（已损坏）", ".zip", "application/zip")

    if "word/document.xml" in names and "[Content_Types].xml" in names:
        return (
            "Microsoft Word 文档（OOXML）",
            ".docx",
            "application/vnd.openxmlformats-officedocument.wordprocessingml.document",
        )
    if "xl/workbook.xml" in names and "[Content_Types].xml" in names:
        return (
            "Microsoft Excel 表格（OOXML）",
            ".xlsx",
            "application/vnd.openxmlformats-officedocument.spreadsheetml.sheet",
        )
    if "ppt/presentation.xml" in names and "[Content_Types].xml" in names:
        return (
            "Microsoft PowerPoint 演示（OOXML）",
            ".pptx",
            "application/vnd.openxmlformats-officedocument.presentationml.presentation",
        )
    if "mimetype" in names and any(n.startswith("META-INF/") for n in names):
        if any(n.endswith(".dex") for n in names) or "classes.dex" in names:
            return ("Android APK 安装包", ".apk", "application/vnd.android.package-archive")
        if any(n.endswith(".class") for n in names) or "classes/" in names:
            return ("Java JAR 包", ".jar", "application/java-archive")
        try:
            mt = zf.read("mimetype").decode("utf-8", errors="ignore")
            if mt.strip() == "application/epub+zip":
                return ("EPUB 电子书", ".epub", "application/epub+zip")
        except Exception:  # noqa: BLE001
            pass
    # 普通 ZIP —— 返回短中文名
    return ("ZIP 压缩包", ".zip", "application/zip")


# 内置魔数表，覆盖法院场景常见格式
_MAGIC_TABLE = [
    (b"MZ", ".exe", "DOS/Windows 可执行文件"),
    (b"\x7fELF", ".elf", "ELF 可执行文件"),
    (b"%PDF", ".pdf", "PDF 文档"),
    (b"\xd0\xcf\x11\xe0\xa1\xb1\x1a\xe1", ".doc", "Microsoft OLE2 复合文档"),
    (b"PK\x03\x04", ".zip", "ZIP 归档"),
    (b"Rar!\x1a\x07", ".rar", "RAR 归档"),
    (b"7z\xbc\xaf\x27\x1c", ".7z", "7-Zip 归档"),
    (b"\x1f\x8b", ".gz", "GZIP 压缩"),
    (b"BZh", ".bz2", "BZIP2 压缩"),
    (b"\xfd7zXZ", ".xz", "XZ 压缩"),
    (b"\x89PNG", ".png", "PNG 图像"),
    (b"\xff\xd8\xff", ".jpg", "JPEG 图像"),
    (b"GIF8", ".gif", "GIF 图像"),
    (b"BM", ".bmp", "BMP 图像"),
    (b"\x00\x00\x01\x00", ".ico", "ICO 图标"),
    (b"{\\rtf", ".rtf", "RTF 文档"),
    (b"\x25\x21PS", ".ps", "PostScript"),
    (b"ustar", ".tar", "TAR 归档"),
    (b"\xd0\xcf\x11\xe0", ".msi", "MSI 安装包"),
    (b"regf", ".dat", "Windows 注册表"),
    (b"SQLite format 3", ".sqlite", "SQLite 数据库"),
]


def _fallback_identify(path: str) -> tuple[str, str, str]:
    """基于魔数的备用识别（libmagic 不可用时）。"""
    try:
        with open(path, "rb") as fh:
            head = fh.read(512)
    except OSError:
        return "未知", "", "application/octet-stream"

    for magic_bytes, ext, desc in _MAGIC_TABLE:
        if head.startswith(magic_bytes):
            return desc, ext, "application/octet-stream"

    if head.startswith(b"\x50\x4b\x03\x04"):
        # 区分 docx/xlsx/pptx（OOXML）与普通 zip
        try:
            import zipfile

            with zipfile.ZipFile(path) as zf:
                names = set(zf.namelist())
            if "word/document.xml" in names:
                return "Microsoft Word 2007+", ".docx", "application/vnd.openxmlformats-officedocument.wordprocessingml.document"
            if "xl/workbook.xml" in names:
                return "Microsoft Excel 2007+", ".xlsx", "application/vnd.openxmlformats-officedocument.spreadsheetml.sheet"
            if "ppt/presentation.xml" in names:
                return "Microsoft PowerPoint 2007+", ".pptx", "application/vnd.openxmlformats-officedocument.presentationml.presentation"
        except Exception:  # noqa: BLE001
            pass
        return "ZIP 归档", ".zip", "application/zip"

    if not head:
        return "空文件", "", "application/x-empty"

    return "未知二进制", "", "application/octet-stream"


def _safe_read_text(path: str, *, max_bytes: int = 65536, encoding: str = "utf-8") -> str:
    """安全读取文件文本内容，做大小和编码保护。"""
    with open(path, "rb") as fh:
        head = fh.read(max_bytes + 1)
    if len(head) > max_bytes:
        raise ValueError(f"文件超过 {max_bytes} 字节，已截断")
    # 尝试 UTF-8，失败回退 GBK
    try:
        return head.decode(encoding)
    except UnicodeDecodeError:
        return head.decode("gbk", errors="replace")