python-3.x - 来自字符串的 Python3 Scapy 自定义负载

标签 python-3.x network-programming scapy

介绍
我正在使用 python3 和 scapy 修改和重新发送捕获的数据包。
问题
当我在 Raw 部分发送修改后的有效负载时,消息的\r 和\n 部分不会被解释为特殊字符。相反,它们作为字符串包含在下面的 cmd 和 wireshark 屏幕截图中。
当前有效载荷
Cmd wrong payload screenshot
Wireshark wrong payload screenshot
预期有效载荷
这是在网络上捕获的正确数据包格式。
Wireshark correct payload description
在wireshark中显示的数据包如下:

SIP/2.0 486 Busy Here Via: SIP/2.0/UDP 192.168.1.5:5060;branch=z9hG4bK226016822;received=192.168.1.5;rport=5060 From: sip:301@192.168.1.2;tag=2032604445 To: sip:300@192.168.1.2;tag=as1b0290be Call-ID: 338695025 CSeq: 21 INVITE Server: Asterisk PBX 16.10.0 Allow: INVITE, ACK, CANCEL, OPTIONS, BYE, REFER, SUBSCRIBE, NOTIFY, INFO, PUBLISH, MESSAGE Supported: replaces, timer X-Asterisk-HangupCause: Call Rejected X-Asterisk-HangupCauseCode: 21 Content-Length: 0


编码
这是到目前为止的代码:
from scapy.all import sniff, Ether, IP, UDP, sendp, ICMP, rdpcap,Raw
import scapy.fields
import re
import codecs
import argparse

def traffic_parser(packet):
    BUSY_1 = 'SIP/2.0 486 Busy Here'
    BUSY_2 = 'X-Asterisk-HangupCause: Call Rejected\\\\r\\\\nX-Asterisk-HangupCauseCode: 21'

    payload = packet[3].command()
    print(bytes(payload))

    header=re.findall("Ringing", payload)
    if header:
        
        eth_attributes={}
        eth_attributes['dst']=packet[0].dst
        eth_attributes['src']=packet[0].src
        eth_attributes['type']=packet[0].type
        
        eth = Ether_layer(eth_attributes)


        udp_attributes={}
        udp_attributes['sport']=packet[2].sport
        udp_attributes['dport']=packet[2].dport
        udp_attributes['len']=444
    
        udp = UDP_layer(udp_attributes)

        # Implement packet modification
        payload = payload.replace("SIP/2.0 180 Ringing", BUSY_1, 1)
        payload = re.sub("Contact\:.*>", BUSY_2, payload,1)
        payload = payload.replace("Raw(load=b\'", '', 1)
        payload = re.sub("\'\)$", '', payload, 1)

        for incr in range(1,5):
            ip_attributes={}
            ip_attributes['version']=packet[1].version
            ip_attributes['tos']=packet[1].tos
            ip_attributes['len']=464 
            ip_attributes['id']=0 #Zero is handled by scapy on send by default
            ip_attributes['flags']=packet[1].flags
            ip_attributes['frag']=packet[1].frag
            ip_attributes['ttl']=packet[1].ttl
            ip_attributes['proto']=packet[1].proto
            ip_attributes['src']=packet[1].src
            ip_attributes['dst']=packet[1].dst

            ip = IP_layer(ip_attributes)

            sendp(eth/ip/udp/Raw(load=payload))

            print(payload)
            print(Raw(load=payload))
            print("\n")

def Ether_layer(attributes):
    layer2=Ether()
    layer2.dst=attributes['dst']
    layer2.src=attributes['src']
    layer2.type=attributes['type']

    return layer2


def IP_layer(attributes):
    layer3=IP()
    layer3.version=attributes['version']
    layer3.tos=attributes['tos']
    layer3.len=attributes['len']
    layer3.id=attributes['id']
    layer3.flags=attributes['flags']
    layer3.frag=attributes['frag']
    layer3.ttl=attributes['ttl']
    layer3.proto=attributes['proto']
    layer3.src=attributes['src']
    layer3.dst=attributes['dst']

    return layer3


def UDP_layer(attributes):
    layer4=UDP()
    layer4.sport=attributes['sport']
    layer4.dport=attributes['dport']
    layer4.len=attributes['len']

    return layer4


parser = argparse.ArgumentParser(description="rtp replay script. Arguments: -i <interface> -f <sniff filter> -o <sniff outputfile> Interface defaults to 'eth0' and filter defaults to 'udp and port 5060'")
parser.add_argument('-i', "--interface", default="eth0", help="interface to use for sniffing")
parser.add_argument('-f', '--filter', default="udp and port 5060", help="filter to be used in scapy")
parser.add_argument('-o', "--outfile", help="output file (optional)")
parser.add_argument('-t', "--testfile", help="parse test file (optional)")
args=parser.parse_args()

if __name__ == '__main__':

    if args.testfile:
        packets = rdpcap(args.testfile)
        for packet in packets:
            traffic_parser(packet)
    else:
        sniff(iface=args.interface, prn=traffic_parser, filter="udp and port 5060", store=0)

如何以所需的形式附加有效载荷?
编辑:
此 pcap 文件可用于测试。 pcap_file
指示性脚本执行:
须藤 python byespam.py -t <文件名.pcapng>

最佳答案

为了能够将有效负载用作字符串,您需要将其转换为一个字符串,并确保您拥有正确的编码。
在您的情况下,您应该:

  • 您应该只从 scapy.packet.Packet 获取 Raw 的加载部分。
  • 将负载部分编码为 UTF-8,以便 Python 正确解释特殊字符。

  • 以下是适用于您的案例的一些工作代码:
    payload = packet[UDP].payload.load *
    payload = payload.decode("utf-8")
    
    然后你可以打印有效载荷,它会被“正确”解释。
    * 我使用 UDP 是因为在你的脚本和 pcapng 文件中你只有 UDP 数据包。如果有 TCP 数据包,您应该使用 payload = packet[TCP].payload.load

    关于python-3.x - 来自字符串的 Python3 Scapy 自定义负载,我们在Stack Overflow上找到一个类似的问题: https://stackoverflow.com/questions/62642175/

    相关文章:

    Python3 使用 BeautifulSoup 在 html 中查找属性值

    c# - 使用 MixedRealityToolkit-Unity 在自定义消息中发送字节数组

    c - 如何使用 SOCK_DGRAM 制作双向 unix 域套接字?

    javascript - 如何在服务器发送的 js 文件到达我的浏览器之前对其进行编辑?

    Scapy:属性错误: 'Tuple' 对象没有属性 'x'

    python - 比较 2 个数据框并找到匹配的行

    python - 使 __str__ 或 __repr__ 以字符串形式返回列表

    python - 当多个子进程使用 `queue.Queue`进行访问时, `concurrent.futures.ProcessPoolExecutor`是线程安全的吗?

    java - 实现网络协议(protocol) - 关于设计和性能的思考

    python - OSX 上的 Scapy Sniff 段错误