zhaolei
5c776785b9
对账: 162/168 通过 (96.4%)，剩余 6 条为样本占位符问题 structure.py: - _check_zip_trailing_garbage(): 新增模块级函数检测 ZIP EOCD 尾部 多余字节（L0-026 corrupt_direntry.zip 伪装损坏模式） - _check_embedded_eicar(): 新增模块级函数，作为 ClamAV 只匹配精确 68B 纯 EICAR 的补充，对文本/代码文件扫描内嵌 EICAR 字节串 （L1-013 大文件尾部含 EICAR） - scan() 主流程在 4b/4c 后加 4d 调用 _check_embedded_eicar binary.py: - _manual_pe_analysis(): 新增模块级函数，pefile 解析失败时手工搜索 PE\\0\\0 签名并解析节表 Characteristics，检测 RWX 可写可执行节 （L5-004 假 PE，e_lfanew 偏移错误导致 pefile 报 Invalid NT Headers） - scan() 在 pefile 异常后 fallback 调用 _manual_pe_analysis bugfix: _check_zip_trailing_garbage 越界 crash (IndexError) - EOCD 被截断/comment_len 溢出时跳过，不抛异常 - 修复 NEG-017 4layer_nested.zip 误触发 engine_failure fail_closed
486 lines
18 KiB
Python
486 lines
18 KiB
Python
"""L2b YARA 规则与二进制分析引擎。
|
||
|
||
包含：
|
||
- YARA 规则匹配（规则库从 ``rules/`` 目录加载，支持热加载）
|
||
- PE 文件结构分析（pefile）：节区异常、导入表、编译信息、签名
|
||
- 可执行文件能力识别（capa，若已安装）
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import os
|
||
import threading
|
||
from typing import Any, Dict, List, Optional
|
||
|
||
from app.engines.base import Engine, EngineResult, Finding, ScanTarget, Severity
|
||
|
||
BINARY_EXTS = {".exe", ".dll", ".sys", ".scr", ".com", ".cpl", ".ocx", ".msi", ".apk", ".jar", ".so", ".bin", ".elf"}
|
||
|
||
|
||
# ---------- pefile 失败时的手工 PE 分析 fallback ----------
|
||
|
||
# PE Section Characteristics 常量（与 pefile.SECTION_CHARACTERISTICS 一致）
|
||
_PE_SCN_MEM_EXECUTE = 0x20000000
|
||
_PE_SCN_MEM_WRITE = 0x80000000
|
||
|
||
|
||
def _manual_pe_analysis(path: str) -> List[Finding]:
|
||
"""pefile 解析失败时，手工搜索 PE\\0\\0 签名并解析节表。
|
||
|
||
覆盖场景：假 PE / 伪装 PE / e_lfanew 偏移错误。pefile 严格校验
|
||
e_lfanew 指向的签名是否为 PE\\0\\0，但攻击者可以让 e_lfanew 偏移
|
||
错几个字节让 pefile 报错，同时文件里仍有真实的 PE\\0\\0 和节表。
|
||
"""
|
||
import struct
|
||
|
||
findings: List[Finding] = []
|
||
try:
|
||
with open(path, "rb") as fh:
|
||
data = fh.read()
|
||
except OSError:
|
||
return findings
|
||
|
||
# 1) MZ 头存在？（PE 文件必需）
|
||
if len(data) < 2 or data[:2] != b"MZ":
|
||
return findings # 根本不是 PE
|
||
|
||
# 2) 尝试从 e_lfanew 取 PE 偏移；不对就全局搜索 PE\0\0
|
||
pe_off = -1
|
||
if len(data) >= 0x40:
|
||
try:
|
||
e_lfanew = struct.unpack_from("<I", data, 0x3C)[0]
|
||
if e_lfanew + 4 <= len(data) and data[e_lfanew:e_lfanew + 4] == b"PE\x00\x00":
|
||
pe_off = e_lfanew
|
||
except struct.error:
|
||
pass
|
||
|
||
if pe_off < 0:
|
||
pe_off = data.find(b"PE\x00\x00")
|
||
if pe_off < 0:
|
||
return findings # 没找到 PE 签名
|
||
|
||
# 3) COFF header 在 PE\0\0 后 4 字节
|
||
coff_off = pe_off + 4
|
||
if coff_off + 24 > len(data):
|
||
return findings
|
||
num_sections = struct.unpack_from("<H", data, coff_off + 2)[0]
|
||
size_opt_header = struct.unpack_from("<H", data, coff_off + 16)[0]
|
||
section_table_off = coff_off + 20 + size_opt_header
|
||
|
||
if num_sections == 0 or num_sections > 96: # 合理上限
|
||
return findings
|
||
if section_table_off + num_sections * 40 > len(data):
|
||
return findings
|
||
|
||
# 4) 逐节扫描 RWX 属性
|
||
rwx_sections = []
|
||
for i in range(num_sections):
|
||
sec_off = section_table_off + i * 40
|
||
name = data[sec_off:sec_off + 8].rstrip(b"\x00")
|
||
characteristics = struct.unpack_from("<I", data, sec_off + 36)[0]
|
||
if (characteristics & _PE_SCN_MEM_EXECUTE) and (characteristics & _PE_SCN_MEM_WRITE):
|
||
rwx_sections.append(name.decode("utf-8", errors="replace"))
|
||
|
||
if rwx_sections:
|
||
findings.append(
|
||
Finding(
|
||
engine="binary", layer="L2B",
|
||
severity=Severity.HIGH.value, rule="RWXSectionDetected",
|
||
detail=f"手工 PE 分析检测到可写且可执行的危险节：{', '.join(rwx_sections)}",
|
||
weight=50.0,
|
||
)
|
||
)
|
||
|
||
return findings
|
||
|
||
|
||
# ---------- 模块级全局 YARA 规则单例 ----------
|
||
# 所有 YaraEngine 实例共享同一个编译结果，避免重复 compile
|
||
# rules_dir 不同时（极少见）才分叉加载
|
||
_YARA_GLOBAL: Dict[str, Dict[str, Any]] = {}
|
||
_YARA_LOCK = threading.Lock()
|
||
|
||
|
||
def _get_global_yara(rules_dir: str) -> Optional[Any]:
|
||
"""获取全局共享的 YARA Rules 对象（自动按目录 mtime 重载）。"""
|
||
with _YARA_LOCK:
|
||
cached = _YARA_GLOBAL.get(rules_dir)
|
||
import yara as _yr
|
||
newest = 0.0
|
||
files: List[str] = []
|
||
if os.path.isdir(rules_dir):
|
||
for name in sorted(os.listdir(rules_dir)):
|
||
if name.endswith((".yar", ".yara")):
|
||
path = os.path.join(rules_dir, name)
|
||
files.append(path)
|
||
newest = max(newest, os.path.getmtime(path))
|
||
|
||
if cached and cached.get("mtime") == newest:
|
||
return cached.get("rules")
|
||
|
||
if not files:
|
||
_YARA_GLOBAL[rules_dir] = {"rules": None, "mtime": newest}
|
||
return None
|
||
|
||
try:
|
||
compiled = _yr.compile(filepaths={os.path.basename(p): p for p in files})
|
||
except Exception:
|
||
_YARA_GLOBAL[rules_dir] = {"rules": None, "mtime": newest}
|
||
return None
|
||
|
||
_YARA_GLOBAL[rules_dir] = {"rules": compiled, "mtime": newest}
|
||
return compiled
|
||
|
||
|
||
def _discover_rules_dir() -> str:
|
||
"""自动探测 YARA 规则目录（优先环境变量 > /app/rules > 项目根 rules/）。"""
|
||
env_dir = os.environ.get("MESHSAFE_RULES_DIR", "").strip()
|
||
if env_dir and os.path.isdir(env_dir):
|
||
return env_dir
|
||
|
||
for candidate in ("/app/rules", "rules", "./rules", "../rules"):
|
||
if os.path.isdir(candidate):
|
||
return os.path.abspath(candidate)
|
||
|
||
return env_dir or "/app/rules"
|
||
|
||
|
||
class YaraEngine(Engine):
|
||
"""YARA 规则匹配引擎（使用全局共享的编译结果）。"""
|
||
|
||
name = "yara"
|
||
layer = "L2B"
|
||
|
||
def __init__(self, rules_dir: str | None = None) -> None:
|
||
# 显式传入时才用传入值，否则走自动探测
|
||
self.rules_dir = rules_dir if rules_dir else _discover_rules_dir()
|
||
|
||
def _load_rules(self) -> Any:
|
||
"""委托给全局单例。"""
|
||
try:
|
||
import yara # noqa: F401 — 只是探测可用性
|
||
except Exception: # noqa: BLE001
|
||
return None
|
||
return _get_global_yara(self.rules_dir)
|
||
|
||
def scan(self, target: ScanTarget, ctx: Any) -> EngineResult:
|
||
result = EngineResult(engine=self.name)
|
||
rules = self._load_rules()
|
||
if rules is None:
|
||
result.ok = False
|
||
result.error = "YARA 规则库不可用或编译失败"
|
||
return result
|
||
|
||
try:
|
||
matches = rules.match(target.path)
|
||
except Exception as exc: # noqa: BLE001
|
||
result.ok = False
|
||
result.error = f"YARA 匹配失败：{exc}"
|
||
return result
|
||
|
||
for match in matches:
|
||
severity, weight = _judge_yara_rule(match.rule, match.tags)
|
||
# 覆盖：如果 ctx 里的策略为该 YARA 规则定义了精细权重，用它
|
||
try:
|
||
policy = ctx.policy_engine.policy
|
||
yara_weights = getattr(policy, "yara_weights", {}) or {}
|
||
if match.rule in yara_weights:
|
||
weight = float(yara_weights[match.rule])
|
||
except Exception:
|
||
pass # ctx 没有 policy 字段时保持默认 weight
|
||
result.findings.append(
|
||
Finding(
|
||
engine=self.name,
|
||
layer=self.layer,
|
||
severity=severity,
|
||
rule=match.rule,
|
||
detail=f"YARA 规则 {match.rule} 命中（标签：{', '.join(match.tags) or '无'}）",
|
||
weight=weight,
|
||
meta={"tags": list(match.tags)},
|
||
)
|
||
)
|
||
# rule_set_version 用全局 compiled 对象的 id（每次 reload 变）
|
||
try:
|
||
_r = _get_global_yara(self.rules_dir)
|
||
result.meta["rule_set_version"] = str(id(_r)) if _r else "unavailable"
|
||
except Exception:
|
||
result.meta["rule_set_version"] = "unknown"
|
||
return result
|
||
|
||
def health(self) -> Dict[str, Any]:
|
||
rules = self._load_rules()
|
||
ok = rules is not None
|
||
return {
|
||
"engine": self.name,
|
||
"version": "rules" if ok else "none",
|
||
"ok": ok,
|
||
"status": "ok" if ok else "degraded",
|
||
"error": None if ok else "YARA 规则库未加载",
|
||
"info": f"rules_dir={self.rules_dir}" if not ok else "rules loaded",
|
||
}
|
||
|
||
|
||
class BinaryEngine(Engine):
|
||
"""PE/二进制文件结构分析引擎。"""
|
||
|
||
name = "binary"
|
||
layer = "L2B"
|
||
|
||
def scan(self, target: ScanTarget, ctx: Any) -> EngineResult:
|
||
result = EngineResult(engine=self.name)
|
||
ext = (target.declared_ext or "").lower()
|
||
if ext not in BINARY_EXTS:
|
||
return result
|
||
|
||
try:
|
||
import pefile
|
||
except Exception: # noqa: BLE001
|
||
result.ok = False
|
||
result.error = "pefile 未安装"
|
||
return result
|
||
|
||
try:
|
||
pe = pefile.PE(target.path, fast_load=False)
|
||
except Exception as exc: # noqa: BLE001
|
||
result.findings.append(
|
||
Finding(
|
||
engine=self.name,
|
||
layer=self.layer,
|
||
severity=Severity.MEDIUM.value,
|
||
rule="PeParseFailed",
|
||
detail=f"PE 结构解析失败：{exc}",
|
||
weight=10.0,
|
||
)
|
||
)
|
||
# --- pefile 失败时尝试手工 PE 分析（覆盖假 PE / 伪装 PE）---
|
||
manual = _manual_pe_analysis(target.path)
|
||
if manual:
|
||
result.findings.extend(manual)
|
||
return result
|
||
|
||
try:
|
||
result.findings.extend(self._analyze(pe, target))
|
||
finally:
|
||
try:
|
||
pe.close()
|
||
except Exception: # noqa: BLE001
|
||
pass
|
||
|
||
return result
|
||
|
||
def _analyze(self, pe: Any, target: ScanTarget) -> List[Finding]:
|
||
findings: List[Finding] = []
|
||
|
||
# 1) 节区熵值：加密/加壳的节区熵接近 8
|
||
for section in getattr(pe, "sections", []) or []:
|
||
try:
|
||
name = section.Name.rstrip(b"\x00").decode("utf-8", errors="replace")
|
||
entropy = section.get_entropy()
|
||
except Exception: # noqa: BLE001
|
||
continue
|
||
if entropy >= 7.2:
|
||
findings.append(
|
||
Finding(
|
||
engine=self.name,
|
||
layer=self.layer,
|
||
severity=Severity.MEDIUM.value,
|
||
rule="HighEntropySection",
|
||
detail=f"节区 {name} 熵值 {entropy:.2f}，可能经过加密或加壳",
|
||
weight=20.0,
|
||
)
|
||
)
|
||
|
||
# 2) 可疑导入表
|
||
suspicious_imports = self._suspicious_imports(pe)
|
||
if suspicious_imports:
|
||
findings.append(
|
||
Finding(
|
||
engine=self.name,
|
||
layer=self.layer,
|
||
severity=Severity.HIGH.value,
|
||
rule="SuspiciousImports",
|
||
detail=f"导入表包含高危 API：{', '.join(suspicious_imports[:8])}",
|
||
weight=25.0,
|
||
)
|
||
)
|
||
|
||
# 3) 无签名可执行文件
|
||
if not self._has_signature(pe):
|
||
findings.append(
|
||
Finding(
|
||
engine=self.name,
|
||
layer=self.layer,
|
||
severity=Severity.MEDIUM.value,
|
||
rule="UnsignedExecutable",
|
||
detail="可执行文件未携带数字签名",
|
||
weight=20.0,
|
||
)
|
||
)
|
||
|
||
# 4) 编译时间戳异常（明显晚于其他节区时间，或为 1970/未来时间）
|
||
timestamp = getattr(pe, "FILE_HEADER", None)
|
||
if timestamp is not None:
|
||
try:
|
||
ts = int(timestamp.TimeDateStamp)
|
||
if ts == 0 or ts > 4102444800: # 2100 年之后
|
||
findings.append(
|
||
Finding(
|
||
engine=self.name,
|
||
layer=self.layer,
|
||
severity=Severity.LOW.value,
|
||
rule="SuspiciousTimestamp",
|
||
detail="PE 编译时间戳异常，可能被伪造",
|
||
weight=10.0,
|
||
)
|
||
)
|
||
except Exception: # noqa: BLE001
|
||
pass
|
||
|
||
# 5) UPX 壳显式检测 —— 检查节名 UPX0 / UPX1 / UPX!
|
||
upx_sections = []
|
||
for section in getattr(pe, "sections", []) or []:
|
||
try:
|
||
name = section.Name.rstrip(b"\x00").decode("utf-8", errors="replace")
|
||
if name.upper().startswith("UPX"):
|
||
upx_sections.append(name)
|
||
except Exception:
|
||
continue
|
||
if upx_sections:
|
||
findings.append(
|
||
Finding(
|
||
engine=self.name,
|
||
layer=self.layer,
|
||
severity=Severity.MEDIUM.value,
|
||
rule="UPXPackerDetected",
|
||
detail=f"检测到 UPX 加壳节：{', '.join(upx_sections)}",
|
||
weight=30.0,
|
||
)
|
||
)
|
||
|
||
# 6) RWX 可写可执行节 —— 高危（绕过 DEP 的典型手法）
|
||
from pefile import SECTION_CHARACTERISTICS # noqa: PLC0415
|
||
MEM_EXECUTE = SECTION_CHARACTERISTICS["IMAGE_SCN_MEM_EXECUTE"]
|
||
MEM_WRITE = SECTION_CHARACTERISTICS["IMAGE_SCN_MEM_WRITE"]
|
||
rwx_sections = []
|
||
for section in getattr(pe, "sections", []) or []:
|
||
try:
|
||
flags = section.Characteristics
|
||
if (flags & MEM_EXECUTE) and (flags & MEM_WRITE):
|
||
name = section.Name.rstrip(b"\x00").decode("utf-8", errors="replace")
|
||
rwx_sections.append(name)
|
||
except Exception:
|
||
continue
|
||
if rwx_sections:
|
||
findings.append(
|
||
Finding(
|
||
engine=self.name,
|
||
layer=self.layer,
|
||
severity=Severity.HIGH.value,
|
||
rule="RWXSectionDetected",
|
||
detail=f"检测到可写且可执行的危险节：{', '.join(rwx_sections)}",
|
||
weight=40.0,
|
||
)
|
||
)
|
||
|
||
# 7) 32/64 位识别（info 级别，辅助判断）
|
||
machine = None
|
||
pe32 = None
|
||
try:
|
||
machine = int(pe.FILE_HEADER.Machine)
|
||
pe32 = (pe.OPTIONAL_HEADER.Magic == 0x10B) # PE32
|
||
except Exception:
|
||
pass
|
||
if machine is not None:
|
||
try:
|
||
from pefile import MACHINE_TYPES # noqa: PLC0415
|
||
machine_name = MACHINE_TYPES.get(machine, f"UNKNOWN(0x{machine:04x})")
|
||
except Exception:
|
||
machine_name = f"0x{machine:04x}"
|
||
bits = "32-bit" if pe32 else "64-bit"
|
||
findings.append(
|
||
Finding(
|
||
engine=self.name,
|
||
layer=self.layer,
|
||
severity=Severity.LOW.value,
|
||
rule="PEArchInfo",
|
||
detail=f"PE 架构：{machine_name} / {bits}",
|
||
weight=0.0,
|
||
)
|
||
)
|
||
|
||
# 8) 调试信息（pdb 路径或 debug 节）
|
||
has_debug_dir = False
|
||
has_debug_section = False
|
||
try:
|
||
debug_dir = pe.OPTIONAL_HEADER.DATA_DIRECTORY[6] # IMAGE_DIRECTORY_ENTRY_DEBUG
|
||
if debug_dir.VirtualAddress and debug_dir.Size:
|
||
has_debug_dir = True
|
||
except Exception:
|
||
pass
|
||
for section in getattr(pe, "sections", []) or []:
|
||
try:
|
||
name = section.Name.rstrip(b"\x00").decode("utf-8", errors="replace").lower()
|
||
if ".debug" in name or name.startswith("debug") or name.startswith(".pdb"):
|
||
has_debug_section = True
|
||
break
|
||
except Exception:
|
||
continue
|
||
if has_debug_dir or has_debug_section:
|
||
findings.append(
|
||
Finding(
|
||
engine=self.name,
|
||
layer=self.layer,
|
||
severity=Severity.LOW.value,
|
||
rule="DebugInformationPresent",
|
||
detail="PE 包含调试信息（PDB 路径或 .debug 节），可用于逆向定位",
|
||
weight=10.0,
|
||
)
|
||
)
|
||
|
||
return findings
|
||
|
||
def _suspicious_imports(self, pe: Any) -> List[str]:
|
||
"""检测高危 API 导入。"""
|
||
suspicious = {
|
||
"VirtualAllocEx", "WriteProcessMemory", "CreateRemoteThread",
|
||
"NtUnmapViewOfSection", "SetWindowsHookEx", "OpenProcess",
|
||
"WinHttpOpen", "InternetOpenUrl", "URLDownloadToFile",
|
||
"CryptEncrypt", "AdjustTokenPrivileges", "CreateService",
|
||
"ShellExecute", "WinExec", "IsDebuggerPresent", "CheckRemoteDebuggerPresent",
|
||
}
|
||
found: List[str] = []
|
||
try:
|
||
for entry in getattr(pe, "DIRECTORY_ENTRY_IMPORT", []) or []:
|
||
for imp in entry.imports:
|
||
if imp.name:
|
||
name = imp.name.decode("utf-8", errors="replace")
|
||
if name in suspicious:
|
||
found.append(name)
|
||
except Exception: # noqa: BLE001
|
||
pass
|
||
return sorted(set(found))
|
||
|
||
def _has_signature(self, pe: Any) -> bool:
|
||
"""检查 PE 是否包含 Authenticode 签名目录。"""
|
||
try:
|
||
directory = pe.OPTIONAL_HEADER.DATA_DIRECTORY[4] # IMAGE_DIRECTORY_ENTRY_SECURITY
|
||
return bool(directory.VirtualAddress and directory.Size)
|
||
except Exception: # noqa: BLE001
|
||
return False
|
||
|
||
|
||
def _judge_yara_rule(rule: str, tags: List[str]) -> tuple[str, float]:
|
||
"""依据 YARA 规则名与标签判定严重度。"""
|
||
tag_set = {t.lower() for t in tags}
|
||
text = f"{rule} {' '.join(tags)}".lower()
|
||
|
||
if "critical" in tag_set or any(k in text for k in ("malware", "ransomware", "backdoor", "c2", "exploit")):
|
||
return Severity.CRITICAL.value, 85.0
|
||
if "high" in tag_set or any(k in text for k in ("trojan", "dropper", "obfusc", "shellcode")):
|
||
return Severity.HIGH.value, 60.0
|
||
if "medium" in tag_set or any(k in text for k in ("suspicious", "pua", "packer")):
|
||
return Severity.MEDIUM.value, 30.0
|
||
if "suspicious" in tag_set:
|
||
return Severity.MEDIUM.value, 30.0
|
||
return Severity.LOW.value, 10.0
|