124 lines
3.1 KiB
Python
124 lines
3.1 KiB
Python
|
"""检测引擎公共数据结构与基类。"""
|
|||
|
|
|
|||
|
|
from __future__ import annotations
|
|||
|
|
|
|||
|
|
from dataclasses import dataclass, field
|
|||
|
|
from enum import Enum
|
|||
|
|
from typing import Any, Dict, List, Optional
|
|||
|
|
|
|||
|
|
|
|||
|
|
class Severity(str, Enum):
|
|||
|
|
INFO = "info"
|
|||
|
|
LOW = "low"
|
|||
|
|
MEDIUM = "medium"
|
|||
|
|
HIGH = "high"
|
|||
|
|
CRITICAL = "critical"
|
|||
|
|
|
|||
|
|
|
|||
|
|
class Layer(str, Enum):
|
|||
|
|
L0 = "L0"
|
|||
|
|
L1 = "L1"
|
|||
|
|
L2 = "L2"
|
|||
|
|
L2B = "L2B"
|
|||
|
|
L3 = "L3"
|
|||
|
|
|
|||
|
|
|
|||
|
|
@dataclass
|
|||
|
|
class Finding:
|
|||
|
|
"""单条检测信号。"""
|
|||
|
|
|
|||
|
|
engine: str
|
|||
|
|
layer: str
|
|||
|
|
severity: str
|
|||
|
|
rule: str
|
|||
|
|
detail: str = ""
|
|||
|
|
extracted_path: str = ""
|
|||
|
|
weight: float = 0.0
|
|||
|
|
meta: Dict[str, Any] = field(default_factory=dict)
|
|||
|
|
|
|||
|
|
def to_dict(self) -> Dict[str, Any]:
|
|||
|
|
return {
|
|||
|
|
"engine": self.engine,
|
|||
|
|
"layer": self.layer,
|
|||
|
|
"severity": self.severity,
|
|||
|
|
"rule": self.rule,
|
|||
|
|
"detail": self.detail,
|
|||
|
|
"extracted_path": self.extracted_path,
|
|||
|
|
"weight": self.weight,
|
|||
|
|
"meta": self.meta,
|
|||
|
|
}
|
|||
|
|
|
|||
|
|
|
|||
|
|
@dataclass
|
|||
|
|
class EngineResult:
|
|||
|
|
"""单个引擎对单个文件的检测结果。"""
|
|||
|
|
|
|||
|
|
engine: str
|
|||
|
|
ok: bool = True
|
|||
|
|
findings: List[Finding] = field(default_factory=list)
|
|||
|
|
meta: Dict[str, Any] = field(default_factory=dict)
|
|||
|
|
error: str = ""
|
|||
|
|
|
|||
|
|
def to_dict(self) -> Dict[str, Any]:
|
|||
|
|
return {
|
|||
|
|
"engine": self.engine,
|
|||
|
|
"ok": self.ok,
|
|||
|
|
"findings": [f.to_dict() for f in self.findings],
|
|||
|
|
"meta": self.meta,
|
|||
|
|
"error": self.error,
|
|||
|
|
}
|
|||
|
|
|
|||
|
|
|
|||
|
|
@dataclass
|
|||
|
|
class ScanTarget:
|
|||
|
|
"""待检测文件。"""
|
|||
|
|
|
|||
|
|
path: str
|
|||
|
|
filename: str
|
|||
|
|
declared_ext: str = ""
|
|||
|
|
logical_path: str = ""
|
|||
|
|
depth: int = 0
|
|||
|
|
is_extracted: bool = False
|
|||
|
|
|
|||
|
|
|
|||
|
|
class Engine:
|
|||
|
|
"""引擎基类。
|
|||
|
|
|
|||
|
|
子类实现 ``scan`` 方法，捕获所有异常并转换为 ``EngineResult.error``，
|
|||
|
|
保证单个引擎故障不会导致整个扫描任务失败。
|
|||
|
|
"""
|
|||
|
|
|
|||
|
|
name: str = "base"
|
|||
|
|
layer: str = "L0"
|
|||
|
|
|
|||
|
|
def scan(self, target: ScanTarget, ctx: Any) -> EngineResult:
|
|||
|
|
raise NotImplementedError
|
|||
|
|
|
|||
|
|
def safe_scan(self, target: ScanTarget, ctx: Any) -> EngineResult:
|
|||
|
|
try:
|
|||
|
|
return self.scan(target, ctx)
|
|||
|
|
except Exception as exc: # noqa: BLE001
|
|||
|
|
return EngineResult(engine=self.name, ok=False, error=f"{type(exc).__name__}: {exc}")
|
|||
|
|
|
|||
|
|
def health(self) -> Dict[str, Any]:
|
|||
|
|
"""引擎健康状态。子类可覆盖以提供更丰富的探测（如 ClamAV/PING）。
|
|||
|
|
|
|||
|
|
返回结构约定（控制台 dashboard 模板依赖此 schema）：
|
|||
|
|
{
|
|||
|
|
"engine": "clamav", # 引擎名（= self.name）
|
|||
|
|
"version": "ClamAV 1.0.x", # 版本描述
|
|||
|
|
"ok": True/False, # 是否可用
|
|||
|
|
"status": "ok"|"error"|"degraded",
|
|||
|
|
"error": None | "...",
|
|||
|
|
"info": None | "...",
|
|||
|
|
}
|
|||
|
|
"""
|
|||
|
|
return {
|
|||
|
|
"engine": self.name,
|
|||
|
|
"version": "builtin",
|
|||
|
|
"ok": True,
|
|||
|
|
"status": "ok",
|
|||
|
|
"error": None,
|
|||
|
|
"info": None,
|
|||
|
|
}
|