Files
MeshSate/app/engines/documents.py

1002 lines
43 KiB
Python
Raw Normal View History

"""L2 文档深度检测引擎。
覆盖格式:
- Office OLE2 (doc/docm/xls/ppt) —— VBA 宏、DDE、嵌入 OLE 对象、外部模板引用
- Office OOXML (docx/xlsx/pptx) —— 外部链接、外部引用、嵌入对象
- PDF —— JavaScript、LaunchAction、OpenAction、嵌入文件、外部 URI
- RTF —— OLE 嵌入对象
- 脚本 —— PowerShell 混淆、base64 解码、下载器特征
olevba / mraptor / msodde 是主力工具,它们对 VBA 宏的启发式判定
(AutoExec + 可疑关键词 + 编码字符串)在法院文书场景中价值最高。
"""
from __future__ import annotations
import os
import re
from typing import Any, Dict, List
from app.engines.base import Engine, EngineResult, Finding, ScanTarget, Severity
# Office 文档类扩展名
# 注意:带宏变体(.docm/.xlsm/.pptm 等)在 Office 2007+ 生成的是 OOXML ZIP 格式
# oletools(olevba/mraptor/msodde/oleobj)只认 CDFV2 OLE2,不认 ZIP
# 所以带宏扩展名统一归 OOXML_EXTS,由 _scan_vba_plaintext 解压后兜底扫描
OLE_EXTS = {
# Microsoft Office 老格式(OLE 复合文档 / CDFV2,oletools 可解析)
".doc", ".dot",
".xls", ".xlt",
".ppt", ".pot", ".pps",
# WPS Office 老格式(OLE 复合文档,olefile/olevba 可直接解析)
".wps", # WPS 文字(对应 .doc)
".wpt", # WPS 文字模板
".et", # WPS 表格(对应 .xls)
".ett", # WPS 表格模板
".dps", # WPS 演示(对应 .ppt)
".dpt", # WPS 演示模板
}
OOXML_EXTS = {
# Office 2007+ / WPS 2019+(OOXML ZIP 包格式,oletools 不能解析)
".docx", ".xlsx", ".pptx",
".docm", ".xlsm", ".pptm", # 带宏的 ZIP 格式
".dotm", ".xltm", ".potm", # 带宏模板
}
PDF_EXTS = {".pdf"}
RTF_EXTS = {".rtf"}
SCRIPT_EXTS = {
".ps1", ".psm1", ".bat", ".cmd", ".vbs", ".vbe",
".js", ".jse", ".wsf", ".hta",
".sh", ".bash", ".zsh", ".py", ".pl", ".rb",
# VBA / VB 纯文本源码
".bas", ".vba", ".vbs", ".cls", ".frm", ".vb",
}
EML_EXTS = {".eml", ".msg"}
# 带 VBA 宏的 Office 扩展名(用于 VBA 纯文本扫描触发)
VBA_MACRO_OFFICE_EXTS = OLE_EXTS | OOXML_EXTS | {".bas", ".vba", ".cls", ".frm"}
# VBA 自动执行入口关键词
VBA_AUTOEXEC_PATTERNS = [
r"Sub\s+AutoOpen\s*\(", r"Sub\s+AutoExec\s*\(", r"Sub\s+AutoNew\s*\(",
r"Sub\s+AutoClose\s*\(", r"Sub\s+Document_Open\s*\(\s*\)\s*\w*",
r"Sub\s+Workbook_Open\s*\(\s*\)", r"Sub\s+Workbook_Activate\s*\(\s*\)",
r"Private\s+Sub\s+Document_Open", r"Private\s+Sub\s+Workbook_Open",
]
# VBA 危险行为关键词
VBA_DANGEROUS_PATTERNS = [
r"\bShell\s*\(\s*\"",
r"\bCreateObject\s*\(\s*\"Wscript",
r"\bCreateObject\s*\(\s*\"Scripting\.Shell",
r"\bWscript\.Shell",
r"\bActiveXObject\s*\(\s*\"Wscript",
r"\bURLDownloadToFile",
r"\bInternetOpen",
r"\bWinHttp\.WinHttpRequest",
r"\bADODB\.Stream",
r"powershell\s+-",
r"cmd\.exe\s+/c",
r"curl\s+http",
r"wget\s+http",
]
# PowerShell 混淆与下载器特征
PS_SUSPICIOUS = [
(r"-e(?:nc(?:odedcommand)?)?\s+[A-Za-z0-9+/=]{40,}", "PowerShell 编码命令执行"),
(r"FromBase64String", "Base64 解码执行载荷"),
(r"Invoke-Expression|\bIEX\b", "Invoke-Expression 动态执行"),
(r"DownloadString|Invoke-WebRequest|Net\.WebClient", "下载远程载荷"),
(r"Start-Process\s+-WindowStyle\s+Hidden", "隐藏窗口启动进程"),
(r"-WindowStyle\s+Hidden", "隐藏窗口启动进程"),
(r"Set-MpPreference\s+.*-Disable", "关闭安全防护"),
(r"Add-MpPreference\s+.*-ExclusionPath", "添加防护排除项"),
(r"vssadmin\s+delete\s+shadows", "删除卷影副本(勒索前置行为)"),
(r"bcdedit\s+.*recoveryenabled\s+no", "禁用系统恢复(勒索前置行为)"),
(r"New-Object\s+System\.Net\.WebClient", "构造 WebClient 下载器"),
]
# 文档中常见的钓鱼/宏滥用关键词(Windows + Linux 双平台)
MACRO_SUSPICIOUS = [
"Shell", "WScript.Shell", "CreateObject", "AutoOpen", "AutoClose", "Document_Open",
"Workbook_Open", "powershell", "cmd.exe", "wscript", "mshta", "rundll32",
"regsvr32", "certutil", "bitsadmin", "http://", "https://",
# === Linux 等价物(WPS on Linux 宏可能调用的系统接口)===
"/bin/sh", "/bin/bash", "SystemShellExecute", "com.sun.star.system",
"wget ", "curl ", "cron ", "crontab", "systemctl enable",
"/dev/tcp/", "bash -i", "nc -e",
]
# 宏混淆 / 加密 / 解码特征 → MacroObfuscated(零容忍)
MACRO_OBFUSCATION = [
# VBA 解码函数链
(r"StrReverse\s*\(", "StrReverse 字符串反转"),
(r"MSDecode\s*\(", "MSDecode 字符串解码"),
(r"CreateObject\s*\([^)]*Scripting\.Decoder", "Scripting.Decoder COM 对象解码"),
(r"HexDecode|FromHex|DeHex", "十六进制解码"),
(r"base64|Base64|FromBase64|DecodeBase64", "Base64 解码"),
# 字符串数组逐字符拼接(常见混淆手段)
(r"Chr\s*\(\s*\d+\s*\)\s*&\s*Chr\s*\(", "Chr 字符拼接"),
(r"Array\s*\(\s*\d+\s*,\s*\d+\s*,\s*\d+", "数组存储字符码(后续 Chr 拼接)"),
# Shell + 内联解码
(r"Shell\s*\(\s*\".*&.*\"", "Shell 参数拼接命令"),
(r"CreateObject.*Script\.Shell.*\.Run", "通过 Script.Shell Run 执行混淆命令"),
]
class DocumentEngine(Engine):
"""文档深度检测引擎。"""
name = "document"
layer = "L2"
def scan(self, target: ScanTarget, ctx: Any) -> EngineResult:
result = EngineResult(engine=self.name)
ext = (target.declared_ext or "").lower()
findings: List[Finding] = []
if ext in OLE_EXTS or ext in OOXML_EXTS or ext in PDF_EXTS or ext in RTF_EXTS:
findings.extend(self._scan_office_and_pdf(target))
if ext in SCRIPT_EXTS:
findings.extend(self._scan_script(target))
if ext in EML_EXTS:
findings.extend(self._scan_eml(target))
# VBA 宏纯文本扫描 —— 兜底:oletools 打不开时仍能靠纯文本命中
if ext in VBA_MACRO_OFFICE_EXTS:
findings.extend(self._scan_vba_plaintext(target))
result.findings = findings
result.meta["scanned_type"] = ext
return result
# ------------------------------------------------------------- Office/PDF
def _scan_office_and_pdf(self, target: ScanTarget) -> List[Finding]:
findings: List[Finding] = []
ext = (target.declared_ext or "").lower()
# 0) Office 加密检测(msoffcrypto)
# 加密 Office 无法完整解析,需要尽早标记 —— 与 PDF 加密同等级处理
if ext in OLE_EXTS or ext in OOXML_EXTS:
findings.extend(self._check_office_encrypted(target))
# 1) oletools:VBA 宏、DDE、嵌入对象
findings.extend(self._run_oletools(target))
# 2) OOXML 外部链接与外部引用
if ext in OOXML_EXTS:
findings.extend(self._scan_ooxml(target))
# 3) PDF 深度检测
if ext in PDF_EXTS:
findings.extend(self._scan_pdf(target))
return findings
def _run_oletools(self, target: ScanTarget) -> List[Finding]:
# --- oletools(olevba/mraptor/msodde/oleobj)只支持 CDFV2 OLE2 格式 ---
# OOXML ZIP 格式(.docx/.xlsx/.pptx/.docm 等)oletools 会抛 "not supported"
# → 直接在入口过滤:只有 OLE_EXTS(.doc/.xls/.ppt/.wps/.et/.dps 等老格式)才跑
# OOXML 格式的 VBA 宏由 _scan_vba_plaintext 解压后纯文本扫描兜底
ext = (target.declared_ext or "").lower()
if ext not in OLE_EXTS:
return []
findings: List[Finding] = []
path = target.path
# olevba:提取并分析 VBA 宏
try:
from oletools.olevba import VBA_Parser
parser = VBA_Parser(path)
if parser.detect_vba_macros():
for _, _, vba_filename, vba_code in parser.extract_macros():
if not vba_code:
continue
analysis = self._analyze_vba(vba_code)
findings.append(
Finding(
engine=self.name,
layer=self.layer,
severity=analysis["severity"],
rule=analysis["rule"],
detail=f"在 {vba_filename or '未知位置'} 检测到 VBA 宏:{analysis['detail']}",
extracted_path=vba_filename or "",
weight=analysis["weight"],
meta={"code_excerpt": analysis["excerpt"]},
)
)
parser.close()
else:
parser.close()
except Exception as exc: # noqa: BLE001
findings.append(
Finding(
engine=self.name,
layer=self.layer,
severity=Severity.LOW.value,
rule="OleToolsUnavailable",
detail=f"VBA 宏解析失败:{exc}",
weight=0.0,
)
)
# mraptor:宏的整体启发式判定
try:
from oletools.mraptor import MRaptor
if os.path.isfile(path):
mraptor = MRaptor(path)
if mraptor.detect_vba_macros():
verdict = mraptor.scan()
is_malicious = bool(getattr(verdict, "is_malicious", False))
findings.append(
Finding(
engine=self.name,
layer=self.layer,
severity=Severity.CRITICAL.value if is_malicious else Severity.HIGH.value,
rule="MRaptorAutoExecSuspicious" if is_malicious else "MRaptorAutoExec",
detail=(
"mraptor 判定宏具备自动执行且包含可疑行为"
if is_malicious
else "宏包含自动执行入口(AutoOpen/AutoExec),需人工确认"
),
weight=70.0 if is_malicious else 40.0,
)
)
except Exception: # noqa: BLE001
pass
# msodde:DDE / DDEAUTO 自动执行
try:
from oletools.msodde import DdeParser
dde = DdeParser(path)
if dde.detect():
for usage in dde.get_uses():
findings.append(
Finding(
engine=self.name,
layer=self.layer,
severity=Severity.CRITICAL.value,
rule="DDEAutoExecution",
detail=f"检测到 DDE/DDEAUTO 字段:{usage}",
weight=65.0,
)
)
except Exception: # noqa: BLE001
pass
# oleobj:嵌入的 OLE 对象
try:
from oletools.oleobj import OleObjectParser
parser = OleObjectParser(path)
for obj in parser.list_ole():
try:
filename, stream_path, _, _ = obj
except Exception: # noqa: BLE001
continue
name = filename or stream_path or ""
ext = os.path.splitext(name)[1].lower()
if ext in {".exe", ".dll", ".scr", ".js", ".vbs", ".bat", ".cmd", ".ps1", ".hta"}:
findings.append(
Finding(
engine=self.name,
layer=self.layer,
severity=Severity.CRITICAL.value,
rule="EmbeddedExecutable",
detail=f"文档内嵌可执行对象:{name}",
weight=75.0,
)
)
else:
findings.append(
Finding(
engine=self.name,
layer=self.layer,
severity=Severity.MEDIUM.value,
rule="EmbeddedOleObject",
detail=f"文档内嵌 OLE 对象:{name}",
weight=20.0,
)
)
except Exception: # noqa: BLE001
pass
return findings
def _analyze_vba(self, code: str) -> Dict[str, Any]:
"""分析 VBA 宏代码,输出严重度、规则名与说明。
判定优先级(混淆 > 自动执行+危险行为 > 自动执行 > 仅有危险行为 > 仅有宏):
1. 混淆特征 → MacroObfuscated CRITICAL(零容忍,意图不可知)
2. AutoExec + Shell/URL → VBAAutoExecShell CRITICAL
3. 有 Shell(Linux 等价物也算) → VBAShellInvocation HIGH
4. AutoExec → VBAAutoExec HIGH
5. URL / 外部链接 → VBAExternalLink MEDIUM
6. 仅有编码字符串 → VBAEncodedString MEDIUM
7. 仅有宏 → VBAMacroPresent LOW
"""
lowered = code.lower()
hits = [kw for kw in MACRO_SUSPICIOUS if kw.lower() in lowered]
has_autoexec = any(
k.lower() in lowered
for k in ("autoopen", "auto_close", "document_open", "workbook_open", "autoexec")
)
# Shell/CreateObject + Linux 等价物
shell_markers = ("shell", "wscript.shell", "createobject", "/bin/sh", "/bin/bash",
"systemshelexecute", "com.sun.star.system")
has_shell = any(k.lower() in lowered for k in shell_markers)
# 下载 + Linux 等价物
download_markers = ("http://", "https://", "wget ", "curl ")
has_url = any(k in lowered for k in download_markers)
has_encoded = bool(re.search(r"[A-Za-z0-9+/]{60,}={0,2}", code))
# --- 混淆检测:优先于所有其他分级(零容忍)---
obfuscation_hits = []
for pattern, desc in MACRO_OBFUSCATION:
if re.search(pattern, code, re.IGNORECASE):
obfuscation_hits.append(desc)
excerpt = ""
for line in code.splitlines():
if any(k.lower() in line.lower() for k in (
"shell", "createobject", "autoopen", "powershell", "url",
"/bin/sh", "/bin/bash", "strreverse", "msdecode", "chr(", "array(",
)):
excerpt = line.strip()[:200]
break
if obfuscation_hits:
return {
"severity": Severity.CRITICAL.value,
"rule": "MacroObfuscated",
"detail": f"宏代码存在混淆/加密特征:{'; '.join(obfuscation_hits)}",
"weight": 100.0,
"excerpt": excerpt,
}
# --- 常规分级 ---
if has_autoexec and (has_shell or has_url):
return {
"severity": Severity.CRITICAL.value,
"rule": "VBAAutoExecShell",
"detail": f"宏含自动执行入口并调用系统命令/外联,命中特征 {len(hits)} 项",
"weight": 80.0,
"excerpt": excerpt,
}
if has_shell:
return {
"severity": Severity.HIGH.value,
"rule": "VBAShellInvocation",
"detail": "宏调用 Shell/CreateObject(含 Linux 等价物)执行外部命令",
"weight": 55.0,
"excerpt": excerpt,
}
if has_autoexec:
return {
"severity": Severity.HIGH.value,
"rule": "VBAAutoExec",
"detail": "宏包含自动执行入口,打开文档即触发",
"weight": 45.0,
"excerpt": excerpt,
}
if has_url:
return {
"severity": Severity.MEDIUM.value,
"rule": "VBAExternalLink",
"detail": "宏中包含外部链接或下载命令",
"weight": 25.0,
"excerpt": excerpt,
}
if has_encoded:
return {
"severity": Severity.MEDIUM.value,
"rule": "VBAEncodedString",
"detail": "宏中存在疑似编码/混淆的长字符串",
"weight": 20.0,
"excerpt": excerpt,
}
return {
"severity": Severity.LOW.value,
"rule": "VBAMacroPresent",
"detail": "文档包含 VBA 宏,需人工确认用途",
"weight": 15.0,
"excerpt": excerpt,
}
def _scan_ooxml(self, target: ScanTarget) -> List[Finding]:
"""检测 OOXML(docx/xlsx/pptx)中的外部引用与外部链接。"""
findings: List[Finding] = []
try:
import zipfile
with zipfile.ZipFile(target.path) as zf:
names = zf.namelist()
# 外部链接部件
for name in names:
if name.startswith("xl/externalLinks/") and name.endswith(".xml"):
findings.append(
Finding(
engine=self.name,
layer=self.layer,
severity=Severity.HIGH.value,
rule="ExternalLinkReference",
detail=f"工作簿包含外部链接部件:{name}",
weight=45.0,
)
)
if name.startswith("word/embeddings/") or name.startswith("xl/embeddings/") or name.startswith("ppt/embeddings/"):
embedded_ext = os.path.splitext(name)[1].lower()
if embedded_ext in {".exe", ".dll", ".scr", ".js", ".vbs", ".bat", ".bin"}:
findings.append(
Finding(
engine=self.name,
layer=self.layer,
severity=Severity.CRITICAL.value,
rule="EmbeddedExecutable",
detail=f"文档内嵌可执行对象:{name}",
weight=75.0,
)
)
else:
findings.append(
Finding(
engine=self.name,
layer=self.layer,
severity=Severity.MEDIUM.value,
rule="EmbeddedObject",
detail=f"文档内嵌对象:{name}",
weight=20.0,
)
)
# XML 层面的外部 URI 与模板注入
for name in names:
if not name.endswith(".xml") and not name.endswith(".rels"):
continue
try:
content = zf.read(name).decode("utf-8", errors="ignore")
except Exception: # noqa: BLE001
continue
for pattern, rule, severity, weight, desc in [
(r'TargetMode="External"[^>]*Target="(https?://[^"]+)"', "ExternalUri", Severity.HIGH.value, 45.0, "包含指向外部地址的链接"),
(r"<w:attachedTemplate[^>]*>", "AttachedTemplate", Severity.HIGH.value, 45.0, "关联了外部模板,可能触发模板注入"),
(r"DDEAUTO", "DDEAutoExecution", Severity.CRITICAL.value, 65.0, "包含 DDEAUTO 字段"),
(r"<w:docVars>", "DocVars", Severity.MEDIUM.value, 20.0, "包含文档变量,可能用于隐藏载荷"),
]:
if re.search(pattern, content, re.IGNORECASE):
findings.append(
Finding(
engine=self.name,
layer=self.layer,
severity=severity,
rule=rule,
detail=f"{name} {desc}",
extracted_path=name,
weight=weight,
)
)
if rule == "ExternalUri":
break
except Exception as exc: # noqa: BLE001
findings.append(
Finding(
engine=self.name,
layer=self.layer,
severity=Severity.LOW.value,
rule="OoxmlParseFailed",
detail=f"OOXML 容器解析失败:{exc}",
weight=0.0,
)
)
return findings
def _scan_pdf(self, target: ScanTarget) -> List[Finding]:
"""PDF 深度检测。"""
findings: List[Finding] = []
try:
from pypdf import PdfReader
reader = PdfReader(target.path)
if reader.is_encrypted:
findings.append(
Finding(
engine=self.name,
layer=self.layer,
severity=Severity.MEDIUM.value,
rule="PdfEncrypted",
detail="PDF 已加密,无法进行完整静态分析",
weight=15.0,
)
)
try:
if reader.decrypt("") == 0:
return findings
except Exception: # noqa: BLE001
return findings
for index, page in enumerate(reader.pages):
try:
annotations = page.get("/Annots") or []
except Exception: # noqa: BLE001
continue
for annot in annotations:
try:
obj = annot.get_object()
subtype = obj.get("/Subtype")
action = obj.get("/A")
except Exception: # noqa: BLE001
continue
if action is not None:
try:
atype = action.get_object().get("/S")
except Exception: # noqa: BLE001
atype = None
if atype == "/JavaScript":
findings.append(
Finding(
engine=self.name,
layer=self.layer,
severity=Severity.CRITICAL.value,
rule="PdfJavaScript",
detail=f"第 {index + 1} 页包含 JavaScript 动作",
weight=70.0,
)
)
elif atype == "/Launch":
findings.append(
Finding(
engine=self.name,
layer=self.layer,
severity=Severity.CRITICAL.value,
rule="PdfLaunchAction",
detail=f"第 {index + 1} 页包含启动外部程序的 LaunchAction",
weight=75.0,
)
)
uri = action.get_object().get("/URI")
if uri:
findings.append(
Finding(
engine=self.name,
layer=self.layer,
severity=Severity.MEDIUM.value,
rule="PdfExternalUri",
detail=f"第 {index + 1} 页包含外部 URI:{uri}",
weight=20.0,
)
)
# 嵌入文件与 OpenAction
try:
root = reader.trailer["/Root"]
if "/OpenAction" in root:
findings.append(
Finding(
engine=self.name,
layer=self.layer,
severity=Severity.HIGH.value,
rule="PdfOpenAction",
detail="PDF 定义了打开时自动执行的动作",
weight=50.0,
)
)
# 跟进 OpenAction 对象本身的类型 —— 可能直接是 /JavaScript 或 /Launch
try:
open_action_obj = root["/OpenAction"]
# OpenAction 可能是 [obj_ref] 数组,也可能是直接 dict
if isinstance(open_action_obj, list) and len(open_action_obj) > 0:
open_action_obj = open_action_obj[0]
action_dict = open_action_obj.get_object()
saction_type = action_dict.get("/S")
if saction_type == "/JavaScript":
findings.append(
Finding(
engine=self.name,
layer=self.layer,
severity=Severity.CRITICAL.value,
rule="PdfJavaScript",
detail="PDF OpenAction 直接包含 JavaScript 动作",
weight=70.0,
)
)
elif saction_type == "/Launch":
findings.append(
Finding(
engine=self.name,
layer=self.layer,
severity=Severity.CRITICAL.value,
rule="PdfLaunchAction",
detail="PDF OpenAction 直接启动外部程序",
weight=75.0,
)
)
except Exception: # noqa: BLE001
pass
except Exception: # noqa: BLE001
pass
try:
names = reader.attachments if hasattr(reader, "attachments") else {}
for fname in names:
findings.append(
Finding(
engine=self.name,
layer=self.layer,
severity=Severity.HIGH.value,
rule="PdfEmbeddedFile",
detail=f"PDF 内嵌文件:{fname}",
weight=40.0,
)
)
except Exception: # noqa: BLE001
pass
except Exception as exc: # noqa: BLE001
findings.append(
Finding(
engine=self.name,
layer=self.layer,
severity=Severity.LOW.value,
rule="PdfParseFailed",
detail=f"PDF 解析失败:{exc}",
weight=0.0,
)
)
return findings
# ------------------------------------------------------------- 脚本
def _scan_script(self, target: ScanTarget) -> List[Finding]:
findings: List[Finding] = []
try:
with open(target.path, "rb") as fh:
raw = fh.read(2 * 1024 * 1024)
text = raw.decode("utf-8", errors="ignore")
except OSError as exc:
return [
Finding(
engine=self.name,
layer=self.layer,
severity=Severity.LOW.value,
rule="ScriptReadFailed",
detail=f"脚本读取失败:{exc}",
weight=0.0,
)
]
for pattern, desc in PS_SUSPICIOUS:
if re.search(pattern, text, re.IGNORECASE):
severity = Severity.CRITICAL.value
weight = 70.0
if any(k in desc for k in ("卷影", "恢复", "防护", "排除项")):
severity = Severity.HIGH.value
weight = 60.0
findings.append(
Finding(
engine=self.name,
layer=self.layer,
severity=severity,
rule="SuspiciousScript",
detail=desc,
weight=weight,
)
)
return findings
# ------------------------------------------------------------- Office 加密
def _check_office_encrypted(self, target: ScanTarget) -> List[Finding]:
"""用 msoffcrypto 检测 Office 文档是否加密(密码保护)。"""
findings: List[Finding] = []
try:
import msoffcrypto
with open(target.path, "rb") as fh:
try:
office = msoffcrypto.OfficeFile(fh)
except Exception: # noqa: BLE001
# msoffcrypto 无法识别格式 → 不是合法 Office → 跳过
return findings
# msoffcrypto 6.0+ is_encrypted 是方法,老版本是属性
is_enc = getattr(office, "is_encrypted", False)
if callable(is_enc):
try:
is_enc = is_enc()
except Exception: # noqa: BLE001
is_enc = False
if is_enc:
findings.append(
Finding(
engine=self.name,
layer=self.layer,
severity=Severity.MEDIUM.value,
rule="OfficeEncrypted",
detail="Office 文档已加密(密码保护),无法进行完整静态分析",
weight=25.0,
)
)
except Exception: # noqa: BLE001
# msoffcrypto 不可用 → 静默跳过
pass
return findings
# ------------------------------------------------------------- EML 邮件
def _scan_eml(self, target: ScanTarget) -> List[Finding]:
"""RFC822/MIME 邮件深度检测。
扫描维度:
1) 附件:危险扩展名 → EmailAttachmentDangerous
2) 附件内容(text/plain 形式的 VBA 宏 / Shell / Webshell) → 触发相应规则
3) 正文中的钓鱼关键词(恶意域名、附件名称伪装)
一期:只扫顶层附件,不递归嵌套 EML 附件。
"""
findings: List[Finding] = []
# 危险附件扩展名列表
DANGEROUS_ATTACHMENT_EXTS = {
# 可执行 / 脚本
".exe", ".dll", ".scr", ".com", ".bat", ".cmd", ".vbs", ".vbe",
".js", ".jse", ".wsf", ".wsh", ".hta", ".ps1", ".psm1", ".jar",
".msi", ".gadget", ".lnk", ".chm",
# Office 宏相关
".docm", ".dotm", ".xlsm", ".xltm", ".pptm", ".potm",
".doc", ".xls", ".ppt", # 老格式(可能有宏)
# Web 诱饵
".html", ".htm", ".mht",
}
# 钓鱼关键词(正文/Subject 中出现)
PHISH_KEYWORDS = [
(r"urgent|immediate action|within 24 hours|asap", "紧急操作催促(钓鱼常用)"),
(r"click\s+here|click\s+this\s+link|verify\s+your\s+account", "要求点击链接验证账号"),
(r"password|credential|login", "索要密码/凭证"),
(r"wire\s+transfer|bank\s+transfer|payment\s+required", "诱导转账/支付"),
(r"invoice|order|receipt", "伪装发票/订单/收据(附件可能带毒)"),
(r"secure\.com|login-secure|verify-account", "仿冒安全链接钓鱼域名模式"),
]
try:
with open(target.path, "rb") as fh:
raw = fh.read(20 * 1024 * 1024) # 限制 20MB
except OSError as exc:
return [
Finding(
engine=self.name,
layer=self.layer,
severity=Severity.LOW.value,
rule="EmailReadFailed",
detail=f"邮件读取失败: {exc}",
weight=0.0,
)
]
# 尝试用 email 标准库解析
import email
from email import policy
msg = email.message_from_bytes(raw, policy=policy.default)
# ---- 基本信息提取 ----
sender = str(msg.get("From", "") or "")
subject = str(msg.get("Subject", "") or "")
sender_domain = ""
if "@" in sender:
sender_domain = sender.split("@")[-1].strip(">").strip()
# ---- 正文钓鱼关键词扫描 ----
# 把 Subject + 所有 text/plain 正文拼一起做关键词匹配
text_parts: List[str] = [subject]
for part in msg.walk():
ctype = part.get_content_type()
disp = str(part.get("Content-Disposition", "") or "")
is_attachment = "attachment" in disp.lower()
if is_attachment:
continue
if ctype in ("text/plain", "text/html"):
try:
payload = part.get_content() or ""
text_parts.append(payload[:50000]) # 单段截断防长文
except Exception: # noqa: BLE001
continue
combined_text = "\n".join(text_parts).lower()
for pat, desc in PHISH_KEYWORDS:
if re.search(pat, combined_text):
findings.append(
Finding(
engine=self.name,
layer=self.layer,
severity=Severity.MEDIUM.value,
rule="EmailPhishingKeyword",
detail=f"邮件正文/Subject 命中钓鱼特征: {desc}",
weight=25.0,
)
)
# ---- 附件扫描 ----
for part in msg.walk():
disp = str(part.get("Content-Disposition", "") or "")
is_attachment = "attachment" in disp.lower() or part.get_filename()
if not is_attachment:
continue
fname = str(part.get_filename() or "")
if not fname:
continue
fext = os.path.splitext(fname)[1].lower()
# 1) 危险扩展名
if fext in DANGEROUS_ATTACHMENT_EXTS:
findings.append(
Finding(
engine=self.name,
layer=self.layer,
severity=Severity.HIGH.value,
rule="EmailAttachmentDangerous",
detail=f"邮件附件为高风险类型: {fname}",
weight=55.0,
)
)
# 2) 附件文件名伪装检测(双扩展名、假扩展名)
# 例子: "invoice.pdf.exe" / "report.docx.vbs"
ext_count = len([e for e in fname.split(".") if e]) - 1
if ext_count >= 2:
findings.append(
Finding(
engine=self.name,
layer=self.layer,
severity=Severity.HIGH.value,
rule="EmailAttachmentDisguised",
detail=f"附件可能存在双扩展名伪装: {fname}",
weight=45.0,
)
)
# 3) 附件内容(如果是 text/plain 或有解码能力)
try:
payload_bytes = part.get_payload(decode=True)
if payload_bytes and len(payload_bytes) < 10 * 1024 * 1024: # <10MB
payload_text = payload_bytes.decode("utf-8", errors="ignore").lower()
# 简易关键词触发 → 交给 YARA/关键词引擎后续覆盖
if ("autoopen" in payload_text or "document_open" in payload_text) and \
("shell" in payload_text or "cmd" in payload_text or "wscript" in payload_text):
findings.append(
Finding(
engine=self.name,
layer=self.layer,
severity=Severity.HIGH.value,
rule="EmailAttachmentMacroLike",
detail=f"附件内容含 VBA 宏特征(AutoOpen + Shell): {fname}",
weight=40.0,
)
)
if (
"bash -i" in payload_text
or "nc -e" in payload_text
or "python -c" in payload_text
) and ("tcp" in payload_text or "/dev/" in payload_text or "socket" in payload_text):
findings.append(
Finding(
engine=self.name,
layer=self.layer,
severity=Severity.HIGH.value,
rule="EmailAttachmentReverseShell",
detail=f"附件内容含反弹 Shell 特征: {fname}",
weight=55.0,
)
)
if (
"eval(" in payload_text
or "assert(" in payload_text
or "system(" in payload_text
or "base64_decode" in payload_text
) and ("<?php" in payload_text or "<%" in payload_text):
findings.append(
Finding(
engine=self.name,
layer=self.layer,
severity=Severity.CRITICAL.value,
rule="EmailAttachmentWebshell",
detail=f"附件内容疑似 Webshell: {fname}",
weight=70.0,
)
)
except Exception: # noqa: BLE001
continue
return findings
# ------------------------------------------------------------- VBA 宏纯文本兜底扫描
def _scan_vba_plaintext(self, target: ScanTarget) -> List[Finding]:
"""对 Office 文件 + .bas/.vba 源码做 VBA 宏特征的纯文本扫描。
为什么需要这个?oletools 的 olevba.VBA_Parser 只能解析合法 OLE2 CFB 里的 VBA 项目流。
对于:
1) OOXML .docm 里的 vbaProject.bin 不合法(olevba 打不开)
2) .bas / .vba 纯文本源码(压根不是 Office 文件)
这些场景 olevba 会失败,返回 OleToolsUnavailable。
本方法作为"最后一道防线",直接读文件(或解压 ZIP 取 vbaProject.bin)做纯文本匹配。
"""
findings: List[Finding] = []
try:
with open(target.path, "rb") as fh:
raw = fh.read(5 * 1024 * 1024) # 限制 5MB
except OSError:
return findings
# 如果是 ZIP(OOXML .docm/.xlsm/.pptm),尝试解压 vbaProject.bin
ext = (target.declared_ext or "").lower()
text_to_scan = ""
if ext in {".docm", ".xlsm", ".pptm", ".dotm", ".xltm", ".potm"}:
try:
import zipfile
with zipfile.ZipFile(target.path) as zf:
for name in zf.namelist():
if name.endswith("vbaProject.bin") or name.endswith(".bas"):
try:
content = zf.read(name)
text_to_scan += content.decode("utf-8", errors="ignore") + "\n"
except Exception: # noqa: BLE001
pass
except zipfile.BadZipFile:
pass
# 把原始文件也当纯文本拼进来(OLE2 文件会有二进制噪声,但 VBA 字符串仍可匹配)
text_to_scan += raw.decode("utf-8", errors="ignore") + "\n"
text_lower = text_to_scan.lower()
has_autoexec = any(re.search(p, text_to_scan, re.IGNORECASE) for p in VBA_AUTOEXEC_PATTERNS)
has_dangerous = any(re.search(p, text_to_scan, re.IGNORECASE) for p in VBA_DANGEROUS_PATTERNS)
if has_autoexec and has_dangerous:
# AutoExec + 危险行为 → VBAAutoExecShell(零容忍)
findings.append(
Finding(
engine=self.name,
layer=self.layer,
severity=Severity.CRITICAL.value,
rule="VBAAutoExecShell",
detail="VBA 宏检测到 AutoOpen/Workbook_Open 等自动执行入口 + Shell/CreateObject 等危险调用(纯文本扫描)",
weight=70.0,
)
)
elif has_autoexec:
# 只有 AutoExec,没有明确危险行为 → VBAAutoExec(零容忍)
findings.append(
Finding(
engine=self.name,
layer=self.layer,
severity=Severity.HIGH.value,
rule="VBAAutoExec",
detail="VBA 宏包含 AutoOpen/Workbook_Open 等自动执行入口(纯文本扫描)",
weight=50.0,
)
)
elif has_dangerous:
# 只有危险行为,没有自动入口 → VBAShellInvocation(走 rule_overrides → manual_review)
findings.append(
Finding(
engine=self.name,
layer=self.layer,
severity=Severity.HIGH.value,
rule="VBAShellInvocation",
detail="VBA 代码包含 Shell/CreateObject 等危险调用但未发现自动执行入口(纯文本扫描)",
weight=45.0,
)
)
return findings