SentinelBlue 技术深度解析：蓝方实时威胁监测与自动化响应平台的架构设计与源码实现
> 项目来源：[https://cnb.cool/unicom/SentinelBlue](https://cnb.cool/unicom/SentinelBlue) 关键词：入侵检测、实时响应、进程树回溯、文件完整性校验、审计留痕、可回滚处置 --- ## 摘要 蓝方（防守方）在对抗红方（攻击方）时，最大的痛点往往不是"看不见"，而是 **“看见了却没记全、记全了却来不及断”**。红方的攻击是一条有严格时间顺序的链路： ```javascript 投递 → 落地 → 执行 → 反弹 → 驻留 ``` SentinelBlue 把上述环节收敛为一条 **实时闭环**，在每一次进程生灭、每一条异常外连、每一次阻断动作发生时，都同步完成"记录—分析—响应—报告"，并让全部行为可回溯、可取证、可复盘。本文从工程实现角度，逐模块拆解其架构设计、关键算法与防御性编程思想，并在文末附上完整源码。 --- ## 一、问题建模：为什么需要一条闭环 传统 HIDS（主机入侵检测系统）常见的两个断点是： 1. **检测与响应解耦**。告警进了 SIEM，但处置还要人去敲命令，时间线在"看到"与"动手"之间断裂。 1. **缺乏可溯源的中间态**。很多脚本只管"杀进程"，却不留审计；一旦误杀，无法还原现场，也无法向合规方证明"我做过什么、为什么做"。 SentinelBlue 的设计目标是把这两件事放进同一条主循环： ```javascript 监测(monitor) → 记录(recorder) → 分析(analyzer) → 阻断(responder) → 报告(report) ↑ │ └──────────────────── 全程留痕 / 可溯源 ─────────────────────────┘ ``` 核心约束是：**无论响应成功、失败、跳过还是被拒绝，动作本身都必须先落盘**；**执行任何破坏性动作前，必须先对原状态打快照**。这两条约束是整个系统的"安全灵魂"。 --- ## 二、整体架构与目录结构 ```javascript sentinelblue/ ├── config.yaml # 作战参数表（端口/进程名/阈值/响应模式/基线/告警） ├── requirements.txt # 运行依赖（psutil、PyYAML） ├── main.py # 主程序：run / baseline-build / rollback 三个子命令 ├── core/ │ ├── __init__.py # 包声明 │ ├── recorder.py # 审计记录（SQLite 全程留痕 + 回滚快照表） │ ├── monitor.py # 监测引擎（进程/网络/文件快照，含学习期基线） │ ├── rules.py # 检测规则引擎 + Finding 数据模型 │ ├── analyzer.py # 分析引擎（攻击链回溯、完整性校验、进程生灭） │ ├── responder.py # 响应阻断（kill/封IP/隔离文件 + 回滚） │ ├── report.py # 报告生成（Markdown / HTML） │ ├── baseline.py # 安全基线哈希（可执行文件完整性校验） │ └── notifier.py # 邮件 / Webhook 实时告警推送 └── logs/ # 运行期生成：audit.db、baseline.json、报告、隔离区 ``` 各模块职责与关键接口： |**模块**|**职责**|**关键类 / 函数**| | --- | --- | --- | |`monitor.Monitor`|每个 tick 采集进程、网络连接、关键目录文件事件；首个快照登记"学习期基线进程名"|`snapshot()`、`_collect_processes()`、`_collect_connections()`、`_collect_fs()`| |`rules.RuleEngine`|按规则产出结构化 `Finding`（含严重等级与处置建议）|`evaluate()`、`Finding`| |`analyzer.Analyzer`|组装上下文、判新进程/消亡进程、补全攻击链与完整性信息、广度扫描篡改|`analyze()`、`_attack_chain()`、`_enrich_findings()`、`_integrity_findings()`| |`responder.Responder`|把处置建议落地为 kill / 封 IP / 隔离文件，阻断前自动快照、支持回滚|`handle()`、`_kill()`、`_block_ip()`、`_quarantine()`、`rollback()`| |`recorder.EventStore`|事件与回滚记录的写入/查询/导出|`record()`、`query()`、`save_rollback()`、`get_rollback()`、`list_rollbacks()`、`export_jsonl()`| |`baseline.BaselineManager`|对系统二进制建立 SHA-256 基线并做 mtime 优先的篡改校验|`build()`、`load()`、`check()`| |`notifier.Notifier`|命中发现的 SMTP 邮件 / Webhook 推送，含等级过滤与节流|`should_alert()`、`alert_finding()`| 设计上严格遵循 **单一职责** 与 **开闭原则（OCP）**：规则引擎对外只暴露 `evaluate(ctx)`，新增检测能力只需在内部加一个 `_check_xxx` 方法并在 `evaluate()` 登记一行，其余模块完全无感。 --- ## 三、核心数据模型 整个系统的数据流转围绕两个结构展开。 ### 3.1 快照（Snapshot） `monitor.Monitor.snapshot()` 每个轮询周期输出一份统一结构的快照，作为下游所有判断的输入： ```javascript { "ts": 1712345678.9, # 采集时刻 "processes": { # pid -> 进程信息 1234: { "pid": 1234, "ppid": 1, "name": "xmrig", "exe": "/tmp/xmrig", "cmdline": ["/tmp/xmrig", "-o", "..."], "username": "www", "cpu_percent": 92.3, "memory_percent": 5.1, "create_time": 1712345600.0, "status": "running", }, ... }, "connections": [ # 扁平网络连接列表 {"pid": 1234, "family": "...", "type": "...", "laddr": {"ip": "10.0.0.5", "port": 41234}, "raddr": {"ip": "1.2.3.4", "port": 4444}, "status": "ESTABLISHED"}, ... ], "fs_events": [ # 关键目录文件事件 {"action": "created", "path": "/tmp/.x", "mtime": 1712345601.0}, ], "baseline_names": {"sshd", "bash", ...}, # 学习期基线进程名 } ``` ### 3.2 发现（Finding） `rules.Finding` 是规则引擎的产物，也是连接"分析"与"响应"的契约对象： ```javascript @dataclass class Finding: rule_id: str # PROC-SUSPECT-NAME / NET-SUSPECT-PORT / ... title: str # 人类可读标题 severity: str # low | medium | high | critical category: str # process | network | resource | cmdline | integrity target: str # pid / ip:port / path evidence: Dict[str, Any] # 取证上下文（命令行、攻击链、完整性哈希等） recommendation: str # 阻断建议，直接驱动 responder timestamp: float = field(default_factory=time.time) ``` `Finding` 通过 `to_dict()` 序列化为 JSON 存入审计库；`evidence` 字段在 analyzer 阶段会被进一步富化（见第五节），从而带出完整的进程树与完整性信息。 --- ## 四、监测引擎（core/monitor.py） 监测引擎的本质是一个 **周期性状态采样器**。它需要解决三个工程问题：进程信息采集的健壮性、CPU 占用的精确计算、以及"学习期基线"的建立。 ### 4.1 进程采集与异常容错 `psutil.process_iter()` 会遍历全部进程；在遍历过程中进程可能瞬间消失、或被拒绝访问。`monitor` 用 `p.oneshot()`（一次系统调用批量读取该进程属性，减少 syscall 次数）配合 `try/except` 吞掉 `NoSuchProcess / AccessDenied / ZombieProcess` 三类常见异常，保证单进程出错不影响整个快照： ```javascript for p in psutil.process_iter(): try: with p.oneshot(): pid = p.pid cpu = self._cpu_percent(pid, p.cpu_times(), interval) procs[pid] = { ... } except (psutil.NoSuchProcess, psutil.AccessDenied, psutil.ZombieProcess): continue ``` ### 4.2 精确 CPU 占用：两次采样的时间差 `psutil` 单次 `cpu_percent()` 在没有历史基准时返回 0，直接拿来做阈值判断会产生误判。`monitor` 改为自行维护每个进程的累计 CPU 时间，用 **两次采样的时间差 / 采样间隔** 推算占用率，并除以核心数归一化到 0–100%： ```javascript def _cpu_percent(self, pid, cpu_times, _interval): cur = (cpu_times.user or 0) + (cpu_times.system or 0) prev = self._prev_cpu.get(pid) now = time.time() if prev is None: self._prev_cpu[pid] = (cur, now) return 0.0 prev_total, prev_ts = prev dt = now - prev_ts self._prev_cpu[pid] = (cur, now) if dt -j DROP` 封禁远端 IP。 - `_quarantine(path)`：把可疑文件移动到 `logs/quarantine/` 并追加 16 位 SHA-256 前缀命名，便于取证与恢复。 ### 6.3 防御性编程：五道安全闸门 `responder` 在真正改变系统前设置了多重保护，确保"阻断动作本身不会反噬系统"： 1. **保护关键进程**：`_kill` 拒绝终止 PID 0/1（调度器/init）、本进程及其父进程： ```javascript _PROTECTED_PIDS = lambda: {0, 1, os.getpid(), os.getppid()} ``` 1. **拒绝隔离系统二进制**：`_quarantine` 拒绝移动 `/bin`、`/sbin`、`/usr/bin`、`/usr/sbin`、`/lib`、`/lib64`、`/usr/lib`、`/usr/libexec`、`/usr/local/bin` 下的文件，避免误隔离导致系统瘫痪。 1. **IP 合法性校验**：`_block_ip` 先用 `ipaddress.ip_address()` 校验格式，非法 IP 不下发 iptables 规则。 1. **非 root 自动降级**：无权限时相关调用抛 `PermissionError`，动作记为 `permission_denied` 而非崩溃。 1. **回滚兜底**（见下节）。 ### 6.4 可回滚处置：误杀可恢复 每次破坏性动作执行前，`_begin_rollback()` 先把原状态快照写入 `recorder.rollbacks` 表（仅非 `dry_run` 模式留痕，避免噪声）。动作完成后 `_end_rollback()` 把状态置为 `applied` 并回填细节（如隔离目标路径）。误杀后执行 `rollback ` 即可逆向恢复： - `quarantine` → `shutil.move` 把文件移回原路径； - `firewall` → `iptables -D INPUT -s -j DROP` 撤销封禁； - `kill` → 按 `evidence.process.cmdline` 以列表形式 `subprocess.Popen` 重启进程（**不使用 shell**，规避命令注入）。 ```javascript def rollback(self, rid): rec = self.store.get_rollback(rid) if not rec: return {"status": "not_found", "msg": f"未找到回滚记录 {rid}"} ... if action == "firewall": ip = snap.get("ip") subprocess.run(["iptables", "-D", "INPUT", "-s", ip, "-j", "DROP"], ...) ``` 这正是第二节强调的"破坏性动作前必须先打快照"约束的代码落地——**哪怕阻断失败，也有据可查；哪怕误杀，也有路可退**。 --- ## 七、审计与留痕（core/recorder.py） 蓝方作战第一铁律是"全程留痕"。`EventStore` 以 SQLite 为底层存储，把监测引擎产生的每一类事件（进程、网络、文件系统、告警、响应动作）以结构化方式落盘。 ### 7.1 表结构与索引 ```javascript CREATE TABLE events ( id INTEGER PRIMARY KEY AUTOINCREMENT, ts REAL NOT NULL, category TEXT NOT NULL, event_type TEXT NOT NULL, pid INTEGER, severity TEXT NOT NULL DEFAULT 'info', data TEXT NOT NULL ); CREATE INDEX idx_evt_ts ON events(ts); CREATE INDEX idx_evt_cat ON events(category); CREATE INDEX idx_evt_sev ON events(severity); CREATE INDEX idx_evt_pid ON events(pid); CREATE TABLE rollbacks ( id INTEGER PRIMARY KEY AUTOINCREMENT, ts REAL NOT NULL, action TEXT NOT NULL, target TEXT NOT NULL, status TEXT NOT NULL, snapshot TEXT NOT NULL ); ``` - `category / event_type / severity / pid` 四维标签支撑多维检索； - `data` 以 JSON 序列化存储，保留任意结构化取证信息； - 所有时间戳统一使用 **UTC**，保证跨时区取证一致性。 ### 7.2 线程安全与导出 `EventStore` 用 `threading.RLock()` 包裹所有读写，确保多模块并发写库不丢数据；`check_same_thread=False` 允许跨线程复用连接。`export_jsonl()` 把事件导出为 JSONL，可直接对接 ELK / Splunk 等 SIEM 平台： ```javascript def export_jsonl(self, path): n = 0 with open(path, "w", encoding="utf-8") as fh, self._lock: for r in self._conn.execute( "SELECT ts,category,event_type,pid,severity,data FROM events " "ORDER BY ts ASC"): rec = {"ts": r[0], "category": r[1], "event_type": r[2], "pid": r[3], "severity": r[4], "data": json.loads(r[5])} fh.write(json.dumps(rec, ensure_ascii=False, default=str) + "\n") n += 1 return n ``` --- ## 八、安全基线哈希（core/baseline.py） 完整性校验是检测"后门替换系统二进制"（如替换 `/bin/bash`）的关键手段。 `BaselineManager` 对 `/bin`、`/sbin`、`/usr/bin`、`/usr/sbin` 等可信目录递归计算 SHA-256 基线并持久化为 `logs/baseline.json`。运行期 `check()` 采用 **mtime 优先** 的优化策略： ```javascript def check(self, path): if path not in self._baseline: return {"tracked": False} entry = self._baseline[path] try: st = Path(path).stat() except OSError: return {"tracked": True, "changed": True, "missing": True, ...} # mtime 未变视为未改动，跳过昂贵的哈希计算 if abs(st.st_mtime - entry.get("mtime", 0)) `。 --- ## 十一、主程序编排（[main.py](http://main.py/)） `main.py` 通过 `argparse` 暴露三个子命令，并把各模块串成闭环： ```javascript def cmd_run(args, cfg): ... while running: snap = monitor.snapshot(interval) result = analyzer.analyze(snap, prev) prev = snap for np in result["new_processes"]: store.record("process", "spawn", np, pid=np["pid"], severity="info") ... for f in result["findings"]: store.record("alert", f["rule_id"], f, severity=f["severity"]) if notifier.should_alert(f, mode): notifier.alert_finding(f) resp = responder.handle(f) ... time.sleep(interval) if args.duration and (time.time() - start) >= args.duration: break ``` 关键工程点： - 通过 `signal.signal(SIGINT/SIGTERM, _stop)` 注册优雅退出，Ctrl+C 后进入 `finally` 生成报告并关闭数据库； - `-d/--duration` 支持限时运行（教学/演示），`0` 表示持续运行； - `--report 路径前缀` 控制报告输出位置（自动生成 `.md` 与 `.html`）。 --- ## 十二、典型实战场景走查 假设红方通过 SSH 暴破进入主机，投放挖矿木马 `xmrig` 并建立到 `1.2.3.4:4444` 的反连： 1. **监测**：monitor 采集到新进程 `xmrig`、一条到 `4444` 的 `ESTABLISHED` 连接、`/tmp` 下新建文件。 1. **分析**：`PROC-SUSPECT-NAME` 命中进程名、`NET-SUSPECT-PORT` 命中高危端口；`_attack_chain()` 还原出 `systemd → sshd → bash → xmrig` 攻击链。 1. **告警**：notifier 经邮件/Webhook 推送高危告警，正文含完整攻击链。 1. **响应（manual 模式）**：终端逐条确认 `kill xmrig` / `封禁 1.2.3.4` / `隔离 /tmp/xmrig`，每条动作前自动写入回滚快照。 1. **报告**：运行结束生成战报与态势图，列出发现、动作、可回滚快照与时间线。 1. **误杀兜底**：若 `bash` 被误杀，执行 `rollback ` 即按其原命令行重启。 --- ## 十三、安装、部署与命令参考 ```javascript # 1. 安装依赖 pip install -r requirements.txt # 2. 建立完整性基线（首次 run 若缺失且 auto_build=true 会自动建立） sudo python main.py -c config.yaml baseline-build # 3. 试运行（dry_run：只记录不执行任何阻断；跑 60 秒后出报告） python main.py -c config.yaml run --mode dry_run -d 60 --report logs/ops # 4. 正式值守（需 root 才能 kill 他人进程 / 写 iptables） sudo python main.py -c config.yaml # 5. 误杀后回滚（示例 id=7） sudo python main.py -c config.yaml rollback 7 ``` |**命令**|**参数**|**说明**| | --- | --- | --- | |`run`（默认）|`-d/--duration 秒`、`-m/--mode dry_run\|manual\|auto`、`--report 路径前缀`|启动实时监测闭环| |`baseline-build`|无|立即建立/重建可执行文件完整性基线| |`rollback`|``（回滚记录自增 id）|按回滚快照恢复一次处置| > ⚠️ 阻断动作（终止进程、iptables 封 IP）需要 **root 权限**；没有 root 时平台自动降级，不会崩溃。 --- ## 十四、合规与安全声明 本工具 **仅面向蓝方 / 防守方**，用于授权环境下的资产监测、入侵检测与应急处置。使用者须确保对该系统拥有合法授权。阻断动作具有破坏性（终止进程、封禁 IP），请在生产环境前先用 `dry_run` 验证规则，并妥善保管 `logs/audit.db` 审计记录与 `logs/baseline.json` 基线。 --- # 附录：完整源码 > 源码来自 [https://cnb.cool/unicom/SentinelBlue](https://cnb.cool/unicom/SentinelBlue)
