简介：这是一套面向网络安全初学者与开发者的主机安全态势感知系统完整实现方案，基于Python与HTML技术栈构建，适合用于课程设计、毕业设计或安全监控类项目参考。系统覆盖数据采集、处理分析、可视化展示与报警机制等核心模块，帮助读者理解从主机日志、进程、网络连接等数据中识别潜在威胁的完整链路。资源包共20个文件，以8个py源码文件与9个pyc编译文件为主，另含1个mmdb地理库、1个gitignore及1个txt说明，压缩包约20.04MB，目录中可见app.py入口及WorldMapChart、AttackStatusChart等图表模块，结构清晰便于二次开发。目前已有1204人学习下载，读者可从中获取态势感知系统的架构设计思路、Python数据采集与前端图表联动实现，以及多线程处理与数据库存储的工程化参考，适合需要快速搭建安全监控原型的开发者借鉴。
1. 主机安全态势感知系统到底在感知什么：从一台被挖矿的测试机说起
去年有台放在机房的测试机，CPU 连续三天跑满，登录上去top一看是个陌生进程在挖矿。问题不在于中招，而在于中招三天没人发现——lastb里几万条爆破记录躺着，/var/log/secure早就写满了异常登录，可这些日志没人看，看了也看不出规律。这就是主机安全态势感知系统要解决的事：把散落在单机上的登录日志、进程行为、端口监听、文件变更这些碎片，聚合成一个能看出「现在整批主机处于什么状态、有没有正在发生的攻击」的视图。
用 Python 做采集与分析、用 HTML 做前端展示，是这类系统最务实的组合。Python 生态里psutil、paramiko、pandas能快速把主机指标和日志拉回来，Flask 或 FastAPI 起一个轻量服务，前端用原生 HTML + ECharts 就能把态势画出来，不需要上重型 SIEM。适合谁做：手里管着十几到上百台 Linux 主机、想自己搭一套能落地的监控告警、又不想被商业产品绑定的运维和开发。下面按「采什么 → 怎么采 → 怎么算 → 怎么展示 → 坑在哪」把整套东西拆开讲。
2. 数据采集层：主机指标、登录日志、进程行为怎么落到 Python 里
态势感知的地基是数据。采不到、采不准，后面画出来的图全是玄学。这一层要解决三件事：采哪些字段、用什么方式采、采回来怎么统一格式。
2.1 先定字段：主机态势最小数据集
不要一上来就想采全，先把能反映「安全状态」的最小字段集定下来。我一般分四类：
| 类别 | 关键字段 | 采集来源 | 反映的问题 |
|---|---|---|---|
| 系统负载 | CPU、内存、磁盘、网络连接数 | psutil | 挖矿、异常进程 |
| 登录行为 | 用户名、源 IP、时间、成功/失败 | /var/log/secure、lastb | 暴力破解、异常登录 |
| 进程信息 | PID、进程名、命令行、父进程 | psutil.process_iter | 可疑进程、反弹 shell |
| 端口监听 | 端口、协议、监听地址、所属进程 | psutil.net_connections | 后门端口、未授权服务 |
字段定完再动手，否则代码写一半发现少采了源 IP，回头补采集逻辑很痛苦。这套字段集覆盖了 80% 的主机入侵特征，剩下的文件完整性、内核模块这些可以二期再加。
2.2 用 psutil 采本机指标：一段能直接跑的采集函数
本机采集用psutil最省事，不用起 agent 进程，一个函数就能把核心指标拿全。
import psutil import time import socket def collect_local_metrics(): """采集本机核心安全指标，返回统一结构的 dict""" metrics = { "hostname": socket.gethostname(), "timestamp": int(time.time()), # CPU 使用率，interval=1 表示采样 1 秒，避免瞬时值抖动 "cpu_percent": psutil.cpu_percent(interval=1), # 内存：total/used/percent 三个值一起存，方便前端算趋势 "mem": dict(psutil.virtual_memory()._asdict()), # 磁盘只取根分区，多分区场景改成遍历 psutil.disk_partitions() "disk": dict(psutil.disk_usage("/")._asdict()), # 网络连接：过滤出 ESTABLISHED 和 LISTEN，其他状态噪声大 "connections": [] } for conn in psutil.net_connections(kind="inet"): if conn.status in ("ESTABLISHED", "LISTEN"): metrics["connections"].append({ "laddr": f"{conn.laddr.ip}:{conn.laddr.port}" if conn.laddr else "", "raddr": f"{conn.raddr.ip}:{conn.raddr.port}" if conn.raddr else "", "status": conn.status, "pid": conn.pid }) return metrics逻辑说明：cpu_percent(interval=1)里的 interval 不能省，省了拿到的是自上次调用以来的平均值，第一次调用基本是 0，这是新手最常翻的车。net_connections在 Linux 上需要 root 或 CAP_NET_ADMIN 权限才能看到其他用户的连接，普通用户跑只能看到自己的，部署时要么用 root，要么给 Python 解释器加 capability。
参数说明：kind="inet"只取 IPv4/IPv6，不取 Unix socket；如果主机连接数上万，这个循环会慢，实际部署时加个limit或只取 LISTEN 状态做端口监控，ESTABLISHED 单独用采样频率控制。
2.3 远程主机怎么采：paramiko 拉日志的边界
管多台主机时，常见做法是用paramiko走 SSH 拉日志和指标，不额外装 agent。核心是封装一个执行远程命令的函数：
import paramiko def run_remote_cmd(host, user, key_path, cmd, timeout=10): """通过 SSH 在远程主机执行命令并返回 stdout""" client = paramiko.SSHClient() # 首次连接自动接受 host key，生产环境应预置 known_hosts client.set_missing_host_key_policy(paramiko.AutoAddPolicy()) try: client.connect(host, username=user, key_filename=key_path, timeout=timeout) stdin, stdout, stderr = client.exec_command(cmd, timeout=timeout) return stdout.read().decode("utf-8", errors="ignore") finally: client.close() # 拉取最近 100 条失败登录 cmd = "lastb -n 100 2>/dev/null || tail -n 100 /var/log/secure" raw = run_remote_cmd("10.0.0.5", "ops", "/home/ops/.ssh/id_rsa", cmd)逻辑说明：AutoAddPolicy方便但有中间人风险，生产环境应该把目标主机指纹预置到known_hosts，用RejectPolicy。lastb读的是/var/log/btmp，需要 root 权限，普通用户拿不到，所以命令里加了 fallback 到/var/log/secure。
参数说明：timeout要设，否则一台主机网络卡住会拖死整个采集循环；采集频率建议 1 分钟一次，登录日志可以 5 分钟一次，太频繁对目标机有压力，也容易触发安全设备的告警。
3. 分析层：把原始日志变成「态势」的四个计算步骤
采回来的数据是原料，态势是算出来的。这一层做四件事：日志解析、暴力破解识别、异常进程判定、态势评分。
3.1 登录日志解析：正则要扛得住格式差异
/var/log/secure的格式在不同发行版、不同 SSH 版本间有差异，正则写死一种格式迟早翻车。稳妥做法是写多条正则依次匹配：
import re # 覆盖常见 sshd 失败登录格式 PATTERNS = [ re.compile(r"Failed password for (?:invalid user )?(?P<user>\S+) from (?P<ip>\d+\.\d+\.\d+\.\d+)"), re.compile(r"Invalid user (?P<user>\S+) from (?P<ip>\d+\.\d+\.\d+\.\d+)"), re.compile(r"Accepted password for (?P<user>\S+) from (?P<ip>\d+\.\d+\.\d+\.\d+)"), ] def parse_auth_log(lines): events = [] for line in lines: for pat in PATTERNS: m = pat.search(line) if m: events.append({ "user": m.group("user"), "ip": m.group("ip"), "success": "Accepted" in line, "raw": line.strip() }) break # 匹配到一条就跳出，避免重复计数 return events逻辑说明：break很关键，一条日志可能同时匹配多条正则（比如 Invalid user 那条也含 IP），不 break 会重复计数，导致后面统计爆破次数虚高。success字段用Accepted判断，比再写一条正则省事。
参数说明：如果日志里有 IPv6，\d+\.\d+\.\d+\.\d+匹配不到，需要补一条 IPv6 的正则，或者用ipaddress模块做校验。日志量大的时候，逐行正则匹配是瓶颈，可以先用if "Failed" in line or "Accepted" in line做粗筛再进正则。
3.2 暴力破解识别：滑动窗口比阈值判断更靠谱
单纯统计「失败次数 > N」会误报，正常用户输错几次密码很常见。用滑动窗口统计单位时间内的失败次数更准：
from collections import defaultdict, deque import time class BruteForceDetector: def __init__(self, window_sec=300, threshold=10): self.window = window_sec # 时间窗口：5 分钟 self.threshold = threshold # 窗口内失败次数阈值 self.records = defaultdict(deque) # key: ip, value: 失败时间戳队列 def add_failure(self, ip, ts=None): ts = ts or time.time() q = self.records[ip] q.append(ts) # 踢掉窗口外的旧记录 while q and ts - q[0] > self.window: q.popleft() return len(q) >= self.threshold # 返回是否触发告警 def top_attackers(self, n=10): return sorted( ((ip, len(q)) for ip, q in self.records.items()), key=lambda x: x[1], reverse=True )[:n]逻辑说明：用deque存时间戳，每次新增时从左侧踢掉过期记录，队列长度就是窗口内的失败次数。这样不用定时清理，内存也不会无限涨。top_attackers直接给前端提供「攻击源 TOP N」的数据。
参数说明：window_sec=300、threshold=10是经验值，内网主机可以放宽到 20，暴露在公网的主机收紧到 5。这两个参数要根据自己环境的基线调，没有万能值。注意records字典会随攻击 IP 增多而膨胀，长期运行要加个定期清理，把窗口内无记录的 IP 删掉。
3.3 异常进程判定：白名单 + 行为特征双管齐下
进程异常判定没有银弹，纯白名单维护成本高，纯行为检测误报多。我的做法是两层：先过白名单，白名单外的再看行为特征。
# 常见合法进程白名单（按需扩充） WHITELIST = {"sshd", "systemd", "cron", "nginx", "mysqld", "python3", "java"} # 可疑行为特征 SUSPICIOUS_PATTERNS = [ "/tmp/", # 从 /tmp 执行的进程 "base64 -d", # 解码执行 "curl | sh", # 远程脚本直接执行 "nc -e", # netcat 反弹 shell "/dev/shm/", # 内存目录执行 ] def check_process(proc_info): """proc_info: {"name":..., "cmdline":..., "exe":...}""" name = proc_info.get("name", "") cmdline = " ".join(proc_info.get("cmdline") or []) if name in WHITELIST: return None for pat in SUSPICIOUS_PATTERNS: if pat in cmdline: return f"命中可疑特征: {pat}" return None逻辑说明：白名单先放行，减少误报；白名单外的进程再看命令行里有没有可疑特征。cmdline可能是 None（内核线程），所以用or []兜底。
参数说明：SUSPICIOUS_PATTERNS要按自己环境调，比如有些业务确实会从/tmp跑脚本，那就把这条去掉或改成更精确的匹配。白名单不要写太宽，python3放进去意味着任何 Python 脚本都不告警，实际应该结合exe路径判断。
3.4 态势评分：把多维指标压成一个 0-100 的数
前端要一个直观的「安全分」，就得把多个维度的指标归一化后加权。做法是先给每个维度算风险分，再加权求和：
def calc_risk_score(metrics): """输入采集指标，输出 0-100 风险分，越高越危险""" score = 0 # CPU 持续高于 80% 加 20 分 if metrics["cpu_percent"] > 80: score += 20 # 内存高于 90% 加 15 分 if metrics["mem"]["percent"] > 90: score += 15 # 磁盘高于 85% 加 10 分 if metrics["disk"]["percent"] > 85: score += 10 # 失败登录窗口内超阈值加 30 分 if metrics.get("brute_force_hit"): score += 30 # 命中可疑进程加 25 分 if metrics.get("suspicious_proc"): score += 25 return min(score, 100)逻辑说明：每个维度的权重反映它对安全的威胁程度，暴力破解和可疑进程权重最高，资源占用权重低。min(score, 100)防止累加超过 100。
参数说明：权重不是固定的，如果环境里 CPU 高是常态（比如跑计算任务），就把 CPU 那项权重降到 5 或去掉。评分只是给个直观参考，真正的告警还是要靠单项阈值触发，不要只盯总分。
4. 展示层：HTML 前端怎么把态势画出来又不卡
后端算完数据，前端要能实时看到。这一层的关键是：数据接口设计、图表选型、刷新策略。
4.1 后端接口：Flask 三个路由搞定
用 Flask 起服务，暴露三个接口：当前态势、历史趋势、攻击源列表。
from flask import Flask, jsonify, render_template app = Flask(__name__) @app.route("/") def index(): # 返回 HTML 页面 return render_template("dashboard.html") @app.route("/api/current") def api_current(): # 返回最新一次采集的态势数据 return jsonify(get_latest_metrics()) @app.route("/api/trend") def api_trend(): # 返回最近 1 小时的趋势数据，供折线图使用 return jsonify(get_trend_data(hours=1)) @app.route("/api/attackers") def api_attackers(): # 返回攻击源 TOP 10 return jsonify(detector.top_attackers(10))逻辑说明：接口和页面分离，前端用fetch拿 JSON 自己渲染，后端不拼 HTML，改前端不用动 Python。/api/current返回单条，/api/trend返回数组，职责分清。
参数说明：get_trend_data(hours=1)里的时间范围要跟采集频率匹配，1 分钟采一次、1 小时就是 60 个点，折线图正好。如果采集频率是 5 分钟，1 小时只有 12 个点，图会很稀，可以拉 6 小时。
4.2 前端图表：ECharts 折线图 + 仪表盘
前端用 ECharts 最省事，折线图看趋势，仪表盘看当前分。核心是定时拉数据：
// 每 30 秒刷新一次态势数据 async function refreshDashboard() { const res = await fetch('/api/current'); const data = await res.json(); // 更新仪表盘 gaugeChart.setOption({ series: [{ data: [{ value: data.risk_score }] }] }); // 更新折线图 const trend = await (await fetch('/api/trend')).json(); lineChart.setOption({ xAxis: { data: trend.map(d => d.time) }, series: [{ data: trend.map(d => d.cpu) }] }); } setInterval(refreshDashboard, 30000); refreshDashboard(); // 首次立即执行，不等 30 秒逻辑说明：setInterval前先手动调一次，否则页面打开后要等 30 秒才有数据，体验差。两个 fetch 可以并行，用Promise.all更快，这里为了可读性分开写。
参数说明：刷新间隔 30 秒是折中值，太短后端压力大，太长态势更新不及时。如果主机数量多，改成 WebSocket 推送比轮询更省资源，但实现复杂度上升，小规模用轮询够了。
4.3 页面结构：一个 HTML 文件的最小骨架
前端不用框架，一个 HTML 文件加 ECharts CDN 就能跑：
<!DOCTYPE html> <html lang="zh-cn"> <head> <meta charset="utf-8"> <title>主机安全态势感知</title> <script src="https://cdn.jsdelivr.net/npm/echarts@5/dist/echarts.min.js"></script> </head> <body> <div id="gauge" style="width:400px;height:300px;"></div> <div id="trend" style="width:800px;height:300px;"></div> <script> const gaugeChart = echarts.init(document.getElementById('gauge')); const lineChart = echarts.init(document.getElementById('trend')); // 初始化配置省略，按 ECharts 文档填 option </script> </body> </html>逻辑说明：<!DOCTYPE html>和meta charset是必须的，少了 charset 中文会乱码，这是新手最常见的翻车点。ECharts 用 CDN 引入，内网环境要换成本地文件。
参数说明：图表容器必须给明确的宽高，否则 ECharts 初始化时拿不到尺寸，图不显示。响应式场景下监听window.resize调chart.resize()。
5. 避坑与排查：这套系统上线后最容易翻车的五个地方
5.1 采集脚本把目标机 CPU 拉满
现象：部署采集后，被监控主机负载反而升高，top里看到 Python 进程占 CPU。
原因：psutil.net_connections()在连接数多时遍历开销大，加上采集频率过高（比如 5 秒一次），累积起来很可观。
解决：采集频率降到 1 分钟一次；net_connections只取 LISTEN 状态做端口监控，ESTABLISHED 用ss -s拿汇总数代替全量遍历；采集脚本加nice降低优先级。
5.2 日志解析正则匹配不到，态势图全是 0
现象：前端图表一直显示 0，检查发现登录事件数为 0。
原因：目标机是 CentOS，日志在/var/log/secure，但脚本默认读的是/var/log/auth.log（Ubuntu 路径），文件不存在，读回来是空。
解决：采集前先判断日志文件存在性，secure和auth.log都试；或者用journalctl -u sshd统一拿，但要注意 journalctl 的输出格式和文本日志不同，正则要另写。
5.3 暴力破解告警风暴，一晚上几千条
现象：告警群里一晚上刷了几千条「暴力破解」告警，全是同一个 IP。
原因：阈值设太低（失败 3 次就告警），加上公网主机每天被扫是常态，导致告警疲劳。
解决：阈值提到 10 次/5 分钟；加告警去重，同一 IP 在冷却期（比如 30 分钟）内只告警一次；对已知的扫描 IP 段做白名单，不告警只记录。
5.4 前端页面打开慢，图表卡顿
现象：态势页面加载要十几秒，折线图拖动卡顿。
原因：/api/trend一次返回了几万条历史数据，前端渲染压力大。
解决：后端做聚合，1 小时的数据按分钟聚合，返回 60 个点而不是几万条；前端 ECharts 开sampling: 'lttb'降采样；历史数据存数据库时加时间索引，查询加LIMIT。
5.5 SSH 采集连接被目标机拒绝
现象：采集脚本报Authentication failed或Connection refused。
原因：目标机sshd配了MaxStartups限制，采集脚本并发连接太多被拒；或者密钥权限不对（id_rsa权限不是 600）。
解决：采集改成串行或限制并发数（比如同时最多 5 台）；检查密钥文件权限chmod 600；目标机sshd_config里适当调大MaxStartups，但要注意这本身也是安全权衡。
6. 进阶技巧：让态势系统从「能看」到「能预警」的两个关键动作
系统跑起来能看图只是第一步，真正有价值的是提前预警。这里说两个我实际用下来最有效的动作。
第一个是给态势评分加基线。固定阈值（CPU > 80% 告警）在业务波动大的环境里误报太多。做法是先用一周数据算出每台主机各指标的均值和标准差，之后用「偏离基线 3 个标准差」作为告警条件。实现上就是采集时多存一份历史统计，判定时算 z-score：
import statistics def is_anomaly(value, history, sigma=3): """history: 该指标过去 N 个采样点的列表""" if len(history) < 30: # 样本太少不判定 return False mean = statistics.mean(history) stdev = statistics.pstdev(history) if stdev == 0: return False return abs(value - mean) / stdev > sigma这个函数对 CPU、内存、连接数都适用，关键是history要滚动更新，只保留最近 N 个点。sigma=3是经验值，调到 2 会更敏感但误报多，调到 4 更稳但可能漏报，按环境调。
第二个是把告警和处置串起来。光告警不处置，时间长了没人看。我的做法是给每条告警附一个「建议动作」，比如暴力破解告警附上封禁命令，可疑进程告警附上进程详情和 kill 命令。前端告警列表里直接给个按钮，点了就执行对应处置脚本。这里要注意权限控制，处置脚本只能由特定角色触发，且要有操作日志，否则误封了自己人很麻烦。
一个具体技巧：态势数据存 SQLite 就够，不用上时序数据库。单机每秒写入几条，SQLite 完全扛得住，查询加个时间索引，1 小时数据毫秒级返回。等主机规模上百、采集频率到秒级，再考虑换 ClickHouse 或 InfluxDB，过早引入复杂组件是给自己找事。
我自己踩过的最大教训是：一开始追求「大而全」，想采所有能采的数据，结果采集脚本复杂到没人敢改，出了 bug 排查半天。后来砍到只采四类核心字段，代码从八百行降到两百行，反而稳定跑了半年没出问题。做安全工具，能持续跑、能看懂、能改，比功能多重要得多。希望帮到你。
本文还有配套的精品资源，点击获取
