简介这份资源是一套基于Python实现的TCP入侵检测系统源码面向计算机、网络安全相关专业的毕业设计、期末大作业与课程设计场景也适合想入门网络攻防的新手学习。项目核心功能是实时检测端口扫描与Dos攻击并联动iptables自动下发防御规则形成检测到响应的完整闭环具有较高的实际应用价值。压缩包共6个文件以5个py源码文件和1个md说明文档为主整体约5KB代码结构清晰、注释完整涵盖数据包嗅探、流量分析、过滤规则与数据库记录等模块新手也能看懂并快速部署运行。目前已有111人学习下载可作为毕设或大作业的参考方案帮助读者理解入侵检测原理、掌握Python网络编程与iptables联动防御的实现思路并在此基础上进行功能扩展与二次开发。1. 从一台被扫穿的测试机说起Python 写的 TCP 入侵检测到底在做什么去年我把一台只开了 22 和 8080 的测试机丢进公网两天后ss -s里的 TIME_WAIT 直接飙到四位数dmesg里全是半连接队列溢出的记录。抓包一看有人在用 SYN 半开扫描逐端口试探紧接着就是一波固定源 IP 的高频连接请求。那一刻我意识到与其等云厂商的安全组告警不如自己用 Python 在 TCP 层做一层轻量入侵检测检测到端口扫描和 DoS 特征后直接联动 iptables 把源 IP 封掉。这套思路就是标题里说的「基于 Python 实现的 TCP 入侵检测系统」核心动作只有三个抓 TCP 包、按行为特征判定端口扫描与 DoS、调用 iptables 做防御。它适合有 Linux 基础、想自己掌控检测规则的安全运维和 Python 后端不适合指望开箱即用扛住 T 级流量的人。下面把我实际落地的路径拆开讲包括抓包选型、特征阈值、iptables 联动命令和几个让我翻过车的坑。2. 抓包与协议解析用 scapy 还是原始 socket 拿到 TCP 头2.1 为什么我最终选了 scapy 而不是纯 socket做 TCP 入侵检测第一步是把网卡上的 TCP 报文变成程序能判断的结构化数据。常见做法有三种libpcap 的 C 绑定、原始 socket 自己解 IP/TCP 头、scapy 这类高层库。纯 socket 性能最好但你要自己处理字节序、IP 选项、TCP 选项字段写起来容易在偏移量上出错scapy 慢一些但IP/TCP层的字段名和协议文档一一对应调试成本低。我的场景是千兆内网边缘峰值几万 ppsscapy 配合 BPF 过滤完全够用。如果你的环境是万兆以上建议用 PF_RING 或 eBPFscapy 会先成为瓶颈。安装依赖时注意scapy 在部分发行版上需要单独装libpcapPython 侧用pip install scapy即可。抓包需要 root 或CAP_NET_RAW能力生产环境别直接用 root 跑整个服务后面会讲降权。# sniff_tcp.py from scapy.all import sniff, IP, TCP def handle_packet(pkt): # 只处理带 IP 和 TCP 层的包过滤 ARP、ICMP 等 if IP in pkt and TCP in pkt: src_ip pkt[IP].src dst_ip pkt[IP].dst sport pkt[TCP].sport dport pkt[TCP].dport flags pkt[TCP].flags # flags 是 FlagValue 对象S 表示 SYNA 表示 ACKR 表示 RST print(f{src_ip}:{sport} - {dst_ip}:{dport} flags{flags}) if __name__ __main__: # filter 用 BPF 语法只抓 tcp减少用户态处理量 sniff(filtertcp, prnhandle_packet, storeFalse)这段代码的逻辑很直白sniff在链路层收包filtertcp让内核 BPF 先过滤掉非 TCP 流量storeFalse避免 scapy 把包全存内存导致 OOM。handle_packet里先判断协议层是否存在再取五元组和标志位。参数上prn是每包回调别在里面做耗时操作否则丢包率会上升iface不指定时 scapy 选默认网卡多网卡机器一定要显式传ifaceeth0。2.2 解析 TCP 标志位时最容易搞混的地方端口扫描和 DoS 的判定极度依赖 TCP 标志位组合这里有几个必须记牢的点。SYN 扫描发的是S只有 SYN目标端口开放回SASYNACK关闭回RARSTACK。如果你只统计S包数量会把正常的三次握手也算进去所以判定扫描时要看「同一源 IP 在短时间窗口内对多个不同目的端口发S且没有后续完整握手」。scapy 里pkt[TCP].flags可以直接和字符串比较比如flags S但更稳的是用位运算flags 0x02判断 SYN 位。另一个坑是分片和 TCP 选项。有些扫描器会发带异常选项的包scapy 解析时不会报错但字段值可能不符合预期。我的做法是在特征提取阶段只取五元组、标志位、窗口大小和包长度不深入解析选项避免被畸形包带偏。窗口大小对 DoS 判定有用正常握手窗口通常几百到几万某些洪水攻击会固定为 0 或极大值。提示抓包程序上线前先用tcpdump -i eth0 -nn tcp对照 scapy 的输出确认五元组和标志位一致否则后面所有阈值都是错的。3. 端口扫描与 DoS 的行为特征阈值怎么定才不误杀3.1 端口扫描的滑动窗口统计法端口扫描的本质是「短时间、多端口、少数据」。我用的判定逻辑是滑动窗口计数维护一个以源 IP 为键的字典记录该 IP 在最近 N 秒内访问过的不同目的端口集合以及发起的 SYN 包数量。当不同端口数超过阈值且 SYN 占比超过 80% 时判定为扫描。窗口大小和阈值需要按你的业务调我给一组实测可用的起点窗口 10 秒不同端口数大于 20SYN 包数大于 30。# scan_detect.py import time from collections import defaultdict, deque WINDOW 10 # 滑动窗口秒数 PORT_THRESHOLD 20 # 不同目的端口数阈值 SYN_THRESHOLD 30 # SYN 包数阈值 # 每个源 IP 保存 (时间戳, 目的端口, 是否SYN) 的队列 history defaultdict(deque) def is_port_scan(src_ip, dport, is_syn): now time.time() q history[src_ip] q.append((now, dport, is_syn)) # 剔除窗口外的记录 while q and now - q[0][0] WINDOW: q.popleft() ports {item[1] for item in q} syn_count sum(1 for item in q if item[2]) if len(ports) PORT_THRESHOLD and syn_count SYN_THRESHOLD: return True return False逻辑说明history用defaultdict(deque)存每个源 IP 的近期行为deque的popleft保证窗口滑动时旧数据及时清理避免内存无限增长。ports集合统计不同目的端口syn_count统计 SYN 包。两个条件同时满足才判定能显著降低误报。参数上WINDOW太小会把正常端口跳转误判太大则响应慢PORT_THRESHOLD设 20 是我在办公网环境下的经验值扫描器通常几百个端口起步正常用户很少 10 秒内碰 20 个端口。3.2 DoS 攻击的两种典型形态与判定差异DoS 在 TCP 层常见两类SYN Flood 和连接耗尽。SYN Flood 的特征是大量 SYN 包、源 IP 可能伪造、几乎没有 ACK 完成握手连接耗尽则是真实 IP 高频建连把服务端的 accept 队列占满。两者判定不能共用一套阈值。SYN Flood 我看两个指标单位时间内某目的 IP 的 SYN 包数以及 SYN 与 ACK 的比例。正常流量 SYN 和 ACK 数量接近Flood 时 SYN 远大于 ACK。连接耗尽则看同一源 IP 在短时间内的新建连接数超过阈值就告警。下面这段在抓包回调里同时统计两类指标。# dos_detect.py import time from collections import defaultdict SYN_FLOOD_WINDOW 5 SYN_FLOOD_THRESHOLD 200 # 5秒内同一目的IP的SYN数 CONN_WINDOW 10 CONN_THRESHOLD 100 # 10秒内同一源IP新建连接数 syn_counter defaultdict(list) # dst_ip - [timestamps] conn_counter defaultdict(list) # src_ip - [timestamps] def check_dos(src_ip, dst_ip, flags): now time.time() # SYN Flood 判定 if flags 0x02 and not flags 0x10: # SYN 置位且 ACK 未置位 syn_counter[dst_ip].append(now) syn_counter[dst_ip] [t for t in syn_counter[dst_ip] if now - t SYN_FLOOD_WINDOW] if len(syn_counter[dst_ip]) SYN_FLOOD_THRESHOLD: return syn_flood, dst_ip # 连接耗尽判定只看 SYN 且 ACK 未置位的新建请求 if flags 0x02 and not flags 0x10: conn_counter[src_ip].append(now) conn_counter[src_ip] [t for t in conn_counter[src_ip] if now - t CONN_WINDOW] if len(conn_counter[src_ip]) CONN_THRESHOLD: return conn_exhaust, src_ip return None, None这里用列表推导式清理过期时间戳比 deque 更直观。flags 0x02判断 SYN 位flags 0x10判断 ACK 位not flags 0x10确保是纯 SYN。参数上SYN_FLOOD_THRESHOLD设 200 是 5 秒内同一目的 IP 的 SYN 数真实业务突发可能接近这个值建议先观察一周再定CONN_THRESHOLD设 100 是 10 秒内同一源 IP 新建连接数对爬虫类业务要放宽。注意阈值没有万能值。上线前先跑「只告警不封禁」模式把日志按源 IP 聚合看一周再决定封禁阈值否则很容易把负载均衡的健康检查封掉。4. 联动 iptables 做防御从判定到封禁的完整链路4.1 用独立链管理封禁规则别直接往 INPUT 里塞检测到攻击后要封源 IP最直接的做法是iptables -A INPUT -s x.x.x.x -j DROP。但这样有两个问题规则越加越多排查时看不清哪些是自动封的解封时要按行号删容易删错。我的做法是建一条自定义链TCP_IDS所有自动封禁规则加到这条链INPUT 里只放一条跳转。# 初始化链只执行一次 iptables -N TCP_IDS 2/dev/null || true iptables -C INPUT -j TCP_IDS 2/dev/null || iptables -A INPUT -j TCP_IDS # 封禁一个源 IP带注释方便追溯 iptables -I TCP_IDS 1 -s 192.168.1.100 -m comment --comment ids_ban_20261005 -j DROP # 查看当前封禁列表 iptables -L TCP_IDS -n --line-numbers # 解封 iptables -D TCP_IDS -s 192.168.1.100 -j DROP-N建链-C检查跳转规则是否已存在避免重复添加。-I TCP_IDS 1把新规则插到链首保证最新封禁优先生效。-m comment加注释后面写自动解封脚本时可以按注释前缀匹配。-D删除时最好带上完整匹配条件只写行号在规则变动后会删错。4.2 Python 调用 iptables 的正确姿势与超时控制Python 里调 iptables 用subprocess.run不要用os.system前者能拿到返回码和 stderr。关键点是加超时和参数列表传参避免命令注入和卡死。# firewall.py import subprocess import logging logger logging.getLogger(__name__) def ban_ip(ip, reasonids): # 参数用列表传递避免 shell 注入 cmd [ iptables, -I, TCP_IDS, 1, -s, ip, -m, comment, --comment, fids_ban_{reason}, -j, DROP ] try: result subprocess.run( cmd, capture_outputTrue, textTrue, timeout3 ) if result.returncode ! 0: logger.error(ban failed ip%s stderr%s, ip, result.stderr) return False logger.info(banned ip%s reason%s, ip, reason) return True except subprocess.TimeoutExpired: logger.error(ban timeout ip%s, ip) return False def unban_ip(ip): cmd [iptables, -D, TCP_IDS, -s, ip, -j, DROP] subprocess.run(cmd, capture_outputTrue, textTrue, timeout3)capture_outputTrue把 stdout/stderr 收进结果对象timeout3防止 iptables 因内核锁卡住导致检测线程阻塞。textTrue让输出是字符串方便日志。参数列表传参杜绝了 IP 字段里混入 shell 元字符的风险。封禁前建议先查一下 IP 是否已在链里避免重复插入导致规则膨胀。4.3 自动解封与封禁时长设计只封不解时间一长TCP_IDS链会有上万条规则匹配性能下降。我一般设两档封禁时长扫描类封 600 秒DoS 类封 3600 秒。用一个后台线程定期扫描链上的注释时间戳到期就删。注释格式统一成ids_ban_reason_expire_ts解析时按_分割取时间戳。解封前再确认一次该 IP 近期没有新的攻击记录有就续期避免攻击者换端口继续打。提示iptables 规则数量超过几千条后匹配延迟会明显上升。生产环境建议配合 ipset把封禁 IP 存进集合iptables 只引用集合增删都是 O(1)。5. 避坑与排查那些让我半夜爬起来改配置的问题5.1 抓包丢包严重检测形同虚设现象sniff回调里打印计数发现实际处理量只有 tcpdump 统计的一半。原因scapy 在用户态逐包解析Python GIL 下回调函数稍慢就丢包尤其开了storeTrue时内存和拷贝开销更大。解决storeFalse必开BPF 过滤尽量精确比如只抓tcp and (tcp[tcpflags] tcp-syn ! 0)把特征提取和封禁动作拆到不同线程回调里只做入队。如果还丢考虑换 PF_RING 或直接用 eBPF 在内核侧聚合。5.2 把负载均衡健康检查封了现象上线第二天业务告警后端全部不可用查 iptables 发现 LB 的 IP 被自动封禁。原因健康检查每秒对多个后端端口发 SYN触发了端口扫描阈值。解决维护白名单抓包回调里先判断源 IP 是否在白名单命中直接返回白名单从配置文件读支持热加载。另外把 LB 网段整体加白别只加单个 IP。5.3 iptables 命令执行失败但没日志现象检测到攻击日志里却没有封禁成功记录攻击持续。原因subprocess.run没检查returncodeiptables 因权限不足或链不存在报错被吞掉。解决每次调用后判断返回码非零就把 stderr 写进日志启动时先检查TCP_IDS链是否存在不存在就创建。另外确认运行用户有CAP_NET_ADMIN用setcap给 Python 解释器授权比直接 root 跑更安全。5.4 阈值调太严导致误封正常用户现象某次促销活动大量用户短时间刷新页面被判定为连接耗尽封了 IP。原因CONN_THRESHOLD按日常流量设的没考虑突发。解决引入动态基线按小时统计历史连接数阈值设为基线的 3 到 5 倍活动前手动调高阈值或临时关闭封禁只告警。血泪经验是任何自动封禁上线前都要有「一键清空封禁链」的开关。5.5 时间戳和时区不一致导致解封失效现象封禁规则到期不删链越来越长。原因注释里写的是本地时间解封脚本用 UTC 比较差 8 小时。解决统一用 Unix 时间戳写入和读取都走time.time()展示时再转本地时区。这个坑很隐蔽日志里看时间都对但逻辑就是不对。6. 进阶用 ipset 扛住十万级封禁与检测规则热更新当封禁 IP 上到几万条iptables 逐条匹配会成为新瓶颈。我的做法是引入 ipset建一个hash:ip集合iptables 规则只写-m set --match-set ids_ban src -j DROPPython 侧用ipset add/del管理集合成员。实测十万级 IP 下匹配延迟从几十毫秒降到微秒级。下面是对应的改造代码。# ipset_firewall.py import subprocess SET_NAME ids_ban def init_ipset(): # 创建集合timeout 0 表示不过期由脚本控制解封 subprocess.run( [ipset, create, SET_NAME, hash:ip, timeout, 0], capture_outputTrue, textTrue ) # iptables 引用集合只需一条规则 subprocess.run( [iptables, -C, INPUT, -m, set, --match-set, SET_NAME, src, -j, DROP], capture_outputTrue, textTrue ) def ban_ip_set(ip, timeout600): # timeout 单位秒到期 ipset 自动移除省去解封脚本 subprocess.run( [ipset, add, SET_NAME, ip, timeout, str(timeout)], capture_outputTrue, textTrue, timeout3 )ipset create的timeout 0是默认不过期ban_ip_set里传timeout参数后ipset 内核模块会自动到期移除比自己在 Python 里维护解封队列可靠得多。iptables -C检查规则是否存在避免重复添加。这套改造后封禁和解封都不再需要遍历规则链性能瓶颈转移到 ipset 的哈希表十万级轻松应对。检测规则的热更新同样重要。我把端口扫描和 DoS 的阈值、白名单、封禁时长抽到一个 YAML 文件用watchdog监听文件变更变更后重新加载到内存不重启抓包进程。这样调阈值不用停服务半夜改配置也不用重新登录机器。# config_loader.py import yaml from watchdog.observers import Observer from watchdog.events import FileSystemEventHandler class Config: def __init__(self, path): self.path path self.data {} self.load() def load(self): with open(self.path, r, encodingutf-8) as f: self.data yaml.safe_load(f) class ReloadHandler(FileSystemEventHandler): def __init__(self, config): self.config config def on_modified(self, event): if event.src_path.endswith(ids_config.yaml): self.config.load() print(config reloaded) config Config(ids_config.yaml) observer Observer() observer.schedule(ReloadHandler(config), path., recursiveFalse) observer.start()watchdog的on_modified在文件保存时触发重新调用load覆盖内存中的配置。注意 YAML 解析失败要捕获异常并保留旧配置否则一次手误就让检测停摆。我一般还会在加载后打印关键阈值方便确认生效。最后说个验证方法用hping3或nmap在测试机模拟扫描观察检测日志和 ipset 成员变化确认从抓包到封禁的链路在秒级完成。别等真被打了才验证那时候你连日志都来不及看。这套方案我前后调了三个月最大的教训是自动封禁一定要有白名单和手动开关检测阈值宁可先松后紧别一上来就追求零误报。希望帮到你。本文还有配套的精品资源点击获取