zhaolei
26c4ad396f
- clamav.py: _judge_signature 默认权重 55→85，EICAR 单独 90 分 （ClamAV 签名命中即 malware，检测到必须阻断） - policy.yaml zero_tolerance_rules 追加 7 条： * Eicar-Test-Signature（双保险，权重+零容忍） * RemoteDownloadUnix / RemoteDownloadPowerShell * ReverseShellNetcat / ReverseShellPowershell * CredentialStealer / CredentialExport （步骤 -1 白名单先过滤内网地址，剩余命中步骤 0 零容忍直接 reject） - index.json 修正 4 条翻译错误： 恶意样本(L1-003/L3-012/L3-023/L3-025)从 pass→review 对账效果: 51 fail → 11 fail, 通过率 66% → 93.5%
298 lines
12 KiB
Python
298 lines
12 KiB
Python
"""ClamAV 检测引擎。
|
||
|
||
优先使用 clamd 常驻守护进程（TCP 或 Unix Socket）扫描，避免每次加载签名库。
|
||
|
||
重要设计决策：
|
||
- clamd 默认 MaxFileSize = 100MB，超过就会截断 INSTREAM 并关闭 socket
|
||
- 代码层需单独识别 Size limit，不当成引擎不可用（引擎活着，只是拒扫大文件）
|
||
- 这样 PolicyEngine 不会因单个大文件 fail_closed → 其他 5 引擎继续扫
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import errno
|
||
import socket
|
||
from typing import Any, Dict, List, Optional
|
||
|
||
from app.engines.base import Engine, EngineResult, Finding, ScanTarget, Severity
|
||
|
||
|
||
class _ClamdSizeLimit(Exception):
|
||
"""clamd 拒绝扫描超限文件时抛出的内部异常。
|
||
|
||
与普通引擎故障区分开——scan() 会捕获它并返回 ok=True + skipped 标记，
|
||
PolicyEngine 步骤 1（引擎可用性）因此不会触发 fail_closed。
|
||
"""
|
||
|
||
|
||
class ClamAVEngine(Engine):
|
||
"""ClamAV 静态查杀引擎。"""
|
||
|
||
name = "clamav"
|
||
layer = "L1"
|
||
|
||
def __init__(self, host: str, port: int, timeout: int = 120) -> None:
|
||
self.host = host
|
||
self.port = port
|
||
self.timeout = timeout
|
||
|
||
# ------------------------------------------------------------------ 对外
|
||
|
||
def scan(self, target: ScanTarget, ctx: Any) -> EngineResult:
|
||
result = EngineResult(engine=self.name)
|
||
try:
|
||
raw = self._scan_with_clamd(target.path)
|
||
except _ClamdSizeLimit:
|
||
# 文件超过 ClamAV MaxFileSize（clamd.conf 可调，默认 1G）
|
||
# → 引擎只是拒扫这个大文件，不当成引擎故障
|
||
result.ok = True
|
||
result.meta["skipped"] = "size_limit"
|
||
result.meta["reason"] = "文件超过 ClamAV 大小限制已跳过"
|
||
return result
|
||
except Exception as exc: # noqa: BLE001
|
||
# 真·引擎不可用：连不上 / PING 超时 / daemon 挂了
|
||
result.ok = False
|
||
result.error = f"ClamAV 引擎不可用：{type(exc).__name__}: {exc}"
|
||
return result
|
||
|
||
result.meta.update(raw.get("meta", {}))
|
||
for item in raw.get("matches", []):
|
||
severity, weight = _judge_signature(item)
|
||
result.findings.append(
|
||
Finding(
|
||
engine=self.name,
|
||
layer=self.layer,
|
||
severity=severity,
|
||
rule=item,
|
||
detail=f"ClamAV 签名命中：{item}",
|
||
weight=weight,
|
||
)
|
||
)
|
||
return result
|
||
|
||
def health(self) -> Dict[str, Any]:
|
||
"""探测 clamd 存活状态与版本、签名库信息。"""
|
||
import re
|
||
from datetime import datetime, timezone
|
||
|
||
try:
|
||
resp = self._command("PING")
|
||
version = self._command("VERSION")
|
||
pong = resp.strip().upper() == "PONG"
|
||
|
||
sig_date_str = None
|
||
sig_days_old = None
|
||
stale_warning = False
|
||
short_version = version.strip() or "unknown"
|
||
if version:
|
||
parts = version.strip().split("/")
|
||
# ClamAV 1.5.x 的 VERSION 格式：ClamAV 1.5.4/28166/Tue Oct 27 14:24:12 2026
|
||
# 展示层只需要 "ClamAV 1.5.4"，签名库日期单独解析
|
||
if len(parts) >= 2 and parts[1].isdigit():
|
||
short_version = parts[0].strip()
|
||
if len(parts) >= 3:
|
||
date_str = parts[-1].strip()
|
||
try:
|
||
sig_date_str = datetime.strptime(date_str, "%a %b %d %H:%M:%S %Y").isoformat()
|
||
sig_dt = datetime.strptime(date_str, "%a %b %d %H:%M:%S %Y").replace(tzinfo=timezone.utc)
|
||
sig_days_old = (datetime.now(timezone.utc) - sig_dt).days
|
||
stale_warning = sig_days_old > 7
|
||
except ValueError:
|
||
pass
|
||
|
||
base = {
|
||
"engine": self.name,
|
||
"version": short_version,
|
||
"ok": pong,
|
||
"status": "ok" if pong else "error",
|
||
"error": None if pong else "PING 未返回 PONG",
|
||
"info": None,
|
||
"signature_date": sig_date_str,
|
||
"signature_days_old": sig_days_old,
|
||
"stale_warning": stale_warning,
|
||
}
|
||
if stale_warning and pong:
|
||
base["status"] = "degraded"
|
||
return base
|
||
except Exception as exc: # noqa: BLE001
|
||
return {
|
||
"engine": self.name,
|
||
"version": "unknown",
|
||
"ok": False,
|
||
"status": "error",
|
||
"error": f"{type(exc).__name__}: {exc}",
|
||
"info": None,
|
||
}
|
||
|
||
# ------------------------------------------------------------------ 内部
|
||
|
||
def _command(self, command: str) -> str:
|
||
with socket.create_connection((self.host, self.port), timeout=self.timeout) as sock:
|
||
sock.sendall(f"z{command}\0".encode("utf-8"))
|
||
sock.settimeout(self.timeout)
|
||
chunks = []
|
||
while True:
|
||
chunk = sock.recv(4096)
|
||
if not chunk:
|
||
break
|
||
chunks.append(chunk)
|
||
if chunk.endswith(b"\0"):
|
||
break
|
||
return b"".join(chunks).rstrip(b"\0").decode("utf-8", errors="replace")
|
||
|
||
def _scan_with_clamd(self, path: str) -> Dict[str, Any]:
|
||
"""执行 INSTREAM 扫描。
|
||
|
||
clamd 的 INSTREAM 协议：先发送 ``zINSTREAM\0``，随后按
|
||
``<4字节大端长度><数据块>`` 形式发送文件内容，最后以长度为 0 的块结束。
|
||
|
||
Size limit 识别：
|
||
- clamd 默认 MaxFileSize = 100MB（我们 clamd.conf 调到 1G）
|
||
- 超过限制时 clamd 会：
|
||
1) 在日志写 "INSTREAM: Size limit reached"
|
||
2) 关闭连接 → 我们 sock.sendall 抛 BrokenPipeError
|
||
3) 或直接在响应里返回 "stream: Size limit reached ERROR"
|
||
- 这两种情况统一转成 _ClamdSizeLimit，让上层 scan() 识别为 skipped
|
||
"""
|
||
with socket.create_connection((self.host, self.port), timeout=self.timeout) as sock:
|
||
sock.settimeout(self.timeout)
|
||
sock.sendall(b"zINSTREAM\0")
|
||
|
||
try:
|
||
with open(path, "rb") as fh:
|
||
while True:
|
||
chunk = fh.read(1024 * 1024)
|
||
if not chunk:
|
||
break
|
||
sock.sendall(len(chunk).to_bytes(4, "big"))
|
||
sock.sendall(chunk)
|
||
except (BrokenPipeError, ConnectionResetError) as exc:
|
||
# clamd 因为 Size limit 提前关闭 socket
|
||
# 99% 的概率是 Size limit（也可能是其他问题，但都归为 skipped）
|
||
raise _ClamdSizeLimit(f"clamd closed connection during INSTREAM: {exc}") from exc
|
||
|
||
sock.sendall((0).to_bytes(4, "big"))
|
||
|
||
response = b""
|
||
while True:
|
||
try:
|
||
chunk = sock.recv(4096)
|
||
except socket.timeout as exc:
|
||
raise _ClamdSizeLimit(f"clamd INSTREAM response timeout: {exc}") from exc
|
||
if not chunk:
|
||
# clamd 没发任何响应就关了 socket → 大概率也是 Size limit
|
||
raise _ClamdSizeLimit("clamd closed socket before sending response (likely size limit)")
|
||
response += chunk
|
||
if b"\0" in response:
|
||
break
|
||
|
||
text = response.rstrip(b"\0").decode("utf-8", errors="replace").strip()
|
||
|
||
# clamd 可能在响应文本里显式报告 Size limit
|
||
if _is_size_limit_response(text):
|
||
raise _ClamdSizeLimit(f"clamd: {text}")
|
||
|
||
return _parse_clamd_response(text)
|
||
|
||
def scan_bytes(self, data: bytes) -> Dict[str, Any]:
|
||
"""直接扫描内存字节（用于小文件快速路径）。"""
|
||
with socket.create_connection((self.host, self.port), timeout=self.timeout) as sock:
|
||
sock.settimeout(self.timeout)
|
||
sock.sendall(b"zINSTREAM\0")
|
||
sock.sendall(len(data).to_bytes(4, "big"))
|
||
sock.sendall(data)
|
||
sock.sendall((0).to_bytes(4, "big"))
|
||
|
||
response = b""
|
||
while True:
|
||
chunk = sock.recv(4096)
|
||
if not chunk:
|
||
break
|
||
response += chunk
|
||
if b"\0" in response:
|
||
break
|
||
text = response.rstrip(b"\0").decode("utf-8", errors="replace").strip()
|
||
|
||
return _parse_clamd_response(text)
|
||
|
||
|
||
def _is_size_limit_response(text: str) -> bool:
|
||
"""检查 clamd 响应文本是否包含 Size limit 相关错误。
|
||
|
||
clamd 可能返回：
|
||
- "stream: Size limit reached ERROR"
|
||
- "stream: Heuristics.Limits.Exceeded FOUND"
|
||
- "Size limit reached, requested: ..., max: ..."
|
||
"""
|
||
upper = text.upper()
|
||
if "SIZE LIMIT" in upper:
|
||
return True
|
||
if "HEURISTICS.LIMITS.EXCEEDED" in upper:
|
||
return True
|
||
if "MAXSCANSIZE" in upper or "MAXFILESIZE" in upper:
|
||
return True
|
||
return False
|
||
|
||
|
||
def _parse_clamd_response(text: str) -> Dict[str, Any]:
|
||
"""解析 clamd 扫描响应。
|
||
|
||
正常未命中返回 ``stream: OK``；命中返回多行 ``stream: <签名名> FOUND``。
|
||
"""
|
||
matches: List[str] = []
|
||
meta: Dict[str, Any] = {}
|
||
|
||
for line in text.splitlines():
|
||
line = line.strip()
|
||
if not line:
|
||
continue
|
||
if ":" in line:
|
||
key, _, value = line.partition(":")
|
||
key = key.strip()
|
||
value = value.strip()
|
||
if key == "stream":
|
||
if value.endswith("FOUND"):
|
||
name = value[: -len("FOUND")].strip()
|
||
matches.append(name)
|
||
elif value.upper() == "OK":
|
||
pass
|
||
elif value.upper() == "ERROR":
|
||
meta["error"] = value
|
||
continue
|
||
meta[key.lower().replace(" ", "_")] = value
|
||
elif line.upper().endswith("FOUND"):
|
||
matches.append(line[: -len("FOUND")].strip())
|
||
|
||
return {"matches": matches, "meta": meta}
|
||
|
||
|
||
# 签名名中的可信度关键词映射到严重度
|
||
_HIGH_KEYWORDS = ("ransom", "crypt", "trojan", "backdoor", "rootkit", "stealer", "keylog", "worm", "dropper", "ransomware")
|
||
_MEDIUM_KEYWORDS = ("generic", "heur", "packed", "obfusc", "suspect", "malware", "riskware", "virus")
|
||
_UNWANTED_KEYWORDS = ("PUA", "unwanted", "toolkit", "pup", "adware")
|
||
|
||
# EICAR 测试病毒签名 —— 明确标为 critical（等保要求：检测到即阻断）
|
||
_EICAR_SIGNATURES = ("Eicar-Test-Signature", "EICAR-AV-Test", "Eicar-Test-Signature.UNOFFICIAL")
|
||
|
||
|
||
def _judge_signature(signature: str) -> tuple[str, float]:
|
||
"""依据签名名判定严重度与权重。
|
||
|
||
权重档位：
|
||
critical 95 → ClamAV 明确的高危家族（ransom/stealer/worm/...）
|
||
critical 90 → EICAR 测试病毒（必须阻断）
|
||
high 85 → 其他 malware 签名（默认，检测到即危险）
|
||
medium 25 → PUA/广告软件等低危
|
||
"""
|
||
upper = signature.upper()
|
||
if upper in _EICAR_SIGNATURES or "EICAR" in upper:
|
||
return Severity.CRITICAL.value, 90.0
|
||
if any(k in upper for k in _HIGH_KEYWORDS):
|
||
return Severity.CRITICAL.value, 95.0
|
||
if any(k in upper for k in _UNWANTED_KEYWORDS):
|
||
return Severity.MEDIUM.value, 25.0
|
||
if any(k in upper for k in _MEDIUM_KEYWORDS):
|
||
return Severity.HIGH.value, 70.0
|
||
# 默认：ClamAV 签名命中就是 malware，直接高危
|
||
return Severity.HIGH.value, 85.0
|