如何使用Python检测TCP Flags，实现TCP Reset攻击识别

前置依赖 #

实现需要用到网络报文处理库scapy，先执行安装命令：
pip install scapy

注意：抓包操作需要高权限，Linux/macOS下执行脚本需要加sudo前缀，Windows下需要以管理员身份运行终端。

实现逻辑 #

• 捕获服务端入站TCP报文，判断TCP Flags字段的RST位是否为1，scapy中可直接通过'R' in tcp_flags快速判断是否为RST报文
• 按客户端源IP维度统计RST报文数量，设置时间窗口和阈值，比如10秒内同一IP发送超过5个RST报文即可判定为疑似RST攻击
• 定期清理过期的RST记录，避免内存持续占用

完整实现代码 #

from scapy.all import sniff, TCP, IP
from collections import defaultdict
import time

# 自定义配置项
RST_THRESHOLD = 5  # 同一IP时间窗口内允许的最大RST报文数量
TIME_WINDOW = 10   # 统计时间窗口，单位：秒

# 存储结构：key为源IP，value为对应IP发送RST报文的时间戳列表
ip_rst_records = defaultdict(list)

def rst_attack_detect_callback(packet):
    # 过滤非TCP/IP报文
    if IP not in packet or TCP not in packet:
        return
    src_ip = packet[IP].src
    tcp_flags = packet[TCP].flags
    
    # 判断是否为RST报文
    if 'R' in tcp_flags:
        current_time = time.time()
        # 清理当前IP的过期RST记录
        ip_rst_records[src_ip] = [t for t in ip_rst_records[src_ip] if current_time - t < TIME_WINDOW]
        # 添加新的RST记录
        ip_rst_records[src_ip].append(current_time)
        
        # 检查是否触发攻击阈值
        rst_count = len(ip_rst_records[src_ip])
        if rst_count >= RST_THRESHOLD:
            print(f"[告警] 疑似TCP Reset攻击，来源IP：{src_ip}，{TIME_WINDOW}秒内收到RST报文数：{rst_count}")
            # 此处可扩展自定义逻辑：写入日志、调用防火墙接口拉黑IP等

if __name__ == "__main__":
    print(f"[启动] 流量监听中，检测规则：{TIME_WINDOW}秒内同一IP发送{RST_THRESHOLD}个以上RST报文触发告警")
    # iface参数可指定监听网卡，比如eth0、WLAN等，填None表示监听所有网卡
    # filter为BPF过滤规则，只捕获TCP报文可大幅提升处理效率
    sniff(prn=rst_attack_detect_callback, filter="tcp", store=0, iface=None)

扩展说明 #

• 如果只需要检测特定端口的RST报文，可把filter参数修改为tcp and dst port 你的端口号，比如tcp and dst port 443
• 高并发业务场景可根据实际情况调高阈值，避免正常业务场景下的RST报文触发误告警
• 生产环境长时间运行建议添加日志持久化、异常IP自动封禁能力，配合防火墙使用效果更好

内容的提问来源于stack exchange，提问作者mmt