简介这是一份面向网络安全初学者与高校学生的Python TCP入侵检测系统源码可直接用于毕业设计、期末大作业或课程设计。项目聚焦端口扫描与DoS攻击的实时检测并联动iptables实现自动防御帮助读者理解入侵检测与防火墙联动的完整链路。压缩包共6个文件以5个py源码文件为主分别承担流量分析、数据嗅探、过滤处理与数据库记录等职责另附1个md说明文档整体约5KB轻量易读代码含注释新手也能快速上手部署。目前已有111人学习下载。项目经过严格调试功能完善、操作简单读者可据此掌握TCP流量分析、攻击特征识别与iptables规则下发等核心思路并可直接作为高分毕设或大作业提交具有较高的实际应用与参考价值。1. 从一台被打穿的测试机说起TCP 入侵检测到底在防什么去年帮朋友看一台暴露在公网的测试服务器日志里全是同一个 /24 段地址在两分钟内扫了 65535 个端口紧接着就是一波 SYN 洪水服务直接卡死。事后复盘发现机器上没有任何东西在盯 TCP 层的异常行为防火墙规则也是静态的攻击者换一批源 IP 就绕过去了。这件事让我下决心用 Python 写一套轻量的 TCP 入侵检测系统核心就干两件事识别端口扫描和 DoS 攻击特征然后联动 iptables 把恶意源 IP 封掉。这套东西适合谁手里有 Linux 服务器、想搞明白入侵检测底层逻辑、又不愿意上重型商业方案的运维和开发。它不依赖任何第三方 IDS 引擎纯 Python 抓包加规则判断代码量可控你能完整看懂每一条判断是怎么来的。下面从抓包原理讲到 iptables 联动再到参数调优和踩坑全部是可复现的操作。2. 抓包与特征提取Python 怎么在 TCP 层看清一次扫描2.1 为什么选原始套接字而不是现成 IDS做 TCP 入侵检测第一步是拿到数据包。常见做法有三种libpcap 绑定、原始套接字raw socket、以及 Scapy 这类高层库。我最终选了原始套接字加手工解析 IP/TCP 头原因是可控。Snort、Suricata 这类引擎功能强但规则语法和内部状态机对想改逻辑的人来说是个黑匣子你想加一条「同一源 IP 在 3 秒内 SYN 包超过 50 个就告警」的规则得先啃它的规则文档。原始套接字虽然要自己拆字节但每个字段在哪、怎么判断全在你自己手里。原始套接字在 Linux 上需要 root 权限因为要绕过内核的 TCP 协议栈直接收链路层帧。绑定的时候用socket.AF_PACKET配合socket.SOCK_RAW这样拿到的是完整的以太网帧包含 IP 头和 TCP 头。如果你只想抓 IP 层以上也可以用AF_INETSOCK_RAWIPPROTO_TCP但那样拿不到以太网头做 MAC 层分析就不够了。2.2 手工解析 IP 头和 TCP 头的最小代码下面这段是抓包和解析的核心跑起来就能在终端看到每个 TCP 包的源 IP、目的端口和标志位。import socket import struct # 绑定到 eth0 网卡抓所有以太网帧 sniffer socket.socket(socket.AF_PACKET, socket.SOCK_RAW, socket.ntohs(3)) def parse_ethernet_frame(data): 解析以太网帧返回 (eth_proto, payload) eth_header data[:14] eth_proto struct.unpack(!H, eth_header[12:14])[0] return eth_proto, data[14:] def parse_ip_header(data): 解析 IP 头返回 (协议号, 源IP, 目的IP, 载荷) iph struct.unpack(!BBHHHBBH4s4s, data[:20]) version_ihl iph[0] ihl (version_ihl 0xF) * 4 # 头长度单位字节 protocol iph[6] # 6 表示 TCP src_ip socket.inet_ntoa(iph[8]) dst_ip socket.inet_ntoa(iph[9]) return protocol, src_ip, dst_ip, data[ihl:] def parse_tcp_header(data): 解析 TCP 头返回 (源端口, 目的端口, 标志位) tcph struct.unpack(!HHLLBBHHH, data[:20]) src_port tcph[0] dst_port tcph[1] flags tcph[5] # 低 6 位是 SYN/FIN/RST 等 return src_port, dst_port, flags while True: raw_data, _ sniffer.recvfrom(65535) eth_proto, payload parse_ethernet_frame(raw_data) if eth_proto ! 0x0800: # 只处理 IPv4 continue protocol, src_ip, dst_ip, ip_payload parse_ip_header(payload) if protocol ! 6: # 只处理 TCP continue src_port, dst_port, flags parse_tcp_header(ip_payload) syn (flags 0x02) 1 # SYN 标志位 print(f{src_ip}:{src_port} - {dst_ip}:{dst_port} SYN{syn})逻辑说明parse_ethernet_frame取前 14 字节第 12 到 14 字节是以太网类型0x0800代表 IPv4。parse_ip_header用!BBHHHBBH4s4s格式拆 20 字节固定头其中ihl字段低 4 位乘以 4 才是真实头长度因为 IP 头可能有选项字段。parse_tcp_header拆 20 字节 TCP 头flags字段的第 1 位从 0 数是 SYN第 4 位是 RST。参数说明socket.ntohs(3)里的 3 是ETH_P_ALL表示抓所有协议。recvfrom(65535)的缓冲区大小设成 65535 是为了容纳最大以太网帧。如果你只想抓特定网卡把AF_PACKET换成AF_INET并去掉以太网解析即可但那样会丢掉 MAC 信息。2.3 端口扫描的特征怎么量化端口扫描的本质是短时间内对大量端口发起连接尝试。最常见的两种SYN 扫描只发 SYN 不回 ACK半开连接全连接扫描会完成三次握手再 RST。检测逻辑我用了滑动窗口计数维护一个字典key 是源 IPvalue 是一个时间戳列表每次收到该源的 SYN 包就追加时间戳然后清理掉窗口外的旧记录。如果窗口内 SYN 包数量超过阈值就判定为扫描。窗口大小和阈值是这套系统的命门。我实测下来3 秒窗口、50 个 SYN 包是个比较稳的起点。低于这个数容易漏报慢速扫描高于这个数会把正常的并发连接误判。慢速扫描比如每秒 1 个端口扫 10 分钟用固定窗口抓不住得配合长周期统计这个放到后面进阶部分讲。3. DoS 攻击识别SYN 洪水的计数逻辑与阈值设定3.1 SYN 洪水为什么能打穿服务TCP 三次握手的第一步是客户端发 SYN服务端回 SYN-ACK 并把连接放进半连接队列等客户端回 ACK 才建立完整连接。SYN 洪水的做法是伪造大量源 IP 发 SYN服务端回了 SYN-ACK 但永远等不到 ACK半连接队列被占满正常用户连不进来。检测的关键指标就是单位时间内来自同一源或发往同一目的端口的 SYN 包数量以及 SYN 与 ACK 的比例。我用的判断逻辑分两层第一层按源 IP 统计 SYN 速率超过阈值就标记第二层按目的端口统计如果某个端口在短时间内收到大量 SYN 但完成的连接很少也标记。两层都命中才触发防御这样能降低误杀。3.2 滑动窗口计数器的实现下面这段是 DoS 检测的核心计数器用collections.deque做滑动窗口比列表 pop(0) 高效。from collections import defaultdict, deque import time class SynFloodDetector: def __init__(self, window_sec3, syn_threshold100, ack_ratio0.1): self.window window_sec self.threshold syn_threshold self.ack_ratio ack_ratio # 每个源 IP 的 SYN 时间戳队列 self.syn_records defaultdict(deque) # 每个源 IP 的 ACK 计数 self.ack_count defaultdict(int) def add_syn(self, src_ip): now time.time() q self.syn_records[src_ip] q.append(now) # 清理窗口外的记录 while q and now - q[0] self.window: q.popleft() return len(q) def add_ack(self, src_ip): self.ack_count[src_ip] 1 def is_attack(self, src_ip): syn_num len(self.syn_records[src_ip]) ack_num self.ack_count.get(src_ip, 0) if syn_num self.threshold: return False # SYN 远多于 ACK判定为洪水 if ack_num 0 or syn_num / max(ack_num, 1) 1 / self.ack_ratio: return True return False逻辑说明add_syn每次收到 SYN 就追加时间戳并清理过期记录返回当前窗口内的 SYN 数。is_attack先看 SYN 数是否过阈值再看 SYN/ACK 比例。如果 ACK 数为 0说明全是半开连接直接判定攻击。参数说明window_sec3是滑动窗口秒数syn_threshold100是窗口内 SYN 上限ack_ratio0.1表示正常流量里 ACK 至少应该是 SYN 的 10%。这三个参数要根据你的业务调整如果是高并发 API 网关阈值可以放到 500如果是内部管理后台50 就够。调参的时候先用tcpdump抓一段正常流量跑一遍看实际 SYN 速率落在哪个区间再定阈值。3.3 和端口扫描检测的联动端口扫描和 SYN 洪水在特征上有重叠都是大量 SYN。区别在于扫描是分散到不同目的端口洪水是集中到一个端口。我在代码里把两个检测器串起来先过端口扫描检测如果某个源 IP 在窗口内访问的不同目的端口数超过 20 个标记为扫描再过 SYN 洪水检测如果同一源对同一目的端口的 SYN 数超过阈值标记为洪水。两个标记都进同一个封禁队列交给 iptables 处理。4. 联动 iptables从检测到封禁的完整链路4.1 为什么用 iptables 而不是应用层拦截检测到攻击后最直接的做法是在 Python 进程里丢弃后续包。但 Python 处理速度有限攻击流量一大进程本身就成了瓶颈。iptables 工作在内核态封禁一个 IP 后后续包在进入协议栈之前就被 DROP 掉不消耗应用层资源。所以正确姿势是Python 只负责检测和下发规则实际拦截交给 iptables。4.2 封禁与解封的命令封装下面这段封装了 iptables 的增删规则用subprocess调用系统命令加了幂等检查避免重复添加。import subprocess import time class IPTablesManager: def __init__(self, chainINPUT, timeout300): self.chain chain self.timeout timeout self.banned {} # ip - 封禁时间戳 def _run(self, cmd): return subprocess.run(cmd, shellTrue, capture_outputTrue, textTrue) def ban(self, ip): if ip in self.banned: return False # 先检查规则是否已存在 check self._run(fiptables -C {self.chain} -s {ip} -j DROP) if check.returncode ! 0: self._run(fiptables -I {self.chain} -s {ip} -j DROP) self.banned[ip] time.time() return True def unban(self, ip): self._run(fiptables -D {self.chain} -s {ip} -j DROP) self.banned.pop(ip, None) def cleanup(self): 清理超时的封禁 now time.time() for ip, ts in list(self.banned.items()): if now - ts self.timeout: self.unban(ip)逻辑说明ban先用iptables -C检查规则是否存在不存在才用-I插入到链首保证优先级最高。unban用-D删除规则。cleanup遍历封禁字典超过timeout秒的自动解封。参数说明chainINPUT表示封禁入站流量如果你要封禁转发流量改成FORWARD。timeout300是封禁时长默认 5 分钟。这个值不能太短否则攻击者换个节奏又能打进来也不能太长误封正常用户会影响业务。我一般设 300 到 600 秒配合白名单使用。注意iptables 规则是即时生效的但如果你在容器里跑容器重启后规则会丢失。生产环境建议把封禁规则持久化或者用 ipset 配合 iptables 做批量管理。4.3 白名单和防误杀误封是这套系统最大的风险。我踩过的坑一次内部压测测试机短时间内发了大量 SYN直接被封了导致压测中断。后来加了白名单机制把内网段和已知的监控 IP 排除在检测之外。白名单在检测层就过滤掉根本不进计数器比在 iptables 层放行更彻底。白名单的实现很简单维护一个 CIDR 列表用ipaddress模块判断源 IP 是否在列表内。import ipaddress WHITELIST [ ipaddress.ip_network(10.0.0.0/8), ipaddress.ip_network(192.168.0.0/16), ipaddress.ip_network(172.16.0.0/12), ] def is_whitelisted(ip): addr ipaddress.ip_address(ip) return any(addr in net for net in WHITELIST)逻辑说明ipaddress.ip_network把字符串转成网络对象addr in net判断 IP 是否属于该网段。any只要命中一个就返回 True。参数说明WHITELIST里放的是内网三段保留地址实际部署时把你自己的办公网段、监控服务器 IP、负载均衡器 IP 都加进去。判断放在抓包解析之后、进检测器之前这样白名单流量完全不参与计数性能也更好。5. 避坑与排查这套系统最容易翻车的五个地方5.1 抓不到包程序静默无输出现象程序跑起来没有任何打印tcpdump却能看到流量。原因原始套接字绑定网卡时用了AF_PACKET但没指定网卡名或者绑到了lo回环口。解决把socket.socket(AF_PACKET, SOCK_RAW, ntohs(3))改成socket.socket(AF_PACKET, SOCK_RAW, ntohs(3))后调用bind((eth0, 0))网卡名用ip link确认。另外确认程序以 root 运行非 root 用户绑定原始套接字会直接抛权限错误。5.2 阈值设太低正常业务被误封现象业务高峰期大量正常用户被 iptables 封禁客服电话被打爆。原因syn_threshold设成了 50而正常业务峰值 SYN 速率就有 80。解决先用tcpdump -i eth0 -w normal.pcap抓 10 分钟正常流量用tshark -r normal.pcap -Y tcp.flags.syn1 | wc -l统计 SYN 总数除以时间得到平均速率阈值设成平均值的 3 到 5 倍。同时把白名单配全内网和监控流量不参与检测。5.3 iptables 规则重复添加链越来越长现象iptables -L INPUT -n | wc -l发现规则数持续增长封禁过的 IP 解封后规则还在。原因ban方法没有做幂等检查或者unban时 IP 已经被清理出banned字典导致删不掉。解决ban之前必须用iptables -C检查unban时不管字典里有没有都执行-D并且-D失败不抛异常。更稳的做法是用 ipset把所有封禁 IP 放进一个集合iptables 只引用集合增删都在集合层面操作链长度恒定。5.4 滑动窗口内存泄漏跑几天就 OOM现象程序运行 48 小时后内存占用从 50MB 涨到 2GB。原因syn_records字典里每个源 IP 的 deque 没有及时清理攻击者用大量伪造源 IP 发 SYN每个 IP 都建一个 deque内存被撑爆。解决加一个全局清理任务每 60 秒遍历syn_records把 deque 为空的 key 删掉。同时限制字典最大 key 数量超过 10 万个就触发告警并清理最旧的记录。5.5 封禁后攻击者换 IP 继续打现象封了一个 IP攻击流量降了几秒又恢复源 IP 全变了。原因攻击者用代理池或僵尸网络单 IP 封禁只能挡住一部分。解决把封禁粒度从单 IP 扩展到 /24 网段用iptables -I INPUT -s 1.2.3.0/24 -j DROP。同时提高检测灵敏度对同一 /24 段内多个 IP 的 SYN 做聚合统计超过阈值就封整个段。但要注意封 /24 误杀风险更大必须配合白名单和人工确认。6. 进阶技巧用 ipset 做批量封禁和长周期慢速扫描检测6.1 ipset 替代逐条 iptables 规则前面提过逐条 iptables 规则在封禁量大时链会变长匹配效率下降。ipset 是内核态的 IP 集合查询是哈希表O(1) 复杂度。用法分三步创建集合、iptables 引用集合、往集合里增删 IP。# 创建一个名为 blacklist 的集合类型 hash:ip超时 600 秒 ipset create blacklist hash:ip timeout 600 # iptables 引用集合匹配到集合内的 IP 直接 DROP iptables -I INPUT -m set --match-set blacklist src -j DROP # 添加和删除 IP ipset add blacklist 1.2.3.4 ipset del blacklist 1.2.3.4 # 查看集合内容 ipset list blacklist逻辑说明hash:ip表示集合存的是单个 IPtimeout 600让每个条目 600 秒后自动过期省去了手动解封的代码。iptables 的-m set --match-set blacklist src表示匹配源 IP 在集合内。参数说明timeout可以按条目单独设置ipset add blacklist 1.2.3.4 timeout 300覆盖默认值。集合类型除了hash:ip还有hash:net存网段hash:ip,port存 IP 加端口组合。生产环境建议用hash:net既能封单 IP 也能封网段。Python 侧只需要把IPTablesManager里的iptables -I换成ipset add-D换成ipset del代码更简洁而且不用担心规则重复。6.2 长周期慢速扫描怎么抓固定 3 秒窗口抓不住慢速扫描。我的做法是加一个长周期统计层用 Redis 的 Sorted Set 记录每个源 IP 访问过的目的端口和时间戳score 用时间戳member 用端口号。每 5 分钟统计一次如果某个源 IP 在 10 分钟内访问的不同端口数超过 100 个判定为慢速扫描。import redis import time r redis.Redis(hostlocalhost, port6379, db0) def record_port(src_ip, dst_port): key fscan:{src_ip} now time.time() # member 用端口号score 用时间戳 r.zadd(key, {dst_port: now}) # 清理 10 分钟前的记录 r.zremrangebyscore(key, 0, now - 600) # 设置 key 过期时间防止内存泄漏 r.expire(key, 600) def check_slow_scan(src_ip, threshold100): key fscan:{src_ip} port_count r.zcard(key) return port_count threshold逻辑说明zadd把端口和时间戳写入 Sorted Set同一端口重复访问会更新 score。zremrangebyscore清理 10 分钟前的记录保证统计窗口滑动。zcard返回集合大小即不同端口数。参数说明threshold100是 10 分钟内不同端口数的上限正常用户访问的端口数一般不超过 20 个。expire设 600 秒和清理窗口一致防止冷 IP 的 key 长期占用内存。Redis 挂了怎么办加一个本地降级Redis 连接失败时退回内存字典虽然重启会丢数据但至少不影响主检测流程。6.3 验证检测效果的一个笨办法写完检测逻辑怎么知道它真的有效我的习惯是搭一个隔离环境用nmap从另一台机器发起扫描看系统能不能在 3 秒内标记并封禁。命令是nmap -sS -p 1-1000 目标IP-sS是 SYN 扫描。然后在本机iptables -L INPUT -n看规则有没有加上ipset list blacklist看 IP 有没有进集合。DoS 用hping3 -S --flood -p 80 目标IP模拟观察封禁是否触发。这个笨办法能覆盖 90% 的逻辑错误比看日志猜靠谱得多。我自己的习惯是每次改完阈值或窗口参数都跑一遍这两个测试确认检测和封禁链路没断。这套系统不复杂但参数和边界条件很多上线前多测几轮比出事后再补救省心得多。希望帮到你。本文还有配套的精品资源点击获取