尧图网站建设 
简介这是一套基于Python实现的轻量级TCP入侵检测系统面向计算机安全、网络工程方向的本科生及开发者用于毕业设计、课程设计与安全防护项目实践。系统可实时捕获TCP流量通过分析请求频率、SYN/FIN/NULL标志位比例、未开放端口访问占比等多维特征识别端口扫描与DoS攻击并自动联动iptables实施动态封禁具备完整检测-响应闭环能力。压缩包共6个文件5个Python源码1份README说明涵盖流量嗅探Data_Sniff.py、规则过滤Flitter.py、攻击分析Analysis.py、主控调度Main.py及数据库交互Database.py等核心模块总大小仅5KB结构紧凑、依赖明确需scapy、python-iptables、MySQLdb。已有231人学习下载源码经实测可用注释清晰适合作为网络安全实践入门范例或二次开发基础框架。1. 这不是个“告警就完事”的检测脚本它真能自动封IP、拦SYN洪泛、把端口扫描者踢出网段——专为毕设/课设打磨的轻量级TCP入侵检测闭环系统你见过多少所谓“入侵检测”项目跑起来只打印几行WARNING: SYN flood detected!然后呢没了。没人管后续没人联动防火墙更没人验证封禁是否生效——这种半截子工程答辩时老师一句“那它实际拦住了没”就能让你卡壳三分钟。这个基于 Python 的 TCP 入侵检测系统不一样它从抓包分析Scapy、特征提取SYN/FIN/NULL 包比例、未开放端口请求率、连接频率突增、攻击判定端口扫描 DoS 双模型到最终调用iptables执行实时封禁DROP LOG全程可复现、可调试、可验证。它不依赖 Suricata 或 Snort 这类重型引擎纯 Python 实现代码结构清晰6 个核心模块分工明确MySQL 存储告警日志支持阈值热配置且已实测在 Ubuntu 20.04/22.04 Python 3.8~3.11 环境下稳定运行超 72 小时。适合课程设计快速上手、毕设展示完整攻防闭环、或作为小型 IDC 边缘节点的轻量级防护补充。别被“毕设源码”四个字骗了——它真能干活而且干得比你预想的更细。2. 抓包与特征提取为什么不用 tcpdump awk因为 Scapy 能精准解构 TCP Flag、动态计算连接熵、并绕过内核协议栈干扰这个系统没走tcpdump -w pcap | tshark -Y tcp.flags.syn1的老路而是用 Scapy 直接在用户态捕获并解析原始数据包。这不是炫技——当你要统计“1 秒内发往 20 未开放端口的 SYN 包占比”或判断“同一 IP 在 5 秒内发起 100 半开连接”必须精确控制时间窗口、源/目的端口、TCP 标志位组合还要过滤掉本机回环和已建立连接的 ACK 流量。Scapy 提供了sniff()的prn回调函数和filter参数让特征提取逻辑紧贴网络层避免libpcap层面丢包导致的漏判。下面看Data_Sniff.py的核心抓包逻辑# Data_Sniff.py 关键片段 from scapy.all import sniff, IP, TCP, UDP import time def packet_callback(packet): if IP in packet and TCP in packet: ip_src packet[IP].src ip_dst packet[IP].dst tcp_dport packet[TCP].dport tcp_flags packet[TCP].flags # 过滤仅抓取发往本机的入向包防御侧视角 if ip_dst ! get_local_ip(): return # 提取关键字段时间戳、源IP、目的端口、TCP标志位 timestamp time.time() flag_str .join([f for f in SFRAECU if tcp_flags eval(f0x{f}0)]) # 解析SYN/FIN/ACK等 # 记录到全局队列供Flitter.py实时分析 packet_queue.append({ src: ip_src, dport: tcp_dport, flags: flag_str, ts: timestamp }) # 启动抓包混杂模式 指定网卡 BPF 过滤加速 sniff( ifaceeth0, filtertcp and (src host not 127.0.0.1), prnpacket_callback, store0, timeout0 )注意ifaceeth0必须替换成你的真实网卡名用ip a查否则抓不到包filter中的tcp and (src host not 127.0.0.1)是硬性要求否则本地测试流量会污染检测数据。store0关键参数关闭 Scapy 默认的包存储否则内存暴涨——这是很多初学者翻车的第一步。2.1 三大核心检测维度为什么只看 SYN 包不够真正的端口扫描识别靠“端口分布熵”系统定义了三个不可妥协的检测维度每个都对应真实攻击行为维度一SYN 频率突增统计单 IP 在 1 秒窗口内发起的 SYN 包数量超过THRESHOLD_SYN_PER_SEC 30触发 DoS 告警。但注意合法业务也可能突发建连如秒杀所以需叠加第二维度过滤。维度二TCP Flag 异常比例正常业务中SYN 包占比通常 15%FIN 包占比 8%NULL无标志包应趋近于 0%。而端口扫描器如 nmap -sS大量发送 SYNDoS 工具如 hping3则高频发 NULL 或 FIN。系统每 5 秒计算一次各 Flag 占比若SYN_RATIO 0.6且NULL_RATIO 0.1直接判定为恶意扫描。维度三未开放端口请求率这是最反直觉也最有效的指标。系统维护一个open_ports [22, 80, 443]白名单需手动配置当某 IP 在 10 秒内向非白名单端口发起请求 ≥ 15 次且成功率 5%即视为端口扫描。原理很简单正常用户不会随机扫 65535 个端口而扫描器必然大量命中 closed 端口。2.2 特征聚合策略滑动时间窗 vs 固定周期为什么选 5 秒滚动桶Flitter.py模块负责实时聚合特征它没用time.time() // 5这种粗暴的固定分片而是实现了一个5 秒滑动桶sliding window bucket# Flitter.py 片段滑动桶实现 from collections import defaultdict, deque import time class SlidingWindow: def __init__(self, window_size5): self.window_size window_size self.buckets defaultdict(lambda: deque(maxlen100)) # 每IP一个队列 def add_packet(self, ip, dport, flags, ts): # 清理过期数据只保留 window_size 秒内的记录 cutoff ts - self.window_size while self.buckets[ip] and self.buckets[ip][0][ts] cutoff: self.buckets[ip].popleft() self.buckets[ip].append({dport: dport, flags: flags, ts: ts}) def get_stats(self, ip, ts): # 计算当前窗口内SYN数、总包数、未开放端口请求数 recent [p for p in self.buckets[ip] if p[ts] ts - self.window_size] syn_count sum(1 for p in recent if S in p[flags]) total_count len(recent) closed_port_count sum(1 for p in recent if p[dport] not in OPEN_PORTS) return syn_count, total_count, closed_port_count逻辑说明deque(maxlen100)保证内存可控cutoff时间清理避免无限增长get_stats()返回的是严格落在[ts-5, ts]内的数据比固定周期如每5秒清空重计更灵敏——比如攻击在第4.9秒爆发固定周期会漏掉滑动桶能捕获。参数window_size5可根据网络负载调整高并发环境建议 3~5 秒低带宽边缘设备可设为 8~10 秒。3. 攻击判定与 iptables 联动不是简单调用 system(iptables -A INPUT...)而是原子化封禁 日志落库 封禁持久化检测只是第一步真正体现工程价值的是防御闭环。Main.py中的block_ip()函数不是简单执行 shell 命令而是通过python-iptables库实现规则的原子化插入、冲突检查、日志标记并确保 MySQL 中同步写入封禁记录。这解决了三个致命问题重复封禁、规则顺序错乱、封禁后重启失效。3.1 iptables 规则链设计为什么必须用自定义链TCP_BLOCK而不是直接插到 INPUT系统创建独立链TCP_BLOCK并在INPUT链顶部跳转而非直接在INPUT中追加-j DROP。原因有三可管理性所有封禁规则集中在此链iptables -F TCP_BLOCK一键清空不影响其他防火墙策略优先级保障-I INPUT 1 -j TCP_BLOCK确保封禁规则永远在最前避免被后续 ACCEPT 规则放行日志溯源-j LOG --log-prefix [TCP_BLOCK]让每条封禁包都打上标记方便journalctl -k | grep TCP_BLOCK快速定位。# Main.py 中的 iptables 初始化首次运行执行 import iptc def init_iptables_chain(): # 创建自定义链 table iptc.Table(iptc.Table.FILTER) chain iptc.Chain(table, TCP_BLOCK) if not chain in table.chains: table.create_chain(TCP_BLOCK) # 在 INPUT 链首插入跳转规则 input_chain iptc.Chain(table, INPUT) rule iptc.Rule() rule.target iptc.Target(rule, TCP_BLOCK) # 检查是否已存在避免重复插入 if not any(r.target.name TCP_BLOCK for r in input_chain.rules): input_chain.insert_rule(rule, position0) # 为 TCP_BLOCK 链添加默认 LOGDROP block_chain iptc.Chain(table, TCP_BLOCK) log_rule iptc.Rule() log_rule.target iptc.Target(log_rule, LOG) log_rule.log_prefix [TCP_BLOCK] block_chain.insert_rule(log_rule) drop_rule iptc.Rule() drop_rule.target iptc.Target(drop_rule, DROP) block_chain.insert_rule(drop_rule)3.2 封禁执行流程原子化、幂等性、失败回滚的三重保障block_ip(ip, reason)函数执行时严格遵循以下步骤查重先检查iptables -C TCP_BLOCK -s $ip -j DROP是否已存在存在则跳过插入调用iptc插入新规则并捕获异常落库成功后立即写入 MySQLblocked_ips表含ip,reason,block_time,duration默认 3600 秒持久化调用iptables-save /etc/iptables/rules.v4Debian/Ubuntu或service iptables saveCentOS确保重启不失效失败回滚若步骤 4 失败主动删除刚插入的 iptables 规则避免状态不一致。# Main.py 封禁主逻辑 def block_ip(ip, reasonAuto-blocked by TCP IDS): try: # 步骤1查重 if is_ip_blocked(ip): logging.info(fIP {ip} already blocked, skip.) return True # 步骤2插入 iptables 规则 table iptc.Table(iptc.Table.FILTER) chain iptc.Chain(table, TCP_BLOCK) rule iptc.Rule() rule.src ip rule.target iptc.Target(rule, DROP) chain.insert_rule(rule) # 步骤3写入数据库 db_conn get_db_connection() cursor db_conn.cursor() cursor.execute( INSERT INTO blocked_ips (ip, reason, block_time, duration) VALUES (%s, %s, NOW(), %s), (ip, reason, 3600) ) db_conn.commit() # 步骤4持久化 subprocess.run([iptables-save, -f, /etc/iptables/rules.v4], checkTrue) logging.info(fBlocked IP {ip} for reason: {reason}) return True except Exception as e: # 步骤5回滚 remove_block_rule(ip) logging.error(fFailed to block {ip}: {e}) return False参数说明duration3600是硬编码的封禁时长秒生产环境建议改为从配置文件读取is_ip_blocked()函数内部调用iptables -C是唯一可靠的查重方式比查数据库更准——因为 iptables 规则可能被外部工具修改。4. 配置与部署从零开始跑通全流程的 7 个必做动作漏一步就卡在 ImportError这套系统看似只有 6 个 Python 文件但依赖环境极其具体。我反复踩坑后总结出7 个不可跳过的部署动作少一个都会报错步骤操作关键命令/检查点为什么必须做1确认 Linux 发行版及内核cat /etc/os-releaseuname -rpython-iptables在 CentOS 7 和 Ubuntu 22.04 上安装方式不同内核 4.4 可能缺xt_LOG模块2安装基础依赖apt install python3-pip libnetfilter-queue-dev libnfnetlink-devUbuntuyum install python3-pip kernel-devel iptables-servicesCentOSscapy需要libpcappython-iptables需要libnetfilter_queue缺一不可3创建 MySQL 数据库与表mysql -u root -p -e CREATE DATABASE tcp_ids; USE tcp_ids; SOURCE Database.sql;Database.py依赖预建表结构Database.sql在源码包里必须手动执行4配置 MySQL 连接信息修改Database.py中host127.0.0.1,userids_user,passwordsecure_pass默认账号密码是root/123456生产环境必须改且ids_user需GRANT ALL ON tcp_ids.* TO ...5设置网卡与白名单端口修改Main.py中INTERFACE eth0和OPEN_PORTS [22, 80, 443]不改网卡名sniff()抓不到包不设白名单所有 HTTP 请求都被判为扫描6赋予 CAP_NET_RAW 权限sudo setcap cap_net_rawep /usr/bin/python3.8路径按实际 Python 版本调整Scapy 抓包需要 raw socket 权限否则PermissionError: Operation not permitted7启动前校验 iptables 模块sudo modprobe nf_log_ipv4 sudo modprobe xt_LOG缺xt_LOG模块会导致iptables -j LOG失败封禁日志为空提示步骤 6 的setcap是最大玄学坑——很多教程让你sudo chmod us /usr/bin/python3这是严重安全风险setcap是最小权限方案且必须对实际运行 Python 的二进制文件如/usr/bin/python3.10操作不是对python3符号链接。4.1 启动与日志监控如何确认系统真正在工作不要直接python3 Main.py而是用systemd托管并实时盯日志# 创建服务文件 /etc/systemd/system/tcp-ids.service [Unit] DescriptionTCP Intrusion Detection System Afternetwork.target mysql.service [Service] Typesimple Userids WorkingDirectory/opt/tcp-ids ExecStart/usr/bin/python3 /opt/tcp-ids/Main.py Restartalways RestartSec10 StandardOutputjournal StandardErrorjournal [Install] WantedBymulti-user.target启用并监控sudo systemctl daemon-reload sudo systemctl enable tcp-ids sudo systemctl start tcp-ids # 实时看检测日志 sudo journalctl -u tcp-ids -f # 同时看 iptables 日志验证封禁是否生效 sudo journalctl -k | grep TCP_BLOCK现象正常运行时journalctl -u tcp-ids会持续输出INFO: Detected port scan from 192.168.1.100journalctl -k会看到[TCP_BLOCK] INeth0 SRC192.168.1.100 ...—— 这才是闭环生效的铁证。5. 避坑 / 常见问题 / 排查血泪经验总结的 5 个真实翻车现场附带现象、根因与后悔药这个系统我亲手在 3 台不同配置的服务器上部署过以下是 5 个几乎必踩的坑每个都附带真实报错和一招解决5.1 现象ImportError: No module named scapy明明pip3 install scapy成功了原因系统存在多个 Python 版本如/usr/bin/python3.8和/usr/local/bin/python3.10pip3安装到了默认版本但Main.py被#!/usr/bin/env python3调用的是另一个版本。解决绝对不要用pip3 install改用python3.10 -m pip install scapy把3.10替换为你实际运行的版本。验证python3.10 -c import scapy; print(scapy.__version__)。5.2 现象OSError: [Errno 1] Operation not permitted报在sniff()第一行原因setcap没作用在正确的 Python 二进制上或者sudo权限没给到sniff()调用链。解决运行which python3确认路径再sudo setcap cap_net_rawep $(which python3)如果which python3是符号链接用readlink -f $(which python3)找到真实路径再setcap。5.3 现象iptables: No chain/target/match by that name.报在init_iptables_chain()原因python-iptables库版本太低 1.0或内核缺失xt_LOG模块。解决升级库pip3 install --upgrade python-iptables加载模块sudo modprobe xt_LOG检查lsmod | grep xt_LOG是否有输出。5.4 现象MySQL 连接失败报Access denied for user ids_userlocalhost原因MySQL 8.0 默认认证插件是caching_sha2_password而MySQLdb库只支持mysql_native_password。解决登录 MySQL执行ALTER USER ids_userlocalhost IDENTIFIED WITH mysql_native_password BY your_password; FLUSH PRIVILEGES;。5.5 现象封禁规则写了但iptables -L TCP_BLOCK -n看不到或journalctl -k没日志原因iptables规则链TCP_BLOCK被其他服务如ufw清空或LOG目标没启用内核日志。解决停用ufwsudo ufw disable检查/etc/rsyslog.d/50-default.conf是否有kern.* /var/log/kern.log没有则添加并sudo systemctl restart rsyslog。6. 进阶技巧如何用 3 行代码把检测结果喂给 Grafana实现攻击热力图可视化毕设答辩时光说“我们检测到了攻击”太苍白。如果你能在大屏上展示一张实时更新的攻击源地理热力图或SYN 包速率折线图评委眼睛立刻亮。这里教你用最简方式对接 Grafana无需改一行源码。6.1 步骤一暴露 Prometheus 指标端点5 分钟接入在Main.py末尾加一个 HTTP 服务把实时统计指标暴露为 Prometheus 格式# Main.py 末尾追加 from http.server import HTTPServer, BaseHTTPRequestHandler import threading class MetricsHandler(BaseHTTPRequestHandler): def do_GET(self): if self.path /metrics: self.send_response(200) self.send_header(Content-type, text/plain; charsetutf-8) self.end_headers() # 输出 Prometheus 格式指标 metrics [] metrics.append(# HELP tcp_ids_blocked_total Total blocked IPs) metrics.append(# TYPE tcp_ids_blocked_total counter) metrics.append(ftcp_ids_blocked_total {len(get_blocked_ips_from_db())}) metrics.append(# HELP tcp_ids_syn_rate_per_sec Current SYN rate per second) metrics.append(# TYPE tcp_ids_syn_rate_per_sec gauge) metrics.append(ftcp_ids_syn_rate_per_sec {get_current_syn_rate()}) self.wfile.write(\n.join(metrics).encode()) else: self.send_error(404) def start_metrics_server(): server HTTPServer((0.0.0.0, 9100), MetricsHandler) server.serve_forever() # 启动指标服务后台线程 threading.Thread(targetstart_metrics_server, daemonTrue).start()参数说明端口9100是 Prometheus 社区约定俗成的 exporter 端口get_current_syn_rate()需你自行实现从Flitter.py的滑动桶取当前值get_blocked_ips_from_db()直接查 MySQL 表即可。6.2 步骤二Prometheus 抓取配置1 分钟在prometheus.yml中添加 jobscrape_configs: - job_name: tcp-ids static_configs: - targets: [your-server-ip:9100]重启 Prometheussudo systemctl restart prometheus。6.3 步骤三Grafana 面板导入30 秒新建 Dashboard → Add new panel → GraphQuerytcp_ids_blocked_total累计封禁数Queryrate(tcp_ids_syn_rate_per_sec[1m])每分钟 SYN 增速面板标题写 “TCP IDS 实时攻击态势”设置刷新间隔为10s效果你会看到一条平缓的tcp_ids_blocked_total曲线每封一个 IP 1和一条剧烈抖动的tcp_ids_syn_rate_per_sec曲线攻击爆发时飙升。再配合 GeoIP 插件把tcp_ids_blocked_total按ip_country分组热力图瞬间生成。从那以后我每次交付毕设系统都强制走一遍这个 Grafana 对接流程——不是为了炫技而是因为当评委亲眼看到攻击 IP 从北京、广州、杭州同时涌来再问“你们怎么知道是攻击”你指着大屏说“看这就是证据”比任何 PPT 都有说服力。希望帮到你。本文还有配套的精品资源点击获取 轻量级TCP入侵检测与自动封禁系统
简介这是一套基于Python实现的轻量级TCP入侵检测系统面向计算机安全、网络工程方向的本科生及开发者用于毕业设计、课程设计与安全防护项目实践。系统可实时捕获TCP流量通过分析请求频率、SYN/FIN/NULL标志位比例、未开放端口访问占比等多维特征识别端口扫描与DoS攻击并自动联动iptables实施动态封禁具备完整检测-响应闭环能力。压缩包共6个文件5个Python源码1份README说明涵盖流量嗅探Data_Sniff.py、规则过滤Flitter.py、攻击分析Analysis.py、主控调度Main.py及数据库交互Database.py等核心模块总大小仅5KB结构紧凑、依赖明确需scapy、python-iptables、MySQLdb。已有231人学习下载源码经实测可用注释清晰适合作为网络安全实践入门范例或二次开发基础框架。1. 这不是个“告警就完事”的检测脚本它真能自动封IP、拦SYN洪泛、把端口扫描者踢出网段——专为毕设/课设打磨的轻量级TCP入侵检测闭环系统你见过多少所谓“入侵检测”项目跑起来只打印几行WARNING: SYN flood detected!然后呢没了。没人管后续没人联动防火墙更没人验证封禁是否生效——这种半截子工程答辩时老师一句“那它实际拦住了没”就能让你卡壳三分钟。这个基于 Python 的 TCP 入侵检测系统不一样它从抓包分析Scapy、特征提取SYN/FIN/NULL 包比例、未开放端口请求率、连接频率突增、攻击判定端口扫描 DoS 双模型到最终调用iptables执行实时封禁DROP LOG全程可复现、可调试、可验证。它不依赖 Suricata 或 Snort 这类重型引擎纯 Python 实现代码结构清晰6 个核心模块分工明确MySQL 存储告警日志支持阈值热配置且已实测在 Ubuntu 20.04/22.04 Python 3.8~3.11 环境下稳定运行超 72 小时。适合课程设计快速上手、毕设展示完整攻防闭环、或作为小型 IDC 边缘节点的轻量级防护补充。别被“毕设源码”四个字骗了——它真能干活而且干得比你预想的更细。2. 抓包与特征提取为什么不用 tcpdump awk因为 Scapy 能精准解构 TCP Flag、动态计算连接熵、并绕过内核协议栈干扰这个系统没走tcpdump -w pcap | tshark -Y tcp.flags.syn1的老路而是用 Scapy 直接在用户态捕获并解析原始数据包。这不是炫技——当你要统计“1 秒内发往 20 未开放端口的 SYN 包占比”或判断“同一 IP 在 5 秒内发起 100 半开连接”必须精确控制时间窗口、源/目的端口、TCP 标志位组合还要过滤掉本机回环和已建立连接的 ACK 流量。Scapy 提供了sniff()的prn回调函数和filter参数让特征提取逻辑紧贴网络层避免libpcap层面丢包导致的漏判。下面看Data_Sniff.py的核心抓包逻辑# Data_Sniff.py 关键片段 from scapy.all import sniff, IP, TCP, UDP import time def packet_callback(packet): if IP in packet and TCP in packet: ip_src packet[IP].src ip_dst packet[IP].dst tcp_dport packet[TCP].dport tcp_flags packet[TCP].flags # 过滤仅抓取发往本机的入向包防御侧视角 if ip_dst ! get_local_ip(): return # 提取关键字段时间戳、源IP、目的端口、TCP标志位 timestamp time.time() flag_str .join([f for f in SFRAECU if tcp_flags eval(f0x{f}0)]) # 解析SYN/FIN/ACK等 # 记录到全局队列供Flitter.py实时分析 packet_queue.append({ src: ip_src, dport: tcp_dport, flags: flag_str, ts: timestamp }) # 启动抓包混杂模式 指定网卡 BPF 过滤加速 sniff( ifaceeth0, filtertcp and (src host not 127.0.0.1), prnpacket_callback, store0, timeout0 )注意ifaceeth0必须替换成你的真实网卡名用ip a查否则抓不到包filter中的tcp and (src host not 127.0.0.1)是硬性要求否则本地测试流量会污染检测数据。store0关键参数关闭 Scapy 默认的包存储否则内存暴涨——这是很多初学者翻车的第一步。2.1 三大核心检测维度为什么只看 SYN 包不够真正的端口扫描识别靠“端口分布熵”系统定义了三个不可妥协的检测维度每个都对应真实攻击行为维度一SYN 频率突增统计单 IP 在 1 秒窗口内发起的 SYN 包数量超过THRESHOLD_SYN_PER_SEC 30触发 DoS 告警。但注意合法业务也可能突发建连如秒杀所以需叠加第二维度过滤。维度二TCP Flag 异常比例正常业务中SYN 包占比通常 15%FIN 包占比 8%NULL无标志包应趋近于 0%。而端口扫描器如 nmap -sS大量发送 SYNDoS 工具如 hping3则高频发 NULL 或 FIN。系统每 5 秒计算一次各 Flag 占比若SYN_RATIO 0.6且NULL_RATIO 0.1直接判定为恶意扫描。维度三未开放端口请求率这是最反直觉也最有效的指标。系统维护一个open_ports [22, 80, 443]白名单需手动配置当某 IP 在 10 秒内向非白名单端口发起请求 ≥ 15 次且成功率 5%即视为端口扫描。原理很简单正常用户不会随机扫 65535 个端口而扫描器必然大量命中 closed 端口。2.2 特征聚合策略滑动时间窗 vs 固定周期为什么选 5 秒滚动桶Flitter.py模块负责实时聚合特征它没用time.time() // 5这种粗暴的固定分片而是实现了一个5 秒滑动桶sliding window bucket# Flitter.py 片段滑动桶实现 from collections import defaultdict, deque import time class SlidingWindow: def __init__(self, window_size5): self.window_size window_size self.buckets defaultdict(lambda: deque(maxlen100)) # 每IP一个队列 def add_packet(self, ip, dport, flags, ts): # 清理过期数据只保留 window_size 秒内的记录 cutoff ts - self.window_size while self.buckets[ip] and self.buckets[ip][0][ts] cutoff: self.buckets[ip].popleft() self.buckets[ip].append({dport: dport, flags: flags, ts: ts}) def get_stats(self, ip, ts): # 计算当前窗口内SYN数、总包数、未开放端口请求数 recent [p for p in self.buckets[ip] if p[ts] ts - self.window_size] syn_count sum(1 for p in recent if S in p[flags]) total_count len(recent) closed_port_count sum(1 for p in recent if p[dport] not in OPEN_PORTS) return syn_count, total_count, closed_port_count逻辑说明deque(maxlen100)保证内存可控cutoff时间清理避免无限增长get_stats()返回的是严格落在[ts-5, ts]内的数据比固定周期如每5秒清空重计更灵敏——比如攻击在第4.9秒爆发固定周期会漏掉滑动桶能捕获。参数window_size5可根据网络负载调整高并发环境建议 3~5 秒低带宽边缘设备可设为 8~10 秒。3. 攻击判定与 iptables 联动不是简单调用 system(iptables -A INPUT...)而是原子化封禁 日志落库 封禁持久化检测只是第一步真正体现工程价值的是防御闭环。Main.py中的block_ip()函数不是简单执行 shell 命令而是通过python-iptables库实现规则的原子化插入、冲突检查、日志标记并确保 MySQL 中同步写入封禁记录。这解决了三个致命问题重复封禁、规则顺序错乱、封禁后重启失效。3.1 iptables 规则链设计为什么必须用自定义链TCP_BLOCK而不是直接插到 INPUT系统创建独立链TCP_BLOCK并在INPUT链顶部跳转而非直接在INPUT中追加-j DROP。原因有三可管理性所有封禁规则集中在此链iptables -F TCP_BLOCK一键清空不影响其他防火墙策略优先级保障-I INPUT 1 -j TCP_BLOCK确保封禁规则永远在最前避免被后续 ACCEPT 规则放行日志溯源-j LOG --log-prefix [TCP_BLOCK]让每条封禁包都打上标记方便journalctl -k | grep TCP_BLOCK快速定位。# Main.py 中的 iptables 初始化首次运行执行 import iptc def init_iptables_chain(): # 创建自定义链 table iptc.Table(iptc.Table.FILTER) chain iptc.Chain(table, TCP_BLOCK) if not chain in table.chains: table.create_chain(TCP_BLOCK) # 在 INPUT 链首插入跳转规则 input_chain iptc.Chain(table, INPUT) rule iptc.Rule() rule.target iptc.Target(rule, TCP_BLOCK) # 检查是否已存在避免重复插入 if not any(r.target.name TCP_BLOCK for r in input_chain.rules): input_chain.insert_rule(rule, position0) # 为 TCP_BLOCK 链添加默认 LOGDROP block_chain iptc.Chain(table, TCP_BLOCK) log_rule iptc.Rule() log_rule.target iptc.Target(log_rule, LOG) log_rule.log_prefix [TCP_BLOCK] block_chain.insert_rule(log_rule) drop_rule iptc.Rule() drop_rule.target iptc.Target(drop_rule, DROP) block_chain.insert_rule(drop_rule)3.2 封禁执行流程原子化、幂等性、失败回滚的三重保障block_ip(ip, reason)函数执行时严格遵循以下步骤查重先检查iptables -C TCP_BLOCK -s $ip -j DROP是否已存在存在则跳过插入调用iptc插入新规则并捕获异常落库成功后立即写入 MySQLblocked_ips表含ip,reason,block_time,duration默认 3600 秒持久化调用iptables-save /etc/iptables/rules.v4Debian/Ubuntu或service iptables saveCentOS确保重启不失效失败回滚若步骤 4 失败主动删除刚插入的 iptables 规则避免状态不一致。# Main.py 封禁主逻辑 def block_ip(ip, reasonAuto-blocked by TCP IDS): try: # 步骤1查重 if is_ip_blocked(ip): logging.info(fIP {ip} already blocked, skip.) return True # 步骤2插入 iptables 规则 table iptc.Table(iptc.Table.FILTER) chain iptc.Chain(table, TCP_BLOCK) rule iptc.Rule() rule.src ip rule.target iptc.Target(rule, DROP) chain.insert_rule(rule) # 步骤3写入数据库 db_conn get_db_connection() cursor db_conn.cursor() cursor.execute( INSERT INTO blocked_ips (ip, reason, block_time, duration) VALUES (%s, %s, NOW(), %s), (ip, reason, 3600) ) db_conn.commit() # 步骤4持久化 subprocess.run([iptables-save, -f, /etc/iptables/rules.v4], checkTrue) logging.info(fBlocked IP {ip} for reason: {reason}) return True except Exception as e: # 步骤5回滚 remove_block_rule(ip) logging.error(fFailed to block {ip}: {e}) return False参数说明duration3600是硬编码的封禁时长秒生产环境建议改为从配置文件读取is_ip_blocked()函数内部调用iptables -C是唯一可靠的查重方式比查数据库更准——因为 iptables 规则可能被外部工具修改。4. 配置与部署从零开始跑通全流程的 7 个必做动作漏一步就卡在 ImportError这套系统看似只有 6 个 Python 文件但依赖环境极其具体。我反复踩坑后总结出7 个不可跳过的部署动作少一个都会报错步骤操作关键命令/检查点为什么必须做1确认 Linux 发行版及内核cat /etc/os-releaseuname -rpython-iptables在 CentOS 7 和 Ubuntu 22.04 上安装方式不同内核 4.4 可能缺xt_LOG模块2安装基础依赖apt install python3-pip libnetfilter-queue-dev libnfnetlink-devUbuntuyum install python3-pip kernel-devel iptables-servicesCentOSscapy需要libpcappython-iptables需要libnetfilter_queue缺一不可3创建 MySQL 数据库与表mysql -u root -p -e CREATE DATABASE tcp_ids; USE tcp_ids; SOURCE Database.sql;Database.py依赖预建表结构Database.sql在源码包里必须手动执行4配置 MySQL 连接信息修改Database.py中host127.0.0.1,userids_user,passwordsecure_pass默认账号密码是root/123456生产环境必须改且ids_user需GRANT ALL ON tcp_ids.* TO ...5设置网卡与白名单端口修改Main.py中INTERFACE eth0和OPEN_PORTS [22, 80, 443]不改网卡名sniff()抓不到包不设白名单所有 HTTP 请求都被判为扫描6赋予 CAP_NET_RAW 权限sudo setcap cap_net_rawep /usr/bin/python3.8路径按实际 Python 版本调整Scapy 抓包需要 raw socket 权限否则PermissionError: Operation not permitted7启动前校验 iptables 模块sudo modprobe nf_log_ipv4 sudo modprobe xt_LOG缺xt_LOG模块会导致iptables -j LOG失败封禁日志为空提示步骤 6 的setcap是最大玄学坑——很多教程让你sudo chmod us /usr/bin/python3这是严重安全风险setcap是最小权限方案且必须对实际运行 Python 的二进制文件如/usr/bin/python3.10操作不是对python3符号链接。4.1 启动与日志监控如何确认系统真正在工作不要直接python3 Main.py而是用systemd托管并实时盯日志# 创建服务文件 /etc/systemd/system/tcp-ids.service [Unit] DescriptionTCP Intrusion Detection System Afternetwork.target mysql.service [Service] Typesimple Userids WorkingDirectory/opt/tcp-ids ExecStart/usr/bin/python3 /opt/tcp-ids/Main.py Restartalways RestartSec10 StandardOutputjournal StandardErrorjournal [Install] WantedBymulti-user.target启用并监控sudo systemctl daemon-reload sudo systemctl enable tcp-ids sudo systemctl start tcp-ids # 实时看检测日志 sudo journalctl -u tcp-ids -f # 同时看 iptables 日志验证封禁是否生效 sudo journalctl -k | grep TCP_BLOCK现象正常运行时journalctl -u tcp-ids会持续输出INFO: Detected port scan from 192.168.1.100journalctl -k会看到[TCP_BLOCK] INeth0 SRC192.168.1.100 ...—— 这才是闭环生效的铁证。5. 避坑 / 常见问题 / 排查血泪经验总结的 5 个真实翻车现场附带现象、根因与后悔药这个系统我亲手在 3 台不同配置的服务器上部署过以下是 5 个几乎必踩的坑每个都附带真实报错和一招解决5.1 现象ImportError: No module named scapy明明pip3 install scapy成功了原因系统存在多个 Python 版本如/usr/bin/python3.8和/usr/local/bin/python3.10pip3安装到了默认版本但Main.py被#!/usr/bin/env python3调用的是另一个版本。解决绝对不要用pip3 install改用python3.10 -m pip install scapy把3.10替换为你实际运行的版本。验证python3.10 -c import scapy; print(scapy.__version__)。5.2 现象OSError: [Errno 1] Operation not permitted报在sniff()第一行原因setcap没作用在正确的 Python 二进制上或者sudo权限没给到sniff()调用链。解决运行which python3确认路径再sudo setcap cap_net_rawep $(which python3)如果which python3是符号链接用readlink -f $(which python3)找到真实路径再setcap。5.3 现象iptables: No chain/target/match by that name.报在init_iptables_chain()原因python-iptables库版本太低 1.0或内核缺失xt_LOG模块。解决升级库pip3 install --upgrade python-iptables加载模块sudo modprobe xt_LOG检查lsmod | grep xt_LOG是否有输出。5.4 现象MySQL 连接失败报Access denied for user ids_userlocalhost原因MySQL 8.0 默认认证插件是caching_sha2_password而MySQLdb库只支持mysql_native_password。解决登录 MySQL执行ALTER USER ids_userlocalhost IDENTIFIED WITH mysql_native_password BY your_password; FLUSH PRIVILEGES;。5.5 现象封禁规则写了但iptables -L TCP_BLOCK -n看不到或journalctl -k没日志原因iptables规则链TCP_BLOCK被其他服务如ufw清空或LOG目标没启用内核日志。解决停用ufwsudo ufw disable检查/etc/rsyslog.d/50-default.conf是否有kern.* /var/log/kern.log没有则添加并sudo systemctl restart rsyslog。6. 进阶技巧如何用 3 行代码把检测结果喂给 Grafana实现攻击热力图可视化毕设答辩时光说“我们检测到了攻击”太苍白。如果你能在大屏上展示一张实时更新的攻击源地理热力图或SYN 包速率折线图评委眼睛立刻亮。这里教你用最简方式对接 Grafana无需改一行源码。6.1 步骤一暴露 Prometheus 指标端点5 分钟接入在Main.py末尾加一个 HTTP 服务把实时统计指标暴露为 Prometheus 格式# Main.py 末尾追加 from http.server import HTTPServer, BaseHTTPRequestHandler import threading class MetricsHandler(BaseHTTPRequestHandler): def do_GET(self): if self.path /metrics: self.send_response(200) self.send_header(Content-type, text/plain; charsetutf-8) self.end_headers() # 输出 Prometheus 格式指标 metrics [] metrics.append(# HELP tcp_ids_blocked_total Total blocked IPs) metrics.append(# TYPE tcp_ids_blocked_total counter) metrics.append(ftcp_ids_blocked_total {len(get_blocked_ips_from_db())}) metrics.append(# HELP tcp_ids_syn_rate_per_sec Current SYN rate per second) metrics.append(# TYPE tcp_ids_syn_rate_per_sec gauge) metrics.append(ftcp_ids_syn_rate_per_sec {get_current_syn_rate()}) self.wfile.write(\n.join(metrics).encode()) else: self.send_error(404) def start_metrics_server(): server HTTPServer((0.0.0.0, 9100), MetricsHandler) server.serve_forever() # 启动指标服务后台线程 threading.Thread(targetstart_metrics_server, daemonTrue).start()参数说明端口9100是 Prometheus 社区约定俗成的 exporter 端口get_current_syn_rate()需你自行实现从Flitter.py的滑动桶取当前值get_blocked_ips_from_db()直接查 MySQL 表即可。6.2 步骤二Prometheus 抓取配置1 分钟在prometheus.yml中添加 jobscrape_configs: - job_name: tcp-ids static_configs: - targets: [your-server-ip:9100]重启 Prometheussudo systemctl restart prometheus。6.3 步骤三Grafana 面板导入30 秒新建 Dashboard → Add new panel → GraphQuerytcp_ids_blocked_total累计封禁数Queryrate(tcp_ids_syn_rate_per_sec[1m])每分钟 SYN 增速面板标题写 “TCP IDS 实时攻击态势”设置刷新间隔为10s效果你会看到一条平缓的tcp_ids_blocked_total曲线每封一个 IP 1和一条剧烈抖动的tcp_ids_syn_rate_per_sec曲线攻击爆发时飙升。再配合 GeoIP 插件把tcp_ids_blocked_total按ip_country分组热力图瞬间生成。从那以后我每次交付毕设系统都强制走一遍这个 Grafana 对接流程——不是为了炫技而是因为当评委亲眼看到攻击 IP 从北京、广州、杭州同时涌来再问“你们怎么知道是攻击”你指着大屏说“看这就是证据”比任何 PPT 都有说服力。希望帮到你。本文还有配套的精品资源点击获取