简介：面向高校计算机网络、信息安全专业的毕业设计、课程设计与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频率：从三次握手看扫描与Flood
TCP连接建立依赖三次握手：客户端发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包（FIN+URG+PSH同时置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平均每小时小于5 | 10秒内产生超过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(prn=handle_tcp_packet, store=False, filter="tcp", count=0)filter="tcp"让scapy在BPF层直接过滤非TCP报文，减少用户态处理压力；store=False表示不保存原始报文，避免长时间运行占用内存；count=0是无限抓包模式。需要特别注意的是，sniff必须用root权限运行，并且网卡要支持promiscuous模式。
3.2 Analysis.py：滑动时间窗口与恶意判断
Analysis模块的核心是维护每个源IP在时间窗口内的指标。标准做法是使用字典存储IP对应的时间戳列表，并定期清理过期数据。下面是一种典型的滑动窗口实现：
class TraficWindow: def __init__(self, window_sec=5): 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，建表语句可以参考下面这条SQL：
CREATE 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_scan；action字段用于标记后续是否执行了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/32；rule.protocol = "tcp"能把封禁范围限制在TCP流量，避免误伤同一IP发来的其他协议请求。插入前遍历chain.rules是为了防止同一IP的规则重复堆叠，因为python-iptables不会自动去重。
4.3 集成Main.py事件循环与自动解封
Main.py承担系统调度职责。实践中我会用一个queue.Queue接收来自Analysis模块的告警事件，然后由独立线程执行封禁和数据库写入。封禁不应是永久的，否则一旦误判，合法用户会被长时间隔离。合理做法是记录封禁截止时间，到达时间后自动解封。下面是一个简化的控制循环：
def main_loop(ban_duration=300): while True: event = alert_queue.get() src_ip = event["src_ip"] action = event["attack_type"] ban_ip(src_ip) database.write(event, action="ban") 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 后自动删除 | 防止误杀共享IP |
5. 部署验证与常见坑：从pcap到iptables全链路排错
5.1 环境准备
部署环境选择CentOS 7或Ubuntu Server，Python版本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变成ban，iptables规则新增对应源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的包，这不算错误，但也意味着防御规则生效后抓包线程仍在统计被丢弃的流量，需要靠时间窗自动衰减。
本文还有配套的精品资源，点击获取
