如何使用Scapy直接从TCP Socket捕获数据包

答案是不能，原因和替代方案如下：

核心原因 #

Scapy的sniff()函数是在网卡层面捕获原始数据包，工作在数据链路层/网络层，它抓取的是未经操作系统TCP栈处理的完整报文（包含IP头、TCP头等底层协议信息）。而你用socket模块创建的TCP Socket是操作系统提供的传输层抽象，已经帮你处理了TCP握手、分片重组、流量控制、重传等逻辑，Socket中能读取到的是已经剥离底层协议头的应用层数据。两者处于不同的网络层级，无法直接关联。

你给出的示例代码只是在accept()后循环调用网卡抓包，本质和之前的实现没有区别——仍然是抓取该端口的所有数据包，无法精准对应到当前建立的Socket连接，还可能出现同一个端口多个连接时抓错包的问题。

替代方案 #

1. 精准过滤特定Socket连接的数据包 #

如果需要获取当前Socket对应连接的原始数据包，可以基于accept()返回的客户端地址（addr），在sniff()的过滤规则中增加源IP和源端口的限制，确保只抓取该特定连接的报文：

import socket
from scapy.all import sniff

HOST = '0.0.0.0'
PORT = 12345

with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as server_sock:
    server_sock.bind((HOST, PORT))
    server_sock.listen()

    while True:
        sock, addr = server_sock.accept()
        # 过滤规则：端口匹配 + 客户端IP和端口匹配
        filter_rule = f"port {PORT} and host {addr[0]} and port {addr[1]}"
        print(f"开始捕获来自 {addr} 的数据包")
        while True:
            # 只抓该连接的数据包
            packets = sniff(iface='eth0', filter=filter_rule, count=1)
            # 处理数据包
            print(packets[0].summary())

2. 直接读取Socket数据并结合Scapy处理 #

如果不需要原始数据包，只是想用Scapy处理Socket收到的应用层数据，可以直接从Socket读取数据，再封装成Scapy的协议包结构：

import socket
from scapy.all import IP, TCP

HOST = '0.0.0.0'
PORT = 12345

with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as server_sock:
    server_sock.bind((HOST, PORT))
    server_sock.listen()

    while True:
        sock, addr = server_sock.accept()
        with sock:
            print(f"已连接客户端 {addr}")
            while True:
                # 从Socket读取应用层数据
                raw_data = sock.recv(4096)
                if not raw_data:
                    break
                # 将数据封装为Scapy的IP/TCP包（用于后续分析）
                packet = IP(dst=addr[0], src=HOST)/TCP(dport=addr[1], sport=PORT)/raw_data
                print(f"处理数据：{packet.summary()}")

3. 用Scapy直接实现TCP监听（进阶） #

如果需要完全控制TCP交互过程并捕获原始数据包，可以直接用Scapy代替socket模块，手动处理TCP握手和数据传输，但这需要自己实现TCP状态机，适合深度定制场景：

from scapy.all import sniff, IP, TCP, send

PORT = 12345
connections = {}  # 维护连接状态：(src_ip, src_port) -> seq/ack

def handle_packet(packet):
    if not packet.haslayer(TCP):
        return
    src_ip = packet[IP].src
    src_port = packet[TCP].sport
    dst_port = packet[TCP].dport

    if dst_port != PORT:
        return

    # 处理SYN包（建立连接请求）
    if packet[TCP].flags == 'S':
        seq_num = 1000
        ack_num = packet[TCP].seq + 1
        # 发送SYN-ACK
        syn_ack = IP(dst=src_ip)/TCP(dport=src_port, sport=PORT, flags='SA', seq=seq_num, ack=ack_num)
        send(syn_ack, verbose=0)
        connections[(src_ip, src_port)] = {'seq': seq_num + 1, 'ack': ack_num}
        print(f"已响应 {src_ip}:{src_port} 的连接请求")
    # 处理ACK包（连接建立确认）
    elif packet[TCP].flags == 'A' and (src_ip, src_port) in connections:
        print(f"与 {src_ip}:{src_port} 的连接已建立")
    # 处理带数据的包
    elif packet[TCP].flags & 0x08 and (src_ip, src_port) in connections:
        data = packet[TCP].payload.load
        print(f"收到 {src_ip}:{src_port} 的数据：{data}")
        # 发送ACK确认
        ack = IP(dst=src_ip)/TCP(dport=src_port, sport=PORT, flags='A', seq=connections[(src_ip, src_port)]['seq'], ack=packet[TCP].seq + len(data))
        send(ack, verbose=0)

sniff(iface='eth0', filter=f"port {PORT}", prn=handle_packet)

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