zhaolei
16f134e328
核心功能: - 6 层检测管线: L0 结构分析 → L1 ClamAV → L2 文档深度检测 → L3 关键词匹配 → L4 YARA → L5 PE/二进制 - FastAPI + PostgreSQL + Redis + Docker Compose 架构 - 客户端 HMAC-SHA256 签名鉴权 + 限流 + 幂等性 - 控制台 UI (登录/Dashboard/诊断中心/策略管理/任务列表/测试中心/产品介绍页) - 三层诊断定时器 (error_scan 30s / db_aggregate 30s / container_check 5min) 并发与稳定性修复: - 诊断定时器 Redis 时间槽位锁 (多 worker 同一槽位只有一个执行) - JSONL 存储 fcntl 跨进程排他锁 + 容错解码 + 原子重写 - 移除 SQLite fallback, 统一 PostgreSQL - API_CLIENTS 脱敏为环境变量, .env 已 .gitignore
1002 lines
43 KiB
Python
1002 lines
43 KiB
Python
"""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
|