freebsd-src/tests/sys/netinet6/test_ip6_output.py

import errno
import ipaddress
import socket
import struct
import time
from ctypes import c_byte
from ctypes import c_uint
from ctypes import Structure

import pytest
from atf_python.sys.net.rtsock import SaHelper
from atf_python.sys.net.tools import ToolsHelper
from atf_python.sys.net.vnet import run_cmd
from atf_python.sys.net.vnet import SingleVnetTestTemplate
from atf_python.sys.net.vnet import VnetTestTemplate


class In6Pktinfo(Structure):
    _fields_ = [
        ("ipi6_addr", c_byte * 16),
        ("ipi6_ifindex", c_uint),
    ]


class VerboseSocketServer:
    def __init__(self, ip: str, port: int, ifname: str = None):
        self.ip = ip
        self.port = port

        s = socket.socket(socket.AF_INET6, socket.SOCK_DGRAM, socket.IPPROTO_UDP)
        s.setsockopt(socket.IPPROTO_IPV6, socket.IPV6_RECVPKTINFO, 1)
        addr = ipaddress.ip_address(ip)
        if addr.is_link_local and ifname:
            ifindex = socket.if_nametoindex(ifname)
            addr_tuple = (ip, port, 0, ifindex)
        elif addr.is_multicast and ifname:
            ifindex = socket.if_nametoindex(ifname)
            mreq = socket.inet_pton(socket.AF_INET6, ip) + struct.pack("I", ifindex)
            s.setsockopt(socket.IPPROTO_IPV6, socket.IPV6_JOIN_GROUP, mreq)
            print("## JOINED group {} % {}".format(ip, ifname))
            addr_tuple = ("::", port, 0, ifindex)
        else:
            addr_tuple = (ip, port, 0, 0)
        print("## Listening on [{}]:{}".format(addr_tuple[0], port))
        s.bind(addr_tuple)
        self.socket = s

    def recv(self):
        # data = self.socket.recv(4096)
        # print("RX: " + data)
        data, ancdata, msg_flags, address = self.socket.recvmsg(4096, 128)
        # Assume ancdata has just 1 item
        info = In6Pktinfo.from_buffer_copy(ancdata[0][2])
        dst_ip = socket.inet_ntop(socket.AF_INET6, info.ipi6_addr)
        dst_iface = socket.if_indextoname(info.ipi6_ifindex)

        tx_obj = {
            "data": data,
            "src_ip": address[0],
            "dst_ip": dst_ip,
            "dst_iface": dst_iface,
        }
        return tx_obj


class BaseTestIP6Ouput(VnetTestTemplate):
    TOPOLOGY = {
        "vnet1": {"ifaces": ["if1", "if2", "if3"]},
        "vnet2": {"ifaces": ["if1", "if2", "if3"]},
        "if1": {"prefixes6": [("2001:db8:a::1/64", "2001:db8:a::2/64")]},
        "if2": {"prefixes6": [("2001:db8:b::1/64", "2001:db8:b::2/64")]},
        "if3": {"prefixes6": [("2001:db8:c::1/64", "2001:db8:c::2/64")]},
    }
    DEFAULT_PORT = 45365

    def _vnet2_handler(self, vnet, ip: str, os_ifname: str = None):
        """Generic listener that sends first received packet with metadata
        back to the sender via pipw
        """
        ll_data = ToolsHelper.get_linklocals()
        # Start listener
        ss = VerboseSocketServer(ip, self.DEFAULT_PORT, os_ifname)
        vnet.pipe.send(ll_data)

        tx_obj = ss.recv()
        tx_obj["dst_iface_alias"] = vnet.iface_map[tx_obj["dst_iface"]].alias
        vnet.pipe.send(tx_obj)


class TestIP6Output(BaseTestIP6Ouput):
    def vnet2_handler(self, vnet):
        ip = str(vnet.iface_alias_map["if2"].first_ipv6.ip)
        self._vnet2_handler(vnet, ip, None)

    @pytest.mark.require_user("root")
    def test_output6_base(self):
        """Tests simple UDP output"""
        second_vnet = self.vnet_map["vnet2"]

        # Pick target on if2 vnet2's end
        ifaddr = ipaddress.ip_interface(self.TOPOLOGY["if2"]["prefixes6"][0][1])
        ip = str(ifaddr.ip)

        s = socket.socket(socket.AF_INET6, socket.SOCK_DGRAM, socket.IPPROTO_UDP)
        data = bytes("AAAA", "utf-8")
        print("## TX packet to {},{}".format(ip, self.DEFAULT_PORT))

        # Wait for the child to become ready
        self.wait_object(second_vnet.pipe)
        s.sendto(data, (ip, self.DEFAULT_PORT))

        # Wait for the received object
        rx_obj = self.wait_object(second_vnet.pipe)
        assert rx_obj["dst_ip"] == ip
        assert rx_obj["dst_iface_alias"] == "if2"

    @pytest.mark.require_user("root")
    def test_output6_nhop(self):
        """Tests UDP output with custom nhop set"""
        second_vnet = self.vnet_map["vnet2"]

        # Pick target on if2 vnet2's end
        ifaddr = ipaddress.ip_interface(self.TOPOLOGY["if2"]["prefixes6"][0][1])
        ip_dst = str(ifaddr.ip)
        # Pick nexthop on if1
        ifaddr = ipaddress.ip_interface(self.TOPOLOGY["if1"]["prefixes6"][0][1])
        ip_next = str(ifaddr.ip)
        sin6_next = SaHelper.ip6_sa(ip_next, 0)

        s = socket.socket(socket.AF_INET6, socket.SOCK_DGRAM, 0)
        s.setsockopt(socket.IPPROTO_IPV6, socket.IPV6_NEXTHOP, sin6_next)

        # Wait for the child to become ready
        self.wait_object(second_vnet.pipe)
        data = bytes("AAAA", "utf-8")
        s.sendto(data, (ip_dst, self.DEFAULT_PORT))

        # Wait for the received object
        rx_obj = self.wait_object(second_vnet.pipe)
        assert rx_obj["dst_ip"] == ip_dst
        assert rx_obj["dst_iface_alias"] == "if1"

    @pytest.mark.parametrize(
        "params",
        [
            # esrc: src-ip, if: src-interface, esrc: expected-src,
            # eif: expected-rx-interface
            pytest.param({"esrc": "2001:db8:b::1", "eif": "if2"}, id="empty"),
            pytest.param(
                {"src": "2001:db8:c::1", "esrc": "2001:db8:c::1", "eif": "if2"},
                id="iponly1",
            ),
            pytest.param(
                {
                    "src": "2001:db8:c::1",
                    "if": "if3",
                    "ex": errno.EHOSTUNREACH,
                },
                id="ipandif",
            ),
            pytest.param(
                {
                    "src": "2001:db8:c::aaaa",
                    "ex": errno.EADDRNOTAVAIL,
                },
                id="nolocalip",
            ),
            pytest.param(
                {"if": "if2", "src": "2001:db8:b::1", "eif": "if2"}, id="ifsame"
            ),
        ],
    )
    @pytest.mark.require_user("root")
    def test_output6_pktinfo(self, params):
        """Tests simple UDP output"""
        second_vnet = self.vnet_map["vnet2"]
        vnet = self.vnet

        # Pick target on if2 vnet2's end
        ifaddr = ipaddress.ip_interface(self.TOPOLOGY["if2"]["prefixes6"][0][1])
        dst_ip = str(ifaddr.ip)

        src_ip = params.get("src", "")
        src_ifname = params.get("if", "")
        expected_ip = params.get("esrc", "")
        expected_ifname = params.get("eif", "")
        errno = params.get("ex", 0)

        pktinfo = In6Pktinfo()
        if src_ip:
            for i, b in enumerate(socket.inet_pton(socket.AF_INET6, src_ip)):
                pktinfo.ipi6_addr[i] = b
        if src_ifname:
            os_ifname = vnet.iface_alias_map[src_ifname].name
            pktinfo.ipi6_ifindex = socket.if_nametoindex(os_ifname)

        # Wait for the child to become ready
        self.wait_object(second_vnet.pipe)
        data = bytes("AAAA", "utf-8")

        s = socket.socket(socket.AF_INET6, socket.SOCK_DGRAM, 0)
        try:
            s.setsockopt(socket.IPPROTO_IPV6, socket.IPV6_PKTINFO, bytes(pktinfo))
            aux = (socket.IPPROTO_IPV6, socket.IPV6_PKTINFO, bytes(pktinfo))
            s.sendto(data, (dst_ip, self.DEFAULT_PORT))
        except OSError as e:
            if not errno:
                raise
            assert e.errno == errno
            print("Correctly raised {}".format(e))
            return

        # Wait for the received object
        rx_obj = self.wait_object(second_vnet.pipe)

        assert rx_obj["dst_ip"] == dst_ip
        if expected_ip:
            assert rx_obj["src_ip"] == expected_ip
        if expected_ifname:
            assert rx_obj["dst_iface_alias"] == expected_ifname


class TestIP6OutputLL(BaseTestIP6Ouput):
    def vnet2_handler(self, vnet):
        """Generic listener that sends first received packet with metadata
        back to the sender via pipw
        """
        os_ifname = vnet.iface_alias_map["if2"].name
        ll_data = ToolsHelper.get_linklocals()
        ll_ip, _ = ll_data[os_ifname][0]
        self._vnet2_handler(vnet, ll_ip, os_ifname)

    @pytest.mark.require_user("root")
    def test_output6_linklocal(self):
        """Tests simple UDP output"""
        second_vnet = self.vnet_map["vnet2"]

        # Wait for the child to become ready
        ll_data = self.wait_object(second_vnet.pipe)

        # Pick LL address on if2 vnet2's end
        ip, _ = ll_data[second_vnet.iface_alias_map["if2"].name][0]
        # Get local interface scope
        os_ifname = self.vnet.iface_alias_map["if2"].name
        scopeid = socket.if_nametoindex(os_ifname)

        s = socket.socket(socket.AF_INET6, socket.SOCK_DGRAM, socket.IPPROTO_UDP)
        data = bytes("AAAA", "utf-8")
        target = (ip, self.DEFAULT_PORT, 0, scopeid)
        print("## TX packet to {}%{},{}".format(ip, scopeid, target[1]))

        s.sendto(data, target)

        # Wait for the received object
        rx_obj = self.wait_object(second_vnet.pipe)
        assert rx_obj["dst_ip"] == ip
        assert rx_obj["dst_iface_alias"] == "if2"


class TestIP6OutputNhopLL(BaseTestIP6Ouput):
    def vnet2_handler(self, vnet):
        """Generic listener that sends first received packet with metadata
        back to the sender via pipw
        """
        ip = str(vnet.iface_alias_map["if2"].first_ipv6.ip)
        self._vnet2_handler(vnet, ip, None)

    @pytest.mark.require_user("root")
    def test_output6_nhop_linklocal(self):
        """Tests UDP output with custom link-local nhop set"""
        second_vnet = self.vnet_map["vnet2"]

        # Wait for the child to become ready
        ll_data = self.wait_object(second_vnet.pipe)

        # Pick target on if2 vnet2's end
        ifaddr = ipaddress.ip_interface(self.TOPOLOGY["if2"]["prefixes6"][0][1])
        ip_dst = str(ifaddr.ip)
        # Pick nexthop on if1
        ip_next, _ = ll_data[second_vnet.iface_alias_map["if1"].name][0]
        # Get local interfaces
        os_ifname = self.vnet.iface_alias_map["if1"].name
        scopeid = socket.if_nametoindex(os_ifname)
        sin6_next = SaHelper.ip6_sa(ip_next, scopeid)

        s = socket.socket(socket.AF_INET6, socket.SOCK_DGRAM, 0)
        s.setsockopt(socket.IPPROTO_IPV6, socket.IPV6_NEXTHOP, sin6_next)

        data = bytes("AAAA", "utf-8")
        s.sendto(data, (ip_dst, self.DEFAULT_PORT))

        # Wait for the received object
        rx_obj = self.wait_object(second_vnet.pipe)
        assert rx_obj["dst_ip"] == ip_dst
        assert rx_obj["dst_iface_alias"] == "if1"


class TestIP6OutputScope(BaseTestIP6Ouput):
    def vnet2_handler(self, vnet):
        """Generic listener that sends first received packet with metadata
        back to the sender via pipw
        """
        bind_ip, bind_ifp = self.wait_object(vnet.pipe)
        if bind_ip is None:
            os_ifname = vnet.iface_alias_map[bind_ifp].name
            ll_data = ToolsHelper.get_linklocals()
            bind_ip, _ = ll_data[os_ifname][0]
        if bind_ifp is not None:
            bind_ifp = vnet.iface_alias_map[bind_ifp].name
        print("## BIND {}%{}".format(bind_ip, bind_ifp))
        self._vnet2_handler(vnet, bind_ip, bind_ifp)

    @pytest.mark.parametrize(
        "params",
        [
            # sif/dif: source/destination interface (for link-local addr)
            # sip/dip: source/destination ip (for non-LL addr)
            # ex: OSError errno that sendto() must raise
            pytest.param({"sif": "if2", "dif": "if2"}, id="same"),
            pytest.param(
                {
                    "sif": "if1",
                    "dif": "if2",
                    "ex": errno.EHOSTUNREACH,
                },
                id="ll_differentif1",
            ),
            pytest.param(
                {
                    "sif": "if1",
                    "dip": "2001:db8:b::2",
                    "ex": errno.EHOSTUNREACH,
                },
                id="ll_differentif2",
            ),
            pytest.param(
                {
                    "sip": "2001:db8:a::1",
                    "dif": "if2",
                },
                id="gu_to_ll",
            ),
        ],
    )
    @pytest.mark.require_user("root")
    def test_output6_linklocal_scope(self, params):
        """Tests simple UDP output"""
        second_vnet = self.vnet_map["vnet2"]

        src_ifp = params.get("sif")
        src_ip = params.get("sip")
        dst_ifp = params.get("dif")
        dst_ip = params.get("dip")
        errno = params.get("ex", 0)

        # Sent ifp/IP to bind on
        second_vnet = self.vnet_map["vnet2"]
        second_vnet.pipe.send((dst_ip, dst_ifp))

        # Wait for the child to become ready
        ll_data = self.wait_object(second_vnet.pipe)

        if dst_ip is None:
            # Pick LL address on dst_ifp vnet2's end
            dst_ip, _ = ll_data[second_vnet.iface_alias_map[dst_ifp].name][0]
            # Get local interface scope
            os_ifname = self.vnet.iface_alias_map[dst_ifp].name
            scopeid = socket.if_nametoindex(os_ifname)
            target = (dst_ip, self.DEFAULT_PORT, 0, scopeid)
        else:
            target = (dst_ip, self.DEFAULT_PORT, 0, 0)

        # Bind
        if src_ip is None:
            ll_data = ToolsHelper.get_linklocals()
            os_ifname = self.vnet.iface_alias_map[src_ifp].name
            src_ip, _ = ll_data[os_ifname][0]
            scopeid = socket.if_nametoindex(os_ifname)
            src = (src_ip, self.DEFAULT_PORT, 0, scopeid)
        else:
            src = (src_ip, self.DEFAULT_PORT, 0, 0)

        s = socket.socket(socket.AF_INET6, socket.SOCK_DGRAM, socket.IPPROTO_UDP)
        s.bind(src)
        data = bytes("AAAA", "utf-8")
        print("## TX packet {} -> {}".format(src, target))

        try:
            s.sendto(data, target)
        except OSError as e:
            if not errno:
                raise
            assert e.errno == errno
            print("Correctly raised {}".format(e))
            return

        # Wait for the received object
        rx_obj = self.wait_object(second_vnet.pipe)
        assert rx_obj["dst_ip"] == dst_ip
        assert rx_obj["src_ip"] == src_ip
        # assert rx_obj["dst_iface_alias"] == "if2"


class TestIP6OutputMulticast(BaseTestIP6Ouput):
    def vnet2_handler(self, vnet):
        group = self.wait_object(vnet.pipe)
        os_ifname = vnet.iface_alias_map["if2"].name
        self._vnet2_handler(vnet, group, os_ifname)

    @pytest.mark.parametrize("group_scope", ["ff02", "ff05", "ff08", "ff0e"])
    @pytest.mark.require_user("root")
    def test_output6_multicast(self, group_scope):
        """Tests simple UDP output"""
        second_vnet = self.vnet_map["vnet2"]

        group = "{}::3456".format(group_scope)
        second_vnet.pipe.send(group)

        # Pick target on if2 vnet2's end
        ip = group
        os_ifname = self.vnet.iface_alias_map["if2"].name
        ifindex = socket.if_nametoindex(os_ifname)
        optval = struct.pack("I", ifindex)

        s = socket.socket(socket.AF_INET6, socket.SOCK_DGRAM, socket.IPPROTO_UDP)
        s.setsockopt(socket.IPPROTO_IPV6, socket.IPV6_MULTICAST_IF, optval)

        data = bytes("AAAA", "utf-8")

        # Wait for the child to become ready
        self.wait_object(second_vnet.pipe)

        print("## TX packet to {},{}".format(ip, self.DEFAULT_PORT))
        s.sendto(data, (ip, self.DEFAULT_PORT))

        # Wait for the received object
        rx_obj = self.wait_object(second_vnet.pipe)
        assert rx_obj["dst_ip"] == ip
        assert rx_obj["dst_iface_alias"] == "if2"


class TestIP6OutputLoopback(SingleVnetTestTemplate):
    IPV6_PREFIXES = ["2001:db8:a::1/64"]
    DEFAULT_PORT = 45365

    @pytest.mark.parametrize(
        "source_validation",
        [
            pytest.param(0, id="no_sav"),
            pytest.param(1, id="sav"),
        ],
    )
    @pytest.mark.parametrize("scope", ["gu", "ll", "lo"])
    def test_output6_self_tcp(self, scope, source_validation):
        """Tests IPv6 TCP connection to the local IPv6 address"""

        ToolsHelper.set_sysctl(
            "net.inet6.ip6.source_address_validation", source_validation
        )

        if scope == "gu":
            ip = "2001:db8:a::1"
            addr_tuple = (ip, self.DEFAULT_PORT)
        elif scope == "ll":
            os_ifname = self.vnet.iface_alias_map["if1"].name
            ifindex = socket.if_nametoindex(os_ifname)
            ll_data = ToolsHelper.get_linklocals()
            ip, _ = ll_data[os_ifname][0]
            addr_tuple = (ip, self.DEFAULT_PORT, 0, ifindex)
        elif scope == "lo":
            ip = "::1"
            ToolsHelper.get_output("route add -6 ::1/128 -iface lo0")
            ifindex = socket.if_nametoindex("lo0")
            addr_tuple = (ip, self.DEFAULT_PORT)
        else:
            assert 0 == 1
        print("address: {}".format(addr_tuple))

        start = time.perf_counter()
        ss = socket.socket(socket.AF_INET6, socket.SOCK_STREAM, socket.IPPROTO_TCP)
        ss.setsockopt(socket.IPPROTO_IPV6, socket.IPV6_RECVPKTINFO, 1)
        ss.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEPORT, 1)
        ss.bind(addr_tuple)
        ss.listen()
        s = socket.socket(socket.AF_INET6, socket.SOCK_STREAM, socket.IPPROTO_TCP)
        s.settimeout(2.0)
        s.connect(addr_tuple)
        conn, from_addr = ss.accept()
        duration = time.perf_counter() - start

        assert from_addr[0] == ip
        assert duration < 1.0

    @pytest.mark.parametrize(
        "source_validation",
        [
            pytest.param(0, id="no_sav"),
            pytest.param(1, id="sav"),
        ],
    )
    @pytest.mark.parametrize("scope", ["gu", "ll", "lo"])
    def test_output6_self_udp(self, scope, source_validation):
        """Tests IPv6 UDP connection to the local IPv6 address"""

        ToolsHelper.set_sysctl(
            "net.inet6.ip6.source_address_validation", source_validation
        )

        if scope == "gu":
            ip = "2001:db8:a::1"
            addr_tuple = (ip, self.DEFAULT_PORT)
        elif scope == "ll":
            os_ifname = self.vnet.iface_alias_map["if1"].name
            ifindex = socket.if_nametoindex(os_ifname)
            ll_data = ToolsHelper.get_linklocals()
            ip, _ = ll_data[os_ifname][0]
            addr_tuple = (ip, self.DEFAULT_PORT, 0, ifindex)
        elif scope == "lo":
            ip = "::1"
            ToolsHelper.get_output("route add -6 ::1/128 -iface lo0")
            ifindex = socket.if_nametoindex("lo0")
            addr_tuple = (ip, self.DEFAULT_PORT)
        else:
            assert 0 == 1
        print("address: {}".format(addr_tuple))

        start = time.perf_counter()
        ss = socket.socket(socket.AF_INET6, socket.SOCK_STREAM, socket.IPPROTO_TCP)
        ss.setsockopt(socket.IPPROTO_IPV6, socket.IPV6_RECVPKTINFO, 1)
        ss.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEPORT, 1)
        ss.bind(addr_tuple)
        ss.listen()
        s = socket.socket(socket.AF_INET6, socket.SOCK_STREAM, socket.IPPROTO_TCP)
        s.settimeout(2.0)
        s.connect(addr_tuple)
        conn, from_addr = ss.accept()
        duration = time.perf_counter() - start

        assert from_addr[0] == ip
        assert duration < 1.0