"""诊断规则引擎。 职责： - 定义 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