简介：这是一套面向高校计算机网络、信息安全专业学生及中小型网络运维人员的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 上需要装 Npcap，Linux 上直接跑。如果你的目标是 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(f"src={ip_layer.src} dst_port={tcp_layer.dport} " f"flags={tcp_layer.flags} win={tcp_layer.window}") # 只抓TCP，超时60秒，store=0表示不驻留内存 sniff(filter="tcp", prn=handle_packet, timeout=60, store=0)这段代码的逻辑是：filter="tcp"让 BPF 过滤器只放行 TCP 报文，prn是每包回调，store=0避免长时间抓包把内存吃满。参数上，timeout控制抓包时长，生产环境一般配合iface指定网卡。tcp_layer.flags返回的是 scapy 的 FlagValue 对象，可以直接和"S"、"SA"、"F"比较，这是后面识别扫描和 Flood 的基础。
2.2 端口扫描的三种典型特征
端口扫描不是单一行为，常见的有 SYN 扫描、FIN 扫描、NULL 扫描。SYN 扫描发S标志位，目标端口开放则回SA，关闭则回RA；FIN 扫描发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(filter="tcp", prn=packet_callback, store=0) # 抓包线程 sniffer = threading.Thread(target=start_sniffing, daemon=True) sniffer.start()这里用queue.Queue做线程间通信，daemon=True保证主程序退出时抓包线程自动结束。实际部署时，sniff的iface参数要指定对外网卡，否则可能抓到无关流量。另外，store=0必须加，否则长时间运行内存会涨到几个 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_output=True ) if check.returncode == 0: return # 已封禁，跳过 # 插入到INPUT链首部，立即生效 subprocess.run( ["iptables", "-I", "INPUT", "-s", ip, "-j", "DROP"], check=True ) print(f"blocked {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_output=True ) del blocked_ips[ip] time.sleep(10)BLOCK_DURATION设 600 秒是折中值，太短攻击者可以反复试探，太长误伤用户影响大。unblock_worker每 10 秒扫一次，用-D删除规则。这里有个坑：如果规则被手动删过，-D会报错，所以用capture_output=True吞掉错误，避免线程崩溃。
4. 避坑与排查：那些让我熬夜的翻车现场
4.1 抓不到包：权限与网卡选错
现象：程序跑起来没有任何输出，日志为空。原因通常是两个：一是没用 root 权限，raw socket 和 scapy 的底层抓包都需要 CAP_NET_RAW；二是iface没指定，scapy 默认选了一个没有流量的虚拟网卡。解决方法是sudo运行，并用ip a或ifconfig确认对外网卡名，在sniff里显式传iface="eth0"。
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_THRESHOLD=5，正常页面加载可能触发多个端口。解决方法是先跑观察模式，只记录不封禁，统计一周的端口访问分布，把阈值调到 P99 以上。另外，维护一个白名单文件，启动时加载，检测到白名单 IP 直接跳过。
4.4 内存持续增长：store 没关或字典没清理
现象：程序跑几小时后内存占用几个 G。原因是sniff的store默认为 1，所有包都驻留内存；或者scan_tracker、syn_records字典只增不减。解决方法是sniff(store=0)，并给字典加定期清理逻辑，比如每 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 数增长 | 线性下降 | 几十个 IP |
| ipset | 固定 1 条 | 常数级 | 几千个 IP |
5.4 验证检测效果的一个笨办法
写完检测逻辑后，别急着上生产。我一般在本机开一个nc -l 8000，然后用另一台机器跑nmap -sS <ip>，看程序能不能在 5 秒内识别并封禁。SYN Flood 可以用hping3 -S --flood -p 80 <ip>模拟，但注意别在公网环境跑，容易把目标打挂。验证通过后再放到测试服务器跑一周观察模式，确认误报率可接受再开启封禁。
这套东西我前后改了三版，第一版误封了办公室出口 IP，第二版内存泄漏跑了一夜就崩，第三版才稳定下来。最大的教训是：检测阈值一定要用真实流量统计出来，不要拍脑袋；封禁一定要有白名单和自动解封，否则就是给自己挖坑。希望帮到你。
本文还有配套的精品资源，点击获取
