]> CyberLeo.Net >> Repos - FreeBSD/FreeBSD.git/blob - tests/sys/netinet6/frag6/frag6_01.py
libevent: Import libevent 2.1.12
[FreeBSD/FreeBSD.git] / tests / sys / netinet6 / frag6 / frag6_01.py
1 #!/usr/bin/env python
2 #-
3 # SPDX-License-Identifier: BSD-2-Clause
4 #
5 # Copyright (c) 2019 Netflix, Inc.
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
30 import argparse
31 import logging
32 logging.getLogger("scapy").setLevel(logging.CRITICAL)
33 import scapy.all as sp
34 import socket
35 import sys
36 from sniffer import Sniffer
37 from time import sleep
38
39 def check_icmp6_error(args, packet):
40         ip6 = packet.getlayer(sp.IPv6)
41         if not ip6:
42                 return False
43         oip6 = sp.IPv6(src=args.src[0], dst=args.to[0])
44         if ip6.dst != oip6.src:
45                 return False
46         icmp6 = packet.getlayer(sp.ICMPv6ParamProblem)
47         if not icmp6:
48                 return False
49         # ICMP6_PARAMPROB_HEADER 0
50         if icmp6.code != 0:
51                 return False
52         # Should we check the payload as well?
53         # We are running in a very isolated environment and nothing else
54         # should trigger an ICMPv6 Param Prob so leave it.
55         #icmp6.display()
56         return True
57
58 def main():
59         parser = argparse.ArgumentParser("frag6.py",
60                 description="IPv6 fragementation test tool")
61         parser.add_argument('--sendif', nargs=1,
62                 required=True,
63                 help='The interface through which the packet will be sent')
64         parser.add_argument('--recvif', nargs=1,
65                 required=True,
66                 help='The interface on which to check for the packet')
67         parser.add_argument('--src', nargs=1,
68                 required=True,
69                 help='The source IP address')
70         parser.add_argument('--to', nargs=1,
71                 required=True,
72                 help='The destination IP address')
73         parser.add_argument('--debug',
74                 required=False, action='store_true',
75                 help='Enable test debugging')
76
77         args = parser.parse_args()
78
79
80         # Start sniffing on recvif
81         sniffer = Sniffer(args, check_icmp6_error)
82
83
84         ########################################################################
85         #
86         # A single start fragment with zero length IPv6 header (jumbo).
87         # Make sure we do hit the Fragment case, which is tricky as the
88         # jumbogram needs to be > 64k.
89         #
90         # A:  Jumbo-Fragment not allowed.
91         # R:  ICMPv6 param problem.
92         #
93         #data = "6" * (65536 - 2 - 6 - 8 - 8)
94         data = "6" * 65512
95         ip6f01 = sp.Ether() / \
96                 sp.IPv6(src=args.src[0], dst=args.to[0], plen=0) / \
97                 sp.IPv6ExtHdrHopByHop(options=sp.Jumbo(jumboplen=65536)) / \
98                 sp.IPv6ExtHdrFragment(offset=0, m=1, id=6) / \
99                 sp.UDP(dport=3456, sport=6543) / \
100                 data
101         if args.debug :
102                 ip6f01.display()
103         sp.sendp(ip6f01, iface=args.sendif[0], verbose=False)
104
105         # We should only need to sleep 0.10 but it seems scapy
106         # takes time for this one.
107         sleep(3)
108         sniffer.setEnd()
109         sniffer.join()
110         if not sniffer.foundCorrectPacket:
111                 sys.exit(1)
112
113         sys.exit(0)
114
115 if __name__ == '__main__':
116         main()