zhaolei
57d89b3d05
A 结构层小缺口 (3/3): ✅ structure.py 加 LongFilename 告警 (>120 字符成员名) ✅ extractor.py _SAFE_NAME_RE 扩展 emoji 范围 (保留 emoji 文件名不崩) ✅ 重生成 L0-023_5layer_nested.zip (5 层真嵌套) B 白名单机制 (2/2): ✅ policy.yaml 加 keyword_whitelist (内网 IP/域名自动放行 RemoteDownloadUnix/PowerShell) ✅ policy.yaml 加 yara_hash_whitelist (MD5 在白名单跳过 YARA 告警) ✅ policy.py Property 类加两个 property 读取白名单 ✅ policy.py PolicyEngine.decide() 步骤 -1: 白名单过滤 (优先级最高) ✅ scanner.py decide() 调用补传 md5 外部资源缺口 (3 条，README 已记录): - L1-009/010 需真实签名 .cvd 文件 - L1-014 需压测脚本 stress_test.py 测试靶场: ✅ rules/test_rules.yar (7 条自动生成 YARA 规则) ✅ README 新增测试靶场待办 + 白名单机制说明 + 脚本清单
1402 lines
56 KiB
Python
1402 lines
56 KiB
Python
"""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）
|
||
safe_then_dangerous = (
|
||
filename.endswith(".exe") or filename.endswith(".dll") or filename.endswith(".scr")
|
||
or filename.endswith(".com") or filename.endswith(".pif") or filename.endswith(".cpl")
|
||
or filename.endswith(".msi") or filename.endswith(".lnk") or filename.endswith(".chm")
|
||
or filename.endswith(".bat") or filename.endswith(".cmd") or filename.endswith(".ps1")
|
||
or filename.endswith(".vbs") or filename.endswith(".hta")
|
||
or filename.endswith(".elf") or filename.endswith(".so")
|
||
or filename.endswith(".desktop") or filename.endswith(".run") or filename.endswith(".bin")
|
||
or filename.endswith(".deb") or filename.endswith(".appimage")
|
||
)
|
||
safe_exts = {".txt", ".pdf", ".jpg", ".jpeg", ".png", ".gif", ".bmp", ".doc", ".docx",
|
||
".xls", ".xlsx", ".ppt", ".pptx", ".rtf", ".csv", ".xml", ".html", ".htm",
|
||
".zip", ".rar", ".7z", ".wps", ".et", ".dps", ".wpt", ".ett", ".dpt"}
|
||
|
||
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
|
||
|
||
# 真实类型危险可执行，声明安全类型 → 伪装
|
||
from app.engines.structure import 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
|
||
|
||
|
||
def _has_double_ext(filename: str) -> bool:
|
||
"""判断文件名是否包含两个以上扩展名。"""
|
||
name = os.path.basename(filename)
|
||
parts = name.split(".")
|
||
if len(parts) < 3:
|
||
return False
|
||
prefix_ext = "." + parts[-2].lower()
|
||
suffix_ext = "." + parts[-1].lower()
|
||
if suffix_ext not in SUSPICIOUS_DOUBLE_EXT:
|
||
return False
|
||
return prefix_ext in DANGEROUS_EXTS or prefix_ext in SUSPICIOUS_DOUBLE_EXT
|
||
|
||
|
||
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")
|