1#!/usr/local/bin/python3 2# send 2 non-overlapping ping6 fragments 3 4import os 5import threading 6from addr import * 7from scapy.all import * 8 9class Sniff1(threading.Thread): 10 filter = None 11 captured = None 12 packet = None 13 def run(self): 14 self.captured = sniff(iface=SRC_IF, filter=self.filter, 15 count=1, timeout=3) 16 if self.captured: 17 self.packet = self.captured[0] 18 19dstaddr=sys.argv[1] 20pid=os.getpid() 21eid=pid & 0xffff 22payload=b"ABCDEFGHIJKLOMNO" 23packet=IPv6(src=SRC_OUT6, dst=dstaddr)/ICMPv6EchoRequest(id=eid, data=payload) 24fid=pid & 0xffffffff 25frag0=IPv6ExtHdrFragment(nh=58, id=fid, m=1)/bytes(packet)[40:56] 26frag1=IPv6ExtHdrFragment(nh=58, id=fid, offset=2)/bytes(packet)[56:64] 27pkt0=IPv6(src=SRC_OUT6, dst=dstaddr)/frag0 28pkt1=IPv6(src=SRC_OUT6, dst=dstaddr)/frag1 29eth=[] 30eth.append(Ether(src=SRC_MAC, dst=PF_MAC)/pkt0) 31eth.append(Ether(src=SRC_MAC, dst=PF_MAC)/pkt1) 32 33sniffer = Sniff1(); 34sniffer.filter = "ip6 and src %s and dst %s and icmp6" % (dstaddr, SRC_OUT6) 35sniffer.start() 36time.sleep(1) 37sendp(eth, iface=SRC_IF) 38sniffer.join(timeout=5) 39a = sniffer.packet 40 41if a and a.type == ETH_P_IPV6 and \ 42 ipv6nh[a.payload.nh] == 'ICMPv6' and \ 43 icmp6types[a.payload.payload.type] == 'Echo Reply': 44 id=a.payload.payload.id 45 print("id=%#x" % (id)) 46 if id != eid: 47 print("WRONG ECHO REPLY ID") 48 exit(2) 49 data=a.payload.payload.data 50 print("payload=%s" % (data)) 51 if data == payload: 52 exit(0) 53 print("PAYLOAD!=%s" % (payload)) 54 exit(1) 55print("NO ECHO REPLY") 56exit(2) 57