git: 29e3bc0f456a - stable/13 - netinet6: add ip6_output() tests.
- Go to: [ bottom of page ] [ top of archives ] [ this month ]
Date: Fri, 13 Jan 2023 21:25:32 UTC
The branch stable/13 has been updated by melifaro: URL: https://cgit.FreeBSD.org/src/commit/?id=29e3bc0f456a6c1ee51cd0dca2c64e8b5b985f92 commit 29e3bc0f456a6c1ee51cd0dca2c64e8b5b985f92 Author: Alexander V. Chernikov <melifaro@FreeBSD.org> AuthorDate: 2022-07-07 10:09:28 +0000 Commit: Alexander V. Chernikov <melifaro@FreeBSD.org> CommitDate: 2023-01-13 21:24:11 +0000 netinet6: add ip6_output() tests. Differential Revision: https://reviews.freebsd.org/D35708 (cherry picked from commit ce414d02c1a0b4a3a68da99348b9962700742d44) --- tests/sys/netinet6/Makefile | 1 + tests/sys/netinet6/test_ip6_output.py | 435 ++++++++++++++++++++++++++++++++++ 2 files changed, 436 insertions(+) diff --git a/tests/sys/netinet6/Makefile b/tests/sys/netinet6/Makefile index dfae9f698ec4..41900b57ce02 100644 --- a/tests/sys/netinet6/Makefile +++ b/tests/sys/netinet6/Makefile @@ -5,6 +5,7 @@ PACKAGE= tests TESTSDIR= ${TESTSBASE}/sys/netinet6 FILESDIR= ${TESTSDIR} +ATF_TESTS_PYTEST= test_ip6_output.py ATF_TESTS_SH= \ exthdr \ mld \ diff --git a/tests/sys/netinet6/test_ip6_output.py b/tests/sys/netinet6/test_ip6_output.py new file mode 100644 index 000000000000..be578fb88c02 --- /dev/null +++ b/tests/sys/netinet6/test_ip6_output.py @@ -0,0 +1,435 @@ +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, obj_map, pipe, 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) + pipe.send(ll_data) + + tx_obj = ss.recv() + tx_obj["dst_iface_alias"] = vnet.iface_map[tx_obj["dst_iface"]].alias + pipe.send(tx_obj) + + +class TestIP6Output(BaseTestIP6Ouput): + def vnet2_handler(self, vnet, obj_map, pipe): + ip = str(vnet.iface_alias_map["if2"].first_ipv6.ip) + self._vnet2_handler(vnet, obj_map, pipe, 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, obj_map, pipe): + """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, obj_map, pipe, 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" + + +@pytest.mark.skip(reason="Currently fails") +class TestIP6OutputNhopLL(BaseTestIP6Ouput): + def vnet2_handler(self, vnet, obj_map, pipe): + """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, obj_map, pipe, 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 TestIP6OutputMulticast(BaseTestIP6Ouput): + def vnet2_handler(self, vnet, obj_map, pipe): + group = self.wait_object(pipe) + os_ifname = vnet.iface_alias_map["if2"].name + self._vnet2_handler(vnet, obj_map, pipe, 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", marks=pytest.mark.skip(reason="fails")), + ], + ) + @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", marks=pytest.mark.skip(reason="fails")), + ], + ) + @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