简介这是一套面向高校计算机网络、信息安全专业学生及中小型网络运维人员的Python TCP入侵检测系统源码可作为毕业设计、课程设计或项目开发的技术参考。系统聚焦TCP层面的安全监测通过分析连接请求的时间序列频率、TCP头部标志位组合如SYN、FIN及NULL包分布以及非监听端口连接占比识别端口扫描与DDoS攻击并借助scapy抓包解析、python-iptables联动防火墙实现自动防御检测日志经MySQLdb结构化存储。资源包共10个文件以5个py源码为主辅以3个zbak备份、1个zip与1个md说明文档整体约10KB结构清晰、模块耦合度低。目前已有41人学习下载适合希望理解入侵检测原理、掌握scapy与iptables联动开发、快速搭建实验原型的读者参考与扩展。1. 从一台被打穿的测试机说起TCP 入侵检测到底在防什么去年帮一个做课程设计的学生看环境他那台跑着 Web 服务的云主机在两天内被扫了四万多次日志里全是半开连接和畸形标志位。他一开始以为是业务量上来了直到ss -s显示 SYN-RECV 队列长期爆满才意识到这是典型的端口扫描加 SYN Flood。这件事让我重新审视一个老话题基于 Python 的 TCP 入侵检测系统端口扫描与 DoS 攻击检测再配合 iptables 联动防御到底该怎么落地。这套方案的核心思路并不复杂用原始套接字或抓包库在链路层拿到 TCP 报文解析三次握手过程中的标志位、窗口大小、时间间隔等特征识别出端口扫描和 DoS 攻击行为然后调用 iptables 把恶意源 IP 拉黑。它适合毕业设计、课程设计也适合中小规模服务器的轻量级防护。Python 在这里的角色是快速原型和规则引擎iptables 负责真正的执行层拦截。理解这个分工后面所有代码和参数才有落脚点。2. 抓包与特征提取Python 怎么拿到 TCP 层的原始数据2.1 原始套接字与 scapy 的选型对比做 TCP 入侵检测第一步是拿到包。常见做法有三种raw socket、scapy、pcap 库。raw socket 最底层需要 root 权限跨平台差异大scapy 封装好了解析层开发效率高但高并发下性能吃紧pcap 库依赖 libpcap性能最好但安装门槛高。我一般会推荐 scapy 做原型验证因为它的TCP和IP层字段直接可读调试成本低。选型时要注意一个现实问题很多毕业设计环境是 Windows而 raw socket 在 Windows 上行为不一致。scapy 在 Windows 上需要装 NpcapLinux 上直接跑。如果你的目标是 Linux 服务器防护直接用 scapy 的sniff函数就能开工。下面是最小可运行代码抓取 60 秒内的 TCP 包并打印关键字段。from scapy.all import sniff, TCP, IP import time def handle_packet(pkt): if IP in pkt and TCP in pkt: ip_layer pkt[IP] tcp_layer pkt[TCP] # 打印源IP、目的端口、TCP标志位、窗口大小 print(fsrc{ip_layer.src} dst_port{tcp_layer.dport} fflags{tcp_layer.flags} win{tcp_layer.window}) # 只抓TCP超时60秒store0表示不驻留内存 sniff(filtertcp, prnhandle_packet, timeout60, store0)这段代码的逻辑是filtertcp让 BPF 过滤器只放行 TCP 报文prn是每包回调store0避免长时间抓包把内存吃满。参数上timeout控制抓包时长生产环境一般配合iface指定网卡。tcp_layer.flags返回的是 scapy 的 FlagValue 对象可以直接和S、SA、F比较这是后面识别扫描和 Flood 的基础。2.2 端口扫描的三种典型特征端口扫描不是单一行为常见的有 SYN 扫描、FIN 扫描、NULL 扫描。SYN 扫描发S标志位目标端口开放则回SA关闭则回RAFIN 扫描发F关闭端口回RA开放端口不回。检测逻辑要区分对待。我一般用滑动窗口统计在 5 秒窗口内同一个源 IP 如果向超过 20 个不同目的端口发了 SYN就判定为端口扫描。这个阈值不是拍脑袋普通用户一次访问最多触发几个端口而 nmap 默认扫描是 1000 个端口起步。下面是一个基于字典的计数器实现。from collections import defaultdict import time # 记录每个源IP在时间窗口内访问过的目的端口集合 scan_tracker defaultdict(lambda: {ports: set(), first_seen: time.time()}) SCAN_WINDOW 5 # 时间窗口单位秒 SCAN_PORT_THRESHOLD 20 # 不同端口数阈值 def detect_port_scan(src_ip, dst_port): now time.time() record scan_tracker[src_ip] # 窗口过期则重置 if now - record[first_seen] SCAN_WINDOW: record[ports].clear() record[first_seen] now record[ports].add(dst_port) if len(record[ports]) SCAN_PORT_THRESHOLD: return True return False这里的关键参数是SCAN_WINDOW和SCAN_PORT_THRESHOLD。窗口太短会漏掉慢速扫描太长会误判正常的多连接应用。20 个端口在 5 秒内对普通 Web 服务来说几乎不可能出现但对负载均衡健康检查可能偏敏感实际部署时建议先跑观察模式统计一周再定阈值。2.3 DoS 攻击的流量特征与计数方法DoS 攻击在 TCP 层的表现主要是 SYN Flood。攻击者发大量 SYN 包但不完成三次握手服务器半开连接队列被占满。检测特征是单位时间内同一源 IP 的 SYN 包数量异常高或者同一目的 IP 收到大量来自不同源的 SYN。我一般同时监控两个维度单源 SYN 速率和全局 SYN 速率。单源超过 100 包/秒就标记全局超过 1000 包/秒就告警。下面代码用时间戳队列做速率计算比简单计数更准。from collections import defaultdict, deque import time syn_records defaultdict(deque) SYN_RATE_WINDOW 1 # 统计窗口1秒 SYN_RATE_THRESHOLD 100 # 单源每秒SYN阈值 def detect_syn_flood(src_ip, flags): if S not in flags or A in flags: return False now time.time() q syn_records[src_ip] q.append(now) # 移除窗口外的记录 while q and now - q[0] SYN_RATE_WINDOW: q.popleft() if len(q) SYN_RATE_THRESHOLD: return True return FalseSYN_RATE_WINDOW设为 1 秒是为了快速响应SYN_RATE_THRESHOLD设为 100 是经验值。注意这里排除了SA包因为那是服务端的正常响应。如果攻击者伪造源 IP单源统计会失效这时候要切到全局统计但全局统计容易把正常突发流量误判需要配合白名单。3. 检测引擎与 iptables 联动从告警到真正拦截3.1 检测主循环的组装方式把前面的检测函数串起来需要一个主循环。我一般用 scapy 的sniff加回调在回调里依次调用端口扫描检测和 SYN Flood 检测。检测到恶意行为后不直接封禁而是先记入待封禁队列由独立线程执行 iptables 命令。这样做的好处是抓包线程不被系统调用阻塞。from scapy.all import sniff, TCP, IP import threading import queue block_queue queue.Queue() def packet_callback(pkt): if IP not in pkt or TCP not in pkt: return src_ip pkt[IP].src dst_port pkt[TCP].dport flags pkt[TCP].flags if detect_port_scan(src_ip, dst_port): block_queue.put((src_ip, port_scan)) if detect_syn_flood(src_ip, flags): block_queue.put((src_ip, syn_flood)) def start_sniffing(): sniff(filtertcp, prnpacket_callback, store0) # 抓包线程 sniffer threading.Thread(targetstart_sniffing, daemonTrue) sniffer.start()这里用queue.Queue做线程间通信daemonTrue保证主程序退出时抓包线程自动结束。实际部署时sniff的iface参数要指定对外网卡否则可能抓到无关流量。另外store0必须加否则长时间运行内存会涨到几个 G。3.2 iptables 封禁命令的封装与幂等处理iptables 封禁本身很简单一条iptables -I INPUT -s ip -j DROP就能搞定。但生产环境要考虑幂等同一个 IP 被多次检测到不能重复插入规则否则规则链会膨胀。我一般先查iptables -C再决定是否插入。import subprocess def block_ip(ip, reason): # 先检查规则是否已存在 check subprocess.run( [iptables, -C, INPUT, -s, ip, -j, DROP], capture_outputTrue ) if check.returncode 0: return # 已封禁跳过 # 插入到INPUT链首部立即生效 subprocess.run( [iptables, -I, INPUT, -s, ip, -j, DROP], checkTrue ) print(fblocked {ip} for {reason})-C是检查规则是否存在返回 0 表示存在。-I INPUT把规则插到链首优先级最高。注意这里没有加-p tcp因为封禁整个 IP 更彻底。如果只想封 TCP可以加-p tcp。另外iptables 规则重启后会丢失需要配合iptables-save或iptables-persistent做持久化。3.3 封禁队列的消费与解封策略封禁不能只进不出否则正常用户被误判后永远无法恢复。我一般给每个封禁记录加一个过期时间默认 600 秒到期自动解封。消费线程从队列取任务执行封禁并记录时间戳后台再起一个清理线程定期解封。import time blocked_ips {} # ip - 解封时间戳 BLOCK_DURATION 600 # 封禁10分钟 def block_consumer(): while True: ip, reason block_queue.get() if ip in blocked_ips: continue block_ip(ip, reason) blocked_ips[ip] time.time() BLOCK_DURATION def unblock_worker(): while True: now time.time() for ip, expire in list(blocked_ips.items()): if now expire: subprocess.run( [iptables, -D, INPUT, -s, ip, -j, DROP], capture_outputTrue ) del blocked_ips[ip] time.sleep(10)BLOCK_DURATION设 600 秒是折中值太短攻击者可以反复试探太长误伤用户影响大。unblock_worker每 10 秒扫一次用-D删除规则。这里有个坑如果规则被手动删过-D会报错所以用capture_outputTrue吞掉错误避免线程崩溃。4. 避坑与排查那些让我熬夜的翻车现场4.1 抓不到包权限与网卡选错现象程序跑起来没有任何输出日志为空。原因通常是两个一是没用 root 权限raw socket 和 scapy 的底层抓包都需要 CAP_NET_RAW二是iface没指定scapy 默认选了一个没有流量的虚拟网卡。解决方法是sudo运行并用ip a或ifconfig确认对外网卡名在sniff里显式传ifaceeth0。4.2 iptables 规则不生效链顺序和表选错现象iptables -L能看到 DROP 规则但攻击流量依然进来。原因可能是规则加在了错误的链比如加到了FORWARD而不是INPUT或者服务器用了 nftables 后端iptables 命令实际写到了 nft 表里但没生效。解决方法是先iptables -L INPUT -n --line-numbers确认规则位置再用iptables -I INPUT 1插到最前面。如果系统是 nftables建议直接用nft命令或确认 iptables 兼容层已启用。4.3 误封正常用户阈值太激进现象封禁列表里出现了公司出口 IP 或 CDN 节点。原因是阈值设得太低比如SCAN_PORT_THRESHOLD5正常页面加载可能触发多个端口。解决方法是先跑观察模式只记录不封禁统计一周的端口访问分布把阈值调到 P99 以上。另外维护一个白名单文件启动时加载检测到白名单 IP 直接跳过。4.4 内存持续增长store 没关或字典没清理现象程序跑几小时后内存占用几个 G。原因是sniff的store默认为 1所有包都驻留内存或者scan_tracker、syn_records字典只增不减。解决方法是sniff(store0)并给字典加定期清理逻辑比如每 60 秒清理一次超过窗口期的记录。4.5 封禁后服务不可用误封了网关或 DNS现象封禁某个 IP 后服务器自己上不了网或解析不了域名。原因是攻击者伪造了源 IP而那个 IP 恰好是网关或 DNS。解决方法是封禁前检查 IP 是否在保护名单里至少排除网关、DNS、回环地址。更稳妥的做法是只封禁目的端口为业务端口的流量而不是整个 IP。5. 进阶技巧让检测更准、封禁更稳的几个习惯5.1 用滑动窗口替代固定窗口做速率统计固定窗口在窗口边界会有突刺问题攻击者在窗口切换瞬间发两倍流量可能被漏掉。滑动窗口用双端队列记录每个包的时间戳精度更高。代价是内存占用略大但现代服务器完全扛得住。我一般把SYN_RATE_WINDOW设为 1 秒队列长度上限设为阈值的 2 倍超过就强制清理。5.2 把检测结果落盘方便事后复盘只打印到控制台程序一关什么都没了。我习惯把每次告警写进 SQLite字段包括时间、源 IP、目的端口、攻击类型、处理动作。这样一周后可以跑 SQL 分析误报率也方便毕业设计写论文时出图表。import sqlite3 conn sqlite3.connect(ids_log.db) conn.execute( CREATE TABLE IF NOT EXISTS alerts ( ts REAL, src_ip TEXT, dst_port INTEGER, attack_type TEXT, action TEXT ) ) def log_alert(src_ip, dst_port, attack_type, action): conn.execute( INSERT INTO alerts VALUES (?, ?, ?, ?, ?), (time.time(), src_ip, dst_port, attack_type, action) ) conn.commit()这张表结构简单但足够支撑后续的统计查询。ts用浮点时间戳方便做时间范围过滤。action字段记录是alert还是block观察模式下只写alert。5.3 用 ipset 替代逐条 iptables 规则当封禁 IP 超过几百个时iptables 规则链会变长匹配性能下降。ipset 是内核态哈希表查询复杂度 O(1)适合大规模封禁。用法是先建集合ipset create blacklist hash:ip然后iptables -I INPUT -m set --match-set blacklist src -j DROP之后只需要ipset add blacklist ip和ipset del blacklist ip。这样 iptables 规则只有一条维护成本大幅降低。方案规则数量匹配性能适用规模逐条 iptables随 IP 数增长线性下降几十个 IPipset固定 1 条常数级几千个 IP5.4 验证检测效果的一个笨办法写完检测逻辑后别急着上生产。我一般在本机开一个nc -l 8000然后用另一台机器跑nmap -sS ip看程序能不能在 5 秒内识别并封禁。SYN Flood 可以用hping3 -S --flood -p 80 ip模拟但注意别在公网环境跑容易把目标打挂。验证通过后再放到测试服务器跑一周观察模式确认误报率可接受再开启封禁。这套东西我前后改了三版第一版误封了办公室出口 IP第二版内存泄漏跑了一夜就崩第三版才稳定下来。最大的教训是检测阈值一定要用真实流量统计出来不要拍脑袋封禁一定要有白名单和自动解封否则就是给自己挖坑。希望帮到你。本文还有配套的精品资源点击获取