xref: /freebsd-src/tests/sys/netinet/igmp.py (revision c2e0b604a3aa2a510f87a0e49badf485585252c5)

#
# 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)