MeshSate/app/engines/clamav.py

"""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）
            # → 引擎只是拒扫这个大文件，不当成引擎故障
            # 但"没查过"必须留下信号：策略里 max_file_size 是 1.5G，1G~1.5G 的文件
            # 原本在 L1 完全裸奔却仍可能 clean/pass。按"不可扫描不放行"落人工复核。
            result.ok = True
            result.meta["skipped"] = "size_limit"
            result.meta["reason"] = "文件超过 ClamAV 大小限制已跳过"
            result.findings.append(
                Finding(
                    engine=self.name,
                    layer=self.layer,
                    severity=Severity.MEDIUM.value,
                    rule="ClamAVSkippedSizeLimit",
                    detail="文件超过 clamd MaxFileSize，L1 签名查杀未执行，按不可扫描内容处置",
                    weight=30.0,
                )
            )
            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")
_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