简介面向高校计算机网络、信息安全专业的毕业设计、课程设计与Python网络编程实践这份资源提供了一套基于Python的TCP入侵检测系统完整源码。系统针对端口扫描与分布式拒绝服务两类典型威胁从TCP连接请求频率、协议头部标志位组合、非监听端口访问占比三个维度建立异常判定机制并通过python-iptables与系统防火墙联动实现自动阻断形成实时检测与防御闭环。实现层面采用scapy完成数据包捕获与协议解析MySQLdb负责日志存储与查询代码模块划分清晰耦合度低适合二次开发。压缩包共10个文件内含5个Python源文件、说明文档、数据库及源码备份整体大小仅10KB轻量易部署。目前已有39人学习下载可作为课程设计答辩参照或中小型网络主动防护的入门参考实现。1. 为什么用Python在TCP层面拦截端口扫描与DoS攻击基于Python的TCP入侵检测系统核心任务是在TCP协议层识别端口扫描与DoS攻击并通过iptables联动完成自动化阻断。源码包里的Analysis.py、Data_Sniff.py、Main.py、Flitter.py和Database.py把抓包、协议解析、特征判定、防火墙操作和日志落库串成一条闭环。对毕业设计而言它展示了一个完整的“流量采集—特征提取—告警—防御”工程框架对实际部署而言它能在不影响应用代码的前提下把高频SYN扫描、异常FIN/NULL探测和半连接攻击源动态隔离。这个方案的独特之处在于不追求机器学习的“黑盒”效果而是围绕TCP三次握手的行为逻辑用三个可解释的指标做判定便于调试和演示。2. TCP异常特征的三层判定SYN频率、标志位组合与暗端口占比2.1 SYN频率从三次握手看扫描与FloodTCP连接建立依赖三次握手客户端发SYN服务端回SYN-ACK客户端再发ACK。正常访问的SYN包频率与业务并发强相关端口扫描和SYN Flood则会在短时间内打破这个基线。扫描器通常对多个端口连续发送SYN而SYN Flood会对同一端口发送海量SYN且不回ACK导致服务端半连接队列被占满。系统需要在滑动时间窗口内统计每个源IP的SYN包数量而不是简单看总包数。实际实现中我一般把时间窗设为30秒每收到一个SYN包就记录当前时间戳并将窗口内过期记录清除。判断逻辑可以简化为某个IP在窗口内SYN数量超过正常基线三倍并且未完成握手的比例高于80%就触发扫描或Flood告警。这里的基线可以通过Database.py历史日志计算比如取过去一小时每分钟平均连接数。对于课程演示也可以直接设一个固定阈值比如5秒内60个SYN。2.2 FIN/NULL/XMAS标志位组合的时序指纹只盯SYN会漏掉很大一部分隐蔽扫描。攻击者会用FIN包、NULL包所有标志位为0或XMAS包FINURGPSH同时置1探测端口状态因为不同系统对非法标志组合的响应方式不同。这类包在正常业务流量里几乎不会出现所以哪怕数量很少也值得重点观察。防御系统应按时间顺序统计同一源IP的非常规标志位包。例如在10秒内出现来自10.0.0.5的FIN包且目标端口依次递增这几乎可以认定是一次TCP FIN扫描。源码包中的Flitter.py从文件名看是过滤器和特征提取器它承担的工作就是把scapy解析出的TCP flags归一化成syn、fin、null、xmas四类并附带端口和时间戳。Analysis.py拿到这些特征行后再结合前一个维度的SYN频率做综合评分而不是单一规则触发。2.3 暗端口非监听端口连接尝试占比第三个维度是计算发往非监听端口的TCP连接请求占比。正常情况下客户端不会频繁访问一个不存在的服务端口除非是被攻击者批量扫描。系统需要维护一张“监听端口白名单”来源可以是本机/proc/net/tcp也可以由Database.py定期同步。凡是目标端口不在白名单内的TCP SYN包都记为暗端口探测。单一暗端口样本不构成威胁因为网络环境里偶尔会有配置错误的客户端。但当同一IP在某个时间窗内访问的非监听端口数量占其总连接数超过30%且尝试端口连续变化时基本可以判定为扫描。这个指标对connect扫描很有效对半开扫描则需要与SYN频率结合因为半开扫描同样不会完整走完三次握手。判定维度正常基线异常特征SYN频率单IP每秒小于10且完整握手占比高单IP每秒超过50半连接占比超过80%FIN/NULL/XMAS极低频单源IP平均每小时小于510秒内产生超过10个连续端口探测暗端口占比小于5%大于30%且目标端口序列连续3. 抓包与检测实现Data_Sniff与Analysis模块的完整流程3.1 使用scapy实现实时流量捕获与TCP层切分scapy是这套系统的基础依赖负责从网卡抓取链路层数据并解析出IP和TCP头。常见做法是用sniff()函数绑定网卡然后通过prn回调处理每个包。源码包Data_Sniff.py的核心逻辑可以简化成下面这段代码from scapy.all import sniff, IP, TCP def handle_tcp_packet(pkt): if not (IP in pkt and TCP in pkt): return src_ip pkt[IP].src dst_port pkt[TCP].dport flags pkt[TCP].flags timestamp float(pkt.time) # 交由Flitter做特征提取 flitter.feed(src_ip, dst_port, flags, timestamp) sniff(prnhandle_tcp_packet, storeFalse, filtertcp, count0)filtertcp让scapy在BPF层直接过滤非TCP报文减少用户态处理压力storeFalse表示不保存原始报文避免长时间运行占用内存count0是无限抓包模式。需要特别注意的是sniff必须用root权限运行并且网卡要支持promiscuous模式。3.2 Analysis.py滑动时间窗口与恶意判断Analysis模块的核心是维护每个源IP在时间窗口内的指标。标准做法是使用字典存储IP对应的时间戳列表并定期清理过期数据。下面是一种典型的滑动窗口实现class TraficWindow: def __init__(self, window_sec5): self.window_sec window_sec self.syn_count {} self.fin_count {} def add_sample(self, src_ip, flags, ts): if flags S: # SYN self.syn_count.setdefault(src_ip, []).append(ts) self.syn_count[src_ip] [ t for t in self.syn_count[src_ip] if ts - t self.window_sec ] if len(self.syn_count[src_ip]) SYN_THRESHOLD: self.trigger_alert(src_ip, syn_flood)这里每次加入新样本时列表推导式会剔除掉超过window_sec的旧样本使len()永远只代表当前窗口内的实时数量。SYN_THRESHOLD是模块顶部定义的全局阈值建议根据场景调整测试环境设30生产环境设到100以上。窗口设为5秒比较灵敏适合课程展示真实部署建议30到60秒避免网站瞬间并发导致误报。3.3 Flitter.py把原始包转成特征向量并落库Flitter.py位于Data_Sniff与Analysis之间作用是协议字段归一化。我通常会在该模块里把服务端收到的包转成扁平结构只保留src_ip、dst_ip、dst_port、flags、timestamp五个字段。这样Analysis模块就不必关心scapy对象细节也方便后续对历史数据做离线回放。聚合后的结果通过Database.py导入MySQL。源码里虽然用的是MySQLdb但Python 3环境下建议替换为PyMySQL建表语句可以参考下面这条SQLCREATE TABLE attack_log ( id INT AUTO_INCREMENT PRIMARY KEY, src_ip VARCHAR(45) NOT NULL, dst_ip VARCHAR(45) NOT NULL, attack_type VARCHAR(20) NOT NULL, tcp_flags VARCHAR(20), dest_port INT, first_seen TIMESTAMP DEFAULT CURRENT_TIMESTAMP, last_seen TIMESTAMP DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP, packet_count INT DEFAULT 1, action VARCHAR(10) DEFAULT alert );attack_type建议枚举syn_scan、syn_flood、fin_scan、null_scanaction字段用于标记后续是否执行了iptables封禁。通过这条表可以完整展示“检测—告警—封禁—审计”的闭环。4. iptables联动防御python-iptables封禁、解封与状态同步4.1 为什么不在抓包层直接丢包抓包属于用户态操作scapy处理大流量时CPU开销很高不适合在用户态强行丢包。更重要的是用户态只能影响后续收到的包无法快速清理内核半连接队列。更可靠的做法是检测到异常后把源IP交给内核态的netfilter去阻断。iptables的INPUT链在协议栈早期处理包DROP规则能把攻击流量挡在用户态之前同时释放SYN Flood产生的半连接资源。这就是“控制平面与数据平面分离”的思路。4.2 python-iptables库动态操作INPUT链python-iptables库封装了iptables的C库libiptc可以在Python进程内直接增删规则。安装命令是pip install python-iptables但必须使用root权限运行。下面是一段封禁示例import iptc def ban_ip(src_ip): table iptc.Table(iptc.Table.FILTER) chain iptc.Chain(table, INPUT) # 检查已有规则 for rule in chain.rules: if rule.src src_ip /32 and rule.target.name DROP: print(rule exists:, src_ip) return rule iptc.Rule() rule.src src_ip /32 rule.protocol tcp rule.target iptc.Target(rule, DROP) chain.insert_rule(rule)这段代码中rule.src必须写成CIDR格式例如203.0.113.7/32rule.protocol tcp能把封禁范围限制在TCP流量避免误伤同一IP发来的其他协议请求。插入前遍历chain.rules是为了防止同一IP的规则重复堆叠因为python-iptables不会自动去重。4.3 集成Main.py事件循环与自动解封Main.py承担系统调度职责。实践中我会用一个queue.Queue接收来自Analysis模块的告警事件然后由独立线程执行封禁和数据库写入。封禁不应是永久的否则一旦误判合法用户会被长时间隔离。合理做法是记录封禁截止时间到达时间后自动解封。下面是一个简化的控制循环def main_loop(ban_duration300): while True: event alert_queue.get() src_ip event[src_ip] action event[attack_type] ban_ip(src_ip) database.write(event, actionban) timer threading.Timer(ban_duration, unban_ip, args(src_ip,)) timer.start()这里的ban_duration单位是秒扫描攻击可以设300秒持续DoS攻击建议至少3600秒。需要注意在关闭iptables服务或重启机器后python-iptables添加的规则不会被持久化需要额外调用iptables-save或维护一份规则起步脚本。封禁粒度规则示例适用场景全IP封禁DROP from IP明确恶意扫描源端口级封禁DROP tcp dport 80 from IP单业务端口被SYN Flood时间窗口解封300s / 3600s 后自动删除防止误杀共享IP5. 部署验证与常见坑从pcap到iptables全链路排错5.1 环境准备部署环境选择CentOS 7或Ubuntu ServerPython版本3.6以上。先安装Python依赖pip install scapy python-iptables PyMySQL如果源码中Database.py直接引用MySQLdb需要在导入前加上pymysql.install_as_MySQLdb()否则会报ModuleNotFoundError。然后创建MySQL数据库和attack_log表确保网络连接权限正确。启动时使用root运行Main.py因为sniff和python-iptables都要求root权限。5.2 用nmap和hping3验证检测与封禁在另一台机器执行nmap -sS -T4 192.0.2.10发起扫描同时观察服务器日志和MySQL表。正常情况下1到2秒内会出现syn_scan告警随后action字段从alert变成baniptables规则新增对应源IP的DROP记录。模拟SYN Flood则用hping3 -S -p 80 --flood 192.0.2.10这时触发的是半连接队列异常告警类型应为syn_flood。如果测试中迟迟没有告警优先检查sniff的filter是否匹配到实际网卡以及SYN_THRESHOLD是否设置得过高。还可以临时把阈值调低到5快速验证全链路。5.3 常见坑第一个坑是scapy抓包权限非root用户连socket都打不开先执行sudo -s再启动Main.py。第二个坑是python-iptables的规则不经过iptables-save持久化重启iptables服务后规则全部丢失需要准备一个恢复脚本。第三个坑是Python 3默认没有MySQLdb必须用PyMySQL兼容。还有一个细节如果分析机同时跑在业务主机上scapy可能读到被iptables丢弃但仍进入raw socket的包这不算错误但也意味着防御规则生效后抓包线程仍在统计被丢弃的流量需要靠时间窗自动衰减。本文还有配套的精品资源点击获取 Python实战：TCP端口扫描与DoS攻击检测及iptables自动封禁
简介面向高校计算机网络、信息安全专业的毕业设计、课程设计与Python网络编程实践这份资源提供了一套基于Python的TCP入侵检测系统完整源码。系统针对端口扫描与分布式拒绝服务两类典型威胁从TCP连接请求频率、协议头部标志位组合、非监听端口访问占比三个维度建立异常判定机制并通过python-iptables与系统防火墙联动实现自动阻断形成实时检测与防御闭环。实现层面采用scapy完成数据包捕获与协议解析MySQLdb负责日志存储与查询代码模块划分清晰耦合度低适合二次开发。压缩包共10个文件内含5个Python源文件、说明文档、数据库及源码备份整体大小仅10KB轻量易部署。目前已有39人学习下载可作为课程设计答辩参照或中小型网络主动防护的入门参考实现。1. 为什么用Python在TCP层面拦截端口扫描与DoS攻击基于Python的TCP入侵检测系统核心任务是在TCP协议层识别端口扫描与DoS攻击并通过iptables联动完成自动化阻断。源码包里的Analysis.py、Data_Sniff.py、Main.py、Flitter.py和Database.py把抓包、协议解析、特征判定、防火墙操作和日志落库串成一条闭环。对毕业设计而言它展示了一个完整的“流量采集—特征提取—告警—防御”工程框架对实际部署而言它能在不影响应用代码的前提下把高频SYN扫描、异常FIN/NULL探测和半连接攻击源动态隔离。这个方案的独特之处在于不追求机器学习的“黑盒”效果而是围绕TCP三次握手的行为逻辑用三个可解释的指标做判定便于调试和演示。2. TCP异常特征的三层判定SYN频率、标志位组合与暗端口占比2.1 SYN频率从三次握手看扫描与FloodTCP连接建立依赖三次握手客户端发SYN服务端回SYN-ACK客户端再发ACK。正常访问的SYN包频率与业务并发强相关端口扫描和SYN Flood则会在短时间内打破这个基线。扫描器通常对多个端口连续发送SYN而SYN Flood会对同一端口发送海量SYN且不回ACK导致服务端半连接队列被占满。系统需要在滑动时间窗口内统计每个源IP的SYN包数量而不是简单看总包数。实际实现中我一般把时间窗设为30秒每收到一个SYN包就记录当前时间戳并将窗口内过期记录清除。判断逻辑可以简化为某个IP在窗口内SYN数量超过正常基线三倍并且未完成握手的比例高于80%就触发扫描或Flood告警。这里的基线可以通过Database.py历史日志计算比如取过去一小时每分钟平均连接数。对于课程演示也可以直接设一个固定阈值比如5秒内60个SYN。2.2 FIN/NULL/XMAS标志位组合的时序指纹只盯SYN会漏掉很大一部分隐蔽扫描。攻击者会用FIN包、NULL包所有标志位为0或XMAS包FINURGPSH同时置1探测端口状态因为不同系统对非法标志组合的响应方式不同。这类包在正常业务流量里几乎不会出现所以哪怕数量很少也值得重点观察。防御系统应按时间顺序统计同一源IP的非常规标志位包。例如在10秒内出现来自10.0.0.5的FIN包且目标端口依次递增这几乎可以认定是一次TCP FIN扫描。源码包中的Flitter.py从文件名看是过滤器和特征提取器它承担的工作就是把scapy解析出的TCP flags归一化成syn、fin、null、xmas四类并附带端口和时间戳。Analysis.py拿到这些特征行后再结合前一个维度的SYN频率做综合评分而不是单一规则触发。2.3 暗端口非监听端口连接尝试占比第三个维度是计算发往非监听端口的TCP连接请求占比。正常情况下客户端不会频繁访问一个不存在的服务端口除非是被攻击者批量扫描。系统需要维护一张“监听端口白名单”来源可以是本机/proc/net/tcp也可以由Database.py定期同步。凡是目标端口不在白名单内的TCP SYN包都记为暗端口探测。单一暗端口样本不构成威胁因为网络环境里偶尔会有配置错误的客户端。但当同一IP在某个时间窗内访问的非监听端口数量占其总连接数超过30%且尝试端口连续变化时基本可以判定为扫描。这个指标对connect扫描很有效对半开扫描则需要与SYN频率结合因为半开扫描同样不会完整走完三次握手。判定维度正常基线异常特征SYN频率单IP每秒小于10且完整握手占比高单IP每秒超过50半连接占比超过80%FIN/NULL/XMAS极低频单源IP平均每小时小于510秒内产生超过10个连续端口探测暗端口占比小于5%大于30%且目标端口序列连续3. 抓包与检测实现Data_Sniff与Analysis模块的完整流程3.1 使用scapy实现实时流量捕获与TCP层切分scapy是这套系统的基础依赖负责从网卡抓取链路层数据并解析出IP和TCP头。常见做法是用sniff()函数绑定网卡然后通过prn回调处理每个包。源码包Data_Sniff.py的核心逻辑可以简化成下面这段代码from scapy.all import sniff, IP, TCP def handle_tcp_packet(pkt): if not (IP in pkt and TCP in pkt): return src_ip pkt[IP].src dst_port pkt[TCP].dport flags pkt[TCP].flags timestamp float(pkt.time) # 交由Flitter做特征提取 flitter.feed(src_ip, dst_port, flags, timestamp) sniff(prnhandle_tcp_packet, storeFalse, filtertcp, count0)filtertcp让scapy在BPF层直接过滤非TCP报文减少用户态处理压力storeFalse表示不保存原始报文避免长时间运行占用内存count0是无限抓包模式。需要特别注意的是sniff必须用root权限运行并且网卡要支持promiscuous模式。3.2 Analysis.py滑动时间窗口与恶意判断Analysis模块的核心是维护每个源IP在时间窗口内的指标。标准做法是使用字典存储IP对应的时间戳列表并定期清理过期数据。下面是一种典型的滑动窗口实现class TraficWindow: def __init__(self, window_sec5): self.window_sec window_sec self.syn_count {} self.fin_count {} def add_sample(self, src_ip, flags, ts): if flags S: # SYN self.syn_count.setdefault(src_ip, []).append(ts) self.syn_count[src_ip] [ t for t in self.syn_count[src_ip] if ts - t self.window_sec ] if len(self.syn_count[src_ip]) SYN_THRESHOLD: self.trigger_alert(src_ip, syn_flood)这里每次加入新样本时列表推导式会剔除掉超过window_sec的旧样本使len()永远只代表当前窗口内的实时数量。SYN_THRESHOLD是模块顶部定义的全局阈值建议根据场景调整测试环境设30生产环境设到100以上。窗口设为5秒比较灵敏适合课程展示真实部署建议30到60秒避免网站瞬间并发导致误报。3.3 Flitter.py把原始包转成特征向量并落库Flitter.py位于Data_Sniff与Analysis之间作用是协议字段归一化。我通常会在该模块里把服务端收到的包转成扁平结构只保留src_ip、dst_ip、dst_port、flags、timestamp五个字段。这样Analysis模块就不必关心scapy对象细节也方便后续对历史数据做离线回放。聚合后的结果通过Database.py导入MySQL。源码里虽然用的是MySQLdb但Python 3环境下建议替换为PyMySQL建表语句可以参考下面这条SQLCREATE TABLE attack_log ( id INT AUTO_INCREMENT PRIMARY KEY, src_ip VARCHAR(45) NOT NULL, dst_ip VARCHAR(45) NOT NULL, attack_type VARCHAR(20) NOT NULL, tcp_flags VARCHAR(20), dest_port INT, first_seen TIMESTAMP DEFAULT CURRENT_TIMESTAMP, last_seen TIMESTAMP DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP, packet_count INT DEFAULT 1, action VARCHAR(10) DEFAULT alert );attack_type建议枚举syn_scan、syn_flood、fin_scan、null_scanaction字段用于标记后续是否执行了iptables封禁。通过这条表可以完整展示“检测—告警—封禁—审计”的闭环。4. iptables联动防御python-iptables封禁、解封与状态同步4.1 为什么不在抓包层直接丢包抓包属于用户态操作scapy处理大流量时CPU开销很高不适合在用户态强行丢包。更重要的是用户态只能影响后续收到的包无法快速清理内核半连接队列。更可靠的做法是检测到异常后把源IP交给内核态的netfilter去阻断。iptables的INPUT链在协议栈早期处理包DROP规则能把攻击流量挡在用户态之前同时释放SYN Flood产生的半连接资源。这就是“控制平面与数据平面分离”的思路。4.2 python-iptables库动态操作INPUT链python-iptables库封装了iptables的C库libiptc可以在Python进程内直接增删规则。安装命令是pip install python-iptables但必须使用root权限运行。下面是一段封禁示例import iptc def ban_ip(src_ip): table iptc.Table(iptc.Table.FILTER) chain iptc.Chain(table, INPUT) # 检查已有规则 for rule in chain.rules: if rule.src src_ip /32 and rule.target.name DROP: print(rule exists:, src_ip) return rule iptc.Rule() rule.src src_ip /32 rule.protocol tcp rule.target iptc.Target(rule, DROP) chain.insert_rule(rule)这段代码中rule.src必须写成CIDR格式例如203.0.113.7/32rule.protocol tcp能把封禁范围限制在TCP流量避免误伤同一IP发来的其他协议请求。插入前遍历chain.rules是为了防止同一IP的规则重复堆叠因为python-iptables不会自动去重。4.3 集成Main.py事件循环与自动解封Main.py承担系统调度职责。实践中我会用一个queue.Queue接收来自Analysis模块的告警事件然后由独立线程执行封禁和数据库写入。封禁不应是永久的否则一旦误判合法用户会被长时间隔离。合理做法是记录封禁截止时间到达时间后自动解封。下面是一个简化的控制循环def main_loop(ban_duration300): while True: event alert_queue.get() src_ip event[src_ip] action event[attack_type] ban_ip(src_ip) database.write(event, actionban) timer threading.Timer(ban_duration, unban_ip, args(src_ip,)) timer.start()这里的ban_duration单位是秒扫描攻击可以设300秒持续DoS攻击建议至少3600秒。需要注意在关闭iptables服务或重启机器后python-iptables添加的规则不会被持久化需要额外调用iptables-save或维护一份规则起步脚本。封禁粒度规则示例适用场景全IP封禁DROP from IP明确恶意扫描源端口级封禁DROP tcp dport 80 from IP单业务端口被SYN Flood时间窗口解封300s / 3600s 后自动删除防止误杀共享IP5. 部署验证与常见坑从pcap到iptables全链路排错5.1 环境准备部署环境选择CentOS 7或Ubuntu ServerPython版本3.6以上。先安装Python依赖pip install scapy python-iptables PyMySQL如果源码中Database.py直接引用MySQLdb需要在导入前加上pymysql.install_as_MySQLdb()否则会报ModuleNotFoundError。然后创建MySQL数据库和attack_log表确保网络连接权限正确。启动时使用root运行Main.py因为sniff和python-iptables都要求root权限。5.2 用nmap和hping3验证检测与封禁在另一台机器执行nmap -sS -T4 192.0.2.10发起扫描同时观察服务器日志和MySQL表。正常情况下1到2秒内会出现syn_scan告警随后action字段从alert变成baniptables规则新增对应源IP的DROP记录。模拟SYN Flood则用hping3 -S -p 80 --flood 192.0.2.10这时触发的是半连接队列异常告警类型应为syn_flood。如果测试中迟迟没有告警优先检查sniff的filter是否匹配到实际网卡以及SYN_THRESHOLD是否设置得过高。还可以临时把阈值调低到5快速验证全链路。5.3 常见坑第一个坑是scapy抓包权限非root用户连socket都打不开先执行sudo -s再启动Main.py。第二个坑是python-iptables的规则不经过iptables-save持久化重启iptables服务后规则全部丢失需要准备一个恢复脚本。第三个坑是Python 3默认没有MySQLdb必须用PyMySQL兼容。还有一个细节如果分析机同时跑在业务主机上scapy可能读到被iptables丢弃但仍进入raw socket的包这不算错误但也意味着防御规则生效后抓包线程仍在统计被丢弃的流量需要靠时间窗自动衰减。本文还有配套的精品资源点击获取