git: c2e0b604a3aa - main - netinet tests: test that we send IGMP join/leave packets as expected

The branch main has been updated by kp:

URL: https://cgit.FreeBSD.org/src/commit/?id=c2e0b604a3aa2a510f87a0e49badf485585252c5

commit c2e0b604a3aa2a510f87a0e49badf485585252c5
Author:     Kristof Provost <kp@FreeBSD.org>
AuthorDate: 2023-12-19 16:47:41 +0000
Commit:     Kristof Provost <kp@FreeBSD.org>
CommitDate: 2023-12-19 16:47:41 +0000

    netinet tests: test that we send IGMP join/leave packets as expected
    
    Sponsored by:   Rubicon Communications, LLC ("Netgate")
---
 tests/sys/netinet/Makefile |   1 +
 tests/sys/netinet/igmp.py  | 108 +++++++++++++++++++++++++++++++++++++++++++++
 2 files changed, 109 insertions(+)

diff --git a/tests/sys/netinet/Makefile b/tests/sys/netinet/Makefile
index 44f76508bf5c..434aadc87f46 100644
--- a/tests/sys/netinet/Makefile
+++ b/tests/sys/netinet/Makefile
@@ -24,6 +24,7 @@ ATF_TESTS_SH=	arp \
 		redirect
 
 ATF_TESTS_PYTEST+=	carp.py
+ATF_TESTS_PYTEST+=	igmp.py
 
 TEST_METADATA.divert+=	required_programs="python"
 TEST_METADATA.forward+=	required_programs="python"
diff --git a/tests/sys/netinet/igmp.py b/tests/sys/netinet/igmp.py
new file mode 100644
index 000000000000..b079c5d18664
--- /dev/null
+++ b/tests/sys/netinet/igmp.py
@@ -0,0 +1,108 @@
+#
+# SPDX-License-Identifier: BSD-2-Clause
+#
+# Copyright (c) 2023 Rubicon Communications, LLC (Netgate)
+#
+# Redistribution and use in source and binary forms, with or without
+# modification, are permitted provided that the following conditions
+# are met:
+# 1. Redistributions of source code must retain the above copyright
+#    notice, this list of conditions and the following disclaimer.
+# 2. Redistributions in binary form must reproduce the above copyright
+#    notice, this list of conditions and the following disclaimer in the
+#    documentation and/or other materials provided with the distribution.
+#
+# THIS SOFTWARE IS PROVIDED BY THE AUTHOR AND CONTRIBUTORS ``AS IS'' AND
+# ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE
+# IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE
+# ARE DISCLAIMED.  IN NO EVENT SHALL THE AUTHOR OR CONTRIBUTORS BE LIABLE
+# FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR CONSEQUENTIAL
+# DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS
+# OR SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS INTERRUPTION)
+# HOWEVER CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT
+# LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY
+# OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF
+# SUCH DAMAGE.
+#
+import pytest
+from atf_python.sys.net.tools import ToolsHelper
+from atf_python.sys.net.vnet import VnetTestTemplate
+import os
+import socket
+import struct
+import sys
+import logging
+logging.getLogger("scapy").setLevel(logging.CRITICAL)
+curdir = os.path.dirname(os.path.realpath(__file__))
+netpfil_common = curdir + "/../netpfil/common"
+sys.path.append(netpfil_common)
+from sniffer import Sniffer
+
+sc = None
+sp = None
+
+def check_igmpv3(args, pkt):
+    igmp = pkt.getlayer(sc.igmpv3.IGMPv3)
+    if igmp is None:
+        return False
+
+    igmpmr = pkt.getlayer(sc.igmpv3.IGMPv3mr)
+    if igmpmr is None:
+        return False
+
+    for r in igmpmr.records:
+        if r.maddr != args["group"]:
+            return False
+        if args["type"] == "join":
+            if r.rtype != 4:
+                return False
+        elif args["type"] == "leave":
+            if r.rtype != 3:
+                return False
+        r.show()
+
+    return True
+
+class TestIGMP(VnetTestTemplate):
+    REQUIRED_MODULES = []
+    TOPOLOGY = {
+        "vnet1": { "ifaces": [ "if1" ] },
+        "if1": { "prefixes4": [ ("192.0.2.1/24", "192.0.2.2/24" ) ] },
+    }
+
+    def setup_method(self, method):
+        global sc
+        if sc is None:
+            import scapy.contrib as _sc
+            import scapy.contrib.igmp
+            import scapy.contrib.igmpv3
+            import scapy.all as _sp
+            sc = _sc
+            sp = _sp
+        super().setup_method(method)
+
+    def test_igmp3_join_leave(self):
+        "Test that we send the expected join/leave IGMPv2 messages"
+
+        if1 = self.vnet.iface_alias_map["if1"]
+
+        # Start a background sniff
+        expected_pkt = { "type": "join", "group": "230.0.0.1" }
+        sniffer = Sniffer(expected_pkt, check_igmpv3, if1.name, timeout=10)
+
+        # Now join a multicast group, and see if we're getting the igmp packet we expect
+        s = socket.socket(socket.AF_INET, socket.SOCK_DGRAM, socket.IPPROTO_UDP)
+        mreq = struct.pack("4sl", socket.inet_aton('230.0.0.1'), socket.INADDR_ANY)
+        s.setsockopt(socket.IPPROTO_IP, socket.IP_ADD_MEMBERSHIP, mreq)
+
+        # Wait for the sniffer to see the join packet
+        sniffer.join()
+        assert(sniffer.correctPackets > 0)
+
+        # Now leave, check for the packet
+        expected_pkt = { "type": "leave", "group": "230.0.0.1" }
+        sniffer = Sniffer(expected_pkt, check_igmpv3, if1.name)
+
+        s.close()
+        sniffer.join()
+        assert(sniffer.correctPackets > 0)