简介这是一套基于 Python 开发的主机安全态势感知系统完整项目资料面向计算机相关专业的毕业设计、课程设计以及项目开发学习者帮助解决安全监控类选题缺少可运行代码与配套文档的问题。压缩包共 663 个文件约 35.55MB其中 15 个 py 文件承载核心业务逻辑619 个 js 与 1 个 css 支撑前端可视化另有 pyc 编译文件、txt 说明、html 页面、json 配置及 mmdb 地理库等结构完整。系统实现实时 IP 溯源与动态展示、攻击来源国家统计、应用程序服务状态监测、攻击事件监测以及 24 小时内进出口流量统计等功能主要模块涉及 flask、pyecharts、geoip2、scapy并包含 brute_analyse、http_analyse、ssh_analyse 等分析脚本。目前已有 369 人学习参考。读者可据此获得一套经过测试的完整源码与 md 文档理解态势感知系统的数据采集、分析与可视化链路并在此基础上延伸改造用于答辩或二次开发。1. 主机安全态势感知系统从一台机器的日志到一张可读的态势图很多做毕业设计或课程设计的同学第一次听到“主机安全态势感知系统”会本能地发怵觉得这是大厂安全团队才玩得转的东西。其实把范围收窄到“单机或小规模主机集群”用 Python 完全能搭出一套能跑、能看、能讲清楚原理的系统。它要解决的核心问题很朴素一台主机上散落着登录日志、进程信息、网络连接、文件变更这些数据单看都是碎片拼起来才能回答“这台机器现在安不安全、过去一小时有没有异常”。这套系统适合谁适合计算机毕业设计选题、课程设计交作业、想入门安全开发但不想一上来就啃 C 的 Python 学习者。它不追求企业级吞吐追求的是链路完整、逻辑自洽、每一层你都能讲出为什么这么设计。下面我按自己搭过的一版思路把采集、分析、评分、可视化这条线拆开讲源码和文档的结构也会顺带说清楚。2. 态势感知系统的四层架构与 Python 选型理由2.1 为什么用 Python 而不是 Go 或 C主机安全态势感知系统在毕设场景里第一约束不是性能而是“能不能在两周内把链路跑通并且答辩时讲明白”。Python 在这个约束下几乎是最优解。采集层要读/var/log/auth.log、调psutil拿进程和网络连接、用watchdog监听文件变化这些库全是现成的一行 import 就能用。分析层要做规则匹配、简单统计、甚至接一个轻量机器学习模型pandas和scikit-learn的生态成熟度远超其他语言。展示层用Flask或FastAPI起一个 Web 服务配ECharts画态势图前端不用写太多。Go 的并发模型确实更适合高吞吐采集但毕设主机数量通常个位数Python 的 GIL 在这个量级下根本不是瓶颈。C 更不用考虑开发周期和调试成本对毕设来说不划算。我一般会跟做毕设的同学说先把 Python 版本跑通如果答辩老师问“为什么不用 Go”你就回答“当前主机规模下 Python 的 IO 密集型采集完全够用架构上采集与分析解耦后续换 Go 重写采集层不影响分析层”——这个回答既诚实又体现了架构思维。2.2 四层架构的职责边界一套能讲清楚的主机安全态势感知系统我习惯拆成四层每层职责必须清晰否则后期加功能会互相污染。层级职责典型 Python 实现输出采集层定时/实时抓取主机原始数据psutil、watchdog、subprocess原始事件 JSON分析层规则匹配、异常检测、关联分析pandas、re、scikit-learn告警与风险标签评分层把多维度告警聚合成态势分值自定义权重模型0-100 态势分展示层Web 界面与态势图渲染Flask ECharts可视化页面采集层的关键设计是“只采集不判断”。很多新手会把判断逻辑写进采集脚本比如“发现 22 端口连接就告警”这样后期规则一改就要动采集代码非常痛苦。正确做法是采集层只负责把原始事件结构化输出分析层再决定什么算异常。评分层是最容易被忽视但答辩时最容易被问的部分。态势分值怎么来的常见做法是加权求和每个告警类型有一个基础分乘以严重程度系数再按时间衰减。比如“暴力破解尝试”基础分 30“异常进程启动”基础分 20过去 5 分钟内的告警权重 1.05 到 30 分钟权重 0.5超过 30 分钟权重 0.1。这样态势分能反映“最近有没有事”而不是把一周前的告警一直挂在上面。2.3 最小可运行链路的搭建步骤先别急着写完整系统用最小链路验证四层能不能串起来。下面这段代码把采集、分析、评分、输出串成一个可运行的脚本跑通之后再拆成模块。# minimal_situation.py # 最小态势感知链路采集 - 分析 - 评分 - 输出 import psutil import time import json from datetime import datetime # ---------- 采集层 ---------- def collect_connections(): 采集当前主机的网络连接返回结构化列表 conns [] for conn in psutil.net_connections(kindinet): # 只保留有远端地址的连接过滤掉本地监听 if conn.raddr: conns.append({ local: f{conn.laddr.ip}:{conn.laddr.port}, remote: f{conn.raddr.ip}:{conn.raddr.port}, status: conn.status, pid: conn.pid }) return conns # ---------- 分析层 ---------- SUSPICIOUS_PORTS {4444, 5555, 6666, 31337} # 常见后门端口 def analyze_connections(conns): 对连接列表做规则匹配返回告警列表 alerts [] for c in conns: remote_port int(c[remote].split(:)[1]) if remote_port in SUSPICIOUS_PORTS: alerts.append({ type: suspicious_port, severity: 3, detail: f连接到可疑端口 {remote_port}, timestamp: datetime.now().isoformat() }) return alerts # ---------- 评分层 ---------- def calculate_score(alerts): 根据告警计算态势分0-100越高越危险 base 0 for a in alerts: base a[severity] * 10 return min(base, 100) # ---------- 主循环 ---------- if __name__ __main__: while True: conns collect_connections() alerts analyze_connections(conns) score calculate_score(alerts) result { timestamp: datetime.now().isoformat(), connection_count: len(conns), alerts: alerts, situation_score: score } print(json.dumps(result, ensure_asciiFalse, indent2)) time.sleep(10) # 每 10 秒采集一次这段代码的逻辑说明collect_connections用psutil.net_connections拿所有网络连接过滤掉没有远端地址的本地监听只保留真正对外通信的连接。analyze_connections用一个硬编码的可疑端口集合做规则匹配命中就生成告警。calculate_score把告警严重程度乘以 10 累加封顶 100。主循环每 10 秒跑一次输出 JSON。参数说明SUSPICIOUS_PORTS这个集合在实际项目中应该从配置文件读取不要硬编码。time.sleep(10)的 10 秒是采集间隔毕设演示场景下 5 到 10 秒比较合适太短会刷屏太长演示时看不到变化。severity我设了 1 到 3 三档实际可以扩展到 1 到 5。跑通这个脚本后你会看到终端每 10 秒输出一段 JSON。如果主机上有连接到可疑端口的进程alerts数组里就会有内容situation_score也会相应升高。这就是态势感知最核心的闭环采集原始数据分析出告警聚合成一个可读的分值。3. 日志采集与解析把 auth.log 变成结构化事件3.1 采集哪些日志、为什么是这些主机安全态势感知系统的数据源选择直接决定了系统能发现什么。毕设场景下我建议聚焦三类日志覆盖大部分主机安全事件第一类是认证日志Linux 下是/var/log/auth.logDebian/Ubuntu或/var/log/secureCentOS/RHEL。这里面有 SSH 登录成功/失败、sudo 提权、用户切换等记录是发现暴力破解和异常登录的核心数据源。第二类是系统日志/var/log/syslog或/var/log/messages包含内核消息、服务启动停止、硬件异常等。这类日志量大且杂建议只提取关键字匹配的行比如error、failed、segfault。第三类是进程与网络快照用psutil定时采集不依赖日志文件。很多恶意行为不一定写系统日志但进程列表和网络连接会暴露痕迹。提示不要试图采集所有日志。毕设答辩时老师问“为什么选这三类”你要能回答“认证日志覆盖访问控制系统日志覆盖服务异常进程网络快照覆盖运行时行为三者互补且采集成本可控”。3.2 用正则解析 auth.log 的登录事件auth.log 的格式在不同发行版和不同服务下略有差异但 SSH 登录相关的行有比较稳定的模式。下面这段代码提取 SSH 登录成功和失败事件。# parse_auth_log.py # 解析 auth.log 中的 SSH 登录事件 import re from datetime import datetime # SSH 登录失败的正则 FAILED_PATTERN re.compile( r(?Pmonth\w)\s(?Pday\d)\s(?Ptime\d:\d:\d)\s r(?Phost\S)\ssshd\[\d\]:\s rFailed password for (?:invalid user )?(?Puser\S) from (?Pip\S) ) # SSH 登录成功的正则 ACCEPTED_PATTERN re.compile( r(?Pmonth\w)\s(?Pday\d)\s(?Ptime\d:\d:\d)\s r(?Phost\S)\ssshd\[\d\]:\s rAccepted password for (?Puser\S) from (?Pip\S) ) def parse_line(line): 解析单行日志返回事件字典或 None m FAILED_PATTERN.search(line) if m: return { event: ssh_login_failed, user: m.group(user), ip: m.group(ip), time: f{m.group(month)} {m.group(day)} {m.group(time)} } m ACCEPTED_PATTERN.search(line) if m: return { event: ssh_login_success, user: m.group(user), ip: m.group(ip), time: f{m.group(month)} {m.group(day)} {m.group(time)} } return None def parse_file(path): 解析整个日志文件返回事件列表 events [] with open(path, r, encodingutf-8, errorsignore) as f: for line in f: ev parse_line(line) if ev: events.append(ev) return events if __name__ __main__: events parse_file(/var/log/auth.log) for e in events[-10:]: # 打印最近 10 条 print(e)逻辑说明两个正则分别匹配失败和成功的 SSH 登录行。(?:invalid user )?这个非捕获组用来兼容“Failed password for invalid user admin”这种格式。parse_file逐行读取忽略编码错误把匹配到的事件收集起来。参数说明日志路径/var/log/auth.log在 CentOS 上要改成/var/log/secure。如果你的系统日志用了 rsyslog 且格式有微调正则可能需要调整建议先用grep Failed password /var/log/auth.log | head -5看实际格式再改正则。errorsignore是为了防止日志里有非 UTF-8 字符导致读取中断。3.3 暴力破解检测滑动窗口统计拿到登录失败事件后检测暴力破解最简单有效的方法是滑动窗口计数同一个 IP 在 5 分钟内失败超过 N 次就告警。# brute_force.py # 滑动窗口检测 SSH 暴力破解 from collections import defaultdict from datetime import datetime, timedelta def detect_brute_force(events, window_minutes5, threshold5): events: parse_file 返回的事件列表 window_minutes: 时间窗口大小 threshold: 窗口内失败次数阈值 返回: 告警列表 # 按 IP 分组失败事件 failures defaultdict(list) for e in events: if e[event] ssh_login_failed: failures[e[ip]].append(e[time]) alerts [] for ip, times in failures.items(): # 简化处理直接用事件数量判断 # 生产环境应解析时间戳做真正的滑动窗口 if len(times) threshold: alerts.append({ type: brute_force, ip: ip, count: len(times), severity: 4, detail: fIP {ip} 在日志中出现 {len(times)} 次登录失败 }) return alerts逻辑说明这段代码做了简化直接用失败次数判断没有真正解析时间戳做滑动窗口。为什么因为 auth.log 的时间格式不带年份解析成datetime需要额外处理年份推断毕设场景下容易在这里翻车。更稳妥的做法是用len(times)做粗粒度判断答辩时说明“当前实现基于日志条数生产环境应引入时间窗口”。参数说明threshold5是经验值实际可以调到 10 减少误报。window_minutes在这个简化版里没用到但保留参数是为了接口清晰后续替换成真正的时间窗口实现时不用改调用方。注意如果你要写真正的滑动窗口用datetime.strptime解析时间时记得补上年份否则跨年日志会出错。我见过有同学在这里卡了一整天最后发现是 12 月的日志被解析成了当年 1 月。4. 态势评分模型把零散告警变成一个可解释的分值4.1 加权求和模型的参数怎么定态势评分是答辩时最容易被追问的部分。老师会问“你这个 85 分是怎么算出来的为什么不是 80 或 90”如果你回答“拍脑袋定的”印象分直接打折。我一般用加权求和模型每个参数都有来源。模型公式态势分 min(100, Σ(告警基础分 × 严重系数 × 时间衰减))告警基础分按类型定暴力破解 30异常进程 25可疑端口连接 20文件篡改 35登录成功但来源异常 15。这些数字不是随便写的参考了常见安全事件的 CVSS 评分量级再按毕设场景压缩到 0-100 区间。严重系数按告警的severity字段映射severity 1 对应 0.52 对应 0.83 对应 1.04 对应 1.35 对应 1.6。这样同一个类型的告警严重程度不同对总分影响不同。时间衰减按告警距当前时间算5 分钟内系数 1.05 到 30 分钟 0.630 分钟到 2 小时 0.3超过 2 小时 0.1。这个设计让态势分能反映“最近有没有事”而不是被历史告警拖住。4.2 评分模块的代码实现# scoring.py # 态势评分模块 from datetime import datetime, timedelta # 告警类型基础分 BASE_SCORES { brute_force: 30, suspicious_process: 25, suspicious_port: 20, file_tamper: 35, abnormal_login: 15, } # 严重系数 SEVERITY_FACTOR {1: 0.5, 2: 0.8, 3: 1.0, 4: 1.3, 5: 1.6} def time_decay(alert_time, nowNone): 根据告警时间计算衰减系数 if now is None: now datetime.now() delta now - alert_time minutes delta.total_seconds() / 60 if minutes 5: return 1.0 elif minutes 30: return 0.6 elif minutes 120: return 0.3 else: return 0.1 def calculate_situation_score(alerts, nowNone): alerts: 告警列表每条包含 type, severity, timestamp 返回: (态势分, 各告警贡献明细) if now is None: now datetime.now() total 0 details [] for a in alerts: base BASE_SCORES.get(a[type], 10) # 未知类型给 10 分基础 sev SEVERITY_FACTOR.get(a[severity], 1.0) # 解析时间戳 if isinstance(a[timestamp], str): at datetime.fromisoformat(a[timestamp]) else: at a[timestamp] decay time_decay(at, now) contribution base * sev * decay total contribution details.append({ type: a[type], contribution: round(contribution, 2), decay: decay }) return min(round(total, 2), 100), details逻辑说明BASE_SCORES和SEVERITY_FACTOR是两个可调参数表答辩时你可以现场改一个值演示态势分变化这比干讲公式有说服力。time_decay把时间差转成分钟再分档。calculate_situation_score遍历告警累加贡献值封顶 100同时返回每条告警的贡献明细方便前端展示“为什么是这个分”。参数说明BASE_SCORES里的数字可以根据你的告警类型调整关键是每个数字要有说法。比如文件篡改给 35 是因为它通常意味着系统已经被入侵比暴力破解可能只是尝试更严重。SEVERITY_FACTOR的 1.6 上限对应 severity 5如果你系统里最高只有 3那实际用到 1.0 封顶。4.3 评分结果的可视化数据准备评分模块输出的details列表可以直接喂给前端画堆叠图或瀑布图。下面这段代码把评分结果转成 ECharts 友好的格式。# prepare_chart_data.py # 把评分明细转成 ECharts 堆叠图数据 def to_echarts_data(details): details: calculate_situation_score 返回的明细列表 categories [] values [] for d in details: categories.append(d[type]) values.append(d[contribution]) return { categories: categories, values: values, total: sum(values) } # 示例 if __name__ __main__: sample_details [ {type: brute_force, contribution: 39.0, decay: 1.0}, {type: suspicious_port, contribution: 16.0, decay: 0.6}, ] print(to_echarts_data(sample_details))逻辑说明这个转换很简单但把评分和展示解耦了。前端拿到categories和values就能画图不需要理解评分模型。参数说明total字段是前端显示总分用的和calculate_situation_score的返回值应该一致如果不一致说明有告警被过滤了需要检查。5. 避坑与排查毕设答辩前最容易翻车的五个点5.1 日志路径写死导致换机器就报错现象在 Ubuntu 上跑得好好的换到 CentOS 或老师的演示环境直接FileNotFoundError。原因/var/log/auth.log是 Debian 系的路径CentOS 用/var/log/secure有些最小化安装的系统甚至没有这些文件。解决在配置里做路径探测按优先级尝试多个路径都不存在就降级到只采集进程和网络数据并在日志里明确提示“认证日志不可用已降级运行”。代码里用os.path.exists判断不要用 try-except 吞掉所有异常。5.2 psutil 权限不足导致进程信息缺失现象采集到的进程列表里很多进程的pid是 None或者net_connections返回空列表。原因普通用户权限下psutil拿不到其他用户的进程详情net_connections也需要 root 或 CAP_NET_ADMIN 才能看到完整连接。解决毕设演示时用sudo跑采集脚本或者在文档里写明“需要 root 权限运行采集层”。如果老师要求不能用 root那就降级到只采集当前用户可见的数据并在态势分里标注“数据完整度 60%”。5.3 时间戳格式不统一导致评分全为 0现象评分模块跑出来全是 0或者time_decay报TypeError。原因采集层输出的时间戳是字符串分析层可能又转成了datetime评分层拿到的是混合类型。datetime.fromisoformat对格式要求严格2024-01-01 12:00:00和2024-01-01T12:00:00处理方式不同。解决全链路统一用 ISO 8601 格式字符串只在评分层做一次解析。采集层输出datetime.now().isoformat()分析层原样传递评分层用datetime.fromisoformat解析。如果日志里的时间格式特殊在解析函数里统一转成 ISO 格式再往下传。5.4 态势分一直 100 导致演示没变化现象演示时态势分一直显示 100老师问“为什么一直是满分”你答不上来。原因时间衰减没生效或者告警没有清理机制历史告警一直累加。另一个常见原因是min(total, 100)封顶太早几条告警就顶到 100。解决检查time_decay是否真的被调用打印每条告警的decay值确认。如果确实告警多把封顶值调到 200 或者改成非线性映射比如100 * (1 - exp(-total/50))让高分区间有区分度。演示前手动清理一次告警表从 0 开始积累。5.5 前端图表数据为空但后端有数据现象后端日志显示有告警态势分也算出来了但前端页面图表空白。原因Flask 返回的 JSON 里datetime对象不能直接序列化或者前端请求的接口路径和实际注册的路由不一致。另一个常见原因是 CORS 没配前端在 8080 端口后端在 5000 端口浏览器拦截了请求。解决后端返回前用json.dumps(data, defaultstr)把datetime转成字符串。Flask 用flask-cors加一行CORS(app)。前端打开浏览器开发者工具的 Network 面板看请求是否 200、响应体是否为空这一步能定位 90% 的前后端联调问题。6. 从毕设到可演示系统三个让答辩加分的技巧6.1 用配置文件替代硬编码现场改参数演示答辩时最加分的动作是“现场改一个参数系统行为立刻变化”。把SUSPICIOUS_PORTS、BASE_SCORES、threshold这些全部抽到config.yaml里用pyyaml加载。演示时打开配置文件把暴力破解阈值从 5 改成 2重启采集进程态势分立刻上升。这个操作能直观证明你的系统是“可配置的”而不是“写死的”。# config_loader.py import yaml def load_config(pathconfig.yaml): with open(path, r, encodingutf-8) as f: return yaml.safe_load(f) # config.yaml 示例内容 # suspicious_ports: [4444, 5555, 6666] # brute_force_threshold: 5 # base_scores: # brute_force: 30 # suspicious_port: 20逻辑说明load_config返回一个字典各模块从字典里取自己的参数。参数说明config.yaml的路径建议用环境变量或命令行参数传入不要写死相对路径否则从不同目录启动会找不到文件。6.2 加一个“模拟攻击”按钮演示不用等毕设演示最尴尬的是“等攻击发生”。你可以在系统里加一个模拟模块点击按钮就往告警表里插入一条模拟的暴力破解告警态势分立刻变化。这个模块在文档里标注“仅用于演示”不影响真实采集逻辑。# simulate.py # 演示用注入模拟告警 from datetime import datetime def inject_demo_alert(alert_typebrute_force, severity4): 往告警存储里插入一条模拟告警 return { type: alert_type, severity: severity, timestamp: datetime.now().isoformat(), detail: 演示用模拟告警 }逻辑说明这个函数返回一条格式和真实告警一致的字典直接传给评分模块就能看到态势分变化。参数说明alert_type和severity可以做成前端下拉框演示时选不同类型看分值差异。6.3 文档结构让老师一眼看到你的工作量毕设文档不要写成代码注释的堆砌。我一般按这个结构组织第一章讲选题背景和同类系统对比体现你调研过第二章讲架构设计和选型理由体现你思考过第三章讲采集与分析实现贴核心代码和运行截图第四章讲评分模型与参数来源体现你有依据第五章讲测试与演示贴态势分变化截图最后附配置文件和部署说明。源码目录按模块分文件夹每个文件夹一个README.md说明该模块的输入输出。这样老师翻文档时能快速定位到“这个学生做了什么”而不是在一堆代码里找亮点。我自己做毕设时踩过最大的坑是“功能做太多但每个都讲不清楚”。后来学乖了把核心链路做扎实每个参数都能说出为什么答辩时反而比堆功能的同学得分高。希望帮到你。本文还有配套的精品资源点击获取