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
229 lines
8.5 KiB
Python
229 lines
8.5 KiB
Python
"""诊断规则引擎。
|
||
|
||
职责：
|
||
- 定义 RuleResult 数据结构（每条规则的输出格式）
|
||
- 提供 DiagnosticEngine.run_all() 入口
|
||
- 规则实现分散在 rules.py 中，engine 负责调度
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import time
|
||
from dataclasses import dataclass, field, asdict
|
||
from datetime import datetime, timedelta, timezone
|
||
from typing import Any, Dict, List, Optional
|
||
|
||
|
||
@dataclass
|
||
class RuleResult:
|
||
"""单条规则的输出。"""
|
||
|
||
id: str # R001, R002 ...
|
||
severity: str # high / medium / low / info / ok
|
||
message: str # 给用户看的中文描述
|
||
hint: str = "" # 建议/下一步操作
|
||
evidence: List[str] = field(default_factory=list) # 证据片段（日志行/DB 统计）
|
||
run_at: str = "" # ISO 格式时间戳
|
||
run_by: str = "page_request" # page_request / timer:xxx / export
|
||
skipped: bool = False # 规则是否跳过
|
||
skip_reason: Optional[str] = None # 为什么跳过（如 Docker 不可用）
|
||
|
||
def __post_init__(self) -> None:
|
||
if not self.run_at:
|
||
# 用北京时间（Asia/Shanghai, UTC+8），避免显示差 8 小时
|
||
try:
|
||
bj = timezone(timedelta(hours=8))
|
||
self.run_at = datetime.now(bj).strftime("%Y-%m-%d %H:%M:%S")
|
||
except Exception: # noqa: BLE001
|
||
self.run_at = datetime.now(timezone.utc).isoformat()
|
||
|
||
def to_dict(self) -> Dict[str, Any]:
|
||
return asdict(self)
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# 严重性 → 徽章 CSS 类映射（和 _nav_style.html 保持一致）
|
||
# ---------------------------------------------------------------------------
|
||
SEVERITY_BADGE: Dict[str, str] = {
|
||
"high": "tag bad", # 红色
|
||
"medium": "tag warn", # 橙色
|
||
"low": "tag notice", # 黄色
|
||
"info": "tag muted", # 灰色
|
||
"ok": "tag ok", # 绿色
|
||
}
|
||
|
||
|
||
def badge_for(severity: str) -> str:
|
||
return SEVERITY_BADGE.get(severity, "tag plain")
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# DiagnosticEngine — 规则调度入口
|
||
# ---------------------------------------------------------------------------
|
||
class DiagnosticEngine:
|
||
"""诊断规则引擎。
|
||
|
||
用法::
|
||
|
||
engine = DiagnosticEngine(db=SessionLocal(), scanner=scanner,
|
||
policy_store=policy_store, docker_status="full")
|
||
results = engine.run_all(run_by="page_request")
|
||
"""
|
||
|
||
def __init__(
|
||
self,
|
||
db, # SQLAlchemy Session
|
||
scanner, # Scanner 实例
|
||
policy_store, # PolicyStore 实例
|
||
docker_status: str = "none", # full / partial / none
|
||
docker_log_tail: int = 500,
|
||
):
|
||
self.db = db
|
||
self.scanner = scanner
|
||
self.policy_store = policy_store
|
||
self.docker_status = docker_status
|
||
self.docker_log_tail = docker_log_tail
|
||
|
||
# ------------------------------------------------------------------
|
||
# 入口方法
|
||
# ------------------------------------------------------------------
|
||
|
||
def run_all(self, run_by: str = "page_request") -> List[RuleResult]:
|
||
"""运行全部 15 条规则，返回结果列表。"""
|
||
# 延迟导入，避免循环依赖
|
||
from app.diagnostic import rules as _rules
|
||
|
||
results: List[RuleResult] = []
|
||
all_funcs = [
|
||
_rules.check_r001_error_ratio,
|
||
_rules.check_r002_clamav_db,
|
||
_rules.check_r003_yara_mtime,
|
||
_rules.check_r004_engine_health,
|
||
_rules.check_r005_no_scans,
|
||
_rules.check_r006_suspicious_over_malicious,
|
||
_rules.check_r007_reject_ratio,
|
||
_rules.check_r008_duplicate_sha,
|
||
# 以下 7 条需要 Docker 能力
|
||
_rules.check_r009_clamav_crash,
|
||
_rules.check_r010_postgres_fatal,
|
||
_rules.check_r011_redis_oom,
|
||
_rules.check_r012_no_space,
|
||
_rules.check_r013_nginx_upstream,
|
||
_rules.check_r014_api_connection,
|
||
_rules.check_r015_container_restart,
|
||
_rules.check_r016_restart_history,
|
||
]
|
||
|
||
for fn in all_funcs:
|
||
try:
|
||
results.append(fn(self, run_by=run_by))
|
||
except Exception as e: # noqa: BLE001
|
||
# 单条规则失败不影响整体
|
||
results.append(RuleResult(
|
||
id=getattr(fn, "_rule_id", "UNKNOWN"),
|
||
severity="info",
|
||
message=f"规则执行异常: {e}",
|
||
skipped=True,
|
||
skip_reason=str(e),
|
||
run_by=run_by,
|
||
))
|
||
|
||
return results
|
||
|
||
def run_page_request_rules(self) -> List[RuleResult]:
|
||
"""页面请求时必跑的轻量规则（R001/R002/R003/R004/R005）。"""
|
||
from app.diagnostic import rules as _rules
|
||
|
||
results: List[RuleResult] = []
|
||
for fn in [
|
||
_rules.check_r001_error_ratio,
|
||
_rules.check_r002_clamav_db,
|
||
_rules.check_r003_yara_mtime,
|
||
_rules.check_r004_engine_health,
|
||
_rules.check_r005_no_scans,
|
||
]:
|
||
try:
|
||
results.append(fn(self, run_by="page_request"))
|
||
except Exception as e: # noqa: BLE001
|
||
results.append(RuleResult(
|
||
id=getattr(fn, "_rule_id", "UNKNOWN"),
|
||
severity="info",
|
||
message=f"规则执行异常: {e}",
|
||
skipped=True,
|
||
skip_reason=str(e),
|
||
))
|
||
return results
|
||
|
||
def run_timer_error_scan(self) -> List[RuleResult]:
|
||
"""定时器 A：错误类日志（30s 周期）。"""
|
||
from app.diagnostic import rules as _rules
|
||
|
||
results: List[RuleResult] = []
|
||
for fn in [
|
||
_rules.check_r001_error_ratio,
|
||
_rules.check_r005_no_scans,
|
||
_rules.check_r009_clamav_crash,
|
||
_rules.check_r010_postgres_fatal,
|
||
_rules.check_r011_redis_oom,
|
||
_rules.check_r012_no_space,
|
||
_rules.check_r013_nginx_upstream,
|
||
_rules.check_r014_api_connection,
|
||
]:
|
||
try:
|
||
results.append(fn(self, run_by="timer:error_scan"))
|
||
except Exception as e: # noqa: BLE001
|
||
results.append(RuleResult(
|
||
id=getattr(fn, "_rule_id", "UNKNOWN"),
|
||
severity="info",
|
||
message=f"规则执行异常: {e}",
|
||
skipped=True,
|
||
skip_reason=str(e),
|
||
))
|
||
return results
|
||
|
||
def run_timer_db_aggregate(self) -> List[RuleResult]:
|
||
"""定时器 B：DB 聚合类（60s 周期）。"""
|
||
from app.diagnostic import rules as _rules
|
||
|
||
results: List[RuleResult] = []
|
||
for fn in [
|
||
_rules.check_r006_suspicious_over_malicious,
|
||
_rules.check_r007_reject_ratio,
|
||
_rules.check_r008_duplicate_sha,
|
||
]:
|
||
try:
|
||
results.append(fn(self, run_by="timer:db_aggregate"))
|
||
except Exception as e: # noqa: BLE001
|
||
results.append(RuleResult(
|
||
id=getattr(fn, "_rule_id", "UNKNOWN"),
|
||
severity="info",
|
||
message=f"规则执行异常: {e}",
|
||
skipped=True,
|
||
skip_reason=str(e),
|
||
))
|
||
return results
|
||
|
||
def run_timer_container(self) -> List[RuleResult]:
|
||
"""定时器 C：容器/引擎类（300s 周期）。"""
|
||
from app.diagnostic import rules as _rules
|
||
|
||
results: List[RuleResult] = []
|
||
for fn in [
|
||
_rules.check_r002_clamav_db,
|
||
_rules.check_r003_yara_mtime,
|
||
_rules.check_r004_engine_health,
|
||
_rules.check_r015_container_restart,
|
||
_rules.check_r016_restart_history,
|
||
]:
|
||
try:
|
||
results.append(fn(self, run_by="timer:container_check"))
|
||
except Exception as e: # noqa: BLE001
|
||
results.append(RuleResult(
|
||
id=getattr(fn, "_rule_id", "UNKNOWN"),
|
||
severity="info",
|
||
message=f"规则执行异常: {e}",
|
||
skipped=True,
|
||
skip_reason=str(e),
|
||
))
|
||
return results
|