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 logging
  31 logging.getLogger("scapy").setLevel(logging.CRITICAL)
  32 import scapy.all as sp
  33 import socket
  34 import sys
  35 from sniffer import Sniffer
  36
  37 PAYLOAD_MAGIC = bytes.fromhex('42c0ffee')
  38
  39 def check_ping_request(args, packet):
  40         if args.ip6:
  41                 return check_ping6_request(args, packet)
  42         else:
  43                 return check_ping4_request(args, packet)
  44
  45 def check_ping4_request(args, packet):
  46         """
  47         Verify that the packet matches what we'd have sent
  48         """
  49         dst_ip = args.to[0]
  50
  51         ip = packet.getlayer(sp.IP)
  52         if not ip:
  53                 return False
  54         if ip.dst != dst_ip:
  55                 return False
  56
  57         icmp = packet.getlayer(sp.ICMP)
  58         if not icmp:
  59                 return False
  60         if sp.icmptypes[icmp.type] != 'echo-request':
  61                 return False
  62
  63         raw = packet.getlayer(sp.Raw)
  64         if not raw:
  65                 return False
  66         if raw.load != PAYLOAD_MAGIC:
  67                 return False
  68
  69         # Wait to check expectations until we've established this is the packet we
  70         # sent.
  71         if args.expect_tos:
  72                 if ip.tos != int(args.expect_tos[0]):
  73                         print("Unexpected ToS value %d, expected %d" \
  74                                 % (ip.tos, int(args.expect_tos[0])))
  75                         return False
  76
  77         return True
  78
  79 def check_ping6_request(args, packet):
  80         """
  81         Verify that the packet matches what we'd have sent
  82         """
  83         dst_ip = args.to[0]
  84
  85         ip = packet.getlayer(sp.IPv6)
  86         if not ip:
  87                 return False
  88         if ip.dst != dst_ip:
  89                 return False
  90
  91         icmp = packet.getlayer(sp.ICMPv6EchoRequest)
  92         if not icmp:
  93                 return False
  94         if icmp.data != PAYLOAD_MAGIC:
  95                 return False
  96
  97         return True
  98
  99 def ping(send_if, dst_ip, args):
 100         ether = sp.Ether()
 101         ip = sp.IP(dst=dst_ip)
 102         icmp = sp.ICMP(type='echo-request')
 103         raw = sp.raw(PAYLOAD_MAGIC)
 104
 105         if args.send_tos:
 106                 ip.tos = int(args.send_tos[0])
 107
 108         req = ether / ip / icmp / raw
 109         sp.sendp(req, iface=send_if, verbose=False)
 110
 111 def ping6(send_if, dst_ip, args):
 112         ether = sp.Ether()
 113         ip6 = sp.IPv6(dst=dst_ip)
 114         icmp = sp.ICMPv6EchoRequest(data=sp.raw(PAYLOAD_MAGIC))
 115
 116         req = ether / ip6 / icmp
 117         sp.sendp(req, iface=send_if, verbose=False)
 118
 119 def check_tcpsyn(args, packet):
 120         dst_ip = args.to[0]
 121
 122         ip = packet.getlayer(sp.IP)
 123         if not ip:
 124                 return False
 125         if ip.dst != dst_ip:
 126                 return False
 127
 128         tcp = packet.getlayer(sp.TCP)
 129         if not tcp:
 130                 return False
 131
 132         # Verify IP checksum
 133         chksum = ip.chksum
 134         ip.chksum = None
 135         new_chksum = sp.IP(sp.raw(ip)).chksum
 136         if chksum != new_chksum:
 137                 print("Expected IP checksum %x but found %x\n" % (new_cshkum, chksum))
 138                 return False
 139
 140         # Verify TCP checksum
 141         chksum = tcp.chksum
 142         packet_raw = sp.raw(packet)
 143         tcp.chksum = None
 144         newpacket = sp.Ether(sp.raw(packet[sp.Ether]))
 145         new_chksum = newpacket[sp.TCP].chksum
 146         if chksum != new_chksum:
 147                 print("Expected TCP checksum %x but found %x\n" % (new_chksum, chksum))
 148                 return False
 149
 150         return True
 151
 152 def tcpsyn(send_if, dst_ip, args):
 153         opts=[('Timestamp', (1, 1)), ('MSS', 1280)]
 154
 155         if args.tcpopt_unaligned:
 156                 opts = [('NOP', 0 )] + opts
 157
 158         ether = sp.Ether()
 159         ip = sp.IP(dst=dst_ip)
 160         tcp = sp.TCP(dport=666, flags='S', options=opts)
 161
 162         req = ether / ip / tcp
 163         sp.sendp(req, iface=send_if, verbose=False)
 164
 165
 166 def main():
 167         parser = argparse.ArgumentParser("pft_ping.py",
 168                 description="Ping test tool")
 169         parser.add_argument('--sendif', nargs=1,
 170                 required=True,
 171                 help='The interface through which the packet(s) will be sent')
 172         parser.add_argument('--recvif', nargs=1,
 173                 help='The interface on which to expect the ICMP echo response')
 174         parser.add_argument('--ip6', action='store_true',
 175                 help='Use IPv6')
 176         parser.add_argument('--to', nargs=1,
 177                 required=True,
 178                 help='The destination IP address for the ICMP echo request')
 179
 180         # TCP options
 181         parser.add_argument('--tcpsyn', action='store_true',
 182                         help='Send a TCP SYN packet')
 183         parser.add_argument('--tcpopt_unaligned', action='store_true',
 184                         help='Include unaligned TCP options')
 185
 186         # Packet settings
 187         parser.add_argument('--send-tos', nargs=1,
 188                 help='Set the ToS value for the transmitted packet')
 189
 190         # Expectations
 191         parser.add_argument('--expect-tos', nargs=1,
 192                 help='The expected ToS value in the received packet')
 193
 194         args = parser.parse_args()
 195
 196         # We may not have a default route. Tell scapy where to start looking for routes
 197         sp.conf.iface6 = args.sendif[0]
 198
 199         sniffer = None
 200         if not args.recvif is None:
 201                 checkfn=check_ping_request
 202                 if args.tcpsyn:
 203                         checkfn=check_tcpsyn
 204
 205                 sniffer = Sniffer(args, checkfn)
 206
 207         if args.tcpsyn:
 208                 tcpsyn(args.sendif[0], args.to[0], args)
 209         else:
 210                 if args.ip6:
 211                         ping6(args.sendif[0], args.to[0], args)
 212                 else:
 213                         ping(args.sendif[0], args.to[0], args)
 214
 215         if sniffer:
 216                 sniffer.join()
 217
 218                 if sniffer.foundCorrectPacket:
 219                         sys.exit(0)
 220                 else:
 221                         sys.exit(1)
 222
 223 if __name__ == '__main__':
 224         main()