【发布时间】:2012-06-05 18:29:04
【问题描述】:
我正在寻找一种工具来记录和重放 TCP 流的一侧以进行测试。 我看到用于测试防火墙等的记录整个 TCP 流(服务器和客户端)的工具,但我正在寻找的是一个工具,它只记录客户端提交的流量(带有时间信息)然后重新提交它到服务器进行测试。
【问题讨论】:
标签: testing networking tcp network-programming
我正在寻找一种工具来记录和重放 TCP 流的一侧以进行测试。 我看到用于测试防火墙等的记录整个 TCP 流(服务器和客户端)的工具,但我正在寻找的是一个工具,它只记录客户端提交的流量(带有时间信息)然后重新提交它到服务器进行测试。
【问题讨论】:
标签: testing networking tcp network-programming
由于 TCP 处理重传、序列号、SACK 和窗口化的方式,这可能比您想象的更困难。
通常人们使用tcpreplay 进行数据包重放;但是,它doesn't support synchronizing TCP sequence numbers。由于您需要双向 TCP 流(这需要同步 seq 编号),因此请使用以下选项之一:
如果这是一个交互性很强的客户端/服务器协议,您可以使用scapy 去除流的 TCP 内容,解析时间和交互性。接下来使用此信息,打开一个新的 TCP 套接字到您的服务器并将该数据反序列化到新的 TCP 套接字中。如果遇到 TCP 重新传输和窗口动态,使用 scapy 解析原始流可能会很棘手。将字节写入新的 TCP 套接字不需要自己处理序列号......操作系统会处理这些。
如果这是一个简单的流并且您可以不使用计时(或想要手动插入计时信息),您可以使用 wireshark 从 TCP 流中获取原始字节而无需担心关于使用scapy 进行解析。获得原始字节后,将这些字节写入新的 TCP 套接字(根据需要考虑交互性)。将字节写入新的 TCP 套接字不需要自己处理序列号......操作系统会处理这些。
如果您的流是严格的文本(但不是 html 或 xml)命令,例如 telnet 会话,则类似Expect 的解决方案可能比上述解析更容易。在此解决方案中,您不会直接从您的代码中打开 TCP 套接字,使用期望 spawn 一个 telnet(或其他)会话并使用 send / expect 重播文本命令。您期望的库/底层操作系统将负责序列编号。
如果您正在测试 Web 服务,我猜想模拟真正的 Web 客户端单击带有 Selenium 或 Splinter 的链接会容易得多。您的 http 库/底层操作系统将负责新流中的序列编号。
【讨论】:
看看 WirePlay code.google.com/p/wireplay 或 github.com/abhisek/wireplay,它承诺重播捕获的 TCP 会话的客户端或服务器端,并根据需要修改所有 SYN/ACK 序列号。
我不知道是否有可用的二进制构建,您需要自己编译。
请注意,我自己还没有尝试过,但正在研究它。
【讨论】:
是的,实现这样的工具是一项艰巨的任务。 两年前我开始实现这种工具,现在工具已经成熟。 试试看,也许你会发现它正是你要找的工具。
【讨论】:
我想要类似的东西,所以我与 scapy 合作了一段时间,并提出了一个适合我的解决方案。我的目标是重放捕获的 pcap 文件的客户端部分。我有兴趣从服务器获得响应——不一定是时间。下面是我的 scapy 解决方案 - 它绝不是经过测试或完成的,但它完成了我想要它做的事情。希望这是一个如何使用 scapy 重放 TCP 流的好例子。
from scapy.all import *
import sys
#NOTE - This script assumes that there is only 1 TCP stream in the PCAP file and that
# you wish to replay the role of the client
#acks
ACK = 0x10
#client closing the connection
RSTACK = 0x14
def replay(infile, inface):
recvSeqNum = 0
first = True
targetIp = None
#send will put the correct src ip and mac in
#this assumes that the client portion of the stream is being replayed
for p in rdpcap(infile):
if 'IP' in p and 'TCP' in p:
ip = p[IP]
eth = p[Ether]
tcp = p[TCP]
if targetIp == None:
#figure out the target ip we're interested in
targetIp = ip.dst
print(targetIp)
elif ip.dst != targetIp:
# don't replay a packet that isn't to our target ip
continue
# delete checksums so that they are recalculated
del ip.chksum
del tcp.chksum
if tcp.flags == ACK or tcp.flags == RSTACK:
tcp.ack = recvSeqNum+1
if first or tcp.flags == RSTACK:
# don't expect a response from these
sendp(p, iface=inface)
first=False
continue
rcv = srp1(p, iface=inface)
recvSeqNum = rcv[TCP].seq
def printUsage(prog):
print("%s <pcapPath> <interface>" % prog)
if __name__ == "__main__":
if 3 != len(sys.argv):
printUsage(sys.argv[0])
exit(1)
replay(sys.argv[1], sys.argv[2])
【讨论】:
iptables -A OUTPUT -p tcp --tcp-flags RST RST -o $INTERFACE -j DROP
记录完整 TCP 客户端/服务器通信的数据包捕获。然后,您可以使用tcpliveplay 将通信的客户端重播到真实服务器。 tcpliveplay 会生成新的序列号、IP 地址、MAC 地址等,所以通信会正常进行。
【讨论】: