基于Python与HTML的主机安全态势感知系统实战

简介这是一套面向网络安全初学者与开发者的主机安全态势感知系统完整实现方案基于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 使用率interval1 表示采样 1 秒避免瞬时值抖动 cpu_percent: psutil.cpu_percent(interval1), # 内存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(kindinet): 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(interval1)里的 interval 不能省省了拿到的是自上次调用以来的平均值第一次调用基本是 0这是新手最常翻的车。net_connections在 Linux 上需要 root 或 CAP_NET_ADMIN 权限才能看到其他用户的连接普通用户跑只能看到自己的部署时要么用 root要么给 Python 解释器加 capability。参数说明kindinet只取 IPv4/IPv6不取 Unix socket如果主机连接数上万这个循环会慢实际部署时加个limit或只取 LISTEN 状态做端口监控ESTABLISHED 单独用采样频率控制。2.3 远程主机怎么采paramiko 拉日志的边界管多台主机时常见做法是用paramiko走 SSH 拉日志和指标不额外装 agent。核心是封装一个执行远程命令的函数import paramiko def run_remote_cmd(host, user, key_path, cmd, timeout10): 通过 SSH 在远程主机执行命令并返回 stdout client paramiko.SSHClient() # 首次连接自动接受 host key生产环境应预置 known_hosts client.set_missing_host_key_policy(paramiko.AutoAddPolicy()) try: client.connect(host, usernameuser, key_filenamekey_path, timeouttimeout) stdin, stdout, stderr client.exec_command(cmd, timeouttimeout) return stdout.read().decode(utf-8, errorsignore) 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(rFailed password for (?:invalid user )?(?Puser\S) from (?Pip\d\.\d\.\d\.\d)), re.compile(rInvalid user (?Puser\S) from (?Pip\d\.\d\.\d\.\d)), re.compile(rAccepted password for (?Puser\S) from (?Pip\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_sec300, threshold10): self.window window_sec # 时间窗口5 分钟 self.threshold threshold # 窗口内失败次数阈值 self.records defaultdict(deque) # key: ip, value: 失败时间戳队列 def add_failure(self, ip, tsNone): 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, n10): return sorted( ((ip, len(q)) for ip, q in self.records.items()), keylambda x: x[1], reverseTrue )[:n]逻辑说明用deque存时间戳每次新增时从左侧踢掉过期记录队列长度就是窗口内的失败次数。这样不用定时清理内存也不会无限涨。top_attackers直接给前端提供「攻击源 TOP N」的数据。参数说明window_sec300、threshold10是经验值内网主机可以放宽到 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(hours1)) 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(hours1)里的时间范围要跟采集频率匹配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 langzh-cn head meta charsetutf-8 title主机安全态势感知/title script srchttps://cdn.jsdelivr.net/npm/echarts5/dist/echarts.min.js/script /head body div idgauge stylewidth:400px;height:300px;/div div idtrend stylewidth: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.logUbuntu 路径文件不存在读回来是空。解决采集前先判断日志文件存在性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-scoreimport statistics def is_anomaly(value, history, sigma3): 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 个点。sigma3是经验值调到 2 会更敏感但误报多调到 4 更稳但可能漏报按环境调。第二个是把告警和处置串起来。光告警不处置时间长了没人看。我的做法是给每条告警附一个「建议动作」比如暴力破解告警附上封禁命令可疑进程告警附上进程详情和 kill 命令。前端告警列表里直接给个按钮点了就执行对应处置脚本。这里要注意权限控制处置脚本只能由特定角色触发且要有操作日志否则误封了自己人很麻烦。一个具体技巧态势数据存 SQLite 就够不用上时序数据库。单机每秒写入几条SQLite 完全扛得住查询加个时间索引1 小时数据毫秒级返回。等主机规模上百、采集频率到秒级再考虑换 ClickHouse 或 InfluxDB过早引入复杂组件是给自己找事。我自己踩过的最大教训是一开始追求「大而全」想采所有能采的数据结果采集脚本复杂到没人敢改出了 bug 排查半天。后来砍到只采四类核心字段代码从八百行降到两百行反而稳定跑了半年没出问题。做安全工具能持续跑、能看懂、能改比功能多重要得多。希望帮到你。本文还有配套的精品资源点击获取