[PATCH net-next v9 10/10] test: Add networking selftest for eh limits

[Tom Herbert <tom@herbertland.com>]

Add a networking selftest for Extension Header limits. The
limits to test are in systcls:

	net.ipv6.enforce_ext_hdr_order
	net.ipv6.max_dst_opts_number
	net.ipv6.max_hbh_opts_number
	net.ipv6.max_hbh_length
	net.ipv6.max_dst_opts_length

The basic idea of the test is to fabricate ICMPv6 Echo Request
packets with various combinations of Extension Headers. The packets
are sent to a host in another namespace. If a an ICMPv6 Echo Reply
is received then the packet wasn't dropped due to a limit being
exceeded, and if it was dropped then we assume that a limit was
exceeded. For each test packet we derive an expectation as to
whether the packet will be dropped or not. Test success depends
on whether our expectation is matched. i.e. if we expect a reply
then the test succeeds if we see a reply, and if we don't expect a
reply then the test succeeds if we don't see a reply.

The test is divided into a frontend bash script (eh_limits.sh) and a
backend Python script (eh_limits.py).

The frontend sets up two network namespaces with IPv6 addresses
configured on veth's. We then invoke the backend to send the
test packets. This first pass is done with default sysctl settings.
On a second pass we change the various sysctl settings and run
again.

The backend runs through the various test cases described in the
Make_Test_Packets function. This function calls Make_Packet for
a test case where arguments provide the Extension Header chain to
be tested. The Run_Test function loops through the various packets
and tests if a reply is received versus the expectation. If a test
case fails then an error status is returned by the backend.

The backend script can also be run with the "-w <pcap_file>" to
write the created packets to a pcap file instead of running the
test.

Signed-off-by: Tom Herbert <tom@herbertland.com>
---
 tools/testing/selftests/net/Makefile     |   1 +
 tools/testing/selftests/net/eh_limits.py | 349 +++++++++++++++++++++++
 tools/testing/selftests/net/eh_limits.sh | 205 +++++++++++++
 3 files changed, 555 insertions(+)
 create mode 100755 tools/testing/selftests/net/eh_limits.py
 create mode 100755 tools/testing/selftests/net/eh_limits.sh

diff --git a/tools/testing/selftests/net/Makefile b/tools/testing/selftests/net/Makefile
index 6bced3ed798b..bd07d5c28905 100644
--- a/tools/testing/selftests/net/Makefile
+++ b/tools/testing/selftests/net/Makefile
@@ -25,6 +25,7 @@ TEST_PROGS := \
 	cmsg_time.sh \
 	double_udp_encap.sh \
 	drop_monitor_tests.sh \
+	eh_limits.sh \
 	fcnal-ipv4.sh \
 	fcnal-ipv6.sh \
 	fcnal-other.sh \
diff --git a/tools/testing/selftests/net/eh_limits.py b/tools/testing/selftests/net/eh_limits.py
new file mode 100755
index 000000000000..46a460b9149e
--- /dev/null
+++ b/tools/testing/selftests/net/eh_limits.py
@@ -0,0 +1,349 @@
+#!/usr/bin/env python3
+# SPDX-License-Identifier: GPL-2.0
+
+# Test of extension header limits
+
+import getopt
+import struct
+import sys
+import socket
+import scapy.all
+import proto_nums
+import ext_hdr
+
+# Constants
+
+VERBOSE = False
+SOURCE_MAC = "00:11:22:33:44:55"
+DESTINATION_MAC = "AA:BB:CC:DD:EE:FF"
+SOURCE_IP = "2001:db8::7"
+DESTINATION_IP = "2001:db8::8"
+PACKET_LIST = []
+PCAP_OUT=""
+GLOB_IDENT = 1111
+ETHER_FRAME = scapy.all.Raw()
+WITH_ETH = False
+
+# Parse command line options
+def cli_args():
+    global VERBOSE, SOURCE_MAC, DESTINATION_MAC, SOURCE_IP
+    global DESTINATION_IP, PCAP_OUT
+
+    args = sys.argv[1:]
+
+    try:
+        opts, args = getopt.getopt(args, "vw:",
+                        ["verbose", "src_eth=", "dst_eth", "src_ip=",
+                         "dst_ip=", "pcap_out="])
+    except getopt.GetoptError as err:
+        # Print error message and exit
+        print(err)
+        sys.exit(2)
+
+    for opt, arg in opts:
+        if opt in ("-v", "--verbose"):
+            VERBOSE = True
+        elif opt in ("--src_eth"):
+            SOURCE_MAC = arg
+        elif opt in ("--dst_eth"):
+            DESTINATION_MAC = arg
+        elif opt in ("--src_ip"):
+            SOURCE_IP = arg
+        elif opt in ("--dst_ip"):
+            DESTINATION_IP = arg
+        elif opt in ("-w", "--pcap_out"):
+            PCAP_OUT = arg
+
+# Make an ICMP echo request packet with the requested Extension Header chain
+def make_packet(text_name, eh_list):
+    global GLOB_IDENT
+
+    hdr = scapy.all.Raw()
+    plen = 0
+
+    hdr = scapy.all.ICMPv6EchoRequest(id=GLOB_IDENT)/hdr
+    plen += 8
+
+    pair = ext_hdr.make_eh_chain(
+            proto_nums.IP_Proto.IP_PROTO_IPv6_ICMP.value, eh_list)
+    hdr = pair[0]/hdr
+    plen += pair[1]
+
+    ipv6_pkt = scapy.all.IPv6(src=SOURCE_IP, dst=DESTINATION_IP,
+                              nh=pair[2], plen=plen)
+    hdr = ipv6_pkt / hdr
+    plen += 40
+
+    if WITH_ETH:
+        hdr = ETHER_FRAME/hdr
+        plen += 14
+
+    PACKET_LIST.append((hdr, plen, GLOB_IDENT, pair[3], text_name))
+    GLOB_IDENT += 1
+
+# Write a pacp file with all the created packets
+def write_pcap(pcap_out):
+    global ETHER_FRAME, WITH_ETH
+
+    ETHER_FRAME = scapy.all.Ether(src=SOURCE_MAC, dst=DESTINATION_MAC,
+                                  type=0x86DD)
+    WITH_ETH = True
+
+    packets=[]
+    for packet in PACKET_LIST:
+        packets.append(packet[0])
+
+    scapy.all.wrpcap(pcap_out, packets)
+
+def process_return(recvd_it, packet):
+    if VERBOSE:
+        if recvd_it:
+            if packet[3]:
+                print(f"TEST: {packet[4]}: Received as expected")
+            else:
+                print(f"TEST: {packet[4]}: Unexpected receive")
+        else:
+            if packet[3]:
+                print(f"TEST: {packet[4]}: Didn't receive, "
+                      "but receive expected")
+            else:
+                print(f"TEST: {packet[4]}: Didn't receive as expected")
+
+    if (recvd_it and packet[3] is not True):
+        # We got a reply but weren't expecting one
+        print(f"FAIL: Receive was unexpected for {packet[4]}")
+        return False
+
+    if (not recvd_it and packet[3]):
+        # We didn't get a reply but weret expecting one
+        print(f"FAIL: Expected to receive for {packet[4]}")
+        return False
+
+    return True
+
+# Run ping test
+def run_test():
+    # Open raw ICMP socket
+    try:
+        sock = socket.socket(socket.AF_INET6, socket.SOCK_RAW,
+                  socket.IPPROTO_ICMPV6)
+    except PermissionError:
+        print("This script requires root privileges.")
+        return 2
+
+    # Bind to interface by its IP address
+    sock.bind((SOURCE_IP, 0))
+
+    # Run through each packet
+    for packet in PACKET_LIST:
+        # Send packet
+        scapy.all.send(packet[0], verbose=False)
+
+        sock.settimeout(0.100)
+        recvd_it = False
+
+        # Try to get ICMP echo reply
+        try:
+            while not recvd_it:
+                rpacket, _addr = sock.recvfrom(1024)
+                icmp_type = rpacket[0]
+                identifier = struct.unpack(">H", rpacket[4:6])
+                if (icmp_type ==
+                    proto_nums.ICMP6_Type.ICMPV6_ECHO_REPLY.value and
+                    identifier[0] == packet[2]):
+                    recvd_it = True
+
+        except socket.timeout:
+            pass
+
+        process_return(recvd_it, packet)
+
+    return 0
+
+# Make packets for various test cases
+def make_test_packets():
+    # Two non-padding options in HBH and DestOpt, should succeed
+    # with default sysctls
+    make_packet("Two non-padding options in HBH and DestOpts",
+        [
+            ("H", [(11, 4), (0, 0), (0, 0), (12, 3)]),
+            ("D", [(1, 4), (12, 3)]),
+            ("R", [ "888::1", "9999::1"]),
+            ("F", (0x89abcdef)),
+            ("D", [(1, 4), (12, 3)]),
+        ])
+
+    # Big destination option, should fail when
+    # net.ipv6.max_dst_opts_length equals 64
+    make_packet("Big destination option",
+        [
+            ("H", [(11, 4), (0, 0), (0, 0), (12, 3)]),
+            ("D", [(1, 4), (12, 255)]),
+            ("R", [ "888::1", "9999::1"]),
+            ("F", (0x89abcdef)),
+        ])
+
+    # Almost Big HBH option should succeed when
+    # net.ipv6.max_hbh_length equals 64
+    make_packet("Almost Big HBH option",
+        [
+            ("H", [(11, 53), (1, 0), (12, 3)]),
+            ("D", [(1, 4), (12, 1)]),
+            ("R", [ "888::1", "9999::1"]),
+            ("F", (0x89abcdef)),
+        ])
+
+    # Big Hop-by-Hop option, should fail when
+    # net.ipv6.max_hbh_length equals 64
+    make_packet("Big HBH option",
+        [
+            ("H", [(11, 53), (1, 0), (0, 0), (12, 3)]),
+            ("D", [(1, 4), (12, 1)]),
+            ("R", [ "888::1", "9999::1"]),
+            ("F", (0x89abcdef)),
+        ])
+
+    # Too much HBH padding, should always fail
+    make_packet("Too much HBH padding",
+        [
+            ("H", [(12, 3), (1, 8)]),
+            ("D", [(1, 4), (12, 3)]),
+            ("R", [ "888::1", "9999::1"]),
+            ("F", (0x89abcdef)),
+            ("D", [(1, 4), (12, 3)]),
+        ])
+
+    # Too much DestOpt padding, should always fail
+    make_packet("Too much DestOpt padding",
+        [
+            ("H", [(12, 3), (1, 8)]),
+            ("D", [(1, 4), (12, 3)]),
+            ("R", [ "888::1", "9999::1"]),
+            ("F", (0x89abcdef)),
+            ("D", [(1, 4), (12, 3), (0, 0), (1, 6), (0, 0), (12, 3)]),
+        ])
+
+    # Too much DestOpt padding, should always fail
+    make_packet("Too much DestOpt padding #2",
+        [
+            ("H", [(12, 3)]),
+            ("D", [(1, 4), (12, 3)]),
+            ("R", [ "888::1", "9999::1"]),
+            ("F", (0x89abcdef)),
+            ("D", [(1, 4), (12, 3), (0, 0), (0, 0), (0, 0), (0, 0),
+                   (0, 0), (0, 0), (0, 0), (0, 0), (12, 3)]),
+        ])
+
+    # Too much DestOpt padding, should always fail
+
+    make_packet("Too much DestOpt padding #3",
+        [
+            ("D", [(0, 0), (0, 0), (0, 0), (0, 0),
+                   (0, 0), (0, 0), (0, 0), (0, 0)]),
+        ])
+
+    # Almost too much DestOpt padding, should succeed with default
+    # sysctl settings
+    make_packet("Almost too much DestOpt padding #2",
+        [
+            ("H", [(12, 3)]),
+            ("D", [(1, 4), (12, 3)]),
+            ("R", [ "888::1", "9999::1"]),
+            ("F", (0x89abcdef)),
+            ("D", [(1, 4), (12, 3), (0, 0), (0, 0), (0, 0), (0, 0),
+                   (0, 0), (0, 0), (0, 0), (12, 3)]),
+        ])
+
+    # Two Dest Ops, should fail unless net.ipv6.enforce_ext_hdr_order
+    # equals 1
+    make_packet("Two Dest Ops",
+        [
+            ("D", []),
+            ("D", []),
+        ])
+
+    # OOO Routing headers, should fail unless
+    # net.ipv6.enforce_ext_hdr_order equals 1
+    make_packet("OOO Routing header",
+        [
+            ("F", (0x89abcdef)),
+            ("R", [ "888::1", "9999::1"]),
+        ])
+
+    # Two Routing headers, should fail unless
+    # net.ipv6.enforce_ext_hdr_order equals 1
+    make_packet("Two Routing headers",
+        [
+            ("R", [ "888::1", "9999::1"]),
+            ("R", [ "888::1", "9999::1"]),
+        ])
+
+    # Two DestOpt headers with Routing header should succeed with default
+    # sysctl settings
+    make_packet("Two Destination options okay",
+        [
+            ("D", [(1, 4), (12, 3)]),
+            ("R", [ "888::1", "9999::1"]),
+            ("D", [(1, 4), (12, 3)]),
+        ])
+
+    # Two DestOpt headers without Routing header should fail unless
+    # net.ipv6.enforce_ext_hdr_order equals 1
+    make_packet("Two Destination options",
+        [
+            ("D", [(1, 4), (12, 3)]),
+            ("D", [(1, 4), (12, 3)]),
+        ])
+
+    # Two DestOpt headers after Routing header, should fail unless
+    # net.ipv6.enforce_ext_hdr_order equals 1
+    make_packet("Two Destination options after RH",
+        [
+            ("R", [ "888::1", "9999::1"]),
+            ("D", [(1, 4), (12, 3)]),
+            ("D", [(1, 4), (12, 3)]),
+        ])
+
+    # Many extension headers, should fail unless
+    # net.ipv6.enforce_ext_hdr_order equals 1
+    make_packet("Many EH OOO",
+        [
+            ("H", [(1, 4), (12, 3)]),
+            ("R", [ "888::1", "9999::1"]),
+            ("D", [(1, 4), (12, 3)]),
+            ("D", [(1, 4), (12, 3), (12, 3)]),
+            ("R", [ "888::1", "9999::1"]),
+            ("D", [(1, 4), (12, 3), (12, 3)]),
+            ("R", [ "888::1", "9999::1"]),
+            ("R", [ "888::1", "9999::1"]),
+            ("D", [(1, 4), (12, 3), (12, 3)]),
+            ("R", [ "888::1", "9999::1"]),
+            ("D", [(1, 4), (12, 3)]),
+            ("F", (0x89abcdef)),
+            ("D", [(1, 4), (12, 3), (12, 3)]),
+            ("R", [ "888::1", "9999::1"]),
+            ("D", [(1, 4), (12, 3), (12, 3)]),
+            ("R", [ "888::1", "9999::1"]),
+            ("R", [ "888::1", "9999::1"]),
+            ("D", [(1, 4), (12, 3), (12, 3)]),
+        ])
+
+    # Two fragment headers, should always fail due to stack
+    # implementation
+    make_packet("Two fragment Headers",
+        [
+            ("F", (0x89abcdef)),
+            ("F", (0x89abcdef)),
+        ])
+
+cli_args()
+
+make_test_packets()
+
+if PCAP_OUT != "":
+    write_pcap(PCAP_OUT)
+    STATUS = 0
+else:
+    STATUS = run_test()
+
+sys.exit(STATUS)
diff --git a/tools/testing/selftests/net/eh_limits.sh b/tools/testing/selftests/net/eh_limits.sh
new file mode 100755
index 000000000000..7703f29f2092
--- /dev/null
+++ b/tools/testing/selftests/net/eh_limits.sh
@@ -0,0 +1,205 @@
+#!/bin/bash
+# SPDX-License-Identifier: GPL-2.0
+
+# Tests extension header limits.
+#
+# We start by setting up two network namespaces with IPv6 addresses.
+# Then we create ICMPv6 Echo Request packets with various combinations of
+# Extension Headers, and send them from one namespace to the other and
+# check if a reply is received. Based on the sysctl settings certain packets
+# are expected to produce echo replies and others are expected to be drop
+# because an Extension Header related limit is exceeded. If an Echo Reply is
+# received or not received per our expectations then the test passes,
+# otherwise if the result is unexpected that's a test failure.
+# Tests extension header limits.
+
+source lib.sh
+
+# all tests in this script. Can be overridden with -t option
+TESTS="eh_limits"
+
+VERBOSE=""
+PAUSE_ON_FAIL=no
+PAUSE=no
+NAME="EH-limits"
+
+IP1="2001:db8::1"
+IP2="2001:db8::2"
+
+log_test()
+{
+	local rc=$1
+	local expected=$2
+	local msg="$3"
+
+	if [ "${rc}" -eq "${expected}" ]; then
+		printf "    TEST: %-60s  [ OK ]\n" "${msg}"
+		nsuccess=$((nsuccess+1))
+	else
+		nfail=$((nfail+1))
+		printf "    TEST: %-60s  [FAIL]\n" "${msg}"
+		if [ "${PAUSE_ON_FAIL}" = "yes" ]; then
+		echo
+			echo "hit enter to continue, 'q' to quit"
+			read -r a
+			[ "$a" = "q" ] && exit 1
+		fi
+	fi
+
+	if [ "${PAUSE}" = "yes" ]; then
+		echo
+		echo "hit enter to continue, 'q' to quit"
+		read -r a
+		[ "$a" = "q" ] && exit 1
+	fi
+}
+
+################################################################################
+# Setup
+
+setup()
+{
+	set -e
+
+	setup_ns ns1 ns2
+
+	NS_EXEC="ip netns exec"
+
+	ip link add veth1 type veth peer name veth2
+
+	ip link set veth1 netns "$ns1"
+	ip link set veth2 netns "$ns2"
+
+	$NS_EXEC "$ns1" ip addr add ${IP1}/64 dev veth1
+	$NS_EXEC "$ns1" ip link set veth1 up
+	$NS_EXEC "$ns1" ip link set lo up
+
+	$NS_EXEC "$ns2" ip addr add ${IP2}/64 dev veth2
+	$NS_EXEC "$ns2" ip link set veth2 up
+	$NS_EXEC "$ns2" ip link set lo up
+
+	# Enable SRv6 on the receiver since that's the type of routing header
+	# used in the test
+	$NS_EXEC "$ns2" sysctl -w net.ipv6.conf.all.seg6_enabled=1 > /dev/null
+	$NS_EXEC "$ns2" sysctl -w net.ipv6.conf.veth2.seg6_enabled=1 > /dev/null
+
+	set +e
+
+	# Send a ping to do neighuor discovery
+	$NS_EXEC "$ns1" ping6 -w 2 $IP2 -c 1 > /dev/null
+}
+
+exit_cleanup_all()
+{
+	cleanup_all_ns
+	exit "${EXIT_STATUS}"
+}
+
+eh_limits_test()
+{
+	local ip_addrs="--src_ip $IP1 --dst_ip $IP2"
+
+	if [ "$VERBOSE" = "-v" ]; then
+		echo ">>>>> Default"
+	fi
+
+	# Run the test with default sysctl settings
+	$NS_EXEC "$ns1" python3 ./eh_limits.py $VERBOSE $ip_addrs
+	$NS_EXEC "$ns1" python3 ./eh_limits.py $ip_addrs
+
+	log_test $? 0 "$NAME - default sysctls"
+
+	if [ "$VERBOSE" = "-v" ]; then
+		echo ">>>>> No order enforce, 8 options, 66 length limit"
+	fi
+
+	# Set extension header limit sysctls. We do this on both sides since
+	# the sender reads the sysctl's to determine pass/fail expectations
+
+	$NS_EXEC "$ns1" sysctl -w net.ipv6.enforce_ext_hdr_order=0 > /dev/null
+	$NS_EXEC "$ns1" sysctl -w net.ipv6.max_dst_opts_number=8 > /dev/null
+	$NS_EXEC "$ns1" sysctl -w net.ipv6.max_hbh_opts_number=8 > /dev/null
+	$NS_EXEC "$ns1" sysctl -w net.ipv6.max_hbh_length=64 > /dev/null
+	$NS_EXEC "$ns1" sysctl -w net.ipv6.max_dst_opts_length=64 > /dev/null
+
+	$NS_EXEC "$ns2" sysctl -w net.ipv6.enforce_ext_hdr_order=0 > /dev/null
+	$NS_EXEC "$ns2" sysctl -w net.ipv6.max_dst_opts_number=8 > /dev/null
+	$NS_EXEC "$ns2" sysctl -w net.ipv6.max_hbh_opts_number=8 > /dev/null
+	$NS_EXEC "$ns2" sysctl -w net.ipv6.max_hbh_length=64 > /dev/null
+	$NS_EXEC "$ns2" sysctl -w net.ipv6.max_dst_opts_length=64 > /dev/null
+
+	# Run the test with modified sysctl settings
+	$NS_EXEC "$ns1" python3 ./eh_limits.py $VERBOSE $ip_addrs
+
+	log_test $? 0 "$NAME - modified sysctls"
+}
+
+################################################################################
+# usage
+
+usage()
+{
+	cat <<EOF
+usage: ${0##*/} OPTS
+
+        -t <test>   Test(s) to run (default: all)
+                    (options: $TESTS)
+        -p          Pause on fail
+        -P          Pause after each test before cleanup
+        -v          verbose mode (show commands and output)
+EOF
+}
+
+################################################################################
+# main
+
+require_command scapy
+
+while getopts :t:pPhv o
+do
+	case $o in
+		t) TESTS=$OPTARG;;
+		p) PAUSE_ON_FAIL=yes;;
+		P) PAUSE=yes;;
+		v) VERBOSE="-v";;
+		h) usage; exit 0;;
+		*) usage; exit 1;;
+	esac
+done
+
+# make sure we don't pause twice
+[ "${PAUSE}" = "yes" ] && PAUSE_ON_FAIL=no
+
+ksft_skip=4
+
+if [ "$(id -u)" -ne 0 ];then
+	echo "SKIP: Need root privileges"
+	exit "$ksft_skip"
+fi
+
+if [ ! -x "$(command -v ip)" ]; then
+	echo "SKIP: Could not run test without ip tool"
+	exit "$ksft_skip"
+fi
+
+if [ ! -x "$(command -v socat)" ]; then
+	echo "SKIP: Could not run test without socat tool"
+	exit "$ksft_skip"
+fi
+
+# start clean
+cleanup &> /dev/null
+
+for t in $TESTS
+do
+	case $t in
+	eh_limits)		setup; eh_limits_test; cleanup_all_ns;;
+
+	help) echo "Test names: $TESTS"; exit 0;;
+	esac
+done
+
+if [ "$TESTS" != "none" ]; then
+	printf "\nTests passed: %3d\n" ${nsuccess}
+	printf "Tests failed: %3d\n"   ${nfail}
+fi
-- 
2.43.0