基于Python的TCP入侵检测系统：抓包解析、扫描与DoS判定及iptables联动

简介这是一份面向网络安全初学者与高校学生的Python TCP入侵检测系统源码可直接用于毕业设计、期末大作业或课程设计。项目聚焦端口扫描与Dos攻击的实时检测并联动iptables实现自动防御帮助读者理解入侵检测与防火墙联动的完整链路。压缩包共6个文件以5个py源码文件为主分别承担流量分析、数据嗅探、过滤处理与数据库记录等职责另附1个md说明文档整体约5KB结构精简、注释清晰新手也能快速读懂。目前已有111人学习下载。代码经过严格调试部署简单下载后即可运行可作为高分项目参考也能在此基础上扩展检测规则或防御策略具有较高的实际应用与二次开发价值。1. 从一台被扫穿的测试机说起Python 写 TCP 入侵检测到底在做什么去年我把一台只开了 22 和 80 端口的测试机丢进公网两天后翻/var/log/iptables.log发现有人在 40 秒内把 1 到 65535 全扫了一遍紧接着就是一波 SYN 洪水。那台机器没挂因为我提前挂了一个自己用 Python 写的 TCP 入侵检测脚本它盯着网卡上的 TCP 包识别端口扫描和 DoS 特征命中规则就调 iptables 把源 IP 丢进黑名单。这套东西就是标题里说的「基于 Python 实现的 TCP 入侵检测系统」核心链路只有三段——抓包解析 TCP 头、按规则判定扫描与 DoS、联动 iptables 做防御。它适合谁适合手上有 Linux 测试机、会一点 Python、想搞明白「检测到攻击之后到底怎么自动封」的运维和开发。不适合直接上生产核心链路因为纯 Python 抓包在高流量下会丢包这是它的边界。下面我把这套方案从环境、抓包、判定到 iptables 联动拆开讲参数和坑都给你标出来照着能跑通一个最小可用版本。2. 抓包与 TCP 头解析Python 侧怎么拿到原始包2.1 为什么选原始套接字而不是现成 IDS做 TCP 层检测第一件事是决定包从哪来。常见做法有三种libpcap 绑定scapy、pcapy、原始套接字socket.AF_PACKET或AF_INETSOCK_RAW、以及直接读/proc/net或 conntrack。我一般选原始套接字理由是依赖少、可控、能直接拿到 IP 头和 TCP 头做字段级判断不用为了一个 SYN 标志位去装一整套 Suricata。scapy 写起来最舒服但它在高包速率下性能掉得厉害而且解析每个包都构造对象内存和 CPU 都吃不消。原始套接字拿到的是 bytes自己按偏移切字段快得多代价是你要手动处理字节序和 IP 头长度。标题里说的是「TCP 入侵检测」检测对象是 TCP 标志位和端口行为原始套接字完全够用。先确认环境。Linux 上需要 root 或CAP_NET_RAW能力Python 3.8 以上都行不需要额外 pip 包这是这套方案的一个优点纯标准库就能跑。# 确认 Python 版本3.8 即可 python3 --version # 确认有抓包权限普通用户会报 Permission denied sudo python3 -c import socket; ssocket.socket(socket.AF_PACKET, socket.SOCK_RAW, socket.ntohs(3)); print(raw socket ok)AF_PACKET是 Linux 特有的能拿到链路层帧包含以太网头。socket.ntohs(3)里的 3 是ETH_P_ALL表示抓所有协议。如果你只想抓 IP 层可以用AF_INETSOCK_RAWIPPROTO_TCP少解析一层以太网头但拿不到 MAC做溯源时不如AF_PACKET方便。2.2 手工解析 IP 头与 TCP 头拿到帧之后要剥两层以太网头 14 字节然后是 IP 头可变长看 IHL 字段再是 TCP 头 20 字节起。下面这段是核心解析函数我把它单独抽出来方便你替换成别的判定逻辑。import socket import struct def parse_tcp_packet(data): 从以太网帧里解析出 IP 和 TCP 关键字段返回 dict 或 None # 以太网头固定 14 字节类型字段在 12:14 eth_type struct.unpack(!H, data[12:14])[0] if eth_type ! 0x0800: # 0x0800 IPv4其他直接丢 return None ip_header data[14:34] # IP 头前 20 字节版本IHL 在第一个字节 ver_ihl ip_header[0] ihl (ver_ihl 0x0F) * 4 # IHL 单位是 4 字节算出真实头长 protocol ip_header[9] # 6 TCP if protocol ! 6: return None src_ip socket.inet_ntoa(ip_header[12:16]) dst_ip socket.inet_ntoa(ip_header[16:20]) tcp_start 14 ihl # TCP 头起点随 IP 选项浮动 tcp_header data[tcp_start:tcp_start 20] src_port, dst_port struct.unpack(!HH, tcp_header[0:4]) flags tcp_header[13] # 低 6 位是 URG/ACK/PSH/RST/SYN/FIN return { src_ip: src_ip, dst_ip: dst_ip, src_port: src_port, dst_port: dst_port, syn: (flags 0x02) ! 0, ack: (flags 0x10) ! 0, rst: (flags 0x04) ! 0, fin: (flags 0x01) ! 0, }逻辑说明ihl必须动态算因为 IP 头可能带选项写死 20 字节在带选项的包上会错位这是新手最常见的翻车点。flags取 TCP 头第 13 字节0x02是 SYN0x10是 ACK0x04是 RST。判定扫描和 DoS 全靠这几个位。参数说明data[12:14]是以太网类型0x0800是 IPv40x86DD是 IPv6本方案只处理 IPv4IPv6 直接返回 None。tcp_start用14 ihl而不是固定 34就是为了兼容 IP 选项。如果你在AF_INET模式下抓包没有以太网头ip_header要从data[0:20]开始tcp_start改成ihl这点切换时容易忘。2.3 主循环与包速率控制抓包主循环要处理两件事持续收包、控制处理节奏。原始套接字recvfrom是阻塞的高流量下会积压所以我会加一个简单的计数窗口而不是每个包都做重判定。def sniff_loop(iface_packets, handler): iface_packets 是已绑定的原始套接字handler 是判定回调 count 0 while True: data, _ iface_packets.recvfrom(65535) pkt parse_tcp_packet(data) if pkt is None: continue count 1 handler(pkt) # 每 10000 个包打印一次心跳方便观察是否丢包 if count % 10000 0: print(f[sniff] processed {count} tcp packets)recvfrom(65535)的缓冲区给到最大避免大包被截断。count心跳是排查「脚本是不是卡死」最直接的手段。注意这里没有做多线程判定逻辑如果重会拖慢收包后面第 4 章讲 iptables 联动时会说怎么把封禁动作异步化。3. 端口扫描与 DoS 的判定规则阈值怎么定才不误杀3.1 端口扫描的两种典型特征端口扫描在 TCP 层有两种最常见形态SYN 扫描半开扫描和全连接扫描。SYN 扫描只发 SYN收到 SYN-ACK 后回 RST不完成三次握手全连接扫描会走完握手再断开。检测上我盯两个维度单位时间内同一源 IP 访问的不同目的端口数以及 SYN 包中「只发 SYN 不回 ACK」的比例。判定端口扫描的核心是滑动窗口计数。下面用一个字典维护每个源 IP 的端口集合和时间戳。import time from collections import defaultdict # 每个源 IP 记录访问过的端口集合 窗口起始时间 scan_tracker defaultdict(lambda: {ports: set(), start: time.time()}) SCAN_WINDOW 10 # 统计窗口秒 SCAN_PORT_THRESHOLD 20 # 窗口内不同端口数超过这个值判定为扫描 def detect_port_scan(pkt): if not pkt[syn] or pkt[ack]: return False # 只看纯 SYN排除 SYN-ACK src pkt[src_ip] rec scan_tracker[src] now time.time() if now - rec[start] SCAN_WINDOW: rec[ports].clear() rec[start] now rec[ports].add(pkt[dst_port]) return len(rec[ports]) SCAN_PORT_THRESHOLD逻辑说明SCAN_WINDOW和SCAN_PORT_THRESHOLD是这套检测最需要调的两个参数。窗口太短会漏掉慢速扫描太长会误杀正常的多连接客户端。SCAN_PORT_THRESHOLD设 20 是我在测试机上的经验值正常用户 10 秒内很少访问超过 20 个不同端口而 nmap 默认扫描一秒就能打几百个端口。参数说明pkt[syn] and not pkt[ack]是纯 SYN 的判据这一步能过滤掉大量正常握手包。rec[ports]用 set 去重同一个端口反复 SYN 只算一次避免重传导致误判。慢速扫描比如每秒 1 个端口、持续几分钟用固定窗口抓不到需要改成基于时间衰减的计数器这是进阶做法第 6 章会提。3.2 DoS 攻击的判定SYN 洪水与连接频率DoS 在 TCP 层最典型的是 SYN 洪水大量源 IP或伪造源 IP向同一目的端口发 SYN耗尽半连接队列。检测维度是单位时间内同一目的 IP端口的 SYN 数量以及同一源 IP 的 SYN 速率。syn_counter defaultdict(list) # key: (dst_ip, dst_port) - [时间戳] SYN_RATE_WINDOW 5 # 秒 SYN_RATE_THRESHOLD 200 # 窗口内 SYN 数阈值 def detect_syn_flood(pkt): if not pkt[syn] or pkt[ack]: return False key (pkt[dst_ip], pkt[dst_port]) now time.time() ts syn_counter[key] ts.append(now) # 清掉窗口外的时间戳保持列表不无限增长 syn_counter[key] [t for t in ts if now - t SYN_RATE_WINDOW] return len(syn_counter[key]) SYN_RATE_THRESHOLD逻辑说明这里按「目的 IP端口」聚合因为 SYN 洪水的目标是打垮某个服务的半连接队列源 IP 可能成千上万按源聚合反而看不出攻击。SYN_RATE_THRESHOLD设 200 是相对保守的值真实业务要按自己服务的正常 QPS 往上调否则促销活动会被自己封掉。参数说明syn_counter[key]用列表存时间戳每次清理窗口外的内存占用和窗口内包数成正比。如果攻击流量极大这个列表会膨胀生产上应该换成环形缓冲或计数器这里为了可读性用列表。3.3 把两类判定接进主循环判定函数写好后接进第 2 章的handler里命中就触发封禁动作第 4 章实现。def handler(pkt): if detect_port_scan(pkt): print(f[ALERT] port scan from {pkt[src_ip]}) block_ip(pkt[src_ip]) # 第 4 章实现 if detect_syn_flood(pkt): print(f[ALERT] syn flood to {pkt[dst_ip]}:{pkt[dst_port]}) block_ip(pkt[src_ip])逻辑说明两个判定独立触发同一个包可能同时命中扫描器往往也带高频 SYNblock_ip要做幂等重复封同一个 IP 不能报错也不能重复插规则。参数说明block_ip的入参是源 IPSYN 洪水场景下如果源 IP 是伪造的封了也没用这点在第 5 章避坑里展开。4. 联动 iptables 做防御从检测到封禁的完整链路4.1 用独立链管理封禁规则直接往INPUT链里插规则规则一多就乱而且不好清理。我一般建一个自定义链TCP_IDS所有封禁规则进这个链INPUT只引用一次。# 建链已存在会报错用 -N 前先判断 iptables -N TCP_IDS 2/dev/null || true # 让 INPUT 的流量先过 TCP_IDS 链 iptables -C INPUT -j TCP_IDS 2/dev/null || iptables -I INPUT 1 -j TCP_IDS # 查看当前封禁列表 iptables -L TCP_IDS -n --line-numbers逻辑说明-C是检查规则是否存在配合||保证脚本重复执行不会插重复规则这是幂等的关键。-I INPUT 1插到第一条保证封禁优先于其他放行规则。参数说明-N建链-I插入-C检查这三个是脚本里最常用的幂等组合。4.2 Python 调 iptables 的两种方式Python 里调 iptables 有两条路subprocess调命令行或者python-iptables库直接操作。我选subprocess因为不引入额外依赖而且命令行行为可预期、好调试。import subprocess import time blocked {} # ip - 封禁时间戳用于自动解封 BLOCK_TTL 600 # 封禁时长秒 def block_ip(ip): now time.time() if ip in blocked and now - blocked[ip] BLOCK_TTL: return # 已在封禁期内幂等返回 # 先删旧规则再插避免重复 subprocess.run([iptables, -D, TCP_IDS, -s, ip, -j, DROP], stderrsubprocess.DEVNULL) subprocess.run([iptables, -I, TCP_IDS, 1, -s, ip, -j, DROP], checkTrue) blocked[ip] now print(f[BLOCK] {ip} dropped for {BLOCK_TTL}s)逻辑说明-D先删再-I插保证同一个 IP 只有一条规则。stderrsubprocess.DEVNULL是因为删不存在的规则会报错这里故意忽略。checkTrue只在插入时用插入失败要抛异常不能静默。参数说明BLOCK_TTL是封禁时长设 600 秒是折中太短攻击者换个时间又来太长可能误封正常用户且不好恢复。4.3 自动解封与规则清理只封不解TCP_IDS链会越积越长而且误封的正常 IP 永远出不来。需要一个后台线程定期清理过期封禁。import threading def unblock_worker(): while True: now time.time() for ip, ts in list(blocked.items()): if now - ts BLOCK_TTL: subprocess.run([iptables, -D, TCP_IDS, -s, ip, -j, DROP], stderrsubprocess.DEVNULL) del blocked[ip] print(f[UNBLOCK] {ip} released) time.sleep(30) # 30 秒扫一次够用且不费 CPU # 启动时挂后台线程 threading.Thread(targetunblock_worker, daemonTrue).start()逻辑说明daemonTrue保证主程序退出时线程跟着退不会挂住。list(blocked.items())先转列表再遍历避免遍历中改字典报错。参数说明time.sleep(30)是清理周期封禁时长 600 秒、清理周期 30 秒最坏情况下 IP 多被封 30 秒可以接受。4.4 完整启动脚本把前面几块拼起来就是一个能跑的最小系统。import socket def main(): # 绑定所有接口的原始套接字 sock socket.socket(socket.AF_PACKET, socket.SOCK_RAW, socket.ntohs(3)) threading.Thread(targetunblock_worker, daemonTrue).start() print([start] tcp ids running) sniff_loop(sock, handler) if __name__ __main__: main()逻辑说明AF_PACKET不指定接口时抓所有网卡指定接口用sock.bind((eth0, 0))。参数说明生产上建议绑定具体业务网卡避免抓到无关流量浪费 CPU。启动前确认iptables -L TCP_IDS链已建好否则block_ip会失败。5. 避坑与排查这套方案最容易翻车的五个地方5.1 抓包权限不足脚本静默退出现象脚本启动就报PermissionError: [Errno 1] Operation not permitted或者用 systemd 拉起时直接 failed。原因AF_PACKET原始套接字需要 root 或CAP_NET_RAW。解决用sudo跑或者给 Python 解释器加能力setcap cap_net_raweip $(which python3)。注意加了能力之后普通用户也能抓包安全上要评估。5.2 IP 头长度写死 20 字节导致解析错位现象某些包解析出来的端口是乱码或者src_ip明显不对。原因IP 头带选项时 IHL 大于 5真实头长超过 20 字节写死偏移就错位。解决按第 2 章的ihl (ver_ihl 0x0F) * 4动态算tcp_start 14 ihl。这个坑在只测普通包时发现不了一上真实流量就暴露。5.3 阈值太激进把正常用户封了现象业务方反馈某个客户突然访问不了查TCP_IDS链发现 IP 在里面。原因SCAN_PORT_THRESHOLD或SYN_RATE_THRESHOLD设太低正常的多连接客户端或健康检查被误判。解决先跑观察模式只打印告警不封禁统计一周的正常峰值阈值设在峰值之上 2 到 3 倍。别一上来就开自动封禁。5.4 封了伪造源 IP规则链被撑爆现象TCP_IDS链规则数暴涨到几千条iptables 匹配变慢。原因SYN 洪水的源 IP 是伪造的每个包源 IP 都不同封禁等于往链里灌垃圾。解决对 SYN 洪水不要按源 IP 封改成限速比如iptables -A INPUT -p tcp --syn -m limit --limit 100/s -j ACCEPT配合默认 DROP或者用hashlimit按目的端口限速。按源 IP 封只适用于源 IP 真实的扫描。5.5 脚本重启后封禁列表丢失现象脚本崩溃重启之前封的 IP 全放出来了攻击者趁这个窗口继续打。原因blocked字典在内存里进程一退就没了。解决启动时从iptables -L TCP_IDS -n读回现有规则重建blocked或者把封禁记录落盘。我一般用前者因为 iptables 本身就是持久化的状态源不用额外维护文件。6. 进阶把固定窗口换成令牌桶顺带聊聊这套东西的边界固定窗口计数有个天然缺陷窗口边界处会漏判。攻击者在窗口末尾打 19 个端口、下个窗口开头再打 19 个两个窗口都没超阈值但实际 1 秒内打了 38 个。换成令牌桶或滑动窗口能解决代价是多一点状态维护。class TokenBucket: 按源 IP 限速rate 是每秒补充的令牌数capacity 是桶容量 def __init__(self, rate, capacity): self.rate rate self.capacity capacity self.tokens capacity self.last time.time() def allow(self): now time.time() # 按流逝时间补充令牌上限是桶容量 self.tokens min(self.capacity, self.tokens (now - self.last) * self.rate) self.last now if self.tokens 1: self.tokens - 1 return True return False # 每个源 IP 一个桶SYN 速率超过 50/s 就判定异常 buckets defaultdict(lambda: TokenBucket(rate50, capacity100)) def detect_by_bucket(pkt): if not pkt[syn] or pkt[ack]: return False return not buckets[pkt[src_ip]].allow()逻辑说明allow()返回 False 表示令牌不够即速率超限。rate50是每秒允许的 SYN 数capacity100是突发容忍量。相比固定窗口令牌桶对边界攻击不敏感也不会因为窗口切换产生误判。参数说明rate按业务正常 SYN 速率设capacity一般设rate的 2 倍容忍短时突发。验证这套检测有没有效我一般做三步先用 nmap 从另一台机器扫测试机看告警和封禁是否触发再用hping3 --flood -S打 SYN 洪水看限速规则是否生效最后跑一段正常业务流量确认没有误封。三步都过才算能上测试环境。这套方案的边界要说清楚纯 Python 抓包在千兆线速下会丢包检测率会掉真要扛高流量得上 C 或 eBPFiptables 规则多了匹配会变慢封禁列表要控制规模伪造源 IP 的 DoS 靠封 IP 治不了本得靠限速和上游清洗。我自己的习惯是这套东西只当测试环境和边缘节点的第一道防线核心业务还是交给专业设备。写它的价值在于你能亲手摸清 TCP 标志位、滑动窗口和 iptables 链是怎么回事这些理解比工具本身值钱。希望帮到你。本文还有配套的精品资源点击获取