Newer
Older
s= b'`\x00\x00\x00\x00(<@\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x01\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x01\x87\x02\x01\x02\x00\x00\xc9\x10\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00;\x01\x05\x00\xee`\x00\x00\xd0\x00\x00\x03\x01\x02\x00\x00'
str(IPv6()/IPv6ExtHdrDestOpt(options=[HAO()])/MIP6MH_BU()) == s
= MIP6MH_BU - dissection (default values)
p = IPv6(s)
p[MIP6MH_BU].len == 1
= MIP6MH_BU - build
s = b'`\x00\x00\x00\x00P<@\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x01\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x01\x87\x02\x01\x02\x00\x00\xc9\x10 \x01\r\xb8\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\xca\xfe;\x06\x05\x00\xea\xf2\x00\x00\xd0\x00\x00*\x01\x00\x03\x10\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x01\x02\x00\x00\x06\x12\x00@\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00'
str(IPv6()/IPv6ExtHdrDestOpt(options=[HAO(hoa='2001:db8::cafe')])/MIP6MH_BU(mhtime=42, options=[MIP6OptAltCoA(),MIP6OptMobNetPrefix()])) == s
= MIP6MH_BU - dissection
p = IPv6(s)
p[MIP6MH_BU].cksum == 0xeaf2 and p[MIP6MH_BU].len == 6 and len(p[MIP6MH_BU].options) == 4 and p[MIP6MH_BU].mhtime == 42
############
############
s = b'`\x00\x00\x00\x00\x10\x87@\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x01\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x01;\x01\x06\x00\xbc\xb9\x00\x80\x00\x00\x00*\x01\x02\x00\x00'
str(IPv6()/MIP6MH_BA(mhtime=42)) == s
= MIP6MH_BA - dissection
p = IPv6(s)
p[MIP6MH_BA].cksum == 0xbcb9 and p[MIP6MH_BA].len == 1 and len(p[MIP6MH_BA].options) == 1 and p[MIP6MH_BA].mhtime == 42
############
############
s = b'`\x00\x00\x00\x00\x18\x87@\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x01\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x01;\x02\x07\x00\xbbY\x02\x00\x00\x01\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x02'
str(IPv6()/MIP6MH_BE(status=2, ha='1::2')) == s
= MIP6MH_BE - dissection
p = IPv6(s)
p[MIP6MH_BE].cksum=0xba10 and p[MIP6MH_BE].len == 1 and len(p[MIP6MH_BE].options) == 1
############
############
+ Netflow v5
= NetflowHeaderV5 - basic building
str(NetflowHeader()/NetflowHeaderV5()) == b'\x00\x05\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00'
str(NetflowHeaderV5(engineID=42)) == b'\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00*\x00\x00'
str(NetflowRecordV5(dst="192.168.0.1")) == b'\x7f\x00\x00\x01\xc0\xa8\x00\x01\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x01\x00\x00\x00<\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x02\x06\x00\x00\x00\x00\x00\x00\x00\x00\x00'
str(NetflowHeader()/NetflowHeaderV5(count=1)/NetflowRecordV5(dst="192.168.0.1")) == b'\x00\x05\x00\x01\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x7f\x00\x00\x01\xc0\xa8\x00\x01\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x01\x00\x00\x00<\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x02\x06\x00\x00\x00\x00\x00\x00\x00\x00\x00'
= NetflowHeaderV5 - basic dissection
nf5 = NetflowHeader(b'\x00\x05\x00\x02\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x7f\x00\x00\x01\x7f\x00\x00\x01\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x01\x00\x00\x00<\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x02\x06\x00\x00\x00\x00\x00\x00\x00\x00\x00\x7f\x00\x00\x01\x7f\x00\x00\x01\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x01\x00\x00\x00<\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x02\x06\x00\x00\x00\x00\x00\x00\x00\x00\x00')
nf5.version == 5 and nf5[NetflowHeaderV5].count == 2 and isinstance(nf5[NetflowRecordV5].payload, NetflowRecordV5)
############
############
+ pcap / pcapng format support
= Variable creations
import cStringIO
pcapfile = cStringIO.StringIO(b'\xd4\xc3\xb2\xa1\x02\x00\x04\x00\x00\x00\x00\x00\x00\x00\x00\x00\xff\xff\x00\x00e\x00\x00\x00\xcf\xc5\xacVo*\n\x00(\x00\x00\x00(\x00\x00\x00E\x00\x00(\x00\x01\x00\x00@\x06|\xcd\x7f\x00\x00\x01\x7f\x00\x00\x01\x00\x14\x00P\x00\x00\x00\x00\x00\x00\x00\x00P\x02 \x00\x91|\x00\x00\xcf\xc5\xacV_-\n\x00\x1c\x00\x00\x00\x1c\x00\x00\x00E\x00\x00\x1c\x00\x01\x00\x00@\x11|\xce\x7f\x00\x00\x01\x7f\x00\x00\x01\x005\x005\x00\x08\x01r\xcf\xc5\xacV\xf90\n\x00\x1c\x00\x00\x00\x1c\x00\x00\x00E\x00\x00\x1c\x00\x01\x00\x00@\x01|\xde\x7f\x00\x00\x01\x7f\x00\x00\x01\x08\x00\xf7\xff\x00\x00\x00\x00')
pcapngfile = cStringIO.StringIO(b'\n\r\r\n\\\x00\x00\x00M<+\x1a\x01\x00\x00\x00\xff\xff\xff\xff\xff\xff\xff\xff\x01\x00,\x00File created by merging: \nFile1: test.pcap \n\x04\x00\x08\x00mergecap\x00\x00\x00\x00\\\x00\x00\x00\x01\x00\x00\x00\\\x00\x00\x00e\x00\x00\x00\xff\xff\x00\x00\x02\x006\x00Unknown/not available in original file format(libpcap)\x00\x00\t\x00\x01\x00\x06\x00\x00\x00\x00\x00\x00\x00\\\x00\x00\x00\x06\x00\x00\x00H\x00\x00\x00\x00\x00\x00\x00\x8d*\x05\x00/\xfc[\xcd(\x00\x00\x00(\x00\x00\x00E\x00\x00(\x00\x01\x00\x00@\x06|\xcd\x7f\x00\x00\x01\x7f\x00\x00\x01\x00\x14\x00P\x00\x00\x00\x00\x00\x00\x00\x00P\x02 \x00\x91|\x00\x00H\x00\x00\x00\x06\x00\x00\x00<\x00\x00\x00\x00\x00\x00\x00\x8d*\x05\x00\x1f\xff[\xcd\x1c\x00\x00\x00\x1c\x00\x00\x00E\x00\x00\x1c\x00\x01\x00\x00@\x11|\xce\x7f\x00\x00\x01\x7f\x00\x00\x01\x005\x005\x00\x08\x01r<\x00\x00\x00\x06\x00\x00\x00<\x00\x00\x00\x00\x00\x00\x00\x8d*\x05\x00\xb9\x02\\\xcd\x1c\x00\x00\x00\x1c\x00\x00\x00E\x00\x00\x1c\x00\x01\x00\x00@\x01|\xde\x7f\x00\x00\x01\x7f\x00\x00\x01\x08\x00\xf7\xff\x00\x00\x00\x00<\x00\x00\x00')
pcapnanofile = cStringIO.StringIO(b"M<\xb2\xa1\x02\x00\x04\x00\x00\x00\x00\x00\x00\x00\x00\x00\xff\xff\x00\x00e\x00\x00\x00\xcf\xc5\xacV\xc9\xc1\xb5'(\x00\x00\x00(\x00\x00\x00E\x00\x00(\x00\x01\x00\x00@\x06|\xcd\x7f\x00\x00\x01\x7f\x00\x00\x01\x00\x14\x00P\x00\x00\x00\x00\x00\x00\x00\x00P\x02 \x00\x91|\x00\x00\xcf\xc5\xacV-;\xc1'\x1c\x00\x00\x00\x1c\x00\x00\x00E\x00\x00\x1c\x00\x01\x00\x00@\x11|\xce\x7f\x00\x00\x01\x7f\x00\x00\x01\x005\x005\x00\x08\x01r\xcf\xc5\xacV\x9aL\xcf'\x1c\x00\x00\x00\x1c\x00\x00\x00E\x00\x00\x1c\x00\x01\x00\x00@\x01|\xde\x7f\x00\x00\x01\x7f\x00\x00\x01\x08\x00\xf7\xff\x00\x00\x00\x00")
= Read a pcapng file
= Read a pcap file with nanosecond precision
pktpcapnano = rdpcap(pcapnanofile)
= Check all packet lists are the same
assert list(pktpcap) == list(pktpcapng) == list(pktpcapnano)
assert [p.time for p in pktpcap] == [p.time for p in pktpcapng] == [p.time for p in pktpcapnano]
= Check packets from pcap file
assert all(IP in pkt for pkt in pktpcap)
assert all(any(proto in pkt for pkt in pktpcap) for proto in [ICMP, UDP, TCP])
5092
5093
5094
5095
5096
5097
5098
5099
5100
5101
5102
5103
5104
5105
5106
5107
5108
5109
5110
5111
5112
5113
5114
5115
5116
5117
5118
5119
5120
= Check wrpcap()
import os, tempfile
fdesc, filename = tempfile.mkstemp()
fdesc = os.fdopen(fdesc, "w")
wrpcap(fdesc, pktpcap)
fdesc.close()
= Check offline sniff() (by filename)
assert list(pktpcap) == list(sniff(offline=filename))
= Check offline sniff() (by file object)
fdesc = open(filename)
assert list(pktpcap) == list(sniff(offline=fdesc))
fdesc.close()
= Check offline sniff() with a filter (by filename)
~ tcpdump
pktpcap_flt = [(proto, sniff(offline=filename, filter=proto.__name__.lower()))
for proto in [ICMP, UDP, TCP]]
assert all(list(pktpcap[proto]) == list(packets) for proto, packets in pktpcap_flt)
= Check offline sniff() with a filter (by file object)
~ tcpdump
fdesc = open(filename)
pktpcap_tcp = sniff(offline=fdesc, filter="tcp")
fdesc.close()
assert list(pktpcap[TCP]) == list(pktpcap_tcp)
os.unlink(filename)
= Check wrpcap(nano=True)
fdesc, filename = tempfile.mkstemp()
fdesc = os.fdopen(fdesc, "w")
pktpcapnano[0].time += 0.000000001
wrpcap(fdesc, pktpcapnano, nano=True)
fdesc.close()
pktpcapnanoread = rdpcap(filename)
assert pktpcapnanoread[0].time == pktpcapnano[0].time
assert pktpcapnanoread[0].time == pktpcap[0].time + 0.000000001
os.unlink(filename)
= Check PcapNg with nanosecond precision using obsolete packet block
* first packet from capture file icmp2.ntar -- https://wiki.wireshark.org/Development/PcapNg?action=AttachFile&do=view&target=icmp2.ntar
pcapngfile = cStringIO.StringIO(b'\n\r\r\n\x1c\x00\x00\x00M<+\x1a\x01\x00\x00\x00\xa8\x03\x00\x00\x00\x00\x00\x00\x1c\x00\x00\x00\x01\x00\x00\x00(\x00\x00\x00\x01\x00\x00\x00\xff\xff\x00\x00\r\x00\x01\x00\x04\x04K\x00\t\x00\x01\x00\tK=N\x00\x00\x00\x00(\x00\x00\x00\x02\x00\x00\x00n\x00\x00\x00\x00\x00\x00\x00e\x14\x00\x00)4\'ON\x00\x00\x00N\x00\x00\x00\x00\x12\xf0\x11h\xd6\x00\x13r\t{\xea\x08\x00E\x00\x00<\x90\xa1\x00\x00\x80\x01\x8e\xad\xc0\xa8M\x07\xc0\xa8M\x1a\x08\x00r[\x03\x00\xd8\x00abcdefghijklmnopqrstuvwabcdefghi\xeay$\xf6\x00\x00n\x00\x00\x00')
pktpcapng = rdpcap(pcapngfile)
assert len(pktpcapng) == 1
pkt = pktpcapng[0]
# weird, but wireshark agrees
assert pkt.time == 22425.352221737
assert isinstance(pkt, Ether)
pkt = pkt.payload
assert isinstance(pkt, IP)
pkt = pkt.payload
assert isinstance(pkt, ICMP)
pkt = pkt.payload
assert isinstance(pkt, Raw) and pkt.load == 'abcdefghijklmnopqrstuvwabcdefghi'
pkt = pkt.payload
assert isinstance(pkt, Padding) and pkt.load == b'\xeay$\xf6'
pkt = pkt.payload
assert isinstance(pkt, NoPayload)
= Check PcapNg using Simple Packet Block
* previous file with the (obsolete) packet block replaced by a Simple Packet Block
pcapngfile = cStringIO.StringIO(b'\n\r\r\n\x1c\x00\x00\x00M<+\x1a\x01\x00\x00\x00\xa8\x03\x00\x00\x00\x00\x00\x00\x1c\x00\x00\x00\x01\x00\x00\x00(\x00\x00\x00\x01\x00\x00\x00\xff\xff\x00\x00\r\x00\x01\x00\x04\x04K\x00\t\x00\x01\x00\tK=N\x00\x00\x00\x00(\x00\x00\x00\x03\x00\x00\x00`\x00\x00\x00N\x00\x00\x00\x00\x12\xf0\x11h\xd6\x00\x13r\t{\xea\x08\x00E\x00\x00<\x90\xa1\x00\x00\x80\x01\x8e\xad\xc0\xa8M\x07\xc0\xa8M\x1a\x08\x00r[\x03\x00\xd8\x00abcdefghijklmnopqrstuvwabcdefghi\xeay$\xf6\x00\x00`\x00\x00\x00')
pktpcapng = rdpcap(pcapngfile)
assert len(pktpcapng) == 1
pkt = pktpcapng[0]
assert isinstance(pkt, Ether)
pkt = pkt.payload
assert isinstance(pkt, IP)
pkt = pkt.payload
assert isinstance(pkt, ICMP)
pkt = pkt.payload
assert isinstance(pkt, Raw) and pkt.load == 'abcdefghijklmnopqrstuvwabcdefghi'
pkt = pkt.payload
assert isinstance(pkt, Padding) and pkt.load == b'\xeay$\xf6'
pkt = pkt.payload
assert isinstance(pkt, NoPayload)
= Check tcpdump()
~ tcpdump
* No very specific tests because we do not want to depend on tcpdump output
pcapfile = cStringIO.StringIO(b'\xd4\xc3\xb2\xa1\x02\x00\x04\x00\x00\x00\x00\x00\x00\x00\x00\x00\xff\xff\x00\x00e\x00\x00\x00\xcf\xc5\xacVo*\n\x00(\x00\x00\x00(\x00\x00\x00E\x00\x00(\x00\x01\x00\x00@\x06|\xcd\x7f\x00\x00\x01\x7f\x00\x00\x01\x00\x14\x00P\x00\x00\x00\x00\x00\x00\x00\x00P\x02 \x00\x91|\x00\x00\xcf\xc5\xacV_-\n\x00\x1c\x00\x00\x00\x1c\x00\x00\x00E\x00\x00\x1c\x00\x01\x00\x00@\x11|\xce\x7f\x00\x00\x01\x7f\x00\x00\x01\x005\x005\x00\x08\x01r\xcf\xc5\xacV\xf90\n\x00\x1c\x00\x00\x00\x1c\x00\x00\x00E\x00\x00\x1c\x00\x01\x00\x00@\x01|\xde\x7f\x00\x00\x01\x7f\x00\x00\x01\x08\x00\xf7\xff\x00\x00\x00\x00')
data = tcpdump(pcapfile, dump=True, args=['-n']).split('\n')
print data
assert 'IP 127.0.0.1.20 > 127.0.0.1.80:' in data[0]
assert 'IP 127.0.0.1.53 > 127.0.0.1.53:' in data[1]
assert 'IP 127.0.0.1 > 127.0.0.1:' in data[2]
= Check tcpdump() command with tshark
~ tshark
pcapfile = cStringIO.StringIO(b'\xd4\xc3\xb2\xa1\x02\x00\x04\x00\x00\x00\x00\x00\x00\x00\x00\x00\xff\xff\x00\x00e\x00\x00\x00\xcf\xc5\xacVo*\n\x00(\x00\x00\x00(\x00\x00\x00E\x00\x00(\x00\x01\x00\x00@\x06|\xcd\x7f\x00\x00\x01\x7f\x00\x00\x01\x00\x14\x00P\x00\x00\x00\x00\x00\x00\x00\x00P\x02 \x00\x91|\x00\x00\xcf\xc5\xacV_-\n\x00\x1c\x00\x00\x00\x1c\x00\x00\x00E\x00\x00\x1c\x00\x01\x00\x00@\x11|\xce\x7f\x00\x00\x01\x7f\x00\x00\x01\x005\x005\x00\x08\x01r\xcf\xc5\xacV\xf90\n\x00\x1c\x00\x00\x00\x1c\x00\x00\x00E\x00\x00\x1c\x00\x01\x00\x00@\x01|\xde\x7f\x00\x00\x01\x7f\x00\x00\x01\x08\x00\xf7\xff\x00\x00\x00\x00')
values = [tuple(int(val) for val in line[:-1].split('\t')) for line in tcpdump(pcapfile, prog=conf.prog.tshark, getfd=True, args=['-T', 'fields', '-e', 'ip.ttl', '-e', 'ip.proto'])]
assert values == [(64, 6), (64, 17), (64, 1)]
= Check Raw IP pcap files
import tempfile
filename = tempfile.mktemp(suffix=".pcap")
wrpcap(filename, [IP()/UDP(), IPv6()/UDP()], linktype=DLT_RAW)
packets = rdpcap(filename)
assert(isinstance(packets[0], IP) and isinstance(packets[1], IPv6))
############
############
+ LLTD protocol
= Simple packet dissection
pkt = Ether(b'\xff\xff\xff\xff\xff\xff\x86\x14\xf0\xc7[.\x88\xd9\x01\x00\x00\x01\xff\xff\xff\xff\xff\xff\x86\x14\xf0\xc7[.\x00\x00\xfe\xe9[\xa9\xaf\xc1\x0bS[\xa9\xaf\xc1\x0bS\x01\x06}[G\x8f\xec.\x02\x04p\x00\x00\x00\x03\x04\x00\x00\x00\x06\x07\x04\xac\x19\x88\xe4\t\x02\x00l\n\x08\x00\x00\x00\x00\x00\x0fB@\x0c\x04\x00\x08=`\x0e\x00\x0f\x0eT\x00E\x00S\x00T\x00-\x00A\x00P\x00\x12\x10\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x14\x04\x00\x00\x00\x00\x15\x01\x02\x18\x00\x19\x02\x04\x00\x1a\x00\x00')
assert pkt.dst == pkt.real_dst
assert pkt.src == pkt.real_src
assert pkt.current_mapper_address == pkt.apparent_mapper_address
assert pkt.mac == '7d:5b:47:8f:ec:2e'
assert pkt.hostname == "TEST-AP"
assert isinstance(pkt[LLTDAttributeEOP].payload, NoPayload)
= Packet build / dissection
pkt = Ether(str(Ether(dst=ETHER_BROADCAST, src=RandMAC()) / LLTD(tos=0, function=0)))
assert LLTD in pkt
assert pkt.dst == pkt.real_dst
assert pkt.src == pkt.real_src
assert pkt.tos == 0
assert pkt.function == 0
= Attribute build / dissection
assert isinstance(LLTDAttribute(), LLTDAttribute)
assert isinstance(LLTDAttribute(str(LLTDAttribute())), LLTDAttribute)
assert all(isinstance(LLTDAttribute(type=i), LLTDAttribute) for i in xrange(256))
assert all(isinstance(LLTDAttribute(str(LLTDAttribute(type=i))), LLTDAttribute) for i in xrange(256))
5222
5223
5224
5225
5226
5227
5228
5229
5230
5231
5232
5233
5234
5235
5236
5237
5238
5239
5240
5241
5242
5243
5244
5245
5246
5247
5248
= Large TLV
m1, m2, seq = RandMAC()._fix(), RandMAC()._fix(), 123
preqbase = Ether(src=m1, dst=m2) / LLTD() / \
LLTDQueryLargeTlv(type="Detailed Icon Image")
prespbase = Ether(src=m2, dst=m1) / LLTD() / \
LLTDQueryLargeTlvResp()
plist = []
pkt = preqbase.copy()
pkt.seq = seq
plist.append(Ether(str(pkt)))
pkt = prespbase.copy()
pkt.seq = seq
pkt.flags = "M"
pkt.value = "abcd"
plist.append(Ether(str(pkt)))
pkt = preqbase.copy()
pkt.seq = seq + 1
pkt.offset = 4
plist.append(Ether(str(pkt)))
pkt = prespbase.copy()
pkt.seq = seq + 1
pkt.value = "efg"
plist.append(Ether(str(pkt)))
builder = LargeTlvBuilder()
builder.parse(plist)
data = builder.get_data()
assert len(data) == 1
key, value = data.popitem()
assert key.endswith(' [Detailed Icon Image]')
assert value == 'abcdefg'
5253
5254
5255
5256
5257
5258
5259
5260
5261
5262
5263
5264
5265
5266
5267
5268
5269
5270
5271
5272
5273
5274
############
############
+ Test fragment() / defragment() functions
= fragment()
payloadlen, fragsize = 100, 8
assert fragsize % 8 == 0
fragcount = (payloadlen / fragsize) + bool(payloadlen % fragsize)
* create the packet
pkt = IP() / ("X" * payloadlen)
* create the fragments
frags = fragment(pkt, fragsize)
* count the fragments
assert len(frags) == fragcount
* each fragment except the last one should have MF set
assert all(p.flags == 1 for p in frags[:-1])
assert frags[-1].flags == 0
* each fragment except the last one should have a payload of fragsize bytes
assert all(len(p.payload) == 8 for p in frags[:-1])
assert len(frags[-1].payload) == ((payloadlen % fragsize) or fragsize)
= fragment() and overloaded_fields
pkt1 = Ether() / IP() / UDP()
pkt2 = fragment(pkt1)[0]
pkt3 = pkt2.__class__(str(pkt2))
assert pkt1[IP].proto == pkt2[IP].proto == pkt3[IP].proto
= fragment() already fragmented packets
payloadlen = 1480 * 3
ffrags = fragment(IP() / ("X" * payloadlen), 1480)
ffrags = fragment(ffrags, 1400)
len(ffrags) == 6
* each fragment except the last one should have MF set
assert all(p.flags == 1 for p in ffrags[:-1])
assert ffrags[-1].flags == 0
* fragment offset should be well computed
plen = 0
for p in ffrags:
assert p.frag == plen / 8
plen += len(p.payload)
assert plen == payloadlen
= defrag()
nonfrag, unfrag, badfrag = defrag(frags)
assert not nonfrag
assert not badfrag
assert len(unfrag) == 1
= defragment()
defrags = defragment(frags)
* we should have one single packet
assert len(defrags) == 1
* which should be the same as pkt reconstructed
assert defrags[0] == IP(str(pkt))
5310
5311
5312
5313
5314
5315
5316
5317
5318
5319
5320
5321
5322
5323
5324
5325
5326
5327
5328
5329
5330
5331
5332
5333
5334
5335
5336
5337
5338
5339
5340
5341
5342
5343
5344
5345
5346
5347
5348
5349
= Packet().fragment()
payloadlen, fragsize = 100, 8
assert fragsize % 8 == 0
fragcount = (payloadlen / fragsize) + bool(payloadlen % fragsize)
* create the packet
pkt = IP() / ("X" * payloadlen)
* create the fragments
frags = pkt.fragment(fragsize)
* count the fragments
assert len(frags) == fragcount
* each fragment except the last one should have MF set
assert all(p.flags == 1 for p in frags[:-1])
assert frags[-1].flags == 0
* each fragment except the last one should have a payload of fragsize bytes
assert all(len(p.payload) == 8 for p in frags[:-1])
assert len(frags[-1].payload) == ((payloadlen % fragsize) or fragsize)
= Packet().fragment() and overloaded_fields
pkt1 = Ether() / IP() / UDP()
pkt2 = pkt1.fragment()[0]
pkt3 = pkt2.__class__(str(pkt2))
assert pkt1[IP].proto == pkt2[IP].proto == pkt3[IP].proto
= Packet().fragment() already fragmented packets
payloadlen = 1480 * 3
ffrags = (IP() / ("X" * payloadlen)).fragment(1480)
ffrags = reduce(lambda x, y: x + y, (pkt.fragment(1400) for pkt in ffrags))
len(ffrags) == 6
* each fragment except the last one should have MF set
assert all(p.flags == 1 for p in ffrags[:-1])
assert ffrags[-1].flags == 0
* fragment offset should be well computed
plen = 0
for p in ffrags:
assert p.frag == plen / 8
plen += len(p.payload)
assert plen == payloadlen
############
############
= TCP options: UTO - basic build
str(TCP(options=[("UTO", 0xffff)])) == b"\x00\x14\x00\x50\x00\x00\x00\x00\x00\x00\x00\x00\x60\x02\x20\x00\x00\x00\x00\x00\x1c\x04\xff\xff"
= TCP options: UTO - basic dissection
uto = TCP(b"\x00\x14\x00\x50\x00\x00\x00\x00\x00\x00\x00\x00\x60\x02\x20\x00\x00\x00\x00\x00\x1c\x04\xff\xff")
uto[TCP].options[0][0] == "UTO" and uto[TCP].options[0][1] == 0xffff
5361
5362
5363
5364
5365
5366
5367
5368
5369
5370
5371
5372
5373
5374
5375
5376
5377
5378
5379
5380
5381
5382
5383
5384
5385
5386
5387
= TCP options: SAck - basic build
str(TCP(options=[(5, "abcdefgh")])) == b"\x00\x14\x00P\x00\x00\x00\x00\x00\x00\x00\x00\x80\x02 \x00\x00\x00\x00\x00\x05\nabcdefgh\x00\x00"
= TCP options: SAck - basic dissection
sack = TCP(b"\x00\x14\x00P\x00\x00\x00\x00\x00\x00\x00\x00\x80\x02 \x00\x00\x00\x00\x00\x05\nabcdefgh\x00\x00")
sack[TCP].options[0][0] == "SAck" and sack[TCP].options[0][1] == (1633837924, 1701209960)
= TCP options: SAckOK - basic build
str(TCP(options=[('SAckOK', '')])) == b"\x00\x14\x00P\x00\x00\x00\x00\x00\x00\x00\x00`\x02 \x00\x00\x00\x00\x00\x04\x02\x00\x00"
= TCP options: SAckOK - basic dissection
sackok = TCP(b"\x00\x14\x00P\x00\x00\x00\x00\x00\x00\x00\x00`\x02 \x00\x00\x00\x00\x00\x04\x02\x00\x00")
sackok[TCP].options[0][0] == "SAckOK" and sackok[TCP].options[0][1] == ''
= TCP options: EOL - basic build
str(TCP(options=[(0, '')])) == b"\x00\x14\x00P\x00\x00\x00\x00\x00\x00\x00\x00`\x02 \x00\x00\x00\x00\x00\x00\x02\x00\x00"
= TCP options: EOL - basic dissection
eol = TCP(b"\x00\x14\x00P\x00\x00\x00\x00\x00\x00\x00\x00`\x02 \x00\x00\x00\x00\x00\x00\x02\x00\x00")
eol[TCP].options[0][0] == "EOL" and eol[TCP].options[0][1] == None
= TCP options: malformed - build
str(TCP(options=[('unknown', '')])) == str(TCP())
= TCP options: malformed - dissection
str(TCP(b"\x00\x14\x00P\x00\x00\x00\x00\x00\x00\x00\x00`\x02 \x00\x00\x00\x00\x00\x03\x00\x00\x00")) == b"\x00\x14\x00P\x00\x00\x00\x00\x00\x00\x00\x00`\x02 \x00\x00\x00\x00\x00\x03\x00\x00\x00"
5388
5389
5390
5391
5392
5393
5394
5395
5396
5397
5398
5399
5400
5401
5402
5403
5404
5405
5406
5407
5408
5409
5410
5411
5412
5413
5414
= IP, TCP & UDP checksums (these tests highly depend on default values)
pkt = IP() / TCP()
bpkt = IP(str(pkt))
assert bpkt.chksum == 0x7ccd and bpkt.payload.chksum == 0x917c
pkt = IP(len=40) / TCP()
bpkt = IP(str(pkt))
assert bpkt.chksum == 0x7ccd and bpkt.payload.chksum == 0x917c
pkt = IP(len=40, ihl=5) / TCP()
bpkt = IP(str(pkt))
assert bpkt.chksum == 0x7ccd and bpkt.payload.chksum == 0x917c
pkt = IP() / TCP() / ("A" * 10)
bpkt = IP(str(pkt))
assert bpkt.chksum == 0x7cc3 and bpkt.payload.chksum == 0x4b2c
pkt = IP(len=50) / TCP() / ("A" * 10)
bpkt = IP(str(pkt))
assert bpkt.chksum == 0x7cc3 and bpkt.payload.chksum == 0x4b2c
pkt = IP(len=50, ihl=5) / TCP() / ("A" * 10)
bpkt = IP(str(pkt))
assert bpkt.chksum == 0x7cc3 and bpkt.payload.chksum == 0x4b2c
pkt = IP(options=[IPOption_RR()]) / TCP() / ("A" * 10)
bpkt = IP(str(pkt))
assert bpkt.chksum == 0x70bc and bpkt.payload.chksum == 0x4b2c
pkt = IP(len=54, options=[IPOption_RR()]) / TCP() / ("A" * 10)
bpkt = IP(str(pkt))
assert bpkt.chksum == 0x70bc and bpkt.payload.chksum == 0x4b2c
pkt = IP(len=54, ihl=6, options=[IPOption_RR()]) / TCP() / ("A" * 10)
bpkt = IP(str(pkt))
assert bpkt.chksum == 0x70bc and bpkt.payload.chksum == 0x4b2c
5424
5425
5426
5427
5428
5429
5430
5431
5432
5433
5434
5435
5436
5437
5438
5439
5440
5441
5442
5443
5444
5445
5446
5447
5448
5449
5450
pkt = IP() / UDP()
bpkt = IP(str(pkt))
assert bpkt.chksum == 0x7cce and bpkt.payload.chksum == 0x0172
pkt = IP(len=28) / UDP()
bpkt = IP(str(pkt))
assert bpkt.chksum == 0x7cce and bpkt.payload.chksum == 0x0172
pkt = IP(len=28, ihl=5) / UDP()
bpkt = IP(str(pkt))
assert bpkt.chksum == 0x7cce and bpkt.payload.chksum == 0x0172
pkt = IP() / UDP() / ("A" * 10)
bpkt = IP(str(pkt))
assert bpkt.chksum == 0x7cc4 and bpkt.payload.chksum == 0xbb17
pkt = IP(len=38) / UDP() / ("A" * 10)
bpkt = IP(str(pkt))
assert bpkt.chksum == 0x7cc4 and bpkt.payload.chksum == 0xbb17
pkt = IP(len=38, ihl=5) / UDP() / ("A" * 10)
bpkt = IP(str(pkt))
assert bpkt.chksum == 0x7cc4 and bpkt.payload.chksum == 0xbb17
pkt = IP(options=[IPOption_RR()]) / UDP() / ("A" * 10)
bpkt = IP(str(pkt))
assert bpkt.chksum == 0x70bd and bpkt.payload.chksum == 0xbb17
pkt = IP(len=42, options=[IPOption_RR()]) / UDP() / ("A" * 10)
bpkt = IP(str(pkt))
assert bpkt.chksum == 0x70bd and bpkt.payload.chksum == 0xbb17
pkt = IP(len=42, ihl=6, options=[IPOption_RR()]) / UDP() / ("A" * 10)
bpkt = IP(str(pkt))
assert bpkt.chksum == 0x70bd and bpkt.payload.chksum == 0xbb17
= DNS
* DNS over UDP
pkt = IP(str(IP(src="10.0.0.1", dst="8.8.8.8")/UDP(sport=RandShort(), dport=53)/DNS(qd=DNSQR(qname="secdev.org."))))
assert UDP in pkt and isinstance(pkt[UDP].payload, DNS)
assert pkt[UDP].dport == 53 and pkt[UDP].length is None
assert pkt[DNS].qdcount == 1 and pkt[DNS].qd.qname == "secdev.org."
* DNS over TCP
pkt = IP(str(IP(src="10.0.0.1", dst="8.8.8.8")/TCP(sport=RandShort(), dport=53, flags="P")/DNS(qd=DNSQR(qname="secdev.org."))))
assert TCP in pkt and isinstance(pkt[TCP].payload, DNS)
assert pkt[TCP].dport == 53 and pkt[DNS].length is not None
assert pkt[DNS].qdcount == 1 and pkt[DNS].qd.qname == "secdev.org."
5475
5476
5477
5478
5479
5480
5481
5482
5483
5484
5485
5486
5487
5488
5489
5490
5491
5492
5493
5494
5495
= Layer binding
* Test DestMACField & DestIPField
pkt = Ether(str(Ether()/IP()/UDP(dport=5353)/DNS()))
assert isinstance(pkt, Ether) and pkt.dst == '01:00:5e:00:00:fb'
pkt = pkt.payload
assert isinstance(pkt, IP) and pkt.dst == '224.0.0.251'
pkt = pkt.payload
assert isinstance(pkt, UDP) and pkt.dport == 5353
pkt = pkt.payload
assert isinstance(pkt, DNS) and isinstance(pkt.payload, NoPayload)
* Same with IPv6
pkt = Ether(str(Ether()/IPv6()/UDP(dport=5353)/DNS()))
assert isinstance(pkt, Ether) and pkt.dst == '33:33:00:00:00:fb'
pkt = pkt.payload
assert isinstance(pkt, IPv6) and pkt.dst == 'ff02::fb'
pkt = pkt.payload
assert isinstance(pkt, UDP) and pkt.dport == 5353
pkt = pkt.payload
assert isinstance(pkt, DNS) and isinstance(pkt.payload, NoPayload)
############
############
+ Mocked read_routes() calls
= Truncated netstat -rn output on OS X
import mock
import StringIO
@mock.patch("scapy.arch.unix.get_if_addr")
@mock.patch("scapy.arch.unix.os")
def test_osx_netstat_truncated(mock_os, mock_get_if_addr):
"""Test read_routes() on OS X 10.? with a long interface name"""
# netstat & ifconfig outputs from https://github.com/secdev/scapy/pull/119
netstat_output = """
Routing tables
Internet:
Destination Gateway Flags Refs Use Netif Expire
default 192.168.1.1 UGSc 460 0 en1
default link#11 UCSI 1 0 bridge1
127 127.0.0.1 UCS 1 0 lo0
127.0.0.1 127.0.0.1 UH 10 2012351 lo0
"""
ifconfig_output = "lo0 en1 bridge10\n"
# Mocked file descriptors
def se_popen(command):
"""Perform specific side effects"""
if command.startswith("netstat -rn"):
elif command == "ifconfig -l":
ret = StringIO.StringIO(ifconfig_output)
def unit():
return ret
ret.__call__ = unit
ret.__enter__ = unit
ret.__exit__ = lambda x,y,z: None
return ret
raise Exception("Command not mocked: %s" % command)
mock_os.popen.side_effect = se_popen
# Mocked get_if_addr() behavior
def se_get_if_addr(iface):
"""Perform specific side effects"""
if iface == "bridge1":
oserror_exc = OSError()
oserror_exc.message = "Device not configured"
raise oserror_exc
return "1.2.3.4"
mock_get_if_addr.side_effect = se_get_if_addr
# Test the function
from scapy.arch.unix import read_routes
routes = read_routes()
assert(len(routes) == 4)
assert([r for r in routes if r[3] == "bridge10"])
test_osx_netstat_truncated()
############
############
+ Mocked read_routes6() calls
= Preliminary definitions
5564
5565
5566
5567
5568
5569
5570
5571
5572
5573
5574
5575
5576
5577
5578
5579
5580
5581
5582
5583
5584
5585
5586
5587
5588
5589
5590
5591
5592
import mock
import StringIO
def valid_output_read_routes6(routes):
""""Return True if 'routes' contains correctly formatted entries, False otherwise"""
for destination, plen, next_hop, dev, cset in routes:
if not in6_isvalid(destination) or not type(plen) == int:
return False
if not in6_isvalid(next_hop) or not type(dev) == str:
return False
for address in cset:
if not in6_isvalid(address):
return False
return True
def check_mandatory_ipv6_routes(routes6):
"""Ensure that mandatory IPv6 routes are present"""
if len(filter(lambda r: r[0] == "::1" and r[-1] == ["::1"], routes6)) < 1:
return False
if len(filter(lambda r: r[0] == "fe80::" and r[1] == 64, routes6)) < 1:
return False
if len(filter(lambda r: in6_islladdr(r[0]) and r[1] == 128 and \
r[-1] == ["::1"], routes6)) < 1:
return False
return True
= Mac OS X 10.9.5
5594
5595
5596
5597
5598
5599
5600
5601
5602
5603
5604
5605
5606
5607
5608
5609
5610
5611
5612
5613
5614
5615
5616
5617
5618
5619
5620
5621
5622
5623
5624
5625
5626
5627
5628
5629
5630
5631
5632
5633
@mock.patch("scapy.arch.unix.in6_getifaddr")
@mock.patch("scapy.arch.unix.os")
def test_osx_10_9_5(mock_os, mock_in6_getifaddr):
"""Test read_routes6() on OS X 10.9.5"""
# 'netstat -rn -f inet6' output
netstat_output = """
Routing tables
Internet6:
Destination Gateway Flags Netif Expire
::1 ::1 UHL lo0
fe80::%lo0/64 fe80::1%lo0 UcI lo0
fe80::1%lo0 link#1 UHLI lo0
fe80::%en0/64 link#4 UCI en0
fe80::ba26:6cff:fe5f:4eee%en0 b8:26:6c:5f:4e:ee UHLWIi en0
fe80::bae8:56ff:fe45:8ce6%en0 b8:e8:56:45:8c:e6 UHLI lo0
ff01::%lo0/32 ::1 UmCI lo0
ff01::%en0/32 link#4 UmCI en0
ff02::%lo0/32 ::1 UmCI lo0
ff02::%en0/32 link#4 UmCI en0
"""
# Mocked file descriptor
strio = StringIO.StringIO(netstat_output)
mock_os.popen = mock.MagicMock(return_value=strio)
# Mocked in6_getifaddr() output
mock_in6_getifaddr.return_value = [("::1", IPV6_ADDR_LOOPBACK, "lo0"),
("fe80::ba26:6cff:fe5f:4eee", IPV6_ADDR_LINKLOCAL, "en0")]
# Test the function
from scapy.arch.unix import read_routes6
routes = read_routes6()
for r in routes:
print r
assert(len(routes) == 6)
assert(check_mandatory_ipv6_routes(routes))
test_osx_10_9_5()
= Mac OS X 10.9.5 with global IPv6 connectivity
5635
5636
5637
5638
5639
5640
5641
5642
5643
5644
5645
5646
5647
5648
5649
5650
5651
5652
5653
5654
5655
5656
5657
5658
5659
5660
5661
5662
5663
5664
5665
5666
5667
5668
5669
5670
5671
5672
5673
5674
5675
5676
5677
5678
5679
5680
5681
@mock.patch("scapy.arch.unix.in6_getifaddr")
@mock.patch("scapy.arch.unix.os")
def test_osx_10_9_5_global(mock_os, mock_in6_getifaddr):
"""Test read_routes6() on OS X 10.9.5 with an IPv6 connectivity"""
# 'netstat -rn -f inet6' output
netstat_output = """
Routing tables
Internet6:
Destination Gateway Flags Netif Expire
default fe80::ba26:8aff:fe5f:4eef%en0 UGc en0
::1 ::1 UHL lo0
2a01:ab09:7d:1f01::/64 link#4 UC en0
2a01:ab09:7d:1f01:420:205c:9fab:5be7 b8:e9:55:44:7c:e5 UHL lo0
2a01:ab09:7d:1f01:ba26:8aff:fe5f:4eef b8:26:8a:5f:4e:ef UHLWI en0
2a01:ab09:7d:1f01:bae9:55ff:fe44:7ce5 b8:e9:55:44:7c:e5 UHL lo0
fe80::%lo0/64 fe80::1%lo0 UcI lo0
fe80::1%lo0 link#1 UHLI lo0
fe80::%en0/64 link#4 UCI en0
fe80::5664:d9ff:fe79:4e00%en0 54:64:d9:79:4e:0 UHLWI en0
fe80::6ead:f8ff:fe74:945a%en0 6c:ad:f8:74:94:5a UHLWI en0
fe80::a2f3:c1ff:fec4:5b50%en0 a0:f3:c1:c4:5b:50 UHLWI en0
fe80::ba26:8aff:fe5f:4eef%en0 b8:26:8a:5f:4e:ef UHLWIir en0
fe80::bae9:55ff:fe44:7ce5%en0 b8:e9:55:44:7c:e5 UHLI lo0
ff01::%lo0/32 ::1 UmCI lo0
ff01::%en0/32 link#4 UmCI en0
ff02::%lo0/32 ::1 UmCI lo
"""
# Mocked file descriptor
strio = StringIO.StringIO(netstat_output)
mock_os.popen = mock.MagicMock(return_value=strio)
# Mocked in6_getifaddr() output
mock_in6_getifaddr.return_value = [("::1", IPV6_ADDR_LOOPBACK, "lo0"),
("fe80::ba26:6cff:fe5f:4eee", IPV6_ADDR_LINKLOCAL, "en0")]
# Test the function
from scapy.arch.unix import read_routes6
routes = read_routes6()
assert(valid_output_read_routes6(routes))
for r in routes:
print r
assert(len(routes) == 11)
assert(check_mandatory_ipv6_routes(routes))
test_osx_10_9_5_global()
= Mac OS X 10.10.4
5683
5684
5685
5686
5687
5688
5689
5690
5691
5692
5693
5694
5695
5696
5697
5698
5699
5700
5701
5702
5703
5704
5705
5706
5707
5708
5709
5710
5711
5712
5713
5714
5715
5716
5717
5718
5719
5720
5721
@mock.patch("scapy.arch.unix.in6_getifaddr")
@mock.patch("scapy.arch.unix.os")
def test_osx_10_10_4(mock_os, mock_in6_getifaddr):
"""Test read_routes6() on OS X 10.10.4"""
# 'netstat -rn -f inet6' output
netstat_output = """
Routing tables
Internet6:
Destination Gateway Flags Netif Expire
::1 ::1 UHL lo0
fe80::%lo0/64 fe80::1%lo0 UcI lo0
fe80::1%lo0 link#1 UHLI lo0
fe80::%en0/64 link#4 UCI en0
fe80::a00:27ff:fe9b:c965%en0 8:0:27:9b:c9:65 UHLI lo0
ff01::%lo0/32 ::1 UmCI lo0
ff01::%en0/32 link#4 UmCI en0
ff02::%lo0/32 ::1 UmCI lo0
ff02::%en0/32 link#4 UmCI en0
"""
# Mocked file descriptor
strio = StringIO.StringIO(netstat_output)
mock_os.popen = mock.MagicMock(return_value=strio)
# Mocked in6_getifaddr() output
mock_in6_getifaddr.return_value = [("::1", IPV6_ADDR_LOOPBACK, "lo0"),
("fe80::a00:27ff:fe9b:c965", IPV6_ADDR_LINKLOCAL, "en0")]
# Test the function
from scapy.arch.unix import read_routes6
routes = read_routes6()
for r in routes:
print r
assert(len(routes) == 5)
assert(check_mandatory_ipv6_routes(routes))
test_osx_10_10_4()
= FreeBSD 10.2
5723
5724
5725
5726
5727
5728
5729
5730
5731
5732
5733
5734
5735
5736
5737
5738
5739
5740
5741
5742
5743
5744
5745
5746
5747
5748
5749
5750
5751
5752
5753
5754
5755
5756
5757
5758
5759
5760
@mock.patch("scapy.arch.unix.in6_getifaddr")
@mock.patch("scapy.arch.unix.os")
def test_freebsd_10_2(mock_os, mock_in6_getifaddr):
"""Test read_routes6() on FreeBSD 10.2"""
# 'netstat -rn -f inet6' output
netstat_output = """
Routing tables
Internet6:
Destination Gateway Flags Netif Expire
::/96 ::1 UGRS lo0
::1 link#2 UH lo0
::ffff:0.0.0.0/96 ::1 UGRS lo0
fe80::/10 ::1 UGRS lo0
fe80::%lo0/64 link#2 U lo0
fe80::1%lo0 link#2 UHS lo0
ff01::%lo0/32 ::1 U lo0
ff02::/16 ::1 UGRS lo0
ff02::%lo0/32 ::1 U lo0
"""
# Mocked file descriptor
strio = StringIO.StringIO(netstat_output)
mock_os.popen = mock.MagicMock(return_value=strio)
# Mocked in6_getifaddr() output
mock_in6_getifaddr.return_value = [("::1", IPV6_ADDR_LOOPBACK, "lo0")]
# Test the function
from scapy.arch.unix import read_routes6
routes = read_routes6()
for r in routes:
print r
assert(len(routes) == 3)
assert(check_mandatory_ipv6_routes(routes))
test_freebsd_10_2()
= OpenBSD 5.5
@mock.patch("scapy.arch.unix.OPENBSD")
5764
5765
5766
5767
5768
5769
5770
5771
5772
5773
5774
5775
5776
5777
5778
5779
5780
5781
5782
5783
5784
5785
5786
5787
5788
5789
5790
5791
5792
5793
5794
5795
5796
5797
5798
5799
5800
5801
5802
5803
5804
5805
5806
5807
5808
5809
5810
5811
5812
5813
5814
5815
5816
5817
5818
@mock.patch("scapy.arch.unix.in6_getifaddr")
@mock.patch("scapy.arch.unix.os")
def test_openbsd_5_5(mock_os, mock_in6_getifaddr, mock_openbsd):
"""Test read_routes6() on OpenBSD 5.5"""
# 'netstat -rn -f inet6' output
netstat_output = """
Routing tables
Internet6:
Destination Gateway Flags Refs Use Mtu Prio Iface
::/104 ::1 UGRS 0 0 - 8 lo0
::/96 ::1 UGRS 0 0 - 8 lo0
::1 ::1 UH 14 0 33144 4 lo0
::127.0.0.0/104 ::1 UGRS 0 0 - 8 lo0
::224.0.0.0/100 ::1 UGRS 0 0 - 8 lo0
::255.0.0.0/104 ::1 UGRS 0 0 - 8 lo0
::ffff:0.0.0.0/96 ::1 UGRS 0 0 - 8 lo0
2002::/24 ::1 UGRS 0 0 - 8 lo0
2002:7f00::/24 ::1 UGRS 0 0 - 8 lo0
2002:e000::/20 ::1 UGRS 0 0 - 8 lo0
2002:ff00::/24 ::1 UGRS 0 0 - 8 lo0
fe80::/10 ::1 UGRS 0 0 - 8 lo0
fe80::%em0/64 link#1 UC 0 0 - 4 em0
fe80::a00:27ff:fe04:59bf%em0 08:00:27:04:59:bf UHL 0 0 - 4 lo0
fe80::%lo0/64 fe80::1%lo0 U 0 0 - 4 lo0
fe80::1%lo0 link#3 UHL 0 0 - 4 lo0
fec0::/10 ::1 UGRS 0 0 - 8 lo0
ff01::/16 ::1 UGRS 0 0 - 8 lo0
ff01::%em0/32 link#1 UC 0 0 - 4 em0
ff01::%lo0/32 fe80::1%lo0 UC 0 0 - 4 lo0
ff02::/16 ::1 UGRS 0 0 - 8 lo0
ff02::%em0/32 link#1 UC 0 0 - 4 em0
ff02::%lo0/32 fe80::1%lo0 UC 0 0 - 4 lo0
"""
# Mocked file descriptor
strio = StringIO.StringIO(netstat_output)
mock_os.popen = mock.MagicMock(return_value=strio)
# Mocked in6_getifaddr() output
mock_in6_getifaddr.return_value = [("::1", IPV6_ADDR_LOOPBACK, "lo0"),
("fe80::a00:27ff:fe04:59bf", IPV6_ADDR_LINKLOCAL, "em0")]
# Mocked OpenBSD parsing behavior
mock_openbsd = True
# Test the function
from scapy.arch.unix import read_routes6
routes = read_routes6()
for r in routes:
print r
assert(len(routes) == 5)
assert(check_mandatory_ipv6_routes(routes))
test_openbsd_5_5()
= NetBSD 7.0
@mock.patch("scapy.arch.unix.NETBSD")
5822
5823
5824
5825
5826
5827
5828
5829
5830
5831
5832
5833
5834
5835
5836
5837
5838
5839
5840
5841
5842
5843
5844
5845
5846
5847
5848
5849
5850
5851
5852
5853
5854
5855
5856
5857
5858
5859
5860
5861
5862
5863
5864
5865
5866
5867
5868
@mock.patch("scapy.arch.unix.in6_getifaddr")
@mock.patch("scapy.arch.unix.os")
def test_netbsd_7_0(mock_os, mock_in6_getifaddr, mock_netbsd):
"""Test read_routes6() on NetBSD 7.0"""
# 'netstat -rn -f inet6' output
netstat_output = """
Routing tables
Internet6:
Destination Gateway Flags Refs Use Mtu Interface
::/104 ::1 UGRS - - - lo0
::/96 ::1 UGRS - - - lo0
::1 ::1 UH - - 33648 lo0
::127.0.0.0/104 ::1 UGRS - - - lo0
::224.0.0.0/100 ::1 UGRS - - - lo0
::255.0.0.0/104 ::1 UGRS - - - lo0
::ffff:0.0.0.0/96 ::1 UGRS - - - lo0
2001:db8::/32 ::1 UGRS - - - lo0
2002::/24 ::1 UGRS - - - lo0
2002:7f00::/24 ::1 UGRS - - - lo0
2002:e000::/20 ::1 UGRS - - - lo0
2002:ff00::/24 ::1 UGRS - - - lo0
fe80::/10 ::1 UGRS - - - lo0
fe80::%wm0/64 link#1 UC - - - wm0
fe80::acd1:3989:180e:fde0 08:00:27:a1:64:d8 UHL - - - lo0
fe80::%lo0/64 fe80::1 U - - - lo0
fe80::1 link#2 UHL - - - lo0
ff01:1::/32 link#1 UC - - - wm0
ff01:2::/32 ::1 UC - - - lo0
ff02::%wm0/32 link#1 UC - - - wm0
ff02::%lo0/32 ::1 UC - - - lo0
"""
# Mocked file descriptor
strio = StringIO.StringIO(netstat_output)
mock_os.popen = mock.MagicMock(return_value=strio)
# Mocked in6_getifaddr() output
mock_in6_getifaddr.return_value = [("::1", IPV6_ADDR_LOOPBACK, "lo0"),
("fe80::acd1:3989:180e:fde0", IPV6_ADDR_LINKLOCAL, "wm0")]
# Test the function
from scapy.arch.unix import read_routes6
routes = read_routes6()
for r in routes:
print r
assert(len(routes) == 5)
assert(check_mandatory_ipv6_routes(routes))
test_netbsd_7_0()
Pierre Lorinquer
committed
+ EAPOL class tests
= EAPOL - Instantiation with specific values
str(EAPOL(version = 3, type = 5)) == b'\x03\x05\x00\x00'
s = b'\x03\x01\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00'
Pierre Lorinquer
committed
assert(eapol.version == 3)
assert(eapol.type == 1)
assert(eapol.len == 0)
s = b'\x03\x00\x00\x05\x01\x01\x00\x05\x01\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00'
Pierre Lorinquer
committed
assert(eapol.version == 3)
assert(eapol.type == 0)
assert(eapol.len == 5)
s = b'\x03\x00\x00\x0e\x02\x01\x00\x0e\x01anonymous\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00'
Pierre Lorinquer
committed
assert(eapol.version == 3)
assert(eapol.type == 0)
assert(eapol.len == 14)
req = EAPOL(b'\x03\x00\x00\x05\x01\x01\x00\x05\x01')
ans = EAPOL(b'\x03\x00\x00\x0e\x02\x01\x00\x0e\x01anonymous')
ans.answers(req)
= EAPOL - Dissection (5)
Pierre Lorinquer
committed
assert(eapol.version == 2)
assert(eapol.type == 0)
assert(eapol.len == 6)
assert(eapol.haslayer(EAP_TLS))
s = b'\x03\x00\x00<\x02\x9e\x00<+\x01\x16\x03\x01\x001\x01\x00\x00-\x03\x01dr1\x93ZS\x0en\xad\x1f\xbaH\xbb\xfe6\xe6\xd0\xcb\xec\xd7\xc0\xd7\xb9\xa5\xc9\x0c\xfd\x98o\xa7T \x00\x00\x04\x004\x00\x00\x01\x00\x00\x00'
Pierre Lorinquer
committed
assert(eapol.version == 3)
assert(eapol.type == 0)
assert(eapol.len == 60)
assert(eapol.haslayer(EAP_FAST))
############
############
+ EAPOL-MKA class tests
= EAPOL-MKA - With Basic parameter set - Dissection
eapol = None
s = b'\x03\x05\x00T\x01\xff\xf0<\x00Bh\xa8\x1e\x03\x00\n\xbcj\x00\x96Ywz\x82:\x90\xd9\xe7\x00\x00\x00\x01\x00\x80\xc2\x01\x11\x11\x11\x11\x11\x11\x11\x11\x11\x11\x11\x11\x11\x11\x11\x11\x11\x11\x11\x11\x11\x11\x11\x11\x11\x11\x11\x11\x11\x11\x11\x11\xff\x00\x00\x10\xe5\xf5j\x86V\\\xb1\xcc\xa9\xb95\x04m*Cj'
eapol = EAPOL(s)
assert(eapol.version == 3)
assert(eapol.type == 5)
assert(eapol.len == 84)
assert(eapol.haslayer(MKAPDU))
assert(eapol[MKAPDU].basic_param_set.actor_member_id == b"\xbcj\x00\x96Ywz\x82:\x90\xd9\xe7")
assert(eapol[MKAPDU].haslayer(MKAICVSet))
assert(eapol[MKAPDU][MKAICVSet].icv == b"\xe5\xf5j\x86V\\\xb1\xcc\xa9\xb95\x04m*Cj")
= EAPOL-MKA - With Potential Peer List parameter set - Dissection
eapol = None
s = b'\x03\x05\x00h\x01\x10\xe0<\xccN$\xc4\xf7\x7f\x00\x80q\x8b\x8a9\x86k/X\x14\xc9\xdc\xf6\x00\x00\x00}\x00\x80\xc2\x01\x11\x11\x11\x11\x11\x11\x11\x11\x11\x11\x11\x11\x11\x11\x11\x11\x11\x11\x11\x11\x11\x11\x11\x11\x11\x11\x11\x11\x11\x11\x11\x11\x02\x00\x00\x10\xbcj\x00\x96Ywz\x82:\x90\xd9\xe7\x00\x00\x00\x01\xff\x00\x00\x105\x01\xdc)\xfd\xd1\xff\xd55\x9c_o\xc9\x9c\xca\xc0'
eapol = EAPOL(s)
assert(eapol.version == 3)
assert(eapol.type == 5)
assert(eapol.len == 104)
assert(eapol.haslayer(MKAPDU))
assert(eapol[MKAPDU].basic_param_set.actor_member_id == b"q\x8b\x8a9\x86k/X\x14\xc9\xdc\xf6")
assert(eapol.haslayer(MKAPotentialPeerListParamSet))
assert(eapol[MKAPDU][MKAPotentialPeerListParamSet].member_id_message_num[0].member_id == b"\xbcj\x00\x96Ywz\x82:\x90\xd9\xe7")
assert(eapol[MKAPDU].haslayer(MKAICVSet))
assert(eapol[MKAPDU][MKAICVSet].icv == b"5\x01\xdc)\xfd\xd1\xff\xd55\x9c_o\xc9\x9c\xca\xc0")
= EAPOL-MKA - With Live Peer List parameter set - Dissection
eapol = None
s = b"\x03\x05\x00h\x01\xffp<\x00Bh\xa8\x1e\x03\x00\n\xbcj\x00\x96Ywz\x82:\x90\xd9\xe7\x00\x00\x00\x02\x00\x80\xc2\x01\x11\x11\x11\x11\x11\x11\x11\x11\x11\x11\x11\x11\x11\x11\x11\x11\x11\x11\x11\x11\x11\x11\x11\x11\x11\x11\x11\x11\x11\x11\x11\x11\x01\x00\x00\x10q\x8b\x8a9\x86k/X\x14\xc9\xdc\xf6\x00\x00\x00\x80\xff\x00\x00\x10\xf4\xa1d\x18\tD\xa2}\x8e'\x0c/\xda,\xea\xb7"
eapol = EAPOL(s)
assert(eapol.version == 3)
assert(eapol.type == 5)
assert(eapol.len == 104)
assert(eapol.haslayer(MKAPDU))
assert(eapol[MKAPDU].basic_param_set.actor_member_id == b'\xbcj\x00\x96Ywz\x82:\x90\xd9\xe7')
assert(eapol.haslayer(MKALivePeerListParamSet))
assert(eapol[MKAPDU][MKALivePeerListParamSet].member_id_message_num[0].member_id == b"q\x8b\x8a9\x86k/X\x14\xc9\xdc\xf6")
assert(eapol[MKAPDU].haslayer(MKAICVSet))
assert(eapol[MKAPDU][MKAICVSet].icv == b"\xf4\xa1d\x18\tD\xa2}\x8e'\x0c/\xda,\xea\xb7")
= EAPOL-MKA - With SAK Use parameter set - Dissection
eapol = None
s = b'\x03\x05\x00\x94\x01\xffp<\x00Bh\xa8\x1e\x03\x00\n\xbcj\x00\x96Ywz\x82:\x90\xd9\xe7\x00\x00\x00\x03\x00\x80\xc2\x01\x11\x11\x11\x11\x11\x11\x11\x11\x11\x11\x11\x11\x11\x11\x11\x11\x11\x11\x11\x11\x11\x11\x11\x11\x11\x11\x11\x11\x11\x11\x11\x11\x03\x10\x00(q\x8b\x8a9\x86k/X\x14\xc9\xdc\xf6\x00\x00\x00\x01\x00\x00\x00\x00q\x8b\x8a9\x86k/X\x14\xc9\xdc\xf6\x00\x00\x00\x01\x00\x00\x00\x00\x01\x00\x00\x10q\x8b\x8a9\x86k/X\x14\xc9\xdc\xf6\x00\x00\x00\x83\xff\x00\x00\x10OF\x84\xf1@%\x95\xe6Fw9\x1a\xfa\x03(\xae'
eapol = EAPOL(s)
assert(eapol.version == 3)
assert(eapol.type == 5)
assert(eapol.len == 148)
assert(eapol.haslayer(MKAPDU))
assert(eapol[MKAPDU].basic_param_set.actor_member_id == b'\xbcj\x00\x96Ywz\x82:\x90\xd9\xe7')
assert(eapol.haslayer(MKASAKUseParamSet))
assert(eapol[MKAPDU][MKASAKUseParamSet].latest_key_key_server_member_id == b"q\x8b\x8a9\x86k/X\x14\xc9\xdc\xf6")
assert(eapol.haslayer(MKALivePeerListParamSet))
assert(eapol[MKAPDU][MKALivePeerListParamSet].member_id_message_num[0].member_id == b"q\x8b\x8a9\x86k/X\x14\xc9\xdc\xf6")
assert(eapol[MKAPDU].haslayer(MKAICVSet))
assert(eapol[MKAPDU][MKAICVSet].icv == b"OF\x84\xf1@%\x95\xe6Fw9\x1a\xfa\x03(\xae")
= EAPOL-MKA - With Distributed SAK parameter set - Dissection
eapol = None
s = b"\x03\x05\x00\xb4\x01\x10\xe0<\xccN$\xc4\xf7\x7f\x00\x80q\x8b\x8a9\x86k/X\x14\xc9\xdc\xf6\x00\x00\x00\x81\x00\x80\xc2\x01\x11\x11\x11\x11\x11\x11\x11\x11\x11\x11\x11\x11\x11\x11\x11\x11\x11\x11\x11\x11\x11\x11\x11\x11\x11\x11\x11\x11\x11\x11\x11\x11\x01\x00\x00\x10\xbcj\x00\x96Ywz\x82:\x90\xd9\xe7\x00\x00\x00\x02\x03\x10\x00(q\x8b\x8a9\x86k/X\x14\xc9\xdc\xf6\x00\x00\x00\x01\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x04\x10\x00\x1c\x00\x00\x00\x01Cz\x05\x88\x9f\xe8-\x94W+?\x13~\xfb\x016yVB?\xbd\xa1\x9fu\xff\x00\x00\x10\xb0H\xcf\xe0:\xa1\x94RD'\x03\xe67\xe1Ur"
eapol = EAPOL(s)
assert(eapol.version == 3)
assert(eapol.type == 5)
assert(eapol.len == 180)
assert(eapol.haslayer(MKAPDU))
assert(eapol[MKAPDU].basic_param_set.actor_member_id == b"q\x8b\x8a9\x86k/X\x14\xc9\xdc\xf6")
assert(eapol.haslayer(MKASAKUseParamSet))
assert(eapol[MKAPDU][MKASAKUseParamSet].latest_key_key_server_member_id == b"q\x8b\x8a9\x86k/X\x14\xc9\xdc\xf6")
assert(eapol.haslayer(MKALivePeerListParamSet))
assert(eapol[MKAPDU][MKALivePeerListParamSet].member_id_message_num[0].member_id == b"\xbcj\x00\x96Ywz\x82:\x90\xd9\xe7")
assert(eapol.haslayer(MKADistributedSAKParamSet))
assert(eapol[MKADistributedSAKParamSet].sak_aes_key_wrap == b"Cz\x05\x88\x9f\xe8-\x94W+?\x13~\xfb\x016yVB?\xbd\xa1\x9fu")
assert(eapol[MKAPDU].haslayer(MKAICVSet))
assert(eapol[MKAPDU][MKAICVSet].icv == b"\xb0H\xcf\xe0:\xa1\x94RD'\x03\xe67\xe1Ur")