tests/sys/netpfil/common/pft_ping.py

   1 #!/usr/bin/env python
   2 #
   3 # SPDX-License-Identifier: BSD-2-Clause
   4 #
   5 # Copyright (c) 2017 Kristof Provost <kp@FreeBSD.org>
   6 #
   7 # Redistribution and use in source and binary forms, with or without
   8 # modification, are permitted provided that the following conditions
   9 # are met:
  10 # 1. Redistributions of source code must retain the above copyright
  11 #    notice, this list of conditions and the following disclaimer.
  12 # 2. Redistributions in binary form must reproduce the above copyright
  13 #    notice, this list of conditions and the following disclaimer in the
  14 #    documentation and/or other materials provided with the distribution.
  15 #
  16 # THIS SOFTWARE IS PROVIDED BY THE AUTHOR AND CONTRIBUTORS ``AS IS'' AND
  17 # ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE
  18 # IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE
  19 # ARE DISCLAIMED.  IN NO EVENT SHALL THE AUTHOR OR CONTRIBUTORS BE LIABLE
  20 # FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR CONSEQUENTIAL
  21 # DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS
  22 # OR SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS INTERRUPTION)
  23 # HOWEVER CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT
  24 # LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY
  25 # OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF
  26 # SUCH DAMAGE.
  27 #
  28
  29 import argparse
  30 import scapy.all as sp
  31 import sys
  32 from sniffer import Sniffer
  33
  34 PAYLOAD_MAGIC = bytes.fromhex('42c0ffee')
  35
  36 def check_ping_request(args, packet):
  37         if args.ip6:
  38                 return check_ping6_request(args, packet)
  39         else:
  40                 return check_ping4_request(args, packet)
  41
  42 def check_ping4_request(args, packet):
  43         """
  44         Verify that the packet matches what we'd have sent
  45         """
  46         dst_ip = args.to[0]
  47
  48         ip = packet.getlayer(sp.IP)
  49         if not ip:
  50                 return False
  51         if ip.dst != dst_ip:
  52                 return False
  53
  54         icmp = packet.getlayer(sp.ICMP)
  55         if not icmp:
  56                 return False
  57         if sp.icmptypes[icmp.type] != 'echo-request':
  58                 return False
  59
  60         raw = packet.getlayer(sp.Raw)
  61         if not raw:
  62                 return False
  63         if raw.load != PAYLOAD_MAGIC:
  64                 return False
  65
  66         # Wait to check expectations until we've established this is the packet we
  67         # sent.
  68         if args.expect_tos:
  69                 if ip.tos != int(args.expect_tos[0]):
  70                         print("Unexpected ToS value %d, expected %d" \
  71                                 % (ip.tos, int(args.expect_tos[0])))
  72                         return False
  73
  74         return True
  75
  76 def check_ping6_request(args, packet):
  77         """
  78         Verify that the packet matches what we'd have sent
  79         """
  80         dst_ip = args.to[0]
  81
  82         ip = packet.getlayer(sp.IPv6)
  83         if not ip:
  84                 return False
  85         if ip.dst != dst_ip:
  86                 return False
  87
  88         icmp = packet.getlayer(sp.ICMPv6EchoRequest)
  89         if not icmp:
  90                 return False
  91         if icmp.data != PAYLOAD_MAGIC:
  92                 return False
  93
  94         return True
  95
  96 def ping(send_if, dst_ip, args):
  97         ether = sp.Ether()
  98         ip = sp.IP(dst=dst_ip)
  99         icmp = sp.ICMP(type='echo-request')
 100         raw = sp.raw(PAYLOAD_MAGIC)
 101
 102         if args.send_tos:
 103                 ip.tos = int(args.send_tos[0])
 104
 105         req = ether / ip / icmp / raw
 106         sp.sendp(req, iface=send_if, verbose=False)
 107
 108 def ping6(send_if, dst_ip, args):
 109         ether = sp.Ether()
 110         ip6 = sp.IPv6(dst=dst_ip)
 111         icmp = sp.ICMPv6EchoRequest(data=sp.raw(PAYLOAD_MAGIC))
 112
 113         req = ether / ip6 / icmp
 114         sp.sendp(req, iface=send_if, verbose=False)
 115
 116 def main():
 117         parser = argparse.ArgumentParser("pft_ping.py",
 118                 description="Ping test tool")
 119         parser.add_argument('--sendif', nargs=1,
 120                 required=True,
 121                 help='The interface through which the packet(s) will be sent')
 122         parser.add_argument('--recvif', nargs=1,
 123                 help='The interface on which to expect the ICMP echo response')
 124         parser.add_argument('--ip6', action='store_true',
 125                 help='Use IPv6')
 126         parser.add_argument('--to', nargs=1,
 127                 required=True,
 128                 help='The destination IP address for the ICMP echo request')
 129
 130         # Packet settings
 131         parser.add_argument('--send-tos', nargs=1,
 132                 help='Set the ToS value for the transmitted packet')
 133
 134         # Expectations
 135         parser.add_argument('--expect-tos', nargs=1,
 136                 help='The expected ToS value in the received packet')
 137
 138         args = parser.parse_args()
 139
 140         # We may not have a default route. Tell scapy where to start looking for routes
 141         sp.conf.iface6 = args.sendif[0]
 142
 143         sniffer = None
 144         if not args.recvif is None:
 145                 sniffer = Sniffer(args, check_ping_request)
 146
 147         if args.ip6:
 148                 ping6(args.sendif[0], args.to[0], args)
 149         else:
 150                 ping(args.sendif[0], args.to[0], args)
 151
 152         if sniffer:
 153                 sniffer.join()
 154
 155                 if sniffer.foundCorrectPacket:
 156                         sys.exit(0)
 157                 else:
 158                         sys.exit(1)
 159
 160 if __name__ == '__main__':
 161         main()