Linux Blog

神奇的发包工具scapy

2019-03-25

之前就写过好几次 scapy 。每次需要自己构造报文的时候都会发现： 它好强大 。

调试VLAN

测试协议vlan的时候，需要构造一些二层报文。刚开始同事推荐了一下ipop。试用了一下老是发不出报来。于是在网上搜到了 scapy 。

原来在测试igmp的时候，我就用过这货。当时还记录在了 2018-03-26-Linux下的工具。

当时完成写在python文件中的，这次我查了一下文档 https://scapy.readthedocs.io 。可以有好几种方式来使用它。

直接运行scapy，然后一行一行地运行其实是很方便的，比如文档的中的例子：

>>> a=IP(ttl=10)
>>> a
< IP ttl=10 |>
>>> a.src
’127.0.0.1’
>>> a.dst="192.168.1.1"
>>> a
< IP ttl=10 dst=192.168.1.1 |>
>>> a.src
’192.168.8.14’
>>> del(a.ttl)
>>> a
< IP dst=192.168.1.1 |>
>>> a.ttl
64

发送报文也很简单：

>>> send(IP(dst="1.2.3.4")/ICMP())
.
Sent 1 packets.
>>> sendp(Ether()/IP(dst="1.2.3.4",ttl=(1,4)), iface="eth1")
....
Sent 4 packets.
>>> sendp("I'm travelling on Ethernet", iface="eth1", loop=1, inter=0.2)
................^C
Sent 16 packets.
>>> sendp(rdpcap("/tmp/pcapfile")) # tcpreplay
...........
Sent 11 packets.

Returns packets sent by send()
>>> send(IP(dst='127.0.0.1'), return_packets=True)
.
Sent 1 packets.
<PacketList: TCP:0 UDP:0 ICMP:0 Other:1>

我写了一个简单地python的例子，这个例子中，主要是修改了ether_type字段，也修改了每个报文的mac地址，这是为了方便我调试协议VLAN：

#!/usr/bin/python

from scapy.all import *
from scapy.contrib.igmp import IGMP
import random

sendp(Ether(src="f8:09:23:00:00:01", dst="f8:07:24:00:00:01", type=0x0801)/IP(dst="192.168.30.2", ttl=(1, 1)), iface="eth0", inter=0.2)
sendp(Ether(src="f8:09:23:00:00:02", dst="f8:07:24:00:00:01", type=0x0802)/IP(dst="192.168.30.2", ttl=(1, 1)), iface="eth0", inter=0.2)
sendp(Ether(src="f8:09:23:00:00:03", dst="f8:07:24:00:00:01", type=0x0803)/IP(dst="192.168.30.2", ttl=(1, 1)), iface="eth0", inter=0.2)
sendp(Ether(src="f8:09:23:00:00:04", dst="f8:07:24:00:00:01", type=0x0804)/IP(dst="192.168.30.2", ttl=(1, 1)), iface="eth0", inter=0.2)
sendp(Ether(src="f8:09:23:00:00:05", dst="f8:07:24:00:00:01", type=0x0805)/IP(dst="192.168.30.2", ttl=(1, 1)), iface="eth0", inter=0.2)
sendp(Ether(src="f8:09:23:00:00:06", dst="f8:07:24:00:00:01", type=0x0806)/IP(dst="192.168.30.2", ttl=(1, 1)), iface="eth0", inter=0.2)
sendp(Ether(src="f8:09:23:00:00:07", dst="f8:07:24:00:00:01", type=0x0807)/IP(dst="192.168.30.2", ttl=(1, 1)), iface="eth0", inter=0.2)
sendp(Ether(src="f8:09:23:00:00:08", dst="f8:07:24:00:00:01", type=0x0808)/IP(dst="192.168.30.2", ttl=(1, 1)), iface="eth0", inter=0.2)
sendp(Ether(src="f8:09:23:00:00:09", dst="f8:07:24:00:00:01", type=0x0809)/IP(dst="192.168.30.2", ttl=(1, 1)), iface="eth0", inter=0.2)
sendp(Ether(src="f8:09:23:00:00:0a", dst="f8:07:24:00:00:01", type=0x080a)/IP(dst="192.168.30.2", ttl=(1, 1)), iface="eth0", inter=0.2)
sendp(Ether(src="f8:09:23:00:00:0b", dst="f8:07:24:00:00:01", type=0x080b)/IP(dst="192.168.30.2", ttl=(1, 1)), iface="eth0", inter=0.2)
sendp(Ether(src="f8:09:23:00:00:0c", dst="f8:07:24:00:00:01", type=0x080c)/IP(dst="192.168.30.2", ttl=(1, 1)), iface="eth0", inter=0.2)
sendp(Ether(src="f8:09:23:00:00:0d", dst="f8:07:24:00:00:01", type=0x080d)/IP(dst="192.168.30.2", ttl=(1, 1)), iface="eth0", inter=0.2)
sendp(Ether(src="f8:09:23:00:00:0e", dst="f8:07:24:00:00:01", type=0x080e)/IP(dst="192.168.30.2", ttl=(1, 1)), iface="eth0", inter=0.2)
sendp(Ether(src="f8:09:23:00:00:0f", dst="f8:07:24:00:00:01", type=0x080f)/IP(dst="192.168.30.2", ttl=(1, 1)), iface="eth0", inter=0.2)
sendp(Ether(src="f8:09:23:00:00:10", dst="f8:07:24:00:00:01", type=0x0810)/IP(dst="192.168.30.2", ttl=(1, 1)), iface="eth0", inter=0.2)

后面可以在这上面再做定制。

调试IGMP

发送report报文：

#!/usr/bin/python

from scapy.all import *
from scapy.contrib.igmp import IGMP
import random


src_ip = "192.168.1.1"
multicast_ip = "239.255.12.45"
interface = "eth0"

def send_igmp_report(dst):
        a=Ether(src="94:c6:91:02:56:d6")
        b=IP(src=src_ip)
        c=IGMP(type=0x16, gaddr=dst)
        c.igmpize(b, a)
        print "Joining IP " + c.gaddr + " MAC " + a.dst
        sendp(a/b/c, iface=interface)

send_igmp_report(multicast_ip)

发送leave报文：

#!/usr/bin/python

from scapy.all import *
from scapy.contrib.igmp import IGMP
import random

src_ip = "192.168.1.1"
src_mac = "94:c6:91:02:56:d6"
interface = "eth0"
multicast_ip = "224.239.12.45"


def send_igmp_leave(dst):
        a=Ether(src=src_mac)
        b=IP(src=src_ip)
        c=IGMP(type=0x17, gaddr=dst)
        c.igmpize(b, a)
        print "Joining IP " + c.gaddr + " MAC " + a.dst
        sendp(a/b/c, iface=interface)


send_igmp_leave(multicast_ip);

发送普通查询：

#!/usr/bin/python

from scapy.all import *
from scapy.contrib.igmp import IGMP
import random


src_ip = "192.168.1.1"
interface = "eth0"
multicast_ip = "224.0.0.1"


def send_igmp_ordinary_query():
        a=Ether(src="94:c6:91:02:56:d6")
        b=IP(src=src_ip, dst=multicast_ip)
        c=IGMP(type=0x11)
        c.igmpize(b, a)
        sendp(a/b/c, iface=interface)

send_igmp_ordinary_query()

发送特定组查询：

#!/usr/bin/python

from scapy.all import *
from scapy.contrib.igmp import IGMP
import random


src_ip = "192.168.1.1"
interface = "eth0"
multicast_ip = "239.255.12.45"


def send_igmp_specified_query(dst):
        a=Ether(src="94:c6:91:02:56:d6")
        b=IP(src=src_ip)
        c=IGMP(type=0x11, gaddr=dst)
        c.igmpize(b, a)
        sendp(a/b/c, iface=interface)


send_igmp_specified_query(multicast_ip)

自己定制协议

如果遇到一些“非常规”的协议，比如这次我需要试一下GVRP协议。可以直接自己定义对应的类。比如：

from scapy.all import *
import random

class Ether8023(Packet):
    name = "8023Ether packet"
    fields_desc = [ DestMACField("dst"),
                    SourceMACField("src"),
                    ShortField("len", 12) ]

class GVRP(Packet):
    name = "gvrp packete"
    fields_desc=[
        ShortField("protocol_id", 1),
        XByteField("type", 1),
        XByteField("len", 4),
        XByteField("event", 1),
        ShortField("value", 100),
        XByteField("endattr", 0),
        XByteField("endmessage", 0)
    ]

# Join Empty
sendp(Ether(src="f8:09:23:00:00:01", dst="01:80:c2:00:00:21")/LLC(dsap=0x42, ssap=0x42, ctrl=0x03)/GVRP(event=1), iface="eth0")

# Join In
# sendp(Ether8023(src="f8:09:23:00:00:01", dst="01:80:c2:00:00:21")/LLC(dsap=0x42, ssap=0x42, ctrl=0x03)/GVRP(event=2), iface="eth0")
# sendp(Ether8023(src="f8:09:23:00:00:01", dst="01:80:c2:00:00:21")/LLC(dsap=0x42, ssap=0x42, ctrl=0x03)/GVRP(event=2), iface="eth0")

Ether8023这个类我是按照Ether类直接写的。GVRP里面，主要就是填fields_desc。 ShortField("protocol_id", 1)表式两个字节的数据，使用protocol_id标识，默认值为1。具体的文档在这里。