zhaolei
b6a004bdb7
检测范围与证据链修复，不改裁决口径（口径改动在下一条提交）。 - extractor：zip/tar 成员在超限分支外漏加回扫描队列，解包后内容永远不参与检测； zip 目录条目被当成内嵌对象；tar 超限成员先落盘不删。递归真正生效后深扫耗时可测。 - clamav：_HIGH/_MEDIUM/_UNWANTED 关键词是小写，而签名名是 "Win.Trojan..." 大小写混排， 分级判定恒不命中全走默认档；改统一大写匹配。超过 clamd MaxFileSize 的裸奔区间 (1G~1.5G) 新增 ClamAVSkippedSizeLimit(30) — 按"不可扫描不放行"进复核。 - yara：pe.yar C1 系列改用导入表判定，不再靠裸子串命中字符串表； linux.yar C7 多条规则条件恒真（裸 nc/fetch 之类）已收紧；C8/C3 相应收窄。 - structure：DangerousExecutableExtension 是载体信号而非恶意行为，critical 权重 45→35， 不再作为把中低置信度信号一起顶过 block_score 的底座。 - scanner/documents：real_type 拼进 libmagic 对 MSI/OLE 的元数据串实测 334+ 字符， 超列宽导致整条 scan_tasks INSERT 连 findings 一起丢（被拒文件不留痕）→ 列宽 256→512，migrate.sql 补 ALTER。
204 lines
7.9 KiB
Python
204 lines
7.9 KiB
Python
"""内网特征词表引擎（P1）。
|
||
|
||
数据驱动：词表从 policy.yaml 的 ``keyword_rules`` 段加载，不在代码中硬编码。
|
||
支持三类匹配模式：
|
||
|
||
* ``contains`` — 子串匹配（区分大小写）
|
||
* ``regex`` — 正则表达式匹配
|
||
* ``any_of`` — 多关键词中任一命中
|
||
|
||
用途：识别内网特有模板、密级标识、命名规律等——辅助宏例外判断和内容审计方向。
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import logging
|
||
import os
|
||
import re
|
||
from typing import Any, Dict, List
|
||
|
||
from app.engines.base import Engine, EngineResult, Finding, ScanTarget, Severity
|
||
|
||
logger = logging.getLogger(__name__)
|
||
|
||
# cross_line 规则用的续行归一：换行（及其两侧空白）压成一个空格
|
||
NEWLINE_RUN_RE = re.compile(r"\s*[\r\n]+\s*")
|
||
|
||
|
||
class KeywordEngine(Engine):
|
||
"""内网特征词表匹配引擎。"""
|
||
|
||
# ---- 跳过列表：二进制 / 压缩封装 / 媒体文件 ----
|
||
# 这些文件有专门引擎处理（PDF/Office/Archive/PE 等），
|
||
# 它们的内容是压缩流或二进制格式，关键词 regex 在里面会大面积误报。
|
||
# 短关键词 (nc/wce/curl/...) 在压缩流里等于伪随机字节必然碰撞。
|
||
BINARY_SKIP_EXTS = frozenset({
|
||
# PDF / Office
|
||
".pdf", ".docx", ".docm", ".doc", ".xlsx", ".xlsm", ".xls",
|
||
".pptx", ".pptm", ".ppt", ".ppsx", ".pps", ".potx",
|
||
# 压缩包
|
||
".rar", ".zip", ".7z", ".tar", ".gz", ".bz2", ".xz", ".tgz",
|
||
".iso", ".img", ".dmg",
|
||
# 图片
|
||
".png", ".jpg", ".jpeg", ".gif", ".bmp", ".tiff", ".tif", ".webp",
|
||
".ico", ".svg",
|
||
# 视频
|
||
".mp4", ".avi", ".mkv", ".mov", ".flv", ".wmv", ".webm",
|
||
# 音频
|
||
".mp3", ".wav", ".flac", ".ogg", ".aac", ".wma", ".m4a",
|
||
})
|
||
|
||
name = "keywords"
|
||
layer = "L2B"
|
||
version = "1.0"
|
||
|
||
def __init__(self, keyword_rules: List[Dict[str, Any]] | None = None) -> None:
|
||
"""初始化。keyword_rules 来自 PolicyStore.current().keyword_rules。"""
|
||
self._rules = keyword_rules or []
|
||
# 预编译正则
|
||
self._compiled: List[Dict[str, Any]] = []
|
||
for rule in self._rules:
|
||
pattern = rule.get("pattern", "")
|
||
if rule.get("match") == "regex" and pattern:
|
||
try:
|
||
compiled = re.compile(pattern, re.IGNORECASE if rule.get("ignore_case") else 0)
|
||
except re.error:
|
||
logger.warning("词表规则正则编译失败: %s", rule.get("name"))
|
||
continue
|
||
self._compiled.append({**rule, "_re": compiled})
|
||
else:
|
||
self._compiled.append(rule)
|
||
|
||
# ------------------------------------------------------------------- 接口实现
|
||
|
||
def is_enabled(self) -> bool:
|
||
return bool(self._rules)
|
||
|
||
def scan(self, target: ScanTarget, ctx: Any) -> EngineResult:
|
||
result = EngineResult(engine=self.name)
|
||
if not self._rules:
|
||
return result
|
||
|
||
findings: List[Finding] = []
|
||
|
||
# ---- 跳过二进制/压缩封装文件 ----
|
||
# 这些文件内容是压缩流或纯二进制，短关键词 regex 在里面必然误报
|
||
ext = (target.declared_ext or "").lower()
|
||
if not ext and target.path:
|
||
ext = os.path.splitext(target.path)[1].lower()
|
||
if ext in self.BINARY_SKIP_EXTS:
|
||
return result
|
||
|
||
try:
|
||
text = self._extract_text(target.path, target.declared_ext)
|
||
except Exception: # noqa: BLE001
|
||
return result
|
||
|
||
if not text:
|
||
return result
|
||
|
||
# ---- 预处理 ----
|
||
# 默认按原始行匹配：把整篇文本的换行全部删除会让上一行行尾和下一行行首
|
||
# 黏成一句话（Ruby 源码 "…breed\n+fetch()…" 曾被当成 curl 特征命中）。
|
||
# cross_line: true 的规则再跑一遍"换行压成空格"的文本，用于兜住
|
||
# wget \ + 换行 + -O payload 这类续行绕过；保留分隔符，不产生黏连。
|
||
text_norm = text.replace("\r\n", "\n").replace("\r", "\n")
|
||
text_joined = NEWLINE_RUN_RE.sub(" ", text_norm)
|
||
|
||
for rule in self._compiled:
|
||
match_mode = rule.get("match", "contains")
|
||
rule_name = rule.get("name", "keyword")
|
||
weight = rule.get("weight", 30.0)
|
||
severity_str = rule.get("severity", "medium")
|
||
severity = getattr(Severity, severity_str.upper(), Severity.MEDIUM).value
|
||
|
||
hits: List[str] = []
|
||
|
||
if match_mode == "regex":
|
||
compiled = rule.get("_re")
|
||
if compiled:
|
||
scan_targets = [text_norm]
|
||
if rule.get("cross_line"):
|
||
scan_targets.append(text_joined)
|
||
for body in scan_targets:
|
||
for m in compiled.finditer(body):
|
||
snippet = body[max(0, m.start() - 10): m.end() + 10]
|
||
hits.append(" ".join(snippet.split()))
|
||
if len(hits) >= 3:
|
||
break
|
||
if len(hits) >= 3:
|
||
break
|
||
|
||
elif match_mode == "any_of":
|
||
keywords = rule.get("keywords", [])
|
||
for kw in keywords:
|
||
if kw.lower() in text_norm.lower():
|
||
hits.append(kw)
|
||
if len(hits) >= 3:
|
||
break
|
||
|
||
elif match_mode == "contains":
|
||
pattern = rule.get("pattern", "")
|
||
if pattern and (pattern.lower() in text_norm.lower()):
|
||
hits.append(pattern)
|
||
|
||
if hits:
|
||
findings.append(
|
||
Finding(
|
||
engine=self.name,
|
||
layer=self.layer,
|
||
severity=severity,
|
||
rule=rule_name,
|
||
detail=f"词表命中：{rule_name}（模式={match_mode}），命中 {len(hits)} 处：{hits[:2]}",
|
||
weight=float(weight),
|
||
)
|
||
)
|
||
|
||
result.findings = findings
|
||
result.meta["scanned_chars"] = len(text)
|
||
return result
|
||
|
||
# ------------------------------------------------------------------- 内部辅助
|
||
|
||
@staticmethod
|
||
def _extract_text(path: str, ext: str) -> str:
|
||
"""从文件中提取可匹配的文本（简化版，够用就行）。
|
||
|
||
OOXML/PDF/DOCX 等结构化文档优先交给对应的引擎处理，
|
||
词表引擎只扫纯文本、代码文件和可直接读的二进制残留字符串。
|
||
"""
|
||
text_exts = {".txt", ".csv", ".json", ".xml", ".html", ".htm",
|
||
".md", ".log", ".ini", ".cfg", ".yaml", ".yml",
|
||
".sh", ".bash", ".zsh", ".py", ".pl", ".rb", ".js", ".ts",
|
||
".bat", ".cmd", ".ps1", ".vbs", ".sql", ".go", ".java",
|
||
".c", ".cpp", ".h", ".cs"}
|
||
|
||
ext_lower = (ext or "").lower()
|
||
if ext_lower in text_exts:
|
||
try:
|
||
with open(path, "r", encoding="utf-8", errors="replace") as fh:
|
||
return fh.read(256 * 1024)
|
||
except OSError:
|
||
return ""
|
||
|
||
# 二进制文件：提取可打印字符串（≥4 字符）
|
||
try:
|
||
with open(path, "rb") as fh:
|
||
data = fh.read(512 * 1024)
|
||
except OSError:
|
||
return ""
|
||
|
||
strings: List[str] = []
|
||
current = bytearray()
|
||
for byte in data:
|
||
if 32 <= byte <= 126 or byte in (9,): # 可打印或 tab
|
||
current.append(byte)
|
||
else:
|
||
if len(current) >= 4:
|
||
strings.append(current.decode("ascii", errors="replace"))
|
||
current = bytearray()
|
||
if len(current) >= 4:
|
||
strings.append(current.decode("ascii", errors="replace"))
|
||
|
||
return "\n".join(strings)
|