拓冰网站定制 
简介这是一份面向网络安全初学者与高校学生的Python TCP入侵检测系统源码可直接用于毕业设计、期末大作业或课程设计。项目聚焦端口扫描与Dos攻击的实时检测并联动iptables实现自动防御帮助读者理解入侵检测与主机防火墙协同的基本思路。压缩包共6个文件以5个py源码文件为主分别承担流量分析、数据嗅探、过滤处理与数据库记录等职责另附1个md说明文档整体约5KB结构精简、便于阅读与二次修改。代码含详细注释新手也能看懂部署后即可运行。目前已有111人学习下载适合想快速搭建可演示安全项目、积累实战经验的同学参考借鉴。1. 用 Python 写一个能联动 iptables 的 TCP 入侵检测系统到底在做什么一台暴露在公网的 Linux 服务器平均上线后几分钟内就会收到第一波端口扫描。如果你只开了 22、80、443日志里却能看到有人在一秒内把 1 到 65535 全敲了一遍——这就是最典型的 TCP 入侵前奏。基于 Python 实现的 TCP 入侵检测系统核心目标就三件事抓包识别异常 TCP 行为、判定端口扫描和 DoS 攻击、联动 iptables 把攻击源封掉。它不依赖 Suricata、Snort 这类重型引擎而是用 Python 直接解析 TCP 报文把检测逻辑和防御动作串成一条闭环。适合有 Python 基础、想理解 IDS 内部判定逻辑、又需要一套能改能扩的轻量方案的运维和后台开发。下面从抓包原理讲到 iptables 联动每一步都能在你自己的机器上复现。2. 抓包与 TCP 解析Python 怎么拿到原始报文2.1 为什么选原始套接字而不是抓现成日志做 TCP 入侵检测第一道坎是数据从哪来。常见做法有三种读/var/log/下的应用日志、用 libpcap 抓包、直接开原始套接字raw socket。日志方案最省事但很多扫描和 DoS 根本不会在应用层留下记录比如半开扫描只发 SYN 不回 ACKWeb 服务压根不知道有人来过。libpcap 功能全但要多装一个依赖跨平台编译偶尔翻车。我一般会选原始套接字因为 Linux 下socket.AF_INET, socket.SOCK_RAW, socket.IPPROTO_TCP就能直接拿到 IP 层往上的 TCP 报文不依赖额外库代码可控。代价是必须用 root 或授予CAP_NET_RAW能力运行否则PermissionError。这一点在部署时要提前想清楚别等到线上才发现权限不够。2.2 手工解析 IP 头和 TCP 头原始套接字拿到的是完整的 IP 数据报需要自己按字节偏移拆出 IP 头和 TCP 头。IP 头固定 20 字节不含选项TCP 头也是 20 字节不含选项。关键字段的偏移必须记牢错一位后面全乱。import socket import struct def parse_packet(data): # IP 头前 20 字节 ip_header struct.unpack(!BBHHHBBH4s4s, data[:20]) ihl (ip_header[0] 0x0F) * 4 # 首部长度单位是 4 字节 protocol ip_header[6] # 6 表示 TCP src_ip socket.inet_ntoa(ip_header[8]) # 源 IP dst_ip socket.inet_ntoa(ip_header[9]) # 目的 IP if protocol ! 6: return None # TCP 头从 ihl 偏移开始固定 20 字节 tcp_start ihl tcp_header struct.unpack(!HHLLBBHHH, data[tcp_start:tcp_start 20]) src_port tcp_header[0] dst_port tcp_header[1] seq tcp_header[2] ack tcp_header[3] offset_flags tcp_header[4] flags tcp_header[5] data_offset (offset_flags 4) * 4 # TCP 首部长度 return { src_ip: src_ip, dst_ip: dst_ip, src_port: src_port, dst_port: dst_port, seq: seq, ack: ack, flags: flags, payload_offset: tcp_start data_offset }struct.unpack的格式串!BBHHHBBH4s4s对应 IP 头的字段顺序!表示网络字节序。ihl那一位同时存了版本号和首部长度用 0x0F取低四位再乘 4 才是真实字节数。TCP 的offset_flags高四位是数据偏移低六位才是标志位很多人第一次写会把这两个搞混导致 SYN、ACK 判断全错。2.3 标志位判定SYN、ACK、RST 各代表什么TCP 三次握手是理解扫描检测的基础。客户端发 SYN服务端回 SYNACK客户端再回 ACK。标志位是 8 位里的低 6 位FIN0x01、SYN0x02、RST0x04、PSH0x08、ACK0x10、URG0x20。SYN 0x02 ACK 0x10 RST 0x04 FIN 0x01 def get_flags(flags): return { syn: bool(flags SYN), ack: bool(flags ACK), rst: bool(flags RST), fin: bool(flags FIN), }判断一个包是不是「只发 SYN 不回 ACK」就是syn and not ack。这是半开扫描SYN scan的指纹。正常握手第一个包也是 SYN 无 ACK区别在于正常客户端收到 SYNACK 后会补一个 ACK而扫描器往往直接换下一个端口。所以单看一个包不够必须按源 IP 聚合一段时间窗口内的行为这就引出下一章的检测逻辑。提示原始套接字在 Windows 上行为不一致这套方案建议跑在 Linux 上内核版本不影响Python 3.6 以上即可。3. 端口扫描与 DoS 的检测逻辑怎么写3.1 端口扫描按源 IP 聚合短时间内的目标端口数端口扫描的本质特征不是单个包而是「同一个源 IP 在很短时间内访问了大量不同的目的端口」。所以检测器要维护一张表{源IP: {时间窗口内的端口集合}}。每来一个 SYN 包把目的端口塞进去然后看集合大小有没有超过阈值。import time from collections import defaultdict SCAN_WINDOW 10 # 时间窗口秒 SCAN_PORT_THRESHOLD 20 # 窗口内不同端口数阈值 scan_tracker defaultdict(lambda: {ports: set(), first_seen: time.time()}) def detect_port_scan(src_ip, dst_port): now time.time() entry scan_tracker[src_ip] # 窗口过期就重置 if now - entry[first_seen] SCAN_WINDOW: entry[ports] set() entry[first_seen] now entry[ports].add(dst_port) if len(entry[ports]) SCAN_PORT_THRESHOLD: return True, len(entry[ports]) return False, len(entry[ports])SCAN_WINDOW和SCAN_PORT_THRESHOLD是最需要按环境调的两个参数。内网做资产扫描的合法工具也会触发所以阈值不能太低。公网服务器上10 秒内超过 20 个不同端口基本可以判定为扫描如果业务本身有大量短连接比如压测环境就要把窗口拉长或阈值调高。判断为扫描后不要立刻封先记录连续多个窗口都超阈值再动手能有效降低误封。3.2 DoS 检测SYN Flood 的连接速率与半开连接堆积DoS 里最常见的是 SYN Flood攻击者发大量 SYN 但不完成握手服务端为每个半开连接分配资源最终耗尽。检测思路是统计单位时间内来自同一源 IP 的 SYN 包数量以及「只发 SYN 不回 ACK」的比例。SYN_RATE_WINDOW 5 # 统计窗口秒 SYN_RATE_THRESHOLD 100 # 窗口内 SYN 包数阈值 syn_tracker defaultdict(lambda: {count: 0, first_seen: time.time()}) def detect_syn_flood(src_ip, flags): now time.time() entry syn_tracker[src_ip] if now - entry[first_seen] SYN_RATE_WINDOW: entry[count] 0 entry[first_seen] now if flags SYN and not (flags ACK): entry[count] 1 if entry[count] SYN_RATE_THRESHOLD: return True, entry[count] return False, entry[count]SYN_RATE_THRESHOLD设 100 意味着 5 秒内同一源 IP 发超过 100 个 SYN 就告警。真实用户不可能有这个速率但 NAT 出口后面的整栋楼可能共享一个公网 IP这时候阈值要放宽或者改成按「源 IP 目的端口」维度统计。SYN Flood 和端口扫描的判定可以共用同一份抓包数据只是聚合维度不同一个看端口集合一个看包计数。3.3 把检测结果落库别只 print检测到异常后最忌讳只print一行日志。你需要留证据谁、什么时候、什么行为、封了多久。用 SQLite 就够不引入额外服务。import sqlite3 def init_db(pathids.db): conn sqlite3.connect(path) conn.execute(CREATE TABLE IF NOT EXISTS alerts ( id INTEGER PRIMARY KEY AUTOINCREMENT, ts REAL, src_ip TEXT, attack_type TEXT, detail TEXT, action TEXT)) conn.commit() return conn def log_alert(conn, src_ip, attack_type, detail, action): conn.execute( INSERT INTO alerts (ts, src_ip, attack_type, detail, action) VALUES (?,?,?,?,?), (time.time(), src_ip, attack_type, detail, action)) conn.commit()attack_type存port_scan或syn_floodaction存blocked或observed。这张表后面排查误封时是唯一的后悔药别省。4. 联动 iptables从检测到封禁的完整链路4.1 用独立链而不是直接往 INPUT 里塞规则直接iptables -A INPUT -s x.x.x.x -j DROP能封但规则越加越多管理混乱解封也麻烦。正确做法是建一条自定义链比如IDS_BLOCK在 INPUT 里引用它所有封禁规则都加到这条链上。# 创建自定义链已存在会报错可忽略 iptables -N IDS_BLOCK # 让 INPUT 流量先经过这条链 iptables -C INPUT -j IDS_BLOCK 2/dev/null || iptables -I INPUT -j IDS_BLOCK-C是检查规则是否存在配合||保证重复执行不会插入多条。这一步建议写进系统启动脚本机器重启后链还在但规则需要重新加载。4.2 Python 调用 iptables 封禁与解封封禁就是把源 IP 加进IDS_BLOCK链解封就是删掉。用subprocess调用注意参数用列表传别拼字符串避免命令注入。import subprocess import time BLOCK_CHAIN IDS_BLOCK BLOCK_TTL 3600 # 封禁时长秒 def block_ip(ip): # 先查是否已封避免重复 check subprocess.run( [iptables, -C, BLOCK_CHAIN, -s, ip, -j, DROP], capture_outputTrue) if check.returncode 0: return False subprocess.run( [iptables, -A, BLOCK_CHAIN, -s, ip, -j, DROP], checkTrue) return True def unblock_ip(ip): subprocess.run( [iptables, -D, BLOCK_CHAIN, -s, ip, -j, DROP], capture_outputTrue)-C返回 0 表示规则已存在直接跳过避免同一条规则堆几十遍。BLOCK_TTL配合一个后台清理线程到点自动解封防止误封永久生效。4.3 封禁队列与自动解封线程封禁动作不能阻塞抓包主循环否则高流量下会丢包。用一个队列把「待封 IP」交给独立线程处理同时记录封禁时间。import threading import queue block_queue queue.Queue() blocked {} # {ip: unblock_timestamp} def block_worker(): while True: ip block_queue.get() if block_ip(ip): blocked[ip] time.time() BLOCK_TTL block_queue.task_done() def unblock_worker(): while True: now time.time() for ip, expire in list(blocked.items()): if now expire: unblock_ip(ip) del blocked[ip] time.sleep(10) threading.Thread(targetblock_worker, daemonTrue).start() threading.Thread(targetunblock_worker, daemonTrue).start()两个线程都是daemon主程序退出时自动结束。unblock_worker每 10 秒扫一次到期列表粒度够用。blocked字典同时充当「当前封禁名单」排查时直接打印它就知道封了谁。4.4 主循环把抓包、检测、封禁串起来def main(): conn init_db() sock socket.socket(socket.AF_INET, socket.SOCK_RAW, socket.IPPROTO_TCP) while True: raw, _ sock.recvfrom(65535) pkt parse_packet(raw) if not pkt: continue flags pkt[flags] src_ip pkt[src_ip] # 端口扫描检测 is_scan, port_count detect_port_scan(src_ip, pkt[dst_port]) if is_scan: log_alert(conn, src_ip, port_scan, fports{port_count}, blocked) block_queue.put(src_ip) continue # SYN Flood 检测 is_flood, syn_count detect_syn_flood(src_ip, flags) if is_flood: log_alert(conn, src_ip, syn_flood, fsyn{syn_count}, blocked) block_queue.put(src_ip)主循环只做解析和判定封禁丢给队列保证抓包不卡。continue让一个包只触发一种告警避免同一 IP 被重复入队。注意iptables 规则默认重启后丢失。生产环境要么用iptables-save/iptables-restore持久化要么在启动脚本里重建IDS_BLOCK链。5. 避坑与排查这套方案最容易翻车的地方5.1 抓不到包recvfrom一直阻塞现象程序跑起来没报错但一个包都收不到。原因通常是绑定了错误的网卡或者服务器上有多个网卡流量走的是另一张。原始套接字默认收所有网卡但如果内核开了 rp_filter 或流量被网桥转发可能收不全。解决先用tcpdump -i any tcp确认有流量再检查程序是否真的以 root 运行。容器里跑要加--nethost和--cap-addNET_RAW否则权限被 namespace 隔离。5.2 误封内网扫描器或监控探针现象封了一批 IP结果发现是公司内部的漏洞扫描器或 Zabbix 探针。原因阈值太激进或者没做白名单。解决维护一个WHITELIST集合检测前先判断src_ip in WHITELIST直接跳过。内网网段、监控服务器、负载均衡健康检查地址都加进去。阈值也别一上来就封先跑observed模式记录几天看正常业务的基线再定。5.3 iptables 规则重复堆积现象iptables -L IDS_BLOCK -n里同一个 IP 出现几十条。原因封禁前没查重或者解封失败后又被重新封。解决block_ip里先用-C检查已存在就返回。解封时用-D删除如果返回非 0 说明规则已不在忽略即可。定期用脚本清理IDS_BLOCK链里超过 TTL 的规则和内存里的blocked字典对账。5.4 高流量下 Python 处理不过来现象流量一大检测延迟高甚至丢包。原因单线程 Python 解析每个包GIL 限制下吞吐有限。解决一是用socket.setsockopt调大接收缓冲区二是把检测逻辑拆到多进程按源 IP 哈希分流三是只对 SYN 包做深度检测其他标志位的包快速放行。真到万兆流量Python 方案就该让位给 C 或 eBPF别硬扛。5.5 封禁后业务方反馈访问不了现象正常用户被误封投诉到运维。原因用户和攻击者共用 NAT 出口 IP。解决封禁粒度从「源 IP」细化到「源 IP 目的端口」只封攻击针对的端口不影响其他服务。同时把BLOCK_TTL调短比如 600 秒误封影响面可控。告警表里记清楚封禁原因解封时有据可查。6. 让检测更准滑动窗口与阈值自适应的实战技巧固定阈值在真实环境里迟早会不够用。白天业务高峰的连接速率和凌晨完全不是一个量级用同一个SYN_RATE_THRESHOLD要么白天误报要么凌晨漏报。我后来改成滑动窗口加基线自适应维护最近 N 个窗口的 SYN 速率算出均值和标准差超过均值加三倍标准差才告警。import statistics from collections import deque class AdaptiveDetector: def __init__(self, window_size12, k3): self.history deque(maxlenwindow_size) # 历史窗口速率 self.k k def update(self, current_rate): if len(self.history) self.history.maxlen: self.history.append(current_rate) return False # 样本不足不判定 mean statistics.mean(self.history) std statistics.stdev(self.history) or 1 threshold mean self.k * std self.history.append(current_rate) return current_rate thresholdwindow_size12表示用最近 12 个统计窗口做基线k3是三倍标准差。样本不足时先积累不判定避免冷启动误报。std为 0 时给个默认值 1防止除零。这套逻辑对周期性流量很友好业务高峰基线自动抬高凌晨基线自动压低。另一个技巧是把端口扫描的判定从「端口集合大小」改成「端口熵值」。顺序扫描 1 到 1000 和随机扫描 1000 个端口集合大小一样但随机扫描的熵更高更像恶意行为。计算熵要引入math.log2对每个端口出现次数归一化后求和。熵值超过阈值再结合端口数判定能过滤掉一些批量健康检查的误报。验证检测效果别只看告警数量。我一般会准备两组流量一组用nmap -sS做半开扫描、hping3 --flood -S做 SYN Flood确认能触发另一组跑正常业务压测确认不误报。两组都过了阈值才算调好。这套方案的价值不在于多先进而在于每一行判定逻辑你都能看懂、能改、能对着自己的流量调。我踩过最深的坑就是一开始阈值拍脑袋定上线第一天封了半个内网后来老老实实跑了一周观察模式才敢开自动封禁。希望帮到你。本文还有配套的精品资源点击获取 Python实现TCP入侵检测系统：抓包解析与iptables联动封禁
简介这是一份面向网络安全初学者与高校学生的Python TCP入侵检测系统源码可直接用于毕业设计、期末大作业或课程设计。项目聚焦端口扫描与Dos攻击的实时检测并联动iptables实现自动防御帮助读者理解入侵检测与主机防火墙协同的基本思路。压缩包共6个文件以5个py源码文件为主分别承担流量分析、数据嗅探、过滤处理与数据库记录等职责另附1个md说明文档整体约5KB结构精简、便于阅读与二次修改。代码含详细注释新手也能看懂部署后即可运行。目前已有111人学习下载适合想快速搭建可演示安全项目、积累实战经验的同学参考借鉴。1. 用 Python 写一个能联动 iptables 的 TCP 入侵检测系统到底在做什么一台暴露在公网的 Linux 服务器平均上线后几分钟内就会收到第一波端口扫描。如果你只开了 22、80、443日志里却能看到有人在一秒内把 1 到 65535 全敲了一遍——这就是最典型的 TCP 入侵前奏。基于 Python 实现的 TCP 入侵检测系统核心目标就三件事抓包识别异常 TCP 行为、判定端口扫描和 DoS 攻击、联动 iptables 把攻击源封掉。它不依赖 Suricata、Snort 这类重型引擎而是用 Python 直接解析 TCP 报文把检测逻辑和防御动作串成一条闭环。适合有 Python 基础、想理解 IDS 内部判定逻辑、又需要一套能改能扩的轻量方案的运维和后台开发。下面从抓包原理讲到 iptables 联动每一步都能在你自己的机器上复现。2. 抓包与 TCP 解析Python 怎么拿到原始报文2.1 为什么选原始套接字而不是抓现成日志做 TCP 入侵检测第一道坎是数据从哪来。常见做法有三种读/var/log/下的应用日志、用 libpcap 抓包、直接开原始套接字raw socket。日志方案最省事但很多扫描和 DoS 根本不会在应用层留下记录比如半开扫描只发 SYN 不回 ACKWeb 服务压根不知道有人来过。libpcap 功能全但要多装一个依赖跨平台编译偶尔翻车。我一般会选原始套接字因为 Linux 下socket.AF_INET, socket.SOCK_RAW, socket.IPPROTO_TCP就能直接拿到 IP 层往上的 TCP 报文不依赖额外库代码可控。代价是必须用 root 或授予CAP_NET_RAW能力运行否则PermissionError。这一点在部署时要提前想清楚别等到线上才发现权限不够。2.2 手工解析 IP 头和 TCP 头原始套接字拿到的是完整的 IP 数据报需要自己按字节偏移拆出 IP 头和 TCP 头。IP 头固定 20 字节不含选项TCP 头也是 20 字节不含选项。关键字段的偏移必须记牢错一位后面全乱。import socket import struct def parse_packet(data): # IP 头前 20 字节 ip_header struct.unpack(!BBHHHBBH4s4s, data[:20]) ihl (ip_header[0] 0x0F) * 4 # 首部长度单位是 4 字节 protocol ip_header[6] # 6 表示 TCP src_ip socket.inet_ntoa(ip_header[8]) # 源 IP dst_ip socket.inet_ntoa(ip_header[9]) # 目的 IP if protocol ! 6: return None # TCP 头从 ihl 偏移开始固定 20 字节 tcp_start ihl tcp_header struct.unpack(!HHLLBBHHH, data[tcp_start:tcp_start 20]) src_port tcp_header[0] dst_port tcp_header[1] seq tcp_header[2] ack tcp_header[3] offset_flags tcp_header[4] flags tcp_header[5] data_offset (offset_flags 4) * 4 # TCP 首部长度 return { src_ip: src_ip, dst_ip: dst_ip, src_port: src_port, dst_port: dst_port, seq: seq, ack: ack, flags: flags, payload_offset: tcp_start data_offset }struct.unpack的格式串!BBHHHBBH4s4s对应 IP 头的字段顺序!表示网络字节序。ihl那一位同时存了版本号和首部长度用 0x0F取低四位再乘 4 才是真实字节数。TCP 的offset_flags高四位是数据偏移低六位才是标志位很多人第一次写会把这两个搞混导致 SYN、ACK 判断全错。2.3 标志位判定SYN、ACK、RST 各代表什么TCP 三次握手是理解扫描检测的基础。客户端发 SYN服务端回 SYNACK客户端再回 ACK。标志位是 8 位里的低 6 位FIN0x01、SYN0x02、RST0x04、PSH0x08、ACK0x10、URG0x20。SYN 0x02 ACK 0x10 RST 0x04 FIN 0x01 def get_flags(flags): return { syn: bool(flags SYN), ack: bool(flags ACK), rst: bool(flags RST), fin: bool(flags FIN), }判断一个包是不是「只发 SYN 不回 ACK」就是syn and not ack。这是半开扫描SYN scan的指纹。正常握手第一个包也是 SYN 无 ACK区别在于正常客户端收到 SYNACK 后会补一个 ACK而扫描器往往直接换下一个端口。所以单看一个包不够必须按源 IP 聚合一段时间窗口内的行为这就引出下一章的检测逻辑。提示原始套接字在 Windows 上行为不一致这套方案建议跑在 Linux 上内核版本不影响Python 3.6 以上即可。3. 端口扫描与 DoS 的检测逻辑怎么写3.1 端口扫描按源 IP 聚合短时间内的目标端口数端口扫描的本质特征不是单个包而是「同一个源 IP 在很短时间内访问了大量不同的目的端口」。所以检测器要维护一张表{源IP: {时间窗口内的端口集合}}。每来一个 SYN 包把目的端口塞进去然后看集合大小有没有超过阈值。import time from collections import defaultdict SCAN_WINDOW 10 # 时间窗口秒 SCAN_PORT_THRESHOLD 20 # 窗口内不同端口数阈值 scan_tracker defaultdict(lambda: {ports: set(), first_seen: time.time()}) def detect_port_scan(src_ip, dst_port): now time.time() entry scan_tracker[src_ip] # 窗口过期就重置 if now - entry[first_seen] SCAN_WINDOW: entry[ports] set() entry[first_seen] now entry[ports].add(dst_port) if len(entry[ports]) SCAN_PORT_THRESHOLD: return True, len(entry[ports]) return False, len(entry[ports])SCAN_WINDOW和SCAN_PORT_THRESHOLD是最需要按环境调的两个参数。内网做资产扫描的合法工具也会触发所以阈值不能太低。公网服务器上10 秒内超过 20 个不同端口基本可以判定为扫描如果业务本身有大量短连接比如压测环境就要把窗口拉长或阈值调高。判断为扫描后不要立刻封先记录连续多个窗口都超阈值再动手能有效降低误封。3.2 DoS 检测SYN Flood 的连接速率与半开连接堆积DoS 里最常见的是 SYN Flood攻击者发大量 SYN 但不完成握手服务端为每个半开连接分配资源最终耗尽。检测思路是统计单位时间内来自同一源 IP 的 SYN 包数量以及「只发 SYN 不回 ACK」的比例。SYN_RATE_WINDOW 5 # 统计窗口秒 SYN_RATE_THRESHOLD 100 # 窗口内 SYN 包数阈值 syn_tracker defaultdict(lambda: {count: 0, first_seen: time.time()}) def detect_syn_flood(src_ip, flags): now time.time() entry syn_tracker[src_ip] if now - entry[first_seen] SYN_RATE_WINDOW: entry[count] 0 entry[first_seen] now if flags SYN and not (flags ACK): entry[count] 1 if entry[count] SYN_RATE_THRESHOLD: return True, entry[count] return False, entry[count]SYN_RATE_THRESHOLD设 100 意味着 5 秒内同一源 IP 发超过 100 个 SYN 就告警。真实用户不可能有这个速率但 NAT 出口后面的整栋楼可能共享一个公网 IP这时候阈值要放宽或者改成按「源 IP 目的端口」维度统计。SYN Flood 和端口扫描的判定可以共用同一份抓包数据只是聚合维度不同一个看端口集合一个看包计数。3.3 把检测结果落库别只 print检测到异常后最忌讳只print一行日志。你需要留证据谁、什么时候、什么行为、封了多久。用 SQLite 就够不引入额外服务。import sqlite3 def init_db(pathids.db): conn sqlite3.connect(path) conn.execute(CREATE TABLE IF NOT EXISTS alerts ( id INTEGER PRIMARY KEY AUTOINCREMENT, ts REAL, src_ip TEXT, attack_type TEXT, detail TEXT, action TEXT)) conn.commit() return conn def log_alert(conn, src_ip, attack_type, detail, action): conn.execute( INSERT INTO alerts (ts, src_ip, attack_type, detail, action) VALUES (?,?,?,?,?), (time.time(), src_ip, attack_type, detail, action)) conn.commit()attack_type存port_scan或syn_floodaction存blocked或observed。这张表后面排查误封时是唯一的后悔药别省。4. 联动 iptables从检测到封禁的完整链路4.1 用独立链而不是直接往 INPUT 里塞规则直接iptables -A INPUT -s x.x.x.x -j DROP能封但规则越加越多管理混乱解封也麻烦。正确做法是建一条自定义链比如IDS_BLOCK在 INPUT 里引用它所有封禁规则都加到这条链上。# 创建自定义链已存在会报错可忽略 iptables -N IDS_BLOCK # 让 INPUT 流量先经过这条链 iptables -C INPUT -j IDS_BLOCK 2/dev/null || iptables -I INPUT -j IDS_BLOCK-C是检查规则是否存在配合||保证重复执行不会插入多条。这一步建议写进系统启动脚本机器重启后链还在但规则需要重新加载。4.2 Python 调用 iptables 封禁与解封封禁就是把源 IP 加进IDS_BLOCK链解封就是删掉。用subprocess调用注意参数用列表传别拼字符串避免命令注入。import subprocess import time BLOCK_CHAIN IDS_BLOCK BLOCK_TTL 3600 # 封禁时长秒 def block_ip(ip): # 先查是否已封避免重复 check subprocess.run( [iptables, -C, BLOCK_CHAIN, -s, ip, -j, DROP], capture_outputTrue) if check.returncode 0: return False subprocess.run( [iptables, -A, BLOCK_CHAIN, -s, ip, -j, DROP], checkTrue) return True def unblock_ip(ip): subprocess.run( [iptables, -D, BLOCK_CHAIN, -s, ip, -j, DROP], capture_outputTrue)-C返回 0 表示规则已存在直接跳过避免同一条规则堆几十遍。BLOCK_TTL配合一个后台清理线程到点自动解封防止误封永久生效。4.3 封禁队列与自动解封线程封禁动作不能阻塞抓包主循环否则高流量下会丢包。用一个队列把「待封 IP」交给独立线程处理同时记录封禁时间。import threading import queue block_queue queue.Queue() blocked {} # {ip: unblock_timestamp} def block_worker(): while True: ip block_queue.get() if block_ip(ip): blocked[ip] time.time() BLOCK_TTL block_queue.task_done() def unblock_worker(): while True: now time.time() for ip, expire in list(blocked.items()): if now expire: unblock_ip(ip) del blocked[ip] time.sleep(10) threading.Thread(targetblock_worker, daemonTrue).start() threading.Thread(targetunblock_worker, daemonTrue).start()两个线程都是daemon主程序退出时自动结束。unblock_worker每 10 秒扫一次到期列表粒度够用。blocked字典同时充当「当前封禁名单」排查时直接打印它就知道封了谁。4.4 主循环把抓包、检测、封禁串起来def main(): conn init_db() sock socket.socket(socket.AF_INET, socket.SOCK_RAW, socket.IPPROTO_TCP) while True: raw, _ sock.recvfrom(65535) pkt parse_packet(raw) if not pkt: continue flags pkt[flags] src_ip pkt[src_ip] # 端口扫描检测 is_scan, port_count detect_port_scan(src_ip, pkt[dst_port]) if is_scan: log_alert(conn, src_ip, port_scan, fports{port_count}, blocked) block_queue.put(src_ip) continue # SYN Flood 检测 is_flood, syn_count detect_syn_flood(src_ip, flags) if is_flood: log_alert(conn, src_ip, syn_flood, fsyn{syn_count}, blocked) block_queue.put(src_ip)主循环只做解析和判定封禁丢给队列保证抓包不卡。continue让一个包只触发一种告警避免同一 IP 被重复入队。注意iptables 规则默认重启后丢失。生产环境要么用iptables-save/iptables-restore持久化要么在启动脚本里重建IDS_BLOCK链。5. 避坑与排查这套方案最容易翻车的地方5.1 抓不到包recvfrom一直阻塞现象程序跑起来没报错但一个包都收不到。原因通常是绑定了错误的网卡或者服务器上有多个网卡流量走的是另一张。原始套接字默认收所有网卡但如果内核开了 rp_filter 或流量被网桥转发可能收不全。解决先用tcpdump -i any tcp确认有流量再检查程序是否真的以 root 运行。容器里跑要加--nethost和--cap-addNET_RAW否则权限被 namespace 隔离。5.2 误封内网扫描器或监控探针现象封了一批 IP结果发现是公司内部的漏洞扫描器或 Zabbix 探针。原因阈值太激进或者没做白名单。解决维护一个WHITELIST集合检测前先判断src_ip in WHITELIST直接跳过。内网网段、监控服务器、负载均衡健康检查地址都加进去。阈值也别一上来就封先跑observed模式记录几天看正常业务的基线再定。5.3 iptables 规则重复堆积现象iptables -L IDS_BLOCK -n里同一个 IP 出现几十条。原因封禁前没查重或者解封失败后又被重新封。解决block_ip里先用-C检查已存在就返回。解封时用-D删除如果返回非 0 说明规则已不在忽略即可。定期用脚本清理IDS_BLOCK链里超过 TTL 的规则和内存里的blocked字典对账。5.4 高流量下 Python 处理不过来现象流量一大检测延迟高甚至丢包。原因单线程 Python 解析每个包GIL 限制下吞吐有限。解决一是用socket.setsockopt调大接收缓冲区二是把检测逻辑拆到多进程按源 IP 哈希分流三是只对 SYN 包做深度检测其他标志位的包快速放行。真到万兆流量Python 方案就该让位给 C 或 eBPF别硬扛。5.5 封禁后业务方反馈访问不了现象正常用户被误封投诉到运维。原因用户和攻击者共用 NAT 出口 IP。解决封禁粒度从「源 IP」细化到「源 IP 目的端口」只封攻击针对的端口不影响其他服务。同时把BLOCK_TTL调短比如 600 秒误封影响面可控。告警表里记清楚封禁原因解封时有据可查。6. 让检测更准滑动窗口与阈值自适应的实战技巧固定阈值在真实环境里迟早会不够用。白天业务高峰的连接速率和凌晨完全不是一个量级用同一个SYN_RATE_THRESHOLD要么白天误报要么凌晨漏报。我后来改成滑动窗口加基线自适应维护最近 N 个窗口的 SYN 速率算出均值和标准差超过均值加三倍标准差才告警。import statistics from collections import deque class AdaptiveDetector: def __init__(self, window_size12, k3): self.history deque(maxlenwindow_size) # 历史窗口速率 self.k k def update(self, current_rate): if len(self.history) self.history.maxlen: self.history.append(current_rate) return False # 样本不足不判定 mean statistics.mean(self.history) std statistics.stdev(self.history) or 1 threshold mean self.k * std self.history.append(current_rate) return current_rate thresholdwindow_size12表示用最近 12 个统计窗口做基线k3是三倍标准差。样本不足时先积累不判定避免冷启动误报。std为 0 时给个默认值 1防止除零。这套逻辑对周期性流量很友好业务高峰基线自动抬高凌晨基线自动压低。另一个技巧是把端口扫描的判定从「端口集合大小」改成「端口熵值」。顺序扫描 1 到 1000 和随机扫描 1000 个端口集合大小一样但随机扫描的熵更高更像恶意行为。计算熵要引入math.log2对每个端口出现次数归一化后求和。熵值超过阈值再结合端口数判定能过滤掉一些批量健康检查的误报。验证检测效果别只看告警数量。我一般会准备两组流量一组用nmap -sS做半开扫描、hping3 --flood -S做 SYN Flood确认能触发另一组跑正常业务压测确认不误报。两组都过了阈值才算调好。这套方案的价值不在于多先进而在于每一行判定逻辑你都能看懂、能改、能对着自己的流量调。我踩过最深的坑就是一开始阈值拍脑袋定上线第一天封了半个内网后来老老实实跑了一周观察模式才敢开自动封禁。希望帮到你。本文还有配套的精品资源点击获取