资讯详情

项目来源：https://cnb.cool/unicom/SentinelBlue
关键词：入侵检测、实时响应、进程树回溯、文件完整性校验、审计留痕、可回滚处置

---

摘要

蓝方（防守方）在对抗红方（攻击方）时，最大的痛点往往不是"看不见"，而是 “看见了却没记全、记全了却来不及断”。红方的攻击是一条有严格时间顺序的链路：

投递 → 落地 → 执行 → 反弹 → 驻留

SentinelBlue 把上述环节收敛为一条 实时闭环，在每一次进程生灭、每一条异常外连、每一次阻断动作发生时，都同步完成"记录—分析—响应—报告"，并让全部行为可回溯、可取证、可复盘。本文从工程实现角度，逐模块拆解其架构设计、关键算法与防御性编程思想，并在文末附上完整源码。

---

一、问题建模：为什么需要一条闭环

传统 HIDS（主机入侵检测系统）常见的两个断点是：

1. 检测与响应解耦。告警进了 SIEM，但处置还要人去敲命令，时间线在"看到"与"动手"之间断裂。

2. 缺乏可溯源的中间态。很多脚本只管"杀进程"，却不留审计；一旦误杀，无法还原现场，也无法向合规方证明"我做过什么、为什么做"。

SentinelBlue 的设计目标是把这两件事放进同一条主循环：

监测(monitor) → 记录(recorder) → 分析(analyzer) → 阻断(responder) → 报告(report)↑                                                                  │└──────────────────── 全程留痕 / 可溯源 ─────────────────────────┘

核心约束是：无论响应成功、失败、跳过还是被拒绝，动作本身都必须先落盘；执行任何破坏性动作前，必须先对原状态打快照。这两条约束是整个系统的"安全灵魂"。

---

二、整体架构与目录结构

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() 每个轮询周期输出一份统一结构的快照，作为下游所有判断的输入：

{"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 是规则引擎的产物，也是连接"分析"与"响应"的契约对象：

@dataclass
class Finding:rule_id: str           # PROC-SUSPECT-NAME / NET-SUSPECT-PORT / ...title: str             # 人类可读标题severity: str          # low | medium | high | criticalcategory: str          # process | network | resource | cmdline | integritytarget: str           # pid / ip:port / pathevidence: Dict[str, Any]   # 取证上下文（命令行、攻击链、完整性哈希等）recommendation: str    # 阻断建议，直接驱动 respondertimestamp: 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 三类常见异常，保证单进程出错不影响整个快照：

for p in psutil.process_iter():try:with p.oneshot():pid = p.pidcpu = 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%：

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.0prev_total, prev_ts = prevdt = now - prev_tsself._prev_cpu[pid] = (cur, now)if dt <= 0:return 0.0cores = psutil.cpu_count() or 1return min(100.0, ((cur - prev_total) / dt) * 100.0 / cores)

这一细节是挖矿类进程（长时间占满单核）能被稳定识别的前提。

4.3 学习期基线进程名

首个快照时，monitor 把当时系统中已存在的进程名登记为"学习期基线"。后续判断"新进程"时，若进程名属于基线集合，则视为正常启动噪声而非入侵信号——这是区分"系统噪声"与"真实异常"的关键一步：

if self._first:self.baseline_names = {p["name"] for p in snap["processes"].values()}self._first = False
snap["baseline_names"] = self.baseline_names

4.4 网络与文件系统采集

• 网络：psutil.net_connections(kind="inet") 返回全量连接。非 root 环境下会抛出 AccessDenied，monitor 捕获后静默降级（仅保留自身进程连接可见），保证平台不崩溃。

• 文件系统：对 fs_watch_paths（默认 /tmp /etc /home /var/www）递归遍历，以 mtime 变化判定 created / modified。这是一条"轻量但有效"的落地点检测——/tmp 下的新文件、/etc 的篡改都会被即时记录。

---

五、规则引擎与分析引擎

规则引擎负责"产出发现"，分析引擎负责"把发现讲清楚"。

5.1 规则引擎的开放扩展点

RuleEngine.evaluate(ctx) 把多个 _check_xxx 方法的产出拼接成 Finding 列表：

def evaluate(self, ctx):findings = []findings += self._check_suspicious_process(ctx)findings += self._check_cmdline(ctx)findings += self._check_network(ctx)findings += self._check_resource(ctx)return findings

内置 6 条规则覆盖进程名、命令行、高危端口、连接数、CPU/内存、文件完整性（最后一类由 analyzer 注入，见 5.3）。每条规则命中即生成一个带 recommendation 的 Finding，recommendation 字段直接决定后续响应动作——这就是"发现即处置"的契约。

扩展只需两步（开闭原则）：

def _check_my_rule(self, ctx):out = []for pid, p in ctx["processes"].items():if "evil" in (p.get("name") or "").lower():out.append(Finding(rule_id="PROC-EVIL", severity="high", category="process",target=str(pid), title=f"命中自定义规则：{p.get('name')}",evidence={"name": p.get("name")},recommendation=f"终止进程 PID={pid}",))return out
# 在 evaluate() 中加一行：findings += self._check_my_rule(ctx)

5.2 进程生灭判定

analyzer.Analyzer 维护 known_pids 集合与上一快照对比，精确给出"本 tick 新出现的进程"与"本 tick 消亡的进程"。新进程的 is_baseline 标记来自 monitor 下发的 baseline_names，用于在执行流上区分"基线内新进程"与"真正的异常进程"：

for pid, p in current["processes"].items():if pid not in self.known_pids and pid not in prev_procs:baseline = current.get("baseline_names", self.baseline_names)new_procs.append({"pid": pid, "name": p["name"], "cmdline": p["cmdline"],"age": round(time.time() - p["create_time"], 1),"is_baseline": p["name"] in baseline,})self.known_pids.add(pid)

5.3 攻击链回溯（进程树关联）

这是本平台最具实战价值的能力之一。analyzer._attack_chain() 从可疑进程出发，沿 ppid 逐级向上回溯到根进程，再 reverse() 还原为"根 → … → 目标"的攻击链。例如一次典型的 SSH 暴破后种植挖矿木马的链路会被还原为：

systemd(PID=1) → sshd(PID=952) → bash(PID=1331) → xmrig(PID=1234)

实现上用 seen 集合防止环状 PID（极端情况下 _kill 自身父链）造成死循环，并在命中 ppid in (0, None) 时终止回溯：

def _attack_chain(self, processes, pid):chain, seen, cur = [], set(), pidwhile cur and cur not in seen and cur in processes:seen.add(cur)p = processes[cur]chain.append({"pid": cur, "name": p.get("name"),"cmdline": p.get("cmdline"),"username": p.get("username"), "ppid": p.get("ppid")})cur = p.get("ppid")if cur in (0, None):breakchain.reverse()return chain

该链路被写入 Finding.evidence["process_tree"]，最终出现在实时告警、报告与邮件正文里，让值守人员一眼看清"这个恶意进程是怎么来的"。

5.4 完整性富化与广度扫描

_enrich_findings() 为每个 Finding 补充进程的完整上下文与完整性校验结果；_integrity_findings() 则对 当前所有进程的可执行文件 做一次广度扫描——只要基线内的系统二进制被替换或删除，立即产出高危 INTEGRITY-TAMPER 告警，即使该进程本身并未触发其他规则：

def _integrity_findings(self, processes):out, seen = [], set()for p in processes.values():exe = p.get("exe")if not exe or exe in seen:continueseen.add(exe)info = self.baseline.check(exe)if info.get("changed"):out.append(Finding(rule_id="INTEGRITY-TAMPER", severity="high", category="integrity",target=exe, title=f"系统二进制完整性被破坏：{exe}", ...))return out

---

六、响应与阻断（core/responder.py）

Responder 把 Finding 翻译成具体动作，是整条链路里 唯一会真正改变系统状态 的环节，因此其防御性编程最重。

6.1 三种响应模式

由 config.yaml 的 response.mode 或命令行 -m 决定：

模式	行为	适用场景
dry_run	仅记录、打印动作，不执行任何阻断	试运行、规则调优、教学演示
manual	每个阻断动作交互确认（y/N）	正式值守、人工把关（默认）
auto	自动执行；require_confirm: true 时关键动作仍需确认	高置信规则下的无人值守

handle() 依据 category 派发动作。注意一个工程细节：网络类发现（如 NET-CONN-FLOOD）的 evidence 里不一定带 pid，因此在回退逻辑里允许直接用 target（其本身即为 PID 字符串）继续处置，避免处置链路断裂：

pid = ev.get("pid")
if pid is None and target and target.isdigit():pid = int(target)

6.2 三类动作实现

• _kill(pid)：os.kill(pid, 9) 强杀进程。

• _block_ip(ip)：iptables -A INPUT -s <ip> -j DROP 封禁远端 IP。

• _quarantine(path)：把可疑文件移动到 logs/quarantine/ 并追加 16 位 SHA-256 前缀命名，便于取证与恢复。

6.3 防御性编程：五道安全闸门

responder 在真正改变系统前设置了多重保护，确保"阻断动作本身不会反噬系统"：

1. 保护关键进程：_kill 拒绝终止 PID 0/1（调度器/init）、本进程及其父进程：

   _PROTECTED_PIDS = lambda: {0, 1, os.getpid(), os.getppid()}
   

2. 拒绝隔离系统二进制：_quarantine 拒绝移动 /bin、/sbin、/usr/bin、/usr/sbin、/lib、/lib64、/usr/lib、/usr/libexec、/usr/local/bin 下的文件，避免误隔离导致系统瘫痪。

3. IP 合法性校验：_block_ip 先用 ipaddress.ip_address() 校验格式，非法 IP 不下发 iptables 规则。

4. 非 root 自动降级：无权限时相关调用抛 PermissionError，动作记为 permission_denied 而非崩溃。

5. 回滚兜底（见下节）。

6.4 可回滚处置：误杀可恢复

每次破坏性动作执行前，_begin_rollback() 先把原状态快照写入 recorder.rollbacks 表（仅非 dry_run 模式留痕，避免噪声）。动作完成后 _end_rollback() 把状态置为 applied 并回填细节（如隔离目标路径）。误杀后执行 rollback <id> 即可逆向恢复：

• quarantine → shutil.move 把文件移回原路径；

• firewall → iptables -D INPUT -s <ip> -j DROP 撤销封禁；

• kill → 按 evidence.process.cmdline 以列表形式 subprocess.Popen 重启进程（不使用 shell，规避命令注入）。

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 表结构与索引

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 平台：

def export_jsonl(self, path):n = 0with 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 += 1return n

---

八、安全基线哈希（core/baseline.py）

完整性校验是检测"后门替换系统二进制"（如替换 /bin/bash）的关键手段。

BaselineManager 对 /bin、/sbin、/usr/bin、/usr/sbin 等可信目录递归计算 SHA-256 基线并持久化为 logs/baseline.json。运行期 check() 采用 mtime 优先 的优化策略：

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)) < 1e-6:return {"tracked": True, "changed": False, "hash": entry["hash"]}cur = self._hash(Path(path))return {"tracked": True, "changed": cur != entry["hash"], ...}

即：绝大多数真实场景下文件 mtime 不变，直接跳过哈希计算，运行期开销极低；只有当 mtime 变化时才重新哈希比对。支持 excludes（*.pyc 等）跳过无关文件。

---

九、实时告警推送（core/notifier.py）

命中发现的即时推送全部走标准库（smtplib / urllib），无第三方依赖，并具备三重鲁棒性设计：

1. 严重等级过滤：_SEV_ORDER 把 info/low/medium/high/critical 映射为数值，仅推送 ≥ min_severity（默认 medium）的发现；

2. 同规则节流：should_alert() 用 _last[rule_id] 记录上次推送时刻，间隔 < min_interval（默认 60s）则抑制，避免告警风暴；

3. 失败静默降级：任何推送异常都被 except 捕获并仅打印提示，绝不阻断主监测链路；dry_run 试运行默认不推送（可由 alert_in_dry_run 开启）。

def should_alert(self, finding, mode):if not self.enabled:return Falseif mode == "dry_run" and not self.alert_in_dry_run:return Falsesev = _SEV_ORDER.get(finding.get("severity"), 0)if sev < _SEV_ORDER.get(self.min_sev, 2):return Falserid = finding.get("rule_id")now = time.time()if rid in self._last and (now - self._last[rid]) < self.min_interval:return Falseself._last[rid] = nowreturn True

Webhook 以 POST JSON 推送，可对接企业微信 / 钉钉 / Slack / 自研平台。

---

十、报告生成（core/report.py）

运行结束时 ReportGenerator.build() 从审计库汇总，同时产出两种形态：

• Markdown 战报：适合即时阅读与向上级汇报，含安全发现清单（带攻击链、完整性差异）、已执行阻断动作、可回滚快照、关键时间线；

• HTML 态势报告：带严重等级色标（low=蓝 / medium=橙 / high=红 / critical=暗红），适合留存与演示。

两者均支持在事件中 rollback_id 字段标注"可回滚 #id"，并在回滚章节给出恢复命令 python main.py -c config.yaml rollback <id>。

---

十一、主程序编排（main.py）

main.py 通过 argparse 暴露三个子命令，并把各模块串成闭环：

def cmd_run(args, cfg):...while running:snap = monitor.snapshot(interval)result = analyzer.analyze(snap, prev)prev = snapfor 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 下新建文件。

2. 分析：PROC-SUSPECT-NAME 命中进程名、NET-SUSPECT-PORT 命中高危端口；_attack_chain() 还原出 systemd → sshd → bash → xmrig 攻击链。

3. 告警：notifier 经邮件/Webhook 推送高危告警，正文含完整攻击链。

4. 响应（manual 模式）：终端逐条确认 kill xmrig / 封禁 1.2.3.4 / 隔离 /tmp/xmrig，每条动作前自动写入回滚快照。

5. 报告：运行结束生成战报与态势图，列出发现、动作、可回滚快照与时间线。

6. 误杀兜底：若 bash 被误杀，执行 rollback <id> 即按其原命令行重启。

---

十三、安装、部署与命令参考

# 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>（回滚记录自增 id）	按回滚快照恢复一次处置

⚠️ 阻断动作（终止进程、iptables 封 IP）需要 root 权限；没有 root 时平台自动降级，不会崩溃。

---

十四、合规与安全声明

本工具 仅面向蓝方 / 防守方，用于授权环境下的资产监测、入侵检测与应急处置。使用者须确保对该系统拥有合法授权。阻断动作具有破坏性（终止进程、封禁 IP），请在生产环境前先用 dry_run 验证规则，并妥善保管 logs/audit.db 审计记录与 logs/baseline.json 基线。

---

附录：完整源码

源码来自 https://cnb.cool/unicom/SentinelBlue