简介这是一套面向高校计算机网络、信息安全专业学生及中小型网络运维人员的Python TCP入侵检测系统源码可作为毕业设计、课程设计或项目开发参考。系统聚焦TCP层安全监测通过分析连接请求的时间序列频率、TCP头部标志位组合SYN、FIN及NULL包分布以及非监听端口连接占比识别端口扫描与DDoS攻击并借助scapy抓包解析、python-iptables联动防火墙实现自动防御检测日志经MySQLdb结构化存储。资源包共10个文件以5个py源码文件为核心辅以3个zbak备份、1个zip与1个md说明文档整体约10KB结构清晰、模块耦合度低。目前已有41人学习。读者可从中获取完整的检测逻辑实现、防火墙联动策略与数据库存储方案理解异常流量判定思路并在此基础上扩展自定义规则或适配更复杂的网络环境。1. 从一份 TCP 入侵检测源码说起端口扫描和 DoS 到底怎么被认出来很多人做网络安全方向的课程设计第一反应是抓包第二反应是懵——包抓到了怎么判断哪个是攻击这份基于 Python 的 TCP 入侵检测系统源码解决的正是这个「抓到了但看不懂」的问题。它盯着 TCP 层做两件事识别端口扫描和 DoS 攻击识别出来之后不是弹个窗就完事而是直接调用 iptables 把攻击源 IP 封掉形成检测到响应的闭环。技术栈上抓包解析靠 scapy防火墙联动靠 python-iptables日志落库走 MySQLdb模块拆得比较干净Analysis、Data_Sniff、Flitter、Database 各管一摊。适合谁计算机网络和信息安全专业的毕业设计、课程设计以及想在中小型网络里搭一套主动防护原型的开发者。下面我按「它凭什么判断 → 怎么跑起来 → 坑在哪 → 怎么调」的顺序拆一遍。2. 检测逻辑拆解三个维度怎么把扫描和 DoS 从正常流量里拎出来这套系统的判断不是靠单一阈值拍脑袋而是从 TCP 连接的三个特征维度交叉验证。理解这三个维度你才知道后面参数该往哪调、误报该从哪查。2.1 时间序列频率正常访问基线和突发高频的分界第一个维度是 TCP 连接请求的时间序列频率。思路很直白正常用户访问服务连接请求是稀疏的、有间隔的端口扫描和 DoS 会在短时间内产生大量连接请求。系统维护一个滑动时间窗口统计窗口内某个源 IP 发起的连接数超过基线就标记异常。这里的关键参数是窗口长度和阈值。窗口太短正常业务的突发流量比如页面加载时并发拉取多个资源会误报窗口太长慢速扫描攻击者故意拉长间隔又检测不到。常见做法是窗口设 5 到 10 秒阈值根据自己网络的正常 QPS 来定。我一般会先跑一段纯正常流量看窗口内连接数的分布取一个比正常峰值高 2 到 3 倍的值当阈值而不是拍一个固定数字。# 滑动窗口频率检测的简化逻辑 from collections import defaultdict import time class FrequencyDetector: def __init__(self, window5, threshold50): self.window window # 时间窗口单位秒 self.threshold threshold # 窗口内连接数阈值 self.records defaultdict(list) # 源IP - 连接时间戳列表 def add_connection(self, src_ip): now time.time() self.records[src_ip].append(now) # 清理窗口外的旧记录避免列表无限增长 self.records[src_ip] [ t for t in self.records[src_ip] if now - t self.window ] # 窗口内连接数超阈值即判定异常 if len(self.records[src_ip]) self.threshold: return True return False这段代码里window和threshold是最需要按环境调的两个参数。records用字典按源 IP 分开存时间戳清理旧记录这一步不能省否则跑久了内存会涨。返回 True 只是标记真正封禁交给后面的 Flitter 模块。2.2 TCP 标志位组合SYN、FIN 和 NULL 包的分布玄学第二个维度是 TCP 协议头部的标志位组合模式。这是这套系统里最有技术含量、也最容易翻车的部分。正常 TCP 连接有固定的握手挥手流程SYN → SYN-ACK → ACK 建立连接FIN/ACK 关闭连接。而扫描工具为了探测端口状态会发出各种非常规组合只发 SYN 的包半开扫描如果目标端口开放会回 SYN-ACK不开放回 RST扫描器据此判断端口状态发 FIN 包FIN 扫描绕过某些只检测 SYN 的防火墙发 NULL 包所有标志位为 0这是典型的扫描特征正常通信不会出现。系统统计这些非常规标志位包在总流量中的占比占比异常就判定为扫描。这里有个血泪经验不同操作系统对非常规标志位的响应不一样Windows 和 Linux 对 NULL 包的处理就有差异所以阈值不能照搬别人的得在自己的目标环境里实测。from scapy.all import TCP def classify_flags(packet): 解析TCP标志位返回标志类型 if not packet.haslayer(TCP): return None flags packet[TCP].flags # 正常握手挥手相关 if flags 0x02: # SYN return SYN elif flags 0x01: # FIN return FIN elif flags 0: # NULL扫描所有标志位为0 return NULL elif flags 0x10: # ACK return ACK return OTHERflags是 scapy 解析出来的标志位字段用位运算判断。0x02 是 SYN0x01 是 FIN0x10 是 ACK这些是 TCP 头部的标准定义。NULL 包判断就是flags 0。实际部署时SYN 包占比高不一定是攻击也可能是正常的连接建立所以要结合源 IP 的连接频率一起看单看标志位容易误判。2.3 非监听端口占比扫描行为的另一个指纹第三个维度是发往非监听端口的连接尝试占比。正常用户只会访问服务实际监听的端口比如 Web 服务 80/443、SSH 22。而端口扫描器会挨个试一堆端口其中大部分是没开服务的。系统统计某个源 IP 发往非监听端口的连接数占总连接数的比例比例高就说明它在扫端口。这个维度的前提是你得知道本机哪些端口在监听。常见做法是启动时读一次netstat或/proc/net/tcp把监听端口列表缓存起来后续比对。注意这个列表不是一成不变的服务重启可能换端口所以要么定期刷新要么在检测到大量非监听端口访问时先告警而不是直接封给自己留个后悔药。三个维度结合起来判定逻辑大致是频率异常 标志位异常 非监听端口占比高三者命中越多判定为攻击的置信度越高。这种多维度交叉的设计比单阈值靠谱也是这套源码值得看的地方。3. 把系统跑起来环境、抓包权限和 iptables 联动配置看懂逻辑之后下一步是让它真的在你机器上跑起来。这一步的坑主要集中在环境依赖和权限上很多人卡在第一步就放弃了。3.1 依赖安装与 Python 版本选择源码依赖三个核心库scapy 负责抓包和协议解析python-iptables 负责防火墙联动MySQLdb 负责日志存储。Python 版本建议 3.8 及以上太老的版本 scapy 兼容性会有问题。# 建议在虚拟环境里装避免污染系统 Python python3 -m venv ids_env source ids_env/bin/activate # 核心依赖 pip install scapy pip install python-iptables pip install mysqlclient # MySQLdb 的现代替代兼容性更好 # 如果 mysqlclient 编译报错先装系统依赖 # Ubuntu/Debian: sudo apt-get install python3-dev default-libmysqlclient-dev build-essential # CentOS/RHEL: sudo yum install python3-devel mysql-devel gccmysqlclient是 MySQLdb 的维护版本接口基本一致源码里import MySQLdb的写法不用改。如果坚持用原版 MySQLdb在新版 Python 上大概率编译不过这是常见翻车点。scapy 装完不需要额外配置但抓包需要 root 权限或者给 Python 解释器加CAP_NET_RAW能力。3.2 抓包权限为什么必须 root 或加 capabilityscapy 抓包走的是原始套接字普通用户没权限。直接sudo python3 Main.py能跑但不推荐长期这么干因为整个进程都以 root 运行风险大。更稳妥的做法是给 Python 解释器单独加网络抓包能力# 给 Python 解释器加原始套接字能力避免整个脚本跑在 root 下 sudo setcap cap_net_raw,cap_net_admineip $(readlink -f $(which python3)) # 验证是否生效 getcap $(readlink -f $(which python3)) # 应输出类似/usr/bin/python3.8 cap_net_raw,cap_net_admineipcap_net_raw允许发原始包cap_net_admin允许改网络配置iptables 联动需要。加完之后普通用户也能抓包和操作 iptables 规则。注意这个能力是加在解释器上的虚拟环境里的 python 如果是指向系统解释器的软链能力会继承如果是独立编译的得单独加。3.3 iptables 联动封禁规则怎么下发和回收检测到攻击后Flitter 模块调用 python-iptables 往 INPUT 链插一条 DROP 规则把攻击源 IP 封掉。核心操作是插入规则和后续清理。import iptc # python-iptables 的导入名 def block_ip(src_ip, timeout300): 封禁指定源IPtimeout秒后由清理任务解封 rule iptc.Rule() rule.src src_ip rule.target iptc.Target(rule, DROP) chain iptc.Chain(iptc.Table(iptc.Table.FILTER), INPUT) if rule not in chain.rules: # 避免重复插入 chain.insert_rule(rule) return rule def unblock_ip(rule): 解封从链里删除规则 chain iptc.Chain(iptc.Table(iptc.Table.FILTER), INPUT) chain.delete_rule(rule)rule.src指定源 IPtarget设为 DROP 表示丢弃。insert_rule插到链首优先级最高。这里必须做重复检查否则同一个 IP 被检测到多次会插入多条相同规则链越来越长性能下降。timeout参数是封禁时长实际项目里一般配一个后台清理任务到点调unblock_ip解封避免误封正常用户后一直封着。提示iptables 规则是即时生效的调试阶段建议先用iptables -L -n确认规则内容再让它自动下发。误封自己管理 IP 的情况我见过不止一次。3.4 数据库初始化与日志落库Database 模块负责把检测日志写进 MySQL。先建库建表再让程序连。CREATE DATABASE ids_log DEFAULT CHARACTER SET utf8mb4; USE ids_log; CREATE TABLE detection_log ( id BIGINT AUTO_INCREMENT PRIMARY KEY, src_ip VARCHAR(45) NOT NULL, attack_type VARCHAR(32) NOT NULL, -- scan / dos detail TEXT, detected_at DATETIME DEFAULT CURRENT_TIMESTAMP, INDEX idx_src_ip (src_ip), INDEX idx_detected_at (detected_at) );src_ip用 VARCHAR(45) 是为了兼容 IPv6。attack_type区分扫描和 DoS。两个索引分别加速按 IP 查和按时间查日志量大了之后这两个查询最常用。建完表把连接信息填进 Database.py 的配置里注意字符集用 utf8mb4否则中文详情可能乱码。4. 避坑与排查那些让系统「看起来在跑其实没检测」的问题系统能启动不代表能检测下面几条是我实际调试时踩过的按现象、原因、解决来写。4.1 抓不到包程序静默无输出现象程序启动正常日志没报错但攻击打过来毫无反应。原因网卡没进混杂模式或者抓包绑定的网卡选错了。多网卡机器上 scapy 默认可能绑到回环或错误的网卡。解决在 Data_Sniff 里显式指定网卡名sniff(ifaceeth0, prncallback)用ifconfig或ip a确认业务流量走的是哪块网卡。虚拟机环境还要确认网卡是不是桥接模式NAT 模式下很多流量抓不到。4.2 iptables 规则加了但没生效现象日志显示已封禁但攻击流量还在进来。原因规则插到了错误的链或者被前面的 ACCEPT 规则放行了。iptables 是从上往下匹配如果 INPUT 链前面有一条ACCEPT all后面插的 DROP 永远轮不到。解决用iptables -L INPUT -n --line-numbers看规则顺序确保 DROP 规则在放行规则之前。python-iptables 的insert_rule默认插到链首但如果代码里用了append_rule就会排到末尾检查一下用的是哪个。4.3 误报把正常用户封了现象正常访问的用户被判定为扫描IP 被封。原因阈值设太低或者把 CDN、负载均衡的回源 IP 当成了攻击源。这些中间节点会集中发起大量连接频率特征和扫描很像。解决维护一个白名单把网关、CDN 回源 IP、监控系统 IP 加进去检测逻辑里先过白名单再判定。阈值也别拍脑袋先跑基线。封禁时长设短一点比如 300 秒给误封留恢复窗口。4.4 MySQLdb 连接报编码错误现象写日志时报UnicodeEncodeError或中文变问号。原因数据库连接没指定字符集默认可能是 latin1。解决连接时显式指定charsetutf8mb4建库建表也用 utf8mb4。如果已经建了表用ALTER TABLE detection_log CONVERT TO CHARACTER SET utf8mb4;改过来。4.5 长时间运行内存持续上涨现象跑几个小时内存占用越来越高。原因频率检测的 records 字典里不活跃 IP 的时间戳列表没被清理或者 iptables 规则只加不删。解决records 里除了清理窗口内旧记录还要定期清理长时间没活动的 IP 条目封禁规则一定要配解封逻辑别只 insert 不 delete。这两处是内存泄漏的高发点。5. 进阶调优让检测更准的几个实操技巧基础功能跑通之后真正决定这套系统好不好用的是检测精度。分享几个我调优时常用的手法。第一基线要动态。固定阈值在流量波动大的环境里必然误报或漏报。可以按小时统计正常连接频率取过去若干天同一时段的均值加标准差作为动态阈值。比如白天业务高峰阈值高凌晨低谷阈值低这样慢速扫描在凌晨也藏不住。第二标志位检测加时序关联。单看一个 NULL 包可能是网络异常但短时间内同一源 IP 连续发多个 NULL 包基本可以确定是扫描。把标志位异常和频率维度做与运算而不是各自独立判定误报会明显下降。第三封禁策略分级。不要一检测到就永久封。可以设三档轻度异常记录日志观察中度异常封 60 秒重度异常封 600 秒。这样既拦住了攻击又给误判留了余地。下面是一个分级封禁的参考实现def decide_action(freq_hit, flag_hit, port_hit): 根据三个维度的命中情况决定处置等级 score sum([freq_hit, flag_hit, port_hit]) if score 3: return (block, 600) # 三维全中重封 elif score 2: return (block, 60) # 两维命中短封 elif score 1: return (log, 0) # 单维命中只记录 return (pass, 0)score是三个维度布尔值的和命中越多处置越重。这种打分制比硬编码 if-else 灵活后续想加维度比如 UDP 扫描、ICMP 探测直接往 score 里加就行。第四日志要能回溯。检测日志除了记源 IP 和攻击类型最好把触发时的关键特征值也存下来比如窗口内连接数、SYN 包占比、非监听端口访问数。出误报时能直接看是哪个特征超了不用重新抓包复现。表结构里那个detail字段就是干这个的存 JSON 字符串最方便。第五验证检测效果别只用真实攻击。自己写个简单的扫描脚本打一下看系统反应比等真实攻击可控得多。用 scapy 发几个 SYN 包到非监听端口观察日志和 iptables 规则变化确认整条链路通了。# 自测用发几个SYN包模拟端口扫描验证检测是否触发 from scapy.all import IP, TCP, send target 192.168.1.100 for port in range(1, 30): pkt IP(dsttarget) / TCP(dportport, flagsS) send(pkt, verbose0) print(模拟扫描完成检查检测日志和iptables规则)这段自测脚本往目标发 29 个 SYN 包覆盖 1 到 29 端口正常情况应该触发频率和标志位两个维度的检测。跑完去看数据库有没有新日志、iptables -L INPUT -n有没有新 DROP 规则。如果没反应回头查抓包网卡和阈值。从那以后我每次部署这类联动防御系统都强制先跑一遍自测脚本确认检测链路通了再放到真实环境。不然你以为它在守其实它是个黑匣子。希望帮到你。本文还有配套的精品资源点击获取