mirror of
https://github.com/freebsd/freebsd-src
synced 2024-07-22 18:56:38 +00:00
109 lines
3.8 KiB
Python
109 lines
3.8 KiB
Python
#
|
|
# 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)
|