From nobody Fri Jan 13 21:25:32 2023 X-Original-To: dev-commits-src-all@mlmmj.nyi.freebsd.org Received: from mx1.freebsd.org (mx1.freebsd.org [IPv6:2610:1c1:1:606c::19:1]) by mlmmj.nyi.freebsd.org (Postfix) with ESMTP id 4NtvbK135Mz2qkxY; Fri, 13 Jan 2023 21:25:33 +0000 (UTC) (envelope-from git@FreeBSD.org) Received: from mxrelay.nyi.freebsd.org (mxrelay.nyi.freebsd.org [IPv6:2610:1c1:1:606c::19:3]) (using TLSv1.3 with cipher TLS_AES_256_GCM_SHA384 (256/256 bits) key-exchange X25519 server-signature RSA-PSS (4096 bits) server-digest SHA256 client-signature RSA-PSS (4096 bits) client-digest SHA256) (Client CN "mxrelay.nyi.freebsd.org", Issuer "R3" (verified OK)) by mx1.freebsd.org (Postfix) with ESMTPS id 4NtvbJ6sXTz434v; Fri, 13 Jan 2023 21:25:32 +0000 (UTC) (envelope-from git@FreeBSD.org) DKIM-Signature: v=1; a=rsa-sha256; c=relaxed/relaxed; d=freebsd.org; s=dkim; t=1673645132; h=from:from:reply-to:subject:subject:date:date:message-id:message-id: to:to:cc:mime-version:mime-version:content-type:content-type: content-transfer-encoding:content-transfer-encoding; bh=Cp6gBTX+h/YoT6nDYJPizf+kOOWmgigzqBGPsHN+N48=; b=JbCyFL6DpL/fF3/M7iymmfZnravxawqrk/PWzGDDKEAml9VVWctdAhAxQRAzeRFcR9qkEg fYPaxi1MarkX53Q4GOaV7ZTzusL2m0qn7TuaQp7udLp0JJ7wQFatsEtJiBa1CV20sveeWN Y69Ft3J0rSKUfQt6pGpfCAY4Gr9uo+6z2iXJzx3kFqCl9wEc5oZMfw0EJ3f1/inK90PK1d 7GzZ/BDU2FRIIhNQrH3Blek7GLQsO0QkeTVHZVH+6sgPyHTu/nYp6sFtrOGB+aMIriZyY0 UWR3Ud/LLoTD1BkDQx/xQR3j72zPdZ+SsT8awls55X9w8wawxW85puEFX8wRrw== ARC-Message-Signature: i=1; a=rsa-sha256; c=relaxed/relaxed; d=freebsd.org; s=dkim; t=1673645133; h=from:from:reply-to:subject:subject:date:date:message-id:message-id: to:to:cc:mime-version:mime-version:content-type:content-type: content-transfer-encoding:content-transfer-encoding; bh=Cp6gBTX+h/YoT6nDYJPizf+kOOWmgigzqBGPsHN+N48=; b=go6R0+pIsNNrt2I7ffqQuOvXosBz4vxkrt3GEnqgz5amEDOmhj/zYVPqh1y9f1K44WADwe +8UX+j/mlPWENNoyU9bPD2V06+oC02xpoa+xe2rxaSPuwyNNAHSPZHrfEKesz+yKCBeLHk UMJpaf0ZTrhslLYwOsNxzzMjs7j8mUTUqrC2/7JPOcgsocMFYdhaUCgDF2jwoLRsoDYKH6 Dnznad7/lPAyDsVQD5BHYlaTduQZzQdhAB7FP3FXUagAq3R4V1To6BWR7KdrAmO+8NwIIa ZzGuFuNiO4PE3wHxvmN/YVuC2IUi9P3DjIi+7tYOKC5BQqLZShsCU/3rlGgPYg== ARC-Authentication-Results: i=1; mx1.freebsd.org; none ARC-Seal: i=1; s=dkim; d=freebsd.org; t=1673645133; a=rsa-sha256; cv=none; b=L32eDryIj2wj7I4FjWBUm9mC08IanBw0rdbPz9JM21spa3zrYD7+B/VQ//sdoeRyQnoAlU niG2VdLK4fVtM4AEbv8gHRRRthH6l6YdUNxMvknXei+5g5ida1wKhR523pgJ2zNbFwMTr9 6mlkDj+U29fgZJFRbl32Mwfa+oiq7ty5kngv15jW3V/fK0fq9v40BdLcmKuNAsCfIsMevW ggls5Thx8/0+7hXvrbD0aK2bNPo80LHQheTcETvnJDshmQtzKtH5s+lgUzzZhaIDhJkvKX CLoPypnGGCYZBrsrnbGoCaL47YmAfAdq73TdFr57CCaAqfTkjObCrXw51tRDdw== Received: from gitrepo.freebsd.org (gitrepo.freebsd.org [IPv6:2610:1c1:1:6068::e6a:5]) (using TLSv1.3 with cipher TLS_AES_256_GCM_SHA384 (256/256 bits) key-exchange X25519 server-signature RSA-PSS (4096 bits) server-digest SHA256) (Client did not present a certificate) by mxrelay.nyi.freebsd.org (Postfix) with ESMTPS id 4NtvbJ5ph7zMvF; Fri, 13 Jan 2023 21:25:32 +0000 (UTC) (envelope-from git@FreeBSD.org) Received: from gitrepo.freebsd.org ([127.0.1.44]) by gitrepo.freebsd.org (8.16.1/8.16.1) with ESMTP id 30DLPWuw041995; Fri, 13 Jan 2023 21:25:32 GMT (envelope-from git@gitrepo.freebsd.org) Received: (from git@localhost) by gitrepo.freebsd.org (8.16.1/8.16.1/Submit) id 30DLPWBK041994; Fri, 13 Jan 2023 21:25:32 GMT (envelope-from git) Date: Fri, 13 Jan 2023 21:25:32 GMT Message-Id: <202301132125.30DLPWBK041994@gitrepo.freebsd.org> To: src-committers@FreeBSD.org, dev-commits-src-all@FreeBSD.org, dev-commits-src-branches@FreeBSD.org From: "Alexander V. Chernikov" Subject: git: 29e3bc0f456a - stable/13 - netinet6: add ip6_output() tests. List-Id: Commit messages for all branches of the src repository List-Archive: https://lists.freebsd.org/archives/dev-commits-src-all List-Help: List-Post: List-Subscribe: List-Unsubscribe: Sender: owner-dev-commits-src-all@freebsd.org X-BeenThere: dev-commits-src-all@freebsd.org MIME-Version: 1.0 Content-Type: text/plain; charset=utf-8 Content-Transfer-Encoding: 8bit X-Git-Committer: melifaro X-Git-Repository: src X-Git-Refname: refs/heads/stable/13 X-Git-Reftype: branch X-Git-Commit: 29e3bc0f456a6c1ee51cd0dca2c64e8b5b985f92 Auto-Submitted: auto-generated X-ThisMailContainsUnwantedMimeParts: N The branch stable/13 has been updated by melifaro: URL: https://cgit.FreeBSD.org/src/commit/?id=29e3bc0f456a6c1ee51cd0dca2c64e8b5b985f92 commit 29e3bc0f456a6c1ee51cd0dca2c64e8b5b985f92 Author: Alexander V. Chernikov AuthorDate: 2022-07-07 10:09:28 +0000 Commit: Alexander V. Chernikov 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