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
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,
|
||
}
|