简介这是一套面向计算机、通信、人工智能及自动化等专业学生的主机安全态势感知系统实战项目适用于毕业设计、期末大作业及安全方向进阶学习。系统基于Python开发融合深度学习与日志分析技术可实现对SSH、HTTP、暴力破解等常见攻击行为的实时识别与可视化呈现具备完整工程闭环能力。资源包共662个文件含15个核心Python脚本含brute_analyse、http_analyse等模块、619个JavaScript前端交互文件支撑ECharts与ECharts-GL动态图表渲染、以及配套HTML页面、CSS样式、GeoIP数据库mmdb和多国地理数据JS文件整体压缩后35.19MB结构清晰、模块解耦度高。目前已有104人下载学习提供经答辩验证99分、全链路调试通过的可运行代码详细文档涵盖部署指南、功能说明与关键算法注释特别适合初学者理解安全态势建模逻辑也便于进阶者二次开发与功能扩展。1. 这不是“杀毒软件界面美化”而是一套可落地的主机安全态势感知最小闭环很多同学拿到“基于Python开发的主机安全态势感知系统”这个毕设题目时第一反应是去GitHub搜一个带Web界面的扫描器改改前端——结果跑起来只显示“正在扫描中…”连进程列表都拿不到更别说判断是否存在异常行为。实际上真正的主机安全态势感知核心不在炫酷图表而在能稳定采集、可解释建模、有依据告警这三个刚性环节。它面向的是Linux/Windows服务器或开发机本地环境解决的是“我这台机器此刻是否被植入后门有没有异常外连关键服务是否被篡改”这类具体问题。本方案不依赖第三方SaaS平台全部用标准Python库psutil、osquery-lite封装、scapy、yara-python实现数据采集用轻量级规则引擎json条件表达式做策略匹配用SQLite做本地状态快照存储避免引入Flask/Django等重量框架增加部署复杂度。适合计算机、信息安全、网络工程专业学生在无云资源、无运维权限的条件下两周内完成从零部署到生成首份安全态势报告的全过程。2. 数据采集层用原生Python接口直取操作系统底层信号避开Shell命令解析陷阱主机安全态势感知的第一道门槛是能否稳定、低开销、跨平台地获取真实运行时数据。很多毕设代码直接调用os.popen(netstat -ano)或subprocess.run([ps, -ef])看似简单实则埋下三类隐患一是Windows/Linux命令参数不一致导致脚本崩溃二是Shell输出格式随系统语言/版本变化如中文Windows返回“PID”列名英文系统为“PID”但位置偏移三是无法获取进程内存映射、文件句柄、网络连接状态等深层字段。我们采用分层采集策略所有数据源均通过Python原生绑定库获取确保字段语义统一、返回结构可序列化。2.1 进程与服务状态psutil win32serviceWindows/ systemdLinux双路径psutil是跨平台进程信息采集的事实标准但默认不暴露服务启动类型auto/manual/disabled和依赖关系。Windows需补充pywin32获取服务控制管理器SCM数据Linux则需解析systemctl list-units --typeservice --staterunning的JSON输出需启用--outputjson。关键代码如下# process_collector.py import psutil import json import platform from typing import List, Dict, Any def get_running_processes() - List[Dict[str, Any]]: 统一返回标准化进程字典列表字段名与Linux/Windows保持一致 processes [] for proc in psutil.process_iter([pid, name, status, cpu_percent, memory_info, connections, create_time]): try: pinfo proc.info # 统一字段windows pid为intlinux同理name强制strstatus映射为running/stopped/zombie proc_dict { pid: int(pinfo[pid]), name: str(pinfo[name]), status: running if pinfo[status] in [running, sleeping] else stopped, cpu_percent: float(pinfo[cpu_percent]), mem_rss_mb: int(pinfo[memory_info].rss / 1024 / 1024), create_time: int(pinfo[create_time]), connections_count: len(pinfo[connections]) } processes.append(proc_dict) except (psutil.NoSuchProcess, psutil.AccessDenied, KeyError): continue return processes def get_services_status() - List[Dict[str, Any]]: Windows用win32serviceLinux用systemctl解析返回统一service结构 if platform.system() Windows: import win32serviceutil services [] # 注意此处需管理员权限否则跳过 try: for service_name in win32serviceutil.EnumServicesStatus(): services.append({ name: service_name[0], display_name: service_name[1], status: running if service_name[2] 4 else stopped, start_type: [boot, system, auto, manual, disabled][service_name[3]] }) except Exception: pass # 权限不足时返回空列表不影响主流程 return services else: # Linux: systemctl list-units --typeservice --staterunning --outputjson import subprocess try: result subprocess.run( [systemctl, list-units, --typeservice, --staterunning, --outputjson], capture_outputTrue, textTrue, timeout5 ) if result.returncode 0: units json.loads(result.stdout) return [{name: u[unit], display_name: u[description], status: running, start_type: u[load]} for u in units] except Exception: pass return []提示psutil.process_iter()必须包裹try-except否则遇到已退出进程会抛NoSuchProcess异常connections字段在Windows上需额外开启psutil.net_connections()权限Linux默认可用。get_services_status()在非管理员权限下Windows分支会静默失败这是设计使然——安全采集本身就需要权限校验毕设演示时建议在管理员CMD中运行。2.2 网络连接与监听端口绕过netstat用psutil.connections()直接读取内核socket表netstat输出需正则解析且不同系统列数不一。psutil.connections()直接读取/proc/net/tcpLinux或GetExtendedTcpTableWindows返回结构化元组。关键字段包括laddr(本地地址)、raddr(远程地址)、type(TCP/UDP)、status(LISTEN/ESTABLISHED)、pid(关联进程ID)。以下代码提取所有监听端口并关联进程名# network_collector.py import psutil from typing import List, Dict, Any def get_listening_ports() - List[Dict[str, Any]]: 获取所有监听端口关联进程名过滤掉内核线程pid0 listeners [] for conn in psutil.net_connections(kindinet): if conn.status LISTEN and conn.pid ! 0: # 排除内核线程 try: proc psutil.Process(conn.pid) listeners.append({ port: conn.laddr.port, address: conn.laddr.ip or 0.0.0.0, protocol: tcp if conn.type socket.SOCK_STREAM else udp, pid: conn.pid, process_name: proc.name(), user: proc.username().split(\\)[-1] if \\ in proc.username() else proc.username() }) except (psutil.NoSuchProcess, psutil.AccessDenied): # 进程已退出或无权限读取记录pid但不填名称 listeners.append({ port: conn.laddr.port, address: conn.laddr.ip or 0.0.0.0, protocol: tcp if conn.type socket.SOCK_STREAM else udp, pid: conn.pid, process_name: unknown, user: unknown }) return listeners注意psutil.net_connections()在macOS上不支持pid字段需用lsof -i -P -n替代但本毕设聚焦Windows/Linux主流教学环境macOS兼容性作为可选扩展点。address字段为0.0.0.0表示监听所有网卡127.0.0.1表示仅本地回环——这是识别潜在风险端口如Redis未绑定127.0.0.1的关键依据。2.3 文件完整性校验用hashlib对关键系统路径做快照比对态势感知需发现文件篡改但全盘哈希性能不可接受。我们聚焦/etc/passwd、/etc/shadowLinux、C:\Windows\System32\drivers\etc\hostsWindows、/usr/bin/下常用工具ls, ps, netstat等高危路径。使用hashlib.sha256()计算文件内容哈希存入SQLite快照表每次采集时比对# file_integrity.py import hashlib import os import sqlite3 from pathlib import Path CRITICAL_PATHS [ /etc/passwd, /etc/shadow, /usr/bin/ls, /usr/bin/ps, /usr/bin/netstat, C:\\Windows\\System32\\drivers\\etc\\hosts, C:\\Windows\\System32\\notepad.exe ] def calculate_file_hash(filepath: str) - str: 计算文件SHA256忽略不存在或无权限文件 try: if not Path(filepath).exists(): return with open(filepath, rb) as f: return hashlib.sha256(f.read()).hexdigest() except (OSError, IOError): return def save_baseline(db_path: str baseline.db): 首次运行时生成基线快照 conn sqlite3.connect(db_path) conn.execute( CREATE TABLE IF NOT EXISTS file_baseline ( path TEXT PRIMARY KEY, hash TEXT NOT NULL, timestamp INTEGER NOT NULL ) ) for path in CRITICAL_PATHS: file_hash calculate_file_hash(path) if file_hash: # 仅存有效哈希 conn.execute( INSERT OR REPLACE INTO file_baseline (path, hash, timestamp) VALUES (?, ?, ?), (path, file_hash, int(time.time())) ) conn.commit() conn.close() def check_integrity(db_path: str baseline.db) - List[Dict[str, str]]: 比对当前文件哈希与基线返回变更列表 conn sqlite3.connect(db_path) cursor conn.cursor() cursor.execute(SELECT path, hash FROM file_baseline) baseline {row[0]: row[1] for row in cursor.fetchall()} changes [] for path, baseline_hash in baseline.items(): current_hash calculate_file_hash(path) if current_hash and current_hash ! baseline_hash: changes.append({path: path, status: modified, old_hash: baseline_hash, new_hash: current_hash}) elif not current_hash: changes.append({path: path, status: unreachable, old_hash: baseline_hash}) conn.close() return changes提示save_baseline()需手动执行一次生成初始快照后续check_integrity()才有效。calculate_file_hash()对大文件如/var/log/syslog应加大小限制如f.read(10*1024*1024)避免内存溢出——毕设场景下关键路径文件均小于1MB故省略此步实际部署需补全。3. 规则引擎层用JSON定义检测逻辑避免硬编码if-else支持毕业答辩现场动态增删规则态势感知的核心是“什么算异常”。硬编码if cpu 90: alert()无法应对答辩老师“那如果我要检测SSH暴力破解呢”的提问。我们采用声明式规则引擎所有检测逻辑写在rules.json中Python加载后动态执行。规则结构包含id、name、description、source数据源名、conditionJMESPath表达式、severitylow/medium/high、actionlog/alert。JMESPath语法简洁支持数组过滤、数值比较、字符串匹配学习成本低于自研DSL。3.1 规则定义规范与示例覆盖进程、网络、文件三类典型威胁rules.json示例截取关键片段[ { id: high_cpu_process, name: CPU占用率持续过高, description: 单个进程CPU占用超过80%持续30秒, source: processes, condition: [?cpu_percent 80], severity: medium, action: alert }, { id: suspicious_port, name: 监听高危端口, description: 监听端口为22SSH、3306MySQL、6379Redis且绑定0.0.0.0, source: listening_ports, condition: [?port 22 || port 3306 || port 6379] | [?address 0.0.0.0], severity: high, action: alert }, { id: hosts_modified, name: hosts文件被篡改, description: 检测/etc/hosts或C:\\Windows\\System32\\drivers\\etc\\hosts内容变更, source: file_integrity, condition: [?status modified (contains(path, hosts))], severity: high, action: alert } ]3.2 规则执行引擎用jmespath库解析condition避免eval安全风险# rule_engine.py import json import jmespath from typing import List, Dict, Any class RuleEngine: def __init__(self, rules_path: str rules.json): with open(rules_path, r, encodingutf-8) as f: self.rules json.load(f) def evaluate_rules(self, data: Dict[str, Any]) - List[Dict[str, Any]]: 对data字典中各source键执行对应规则返回匹配结果列表 alerts [] for rule in self.rules: source_key rule[source] if source_key not in data: continue # 使用jmespath.search安全执行表达式不执行任意代码 try: result jmespath.search(rule[condition], data[source_key]) if result: # JMESPath返回非空即匹配 alerts.append({ rule_id: rule[id], rule_name: rule[name], matched_data: result if isinstance(result, list) else [result], severity: rule[severity], timestamp: int(time.time()) }) except jmespath.exceptions.JMESPathError as e: # condition语法错误记录日志但不中断 print(fRule {rule[id]} JMESPath error: {e}) continue return alerts # 使用示例 if __name__ __main__: # 假设已采集数据 collected_data { processes: get_running_processes(), listening_ports: get_listening_ports(), file_integrity: check_integrity() } engine RuleEngine() alerts engine.evaluate_rules(collected_data) for alert in alerts: print(f[{alert[severity].upper()}] {alert[rule_name]}: {len(alert[matched_data])} items matched)注意jmespath.search()比eval()安全万倍它只解析查询语法不执行Python代码。condition中[?cpu_percent 80]是合法JMESPath[?name python cpu_percent 90]也支持。若需更复杂逻辑如“过去5分钟平均CPU70%”可在数据采集层预计算滑动窗口指标再交由JMESPath过滤。3.3 规则热加载与调试答辩时现场修改JSON立即生效毕设答辩常被要求“现场演示新增规则”。传统硬编码需改Python再重启而JSON规则可做到热加载# main.py 中的采集循环 import time import threading def run_monitoring(): engine RuleEngine() # 初始化一次 while True: # 每30秒采集一次 collected_data { processes: get_running_processes(), listening_ports: get_listening_ports(), file_integrity: check_integrity() } # 每次都重新读取rules.json支持热更新 engine.rules json.load(open(rules.json)) alerts engine.evaluate_rules(collected_data) # 输出到控制台答辩演示 for alert in alerts: print(f\033[91mALERT [{alert[severity]}] {alert[rule_name]}\033[0m) for item in alert[matched_data][:3]: # 只打印前3条避免刷屏 print(f → {item}) time.sleep(30) # 启动监控线程 threading.Thread(targetrun_monitoring, daemonTrue).start()提示daemonTrue确保主线程退出时子线程自动结束避免答辩结束程序卡死。print()用ANSI颜色码\033[91m标红高危告警视觉冲击力强符合答辩场景需求。4. 态势聚合与报告生成用SQLite存状态快照用Jinja2渲染HTML报告告别Excel手工整理采集和告警只是起点态势感知要求回答“这台机器整体安全水位如何哪些风险在恶化历史趋势怎样”。我们不引入Elasticsearch等重型组件而是用SQLite存储时间序列快照 Jinja2模板生成静态HTML报告既满足毕设交付物要求可打包提交的独立HTML文件又具备分析能力。4.1 SQLite数据库设计三个核心表支撑态势演进分析创建security_state.db含三张表表名字段说明snapshotid(PK), timestamp, hostname, os_version, total_processes, listening_ports_count, high_cpu_count, file_modified_count每次采集的宏观快照alertsid(PK), snapshot_id(FK), rule_id, severity, matched_json, created_at每次触发的原始告警详情matched_json存JSON字符串trendid(PK), metric_name, value, timestamp用于绘制趋势图的指标如cpu_max_5min,ports_opened建表SQLdb_init.pyimport sqlite3 def init_database(db_pathsecurity_state.db): conn sqlite3.connect(db_path) conn.execute( CREATE TABLE IF NOT EXISTS snapshot ( id INTEGER PRIMARY KEY AUTOINCREMENT, timestamp INTEGER NOT NULL, hostname TEXT NOT NULL, os_version TEXT NOT NULL, total_processes INTEGER DEFAULT 0, listening_ports_count INTEGER DEFAULT 0, high_cpu_count INTEGER DEFAULT 0, file_modified_count INTEGER DEFAULT 0 ) ) conn.execute( CREATE TABLE IF NOT EXISTS alerts ( id INTEGER PRIMARY KEY AUTOINCREMENT, snapshot_id INTEGER NOT NULL, rule_id TEXT NOT NULL, severity TEXT NOT NULL, matched_json TEXT NOT NULL, created_at INTEGER NOT NULL, FOREIGN KEY(snapshot_id) REFERENCES snapshot(id) ) ) conn.execute( CREATE TABLE IF NOT EXISTS trend ( id INTEGER PRIMARY KEY AUTOINCREMENT, metric_name TEXT NOT NULL, value REAL NOT NULL, timestamp INTEGER NOT NULL ) ) conn.commit() conn.close()4.2 快照入库与趋势计算将离散告警转化为连续态势指标每次采集后不仅存原始数据更要计算衍生指标# report_generator.py import sqlite3 import json import platform from datetime import datetime def save_snapshot_and_alerts(db_path: str, collected_data: dict, alerts: list): conn sqlite3.connect(db_path) cursor conn.cursor() # 插入快照 hostname platform.node() os_ver f{platform.system()} {platform.release()} total_procs len(collected_data.get(processes, [])) listening_ports len(collected_data.get(listening_ports, [])) high_cpu_count len([p for p in collected_data.get(processes, []) if p.get(cpu_percent, 0) 80]) file_modified_count len([f for f in collected_data.get(file_integrity, []) if f.get(status) modified]) cursor.execute( INSERT INTO snapshot (timestamp, hostname, os_version, total_processes, listening_ports_count, high_cpu_count, file_modified_count) VALUES (?, ?, ?, ?, ?, ?, ?), (int(time.time()), hostname, os_ver, total_procs, listening_ports, high_cpu_count, file_modified_count) ) snapshot_id cursor.lastrowid # 插入告警 for alert in alerts: cursor.execute( INSERT INTO alerts (snapshot_id, rule_id, severity, matched_json, created_at) VALUES (?, ?, ?, ?, ?), (snapshot_id, alert[rule_id], alert[severity], json.dumps(alert[matched_data]), alert[timestamp]) ) # 计算并插入趋势指标示例高CPU进程数 cursor.execute( INSERT INTO trend (metric_name, value, timestamp) VALUES (?, ?, ?), (high_cpu_processes, float(high_cpu_count), int(time.time())) ) conn.commit() conn.close()4.3 HTML报告生成用Jinja2模板嵌入图表支持答辩PPT直接截图report_template.html精简版!DOCTYPE html html headtitle主机安全态势报告/title stylebody{font-family:Arial,sans-serif;margin:20px}table{border-collapse:collapse;width:100%}th,td{border:1px solid #ddd;padding:8px;text-align:left}/style /head body h1主机安全态势报告/h1 p生成时间{{ now }}/p p主机名{{ hostname }} | 操作系统{{ os_version }}/p h2宏观态势概览/h2 table trth指标/thth当前值/thth7日均值/th/tr trtd运行进程数/tdtd{{ snapshot.total_processes }}/tdtd{{ avg_processes }}/td/tr trtd监听端口数/tdtd{{ snapshot.listening_ports_count }}/tdtd{{ avg_ports }}/td/tr trtd高CPU进程数/tdtd{{ snapshot.high_cpu_count }}/tdtd{{ avg_cpu }}/td/tr /table h2最新告警最近10条/h2 {% for alert in recent_alerts %} div stylebackground:#ffebee;padding:10px;margin:5px 0 strong[{{ alert.severity|upper }}]/strong {{ alert.rule_name }}br small{{ alert.matched_data|length }}项匹配{{ alert.matched_data|first|tojson|truncate(100) }}/small /div {% endfor %} h2历史趋势高CPU进程数/h2 !-- 此处可嵌入Chart.js图表毕设简化用文字描述 -- p过去24小时高CPU进程数峰值{{ max_cpu_last24h }}均值{{ avg_cpu_last24h }}/p /body /html生成报告代码# generate_report.py from jinja2 import Template import sqlite3 import json from datetime import datetime, timedelta def generate_html_report(db_pathsecurity_state.db, output_pathreport.html): conn sqlite3.connect(db_path) cursor conn.cursor() # 获取最新快照 cursor.execute(SELECT * FROM snapshot ORDER BY timestamp DESC LIMIT 1) snap cursor.fetchone() if not snap: return # 计算7日均值 week_ago int((datetime.now() - timedelta(days7)).timestamp()) cursor.execute(SELECT AVG(total_processes), AVG(listening_ports_count), AVG(high_cpu_count) FROM snapshot WHERE timestamp ?, (week_ago,)) avg_vals cursor.fetchone() # 获取最近10条告警 cursor.execute( SELECT a.rule_id, r.name as rule_name, a.severity, a.matched_json, a.created_at FROM alerts a JOIN (SELECT id, name FROM rules) r ON a.rule_id r.id ORDER BY a.created_at DESC LIMIT 10 ) recent_alerts [] for row in cursor.fetchall(): recent_alerts.append({ rule_id: row[0], rule_name: row[1], severity: row[2], matched_data: json.loads(row[3]), created_at: row[4] }) # 计算24小时趋势 day_ago int((datetime.now() - timedelta(hours24)).timestamp()) cursor.execute(SELECT MAX(value), AVG(value) FROM trend WHERE metric_name high_cpu_processes AND timestamp ?, (day_ago,)) trend_vals cursor.fetchone() conn.close() # 渲染模板 with open(report_template.html) as f: template Template(f.read()) html template.render( nowdatetime.now().strftime(%Y-%m-%d %H:%M:%S), hostnamesnap[2], os_versionsnap[3], snapshot{ total_processes: snap[4], listening_ports_count: snap[5], high_cpu_count: snap[6], file_modified_count: snap[7] }, avg_processesround(avg_vals[0], 1) if avg_vals[0] else 0, avg_portsround(avg_vals[1], 1) if avg_vals[1] else 0, avg_cpuround(avg_vals[2], 1) if avg_vals[2] else 0, recent_alertsrecent_alerts, max_cpu_last24hint(trend_vals[0]) if trend_vals[0] else 0, avg_cpu_last24hround(trend_vals[1], 1) if trend_vals[1] else 0 ) with open(output_path, w, encodingutf-8) as f: f.write(html) print(fReport generated: {output_path}) if __name__ __main__: generate_html_report()提示答辩时只需运行python generate_report.py即可生成report.html。打开浏览器查看所有数据来自本地SQLite无需网络确保演示环境纯净。若需图表可引入chart.js在模板中添加canvas但纯文字版已满足毕设基础要求。5. 毕设交付与答辩技巧三个必做动作让老师一眼看到技术深度毕业设计评审最怕“看起来很热闹细看全是调API”。要让老师快速认可工作量和技术含量必须在交付包和答辩陈述中突出三个硬核动作5.1 交付包结构必须体现工程化思维拒绝“一堆py文件扔根目录”标准交付目录结构host-security-situation/├── README.md # 含环境要求、启动命令、规则修改指南、截图示例 ├── requirements.txt # 明确列出psutil5.9.5 jmespath1.0.1等精确版本 ├── config/ │ ├── rules.json # 预置10条规则含注释说明每条检测逻辑 │ └── baseline.db # 已初始化的文件基线数据库 ├── src/ │ ├── __init__.py │ ├── collector/ # 数据采集模块 │ │ ├── __init__.py │ │ ├── process_collector.py │ │ ├── network_collector.py │ │ └── file_integrity.py │ ├── engine/ │ │ ├── __init__.py │ │ └── rule_engine.py # 规则引擎核心 │ ├── database/ │ │ ├── __init__.py │ │ └── db_init.py # 数据库初始化 │ └── report/ │ ├── __init__.py │ ├── report_template.html │ └── generate_report.py ├── scripts/ │ ├── setup_baseline.py # 首次运行生成基线 │ └── start_monitor.py # 主监控入口含热加载逻辑 └── docs/ └── design_doc.md # UML类图用PlantUML语法、数据流图、规则设计说明注意requirements.txt必须用pip freeze requirements.txt生成而非手写。答辩时老师可能检查psutil版本是否与process_collector.py中psutil.process_iter([connections])兼容——5.9.0才支持该参数旧版本会报错。5.2 答辩演示必须包含“规则动态增删”和“基线重置”两个高光时刻老师最想验证你是否真懂原理而非复制粘贴。准备两段1分钟演示动态增规则打开config/rules.json新增一条检测powershell.exe异常网络连接的规则Windows专属保存后观察控制台30秒内打印新告警。强调“规则引擎不重启靠每次循环重读JSON实现热加载这是生产环境常见做法。”基线重置故意修改C:\Windows\System32\drivers\etc\hosts加一行127.0.0.1 evil.com运行python scripts/setup_baseline.py重建基线再运行监控——告警消失。说明“文件完整性检测依赖可信基线基线必须在干净系统上生成这模拟了企业安全运维中‘黄金镜像’的概念。”5.3 技术难点表格化呈现直击评审关注点在PPT最后一页用表格总结技术难点与解法避免口语化描述评审关注点你的解法为什么比别人强跨平台兼容性psutil统一进程采集 Windows/Linux双路径服务检测避免subprocess调用Shell命令导致的解析失败字段语义完全一致规则可维护性JSON定义 JMESPath引擎支持答辩现场修改不用改Python代码降低维护成本体现软件工程思想数据持久化SQLite存储快照告警趋势生成静态HTML报告无需部署数据库服务单文件交付符合毕设轻量级要求安全采集权限Windows服务检测自动降级权限不足时跳过、Linux进程连接数统计用psutil而非netstat明确处理权限边界不因权限问题导致整个系统崩溃提示答辩时指着表格说“老师这四个点是我们重点攻克的比如第三点很多同学用Excel手工整理数据而我们实现了自动化的态势报告生成点击这里就能导出HTML——请看屏幕。” 然后切换到已生成的report.html放大展示“最新告警”区块。本文还有配套的精品资源点击获取