1#!/usr/local/bin/python2.7 2 3import argparse 4import scapy.all as sp 5import sys 6import threading 7 8PAYLOAD_MAGIC = 0x42c0ffee 9 10class Sniffer(threading.Thread): 11 def __init__(self, args, check_function): 12 threading.Thread.__init__(self) 13 14 self._args = args 15 self._recvif = args.recvif[0] 16 self._check_function = check_function 17 self.foundCorrectPacket = False 18 19 self.start() 20 21 def _checkPacket(self, packet): 22 ret = self._check_function(self._args, packet) 23 if ret: 24 self.foundCorrectPacket = True 25 return ret 26 27 def run(self): 28 self.packets = sp.sniff(iface=self._recvif, 29 stop_filter=self._checkPacket, timeout=3) 30 31def check_ping_request(args, packet): 32 if args.ip6: 33 return check_ping6_request(args, packet) 34 else: 35 return check_ping4_request(args, packet) 36 37def check_ping4_request(args, packet): 38 """ 39 Verify that the packet matches what we'd have sent 40 """ 41 dst_ip = args.to[0] 42 43 ip = packet.getlayer(sp.IP) 44 if not ip: 45 return False 46 if ip.dst != dst_ip: 47 return False 48 49 icmp = packet.getlayer(sp.ICMP) 50 if not icmp: 51 return False 52 if sp.icmptypes[icmp.type] != 'echo-request': 53 return False 54 55 raw = packet.getlayer(sp.Raw) 56 if not raw: 57 return False 58 if raw.load != str(PAYLOAD_MAGIC): 59 return False 60 61 # Wait to check expectations until we've established this is the packet we 62 # sent. 63 if args.expect_tos: 64 if ip.tos != int(args.expect_tos[0]): 65 print "Unexpected ToS value %d, expected %s" \ 66 % (ip.tos, args.expect_tos[0]) 67 return False 68 69 return True 70 71def check_ping6_request(args, packet): 72 """ 73 Verify that the packet matches what we'd have sent 74 """ 75 dst_ip = args.to[0] 76 77 ip = packet.getlayer(sp.IPv6) 78 if not ip: 79 return False 80 if ip.dst != dst_ip: 81 return False 82 83 icmp = packet.getlayer(sp.ICMPv6EchoRequest) 84 if not icmp: 85 return False 86 if icmp.data != str(PAYLOAD_MAGIC): 87 return False 88 89 return True 90 91def ping(send_if, dst_ip, args): 92 ether = sp.Ether() 93 ip = sp.IP(dst=dst_ip) 94 icmp = sp.ICMP(type='echo-request') 95 raw = sp.Raw(str(PAYLOAD_MAGIC)) 96 97 if args.send_tos: 98 ip.tos = int(args.send_tos[0]) 99 100 req = ether / ip / icmp / raw 101 sp.sendp(req, iface=send_if, verbose=False) 102 103def ping6(send_if, dst_ip, args): 104 ether = sp.Ether() 105 ip6 = sp.IPv6(dst=dst_ip) 106 icmp = sp.ICMPv6EchoRequest(data=PAYLOAD_MAGIC) 107 108 req = ether / ip6 / icmp 109 sp.sendp(req, iface=send_if, verbose=False) 110 111def main(): 112 parser = argparse.ArgumentParser("pft_ping.py", 113 description="Ping test tool") 114 parser.add_argument('--sendif', nargs=1, 115 required=True, 116 help='The interface through which the packet(s) will be sent') 117 parser.add_argument('--recvif', nargs=1, 118 help='The interface on which to expect the ICMP echo response') 119 parser.add_argument('--ip6', action='store_true', 120 help='Use IPv6') 121 parser.add_argument('--to', nargs=1, 122 required=True, 123 help='The destination IP address for the ICMP echo request') 124 125 # Packet settings 126 parser.add_argument('--send-tos', nargs=1, 127 help='Set the ToS value for the transmitted packet') 128 129 # Expectations 130 parser.add_argument('--expect-tos', nargs=1, 131 help='The expected ToS value in the received packet') 132 133 args = parser.parse_args() 134 135 # We may not have a default route. Tell scapy where to start looking for routes 136 sp.conf.iface6 = args.sendif[0] 137 138 sniffer = None 139 if not args.recvif is None: 140 sniffer = Sniffer(args, check_ping_request) 141 142 if args.ip6: 143 ping6(args.sendif[0], args.to[0], args) 144 else: 145 ping(args.sendif[0], args.to[0], args) 146 147 if sniffer: 148 sniffer.join() 149 150 if sniffer.foundCorrectPacket: 151 sys.exit(0) 152 else: 153 sys.exit(1) 154 155if __name__ == '__main__': 156 main() 157