These tests also verify functionality in TCP-MD5 and unsigned traffic
modes. They were used to find the issue fixed in commit 86f1e3a8489f
("tcp: md5: Fix overlap between vrf and non-vrf keys")
Signed-off-by: Leonard Crestez <cdleonard@xxxxxxxxx>
---
.../tcp_authopt_test/linux_tcp_authopt.py | 9 +
.../tcp_authopt_test/test_vrf_bind.py | 492 ++++++++++++++++++
.../tcp_authopt_test/vrf_netns_fixture.py | 127 +++++
3 files changed, 628 insertions(+)
create mode 100644 tools/testing/selftests/tcp_authopt/tcp_authopt_test/test_vrf_bind.py
create mode 100644 tools/testing/selftests/tcp_authopt/tcp_authopt_test/vrf_netns_fixture.py
diff --git a/tools/testing/selftests/tcp_authopt/tcp_authopt_test/linux_tcp_authopt.py b/tools/testing/selftests/tcp_authopt/tcp_authopt_test/linux_tcp_authopt.py
index 75cf5f993ccb..2a720d49cba2 100644
--- a/tools/testing/selftests/tcp_authopt/tcp_authopt_test/linux_tcp_authopt.py
+++ b/tools/testing/selftests/tcp_authopt/tcp_authopt_test/linux_tcp_authopt.py
@@ -38,10 +38,11 @@ class TCP_AUTHOPT_FLAG(IntFlag):
class TCP_AUTHOPT_KEY_FLAG(IntFlag):
DEL = BIT(0)
EXCLUDE_OPTS = BIT(1)
BIND_ADDR = BIT(2)
+ IFINDEX = BIT(3)
class TCP_AUTHOPT_ALG(IntEnum):
HMAC_SHA_1_96 = 1
AES_128_CMAC_96 = 2
@@ -102,25 +103,31 @@ class tcp_authopt_key:
recv_id: int = 0,
alg=TCP_AUTHOPT_ALG.HMAC_SHA_1_96,
key: KeyArgType = b"",
addr: AddrArgType = None,
auto_flags: bool = True,
+ ifindex: typing.Optional[int] = None,
include_options=None,
):
self.flags = flags
self.send_id = send_id
self.recv_id = recv_id
self.alg = alg
self.key = key
+ self.ifindex = ifindex
self.addr = addr
self.auto_flags = auto_flags
if include_options is not None:
self.include_options = include_options
def get_real_flags(self) -> TCP_AUTHOPT_KEY_FLAG:
result = self.flags
if self.auto_flags:
+ if self.ifindex is not None:
+ result |= TCP_AUTHOPT_KEY_FLAG.IFINDEX
+ else:
+ result &= ~TCP_AUTHOPT_KEY_FLAG.IFINDEX
if self.addr is not None:
result |= TCP_AUTHOPT_KEY_FLAG.BIND_ADDR
else:
result &= ~TCP_AUTHOPT_KEY_FLAG.BIND_ADDR
return result
@@ -136,10 +143,12 @@ class tcp_authopt_key:
self.alg,
len(self.key),
self.key,
)
data += bytes(self.addrbuf.ljust(sockaddr_storage.sizeof, b"\x00"))
+ if self.ifindex is not None:
+ data += bytes(struct.pack("I", self.ifindex))
return data
def __bytes__(self):
return self.pack()
diff --git a/tools/testing/selftests/tcp_authopt/tcp_authopt_test/test_vrf_bind.py b/tools/testing/selftests/tcp_authopt/tcp_authopt_test/test_vrf_bind.py
new file mode 100644
index 000000000000..da43ac8842e5
--- /dev/null
+++ b/tools/testing/selftests/tcp_authopt/tcp_authopt_test/test_vrf_bind.py
@@ -0,0 +1,492 @@
+# SPDX-License-Identifier: GPL-2.0
+"""Test VRF overlap behavior
+
+With tcp_l3mdev_accept single server should be able to differentiate multiple
+clients with same IP coming from different VRFs.
+"""
+import errno
+import logging
+import socket
+from contextlib import ExitStack
+
+import pytest
+
+from . import linux_tcp_md5sig
+from .conftest import parametrize_product, skipif_missing_tcp_authopt
+from .linux_tcp_authopt import (
+ set_tcp_authopt_key,
+ set_tcp_authopt_key_kwargs,
+ tcp_authopt_key,
+)
+from .server import SimpleServerThread
+from .utils import (
+ DEFAULT_TCP_SERVER_PORT,
+ check_socket_echo,
+ create_client_socket,
+ create_listen_socket,
+)
+from .vrf_netns_fixture import VrfNamespaceFixture
+
+logger = logging.getLogger(__name__)
+
+
+class VrfFixture:
+ """Fixture for VRF testing
+
+ Single server has two interfaces with same IP addr: one inside VRF and one
+ outside. Two clients two namespaces have same client IP, one connected to
+ VRF and one outside.
+ """
+
+ def __init__(
+ self,
+ address_family=socket.AF_INET,
+ tcp_l3mdev_accept=1,
+ init_default_listen_socket=True,
+ ):
+ self.address_family = address_family
+ self.tcp_l3mdev_accept = tcp_l3mdev_accept
+ self.init_default_listen_socket = init_default_listen_socket
+
+ @property
+ def server_addr(self):
+ if self.address_family == socket.AF_INET:
+ return self.nsfixture.server_ipv4_addr
+ else:
+ return self.nsfixture.server_ipv6_addr
+
+ @property
+ def client_addr(self):
+ if self.address_family == socket.AF_INET:
+ return self.nsfixture.client_ipv4_addr
+ else:
+ return self.nsfixture.client_ipv6_addr
+
+ @property
+ def server_addr_port(self):
+ return (str(self.server_addr), DEFAULT_TCP_SERVER_PORT)
+
+ @property
+ def vrf1_ifindex(self):
+ return self.nsfixture.server_vrf1_ifindex
+
+ @property
+ def vrf2_ifindex(self):
+ return self.nsfixture.server_vrf2_ifindex
+
+ def create_listen_socket(self, **kw):
+ result = create_listen_socket(
+ family=self.address_family,
+ ns=self.nsfixture.server_netns_name,
+ bind_addr=self.server_addr,
+ **kw
+ )
+ self.exit_stack.enter_context(result)
+ return result
+
+ def create_client_socket(self, ns):
+ result = create_client_socket(
+ ns=ns, family=self.address_family, bind_addr=self.client_addr
+ )
+ self.exit_stack.enter_context(result)
+ return result
+
+ def __enter__(self):
+ self.exit_stack = ExitStack()
+ self.exit_stack.__enter__()
+ self.nsfixture = self.exit_stack.enter_context(
+ VrfNamespaceFixture(tcp_l3mdev_accept=self.tcp_l3mdev_accept)
+ )
+
+ self.server_thread = SimpleServerThread(mode="echo")
+ if self.init_default_listen_socket:
+ self.listen_socket = self.create_listen_socket()
+ self.server_thread.add_listen_socket(self.listen_socket)
+ self.exit_stack.enter_context(self.server_thread)
+ return self
+
+ def __exit__(self, *args):
+ self.exit_stack.__exit__(*args)
+
+
+@pytest.mark.parametrize("address_family", [socket.AF_INET, socket.AF_INET6])
+def test_vrf_overlap_unsigned(exit_stack: ExitStack, address_family):
+ """Test without any signature support"""
+ fix = VrfFixture(address_family)
+ exit_stack.enter_context(fix)
+
+ client_socket0 = fix.create_client_socket(fix.nsfixture.client1_netns_name)
+ client_socket1 = fix.create_client_socket(fix.nsfixture.client1_netns_name)
+ client_socket2 = fix.create_client_socket(fix.nsfixture.client2_netns_name)
+
+ client_socket2.connect(fix.server_addr_port)
+ client_socket1.connect(fix.server_addr_port)
+ client_socket0.connect(fix.server_addr_port)
+ check_socket_echo(client_socket1)
+ check_socket_echo(client_socket2)
+ check_socket_echo(client_socket0)
+ check_socket_echo(client_socket1)
+ check_socket_echo(client_socket2)
+ check_socket_echo(client_socket0)
+ check_socket_echo(client_socket2)
+
+
+KEY0 = b"00000"
+KEY1 = b"1"
+KEY2 = b"22"
+
+
+def set_server_md5(fix, key=KEY0, **kw):
+ linux_tcp_md5sig.setsockopt_md5sig_kwargs(
+ fix.listen_socket, key=key, addr=fix.client_addr, **kw
+ )
+
+
+def set_server_md5_key0(fix, key=KEY0):
+ return set_server_md5(fix, key=key)
+
+
+def set_server_md5_key1(fix, key=KEY1):
+ return set_server_md5(fix, key=key, ifindex=fix.vrf1_ifindex)
+
+
+def set_server_md5_key2(fix, key=KEY2):
+ return set_server_md5(fix, key=key, ifindex=fix.vrf2_ifindex)
+
+
+def set_client_md5_key(fix, client_socket, key):
+ linux_tcp_md5sig.setsockopt_md5sig_kwargs(
+ client_socket, key=key, addr=fix.server_addr
+ )
+
+
+@pytest.mark.parametrize("address_family", [socket.AF_INET, socket.AF_INET6])
+def test_vrf_overlap_md5_samekey(exit_stack: ExitStack, address_family):
+ """Test overlapping keys that are identical"""
+ fix = VrfFixture(address_family)
+ exit_stack.enter_context(fix)
+ set_server_md5_key0(fix, b"same")
+ set_server_md5_key1(fix, b"same")
+ set_server_md5_key2(fix, b"same")
+ client_socket0 = fix.create_client_socket(fix.nsfixture.client0_netns_name)
+ client_socket1 = fix.create_client_socket(fix.nsfixture.client1_netns_name)
+ client_socket2 = fix.create_client_socket(fix.nsfixture.client2_netns_name)
+ set_client_md5_key(fix, client_socket0, b"same")
+ set_client_md5_key(fix, client_socket1, b"same")
+ set_client_md5_key(fix, client_socket2, b"same")
+ client_socket0.connect(fix.server_addr_port)
+ client_socket1.connect(fix.server_addr_port)
+ client_socket2.connect(fix.server_addr_port)
+ check_socket_echo(client_socket1)
+ check_socket_echo(client_socket2)
+ check_socket_echo(client_socket0)
+
+
+@pytest.mark.parametrize("address_family", [socket.AF_INET, socket.AF_INET6])
+def test_vrf_overlap12_md5(exit_stack: ExitStack, address_family):
+ """Test overlapping keys between vrfs"""
+ fix = VrfFixture(address_family)
+ exit_stack.enter_context(fix)
+ set_server_md5_key1(fix)
+ set_server_md5_key2(fix)
+ client_socket1 = fix.create_client_socket(fix.nsfixture.client1_netns_name)
+ client_socket2 = fix.create_client_socket(fix.nsfixture.client2_netns_name)
+ set_client_md5_key(fix, client_socket1, KEY1)
+ set_client_md5_key(fix, client_socket2, KEY2)
+ client_socket1.connect(fix.server_addr_port)
+ client_socket2.connect(fix.server_addr_port)
+ check_socket_echo(client_socket1)
+ check_socket_echo(client_socket2)
+
+
+@pytest.mark.parametrize("address_family", [socket.AF_INET, socket.AF_INET6])
+def test_vrf_overlap01_md5(exit_stack: ExitStack, address_family):
+ """Test overlapping keys inside and outside vrf, VRF key added second"""
+ fix = VrfFixture(address_family)
+ exit_stack.enter_context(fix)
+ set_server_md5_key0(fix)
+ set_server_md5_key1(fix)
+ client_socket0 = fix.create_client_socket(fix.nsfixture.client0_netns_name)
+ client_socket1 = fix.create_client_socket(fix.nsfixture.client1_netns_name)
+ set_client_md5_key(fix, client_socket0, KEY0)
+ set_client_md5_key(fix, client_socket1, KEY1)
+ client_socket1.connect(fix.server_addr_port)
+ client_socket0.connect(fix.server_addr_port)
+ check_socket_echo(client_socket0)
+ check_socket_echo(client_socket1)
+
+
+@pytest.mark.parametrize("address_family", [socket.AF_INET, socket.AF_INET6])
+def test_vrf_overlap10_md5(exit_stack: ExitStack, address_family):
+ """Test overlapping keys inside and outside vrf, VRF key added first"""
+ fix = VrfFixture(address_family)
+ exit_stack.enter_context(fix)
+ set_server_md5_key1(fix)
+ set_server_md5_key0(fix)
+ client_socket0 = fix.create_client_socket(fix.nsfixture.client0_netns_name)
+ client_socket1 = fix.create_client_socket(fix.nsfixture.client1_netns_name)
+ set_client_md5_key(fix, client_socket0, KEY0)
+ set_client_md5_key(fix, client_socket1, KEY1)
+ client_socket1.connect(fix.server_addr_port)
+ client_socket0.connect(fix.server_addr_port)
+ check_socket_echo(client_socket0)
+ check_socket_echo(client_socket1)
+
+
+@pytest.mark.parametrize("address_family", [socket.AF_INET])
+def test_vrf_overlap_md5_prefix(exit_stack: ExitStack, address_family):
+ """VRF keys should take precedence even if prefixlen is low"""
+ fix = VrfFixture(address_family)
+ exit_stack.enter_context(fix)
+ set_server_md5(fix, key=b"fail", prefixlen=16)
+ set_server_md5(
+ fix, key=b"pass", ifindex=fix.nsfixture.server_vrf1_ifindex, prefixlen=1
+ )
+ set_server_md5(fix, key=b"fail", prefixlen=24)
+
+ # connect via VRF
+ client_socket = fix.create_client_socket(fix.nsfixture.client1_netns_name)
+ set_client_md5_key(fix, client_socket, b"pass")
+ client_socket.connect(fix.server_addr_port)
+
+
+class TestVRFOverlapAOBoundKeyPrecedence:
+ """Keys bound to VRF should take precedence over unbound keys.
+
+ KEY0 is unbound (accepts all vrfs)
+ KEY1 is bound to vrf1
+ """
+
+ fix: VrfFixture
+
+ @pytest.fixture(
+ autouse=True,
+ scope="class",
+ params=[socket.AF_INET, socket.AF_INET6],
+ )
+ def init(self, request: pytest.FixtureRequest):
+ address_family = request.param
+ logger.info("init address_family=%s", address_family)
+ with ExitStack() as exit_stack:
+ fix = exit_stack.enter_context(VrfFixture(address_family))
+ set_tcp_authopt_key_kwargs(
+ fix.listen_socket,
+ key=KEY0,
+ ifindex=None,
+ )
+ set_tcp_authopt_key_kwargs(
+ fix.listen_socket,
+ key=KEY1,
+ ifindex=fix.vrf1_ifindex,
+ )
+ self.__class__.fix = fix
+ yield
+ logger.info("done address_family=%s", address_family)
+
+ def test_vrf1_key0(self):
+ client_socket = self.fix.create_client_socket(
+ self.fix.nsfixture.client1_netns_name
+ )
+ set_tcp_authopt_key_kwargs(client_socket, key=KEY0)
+ with pytest.raises(socket.timeout):
+ client_socket.connect(self.fix.server_addr_port)
+
+ def test_vrf1_key1(self):
+ client_socket = self.fix.create_client_socket(
+ self.fix.nsfixture.client1_netns_name
+ )
+ set_tcp_authopt_key_kwargs(client_socket, key=KEY1)
+ client_socket.connect(self.fix.server_addr_port)
+
+ def test_vrf2_key0(self):
+ client_socket = self.fix.create_client_socket(
+ self.fix.nsfixture.client2_netns_name
+ )
+ set_tcp_authopt_key_kwargs(client_socket, key=KEY0)
+ client_socket.connect(self.fix.server_addr_port)
+
+ def test_vrf2_key1(self):
+ client_socket = self.fix.create_client_socket(
+ self.fix.nsfixture.client2_netns_name
+ )
+ set_tcp_authopt_key_kwargs(client_socket, key=KEY1)
+ with pytest.raises(socket.timeout):
+ client_socket.connect(self.fix.server_addr_port)
+
+
+def assert_raises_enoent(func):
+ with pytest.raises(OSError) as e:
+ func()
+ assert e.value.errno == errno.ENOENT
+
+
+def test_vrf_overlap_md5_del_0110():
+ """Removing keys should not raise ENOENT because they are distinct"""
+ with VrfFixture() as fix:
+ set_server_md5(fix, key=KEY0)
+ set_server_md5(fix, key=KEY1, ifindex=fix.vrf1_ifindex)
+ set_server_md5(fix, key=b"", ifindex=fix.vrf1_ifindex)
+ set_server_md5(fix, key=b"")
+ assert_raises_enoent(lambda: set_server_md5(fix, key=b""))
+
+
+def test_vrf_overlap_md5_del_1001():
+ """Removing keys should not raise ENOENT because they are distinct"""
+ with VrfFixture() as fix:
+ set_server_md5(fix, key=KEY1, ifindex=fix.vrf1_ifindex)
+ set_server_md5(fix, key=KEY0)
+ set_server_md5(fix, key=b"")
+ set_server_md5(fix, key=b"", ifindex=fix.vrf1_ifindex)
+ assert_raises_enoent(lambda: set_server_md5(fix, key=b""))
+
+
+def test_vrf_overlap_md5_del_1010():
+ """Removing keys should not raise ENOENT because they are distinct"""
+ with VrfFixture() as fix:
+ set_server_md5(fix, key=KEY1, ifindex=fix.vrf1_ifindex)
+ set_server_md5(fix, key=KEY0)
+ set_server_md5(fix, key=b"", ifindex=fix.vrf1_ifindex)
+ set_server_md5(fix, key=b"")
+ assert_raises_enoent(lambda: set_server_md5(fix, key=b""))
+
+
+@skipif_missing_tcp_authopt
+@pytest.mark.parametrize("address_family", [socket.AF_INET, socket.AF_INET6])
+def test_vrf_overlap_ao_samekey(exit_stack: ExitStack, address_family):
+ """Single server serving both VRF and non-VRF client with same password.
+
+ This requires no special support from TCP-AO.
+ """
+ fix = VrfFixture(address_family)
+ exit_stack.enter_context(fix)
+ set_tcp_authopt_key(fix.listen_socket, tcp_authopt_key(key="11111"))
+
+ client_socket1 = fix.create_client_socket(fix.nsfixture.client1_netns_name)
+ client_socket2 = fix.create_client_socket(fix.nsfixture.client2_netns_name)
+
+ set_tcp_authopt_key(client_socket1, tcp_authopt_key(key="11111"))
+ set_tcp_authopt_key(client_socket2, tcp_authopt_key(key="11111"))
+ client_socket1.connect(fix.server_addr_port)
+ client_socket2.connect(fix.server_addr_port)
+ check_socket_echo(client_socket1)
+ check_socket_echo(client_socket2)
+ check_socket_echo(client_socket1)
+ check_socket_echo(client_socket2)
+
+
+@skipif_missing_tcp_authopt
+@pytest.mark.parametrize("address_family", [socket.AF_INET, socket.AF_INET6])
+def test_vrf_overlap_ao(exit_stack: ExitStack, address_family):
+ """Single server serving both VRF and non-VRF client with different passwords
+
+ This requires kernel to handle ifindex
+ """
+ fix = VrfFixture(address_family)
+ exit_stack.enter_context(fix)
+ set_tcp_authopt_key(
+ fix.listen_socket,
+ tcp_authopt_key(key=KEY0, ifindex=0),
+ )
+ set_tcp_authopt_key(
+ fix.listen_socket,
+ tcp_authopt_key(key=KEY1, ifindex=fix.vrf1_ifindex),
+ )
+ set_tcp_authopt_key(
+ fix.listen_socket,
+ tcp_authopt_key(key=KEY2, ifindex=fix.vrf2_ifindex),
+ )
+
+ client_socket0 = fix.create_client_socket(fix.nsfixture.client0_netns_name)
+ client_socket1 = fix.create_client_socket(fix.nsfixture.client1_netns_name)
+ client_socket2 = fix.create_client_socket(fix.nsfixture.client2_netns_name)
+ set_tcp_authopt_key(client_socket0, tcp_authopt_key(key=KEY0))
+ set_tcp_authopt_key(client_socket1, tcp_authopt_key(key=KEY1))
+ set_tcp_authopt_key(client_socket2, tcp_authopt_key(key=KEY2))
+ client_socket0.connect(fix.server_addr_port)
+ client_socket1.connect(fix.server_addr_port)
+ client_socket2.connect(fix.server_addr_port)
+ check_socket_echo(client_socket0)
+ check_socket_echo(client_socket1)
+ check_socket_echo(client_socket2)
+ check_socket_echo(client_socket0)
+ check_socket_echo(client_socket1)
+ check_socket_echo(client_socket2)
+
+
+@parametrize_product(
+ address_family=(socket.AF_INET, socket.AF_INET6),
+ tcp_l3mdev_accept=(0, 1),
+ bind_key_to_vrf=(0, 1),
+)
+def test_md5_pervrf(
+ exit_stack: ExitStack, address_family, tcp_l3mdev_accept, bind_key_to_vrf
+):
+ """Test one VRF-bound socket.
+
+ Since the socket is already bound to the vrf binding the key should not be required.
+ """
+ fix = VrfFixture(
+ address_family,
+ tcp_l3mdev_accept=tcp_l3mdev_accept,
+ init_default_listen_socket=False,
+ )
+ exit_stack.enter_context(fix)
+ listen_socket1 = fix.create_listen_socket(bind_device="veth1")
+ linux_tcp_md5sig.setsockopt_md5sig_kwargs(
+ listen_socket1,
+ key=KEY1,
+ addr=fix.client_addr,
+ ifindex=fix.vrf1_ifindex if bind_key_to_vrf else None,
+ )
+ fix.server_thread.add_listen_socket(listen_socket1)
+ client_socket1 = fix.create_client_socket(fix.nsfixture.client1_netns_name)
+ set_client_md5_key(fix, client_socket1, KEY1)
+ client_socket1.connect(fix.server_addr_port)
+ check_socket_echo(client_socket1)
+
+
+@pytest.mark.parametrize(
+ "address_family",
+ (socket.AF_INET, socket.AF_INET6),
+)
+def test_vrf_overlap_md5_pervrf(exit_stack: ExitStack, address_family):
+ """Test overlapping via per-VRF sockets"""
+ fix = VrfFixture(
+ address_family,
+ tcp_l3mdev_accept=0,
+ init_default_listen_socket=False,
+ )
+ exit_stack.enter_context(fix)
+ listen_socket0 = fix.create_listen_socket()
+ listen_socket1 = fix.create_listen_socket(bind_device="veth1")
+ listen_socket2 = fix.create_listen_socket(bind_device="veth2")
+ linux_tcp_md5sig.setsockopt_md5sig_kwargs(
+ listen_socket0,
+ key=KEY0,
+ addr=fix.client_addr,
+ )
+ linux_tcp_md5sig.setsockopt_md5sig_kwargs(
+ listen_socket1,
+ key=KEY1,
+ addr=fix.client_addr,
+ )
+ linux_tcp_md5sig.setsockopt_md5sig_kwargs(
+ listen_socket2,
+ key=KEY2,
+ addr=fix.client_addr,
+ )
+ fix.server_thread.add_listen_socket(listen_socket0)
+ fix.server_thread.add_listen_socket(listen_socket1)
+ fix.server_thread.add_listen_socket(listen_socket2)
+ client_socket0 = fix.create_client_socket(fix.nsfixture.client0_netns_name)
+ client_socket1 = fix.create_client_socket(fix.nsfixture.client1_netns_name)
+ client_socket2 = fix.create_client_socket(fix.nsfixture.client2_netns_name)
+ set_client_md5_key(fix, client_socket0, KEY0)
+ set_client_md5_key(fix, client_socket1, KEY1)
+ set_client_md5_key(fix, client_socket2, KEY2)
+ client_socket0.connect(fix.server_addr_port)
+ client_socket1.connect(fix.server_addr_port)
+ client_socket2.connect(fix.server_addr_port)
+ check_socket_echo(client_socket1)
+ check_socket_echo(client_socket1)
+ check_socket_echo(client_socket0)
diff --git a/tools/testing/selftests/tcp_authopt/tcp_authopt_test/vrf_netns_fixture.py b/tools/testing/selftests/tcp_authopt/tcp_authopt_test/vrf_netns_fixture.py
new file mode 100644
index 000000000000..ff9c0959a268
--- /dev/null
+++ b/tools/testing/selftests/tcp_authopt/tcp_authopt_test/vrf_netns_fixture.py
@@ -0,0 +1,127 @@
+# SPDX-License-Identifier: GPL-2.0
+import subprocess
+from ipaddress import IPv4Address, IPv6Address
+
+
+def ip_link_get_ifindex(dev: str, prefix: str = "") -> int:
+ out = subprocess.check_output(
+ f"{prefix}ip -o link show {dev}", text=True, shell=True
+ )
+ return int(out.split(":", 1)[0])
+
+
+def get_ipv4_addr(ns=1, index=1) -> IPv4Address:
+ return IPv4Address("10.10.0.0") + (ns << 8) + index
+
+
+def get_ipv6_addr(ns=1, index=1) -> IPv6Address:
+ return IPv6Address("fd00::") + (ns << 16) + index
+
+
+class VrfNamespaceFixture:
+ """Namespace fixture for VRF testing.
+
+ Single server has two interfaces with same IP addr: one inside VRF and one
+ outside.
+
+ Two clients two namespaces have same client IP, one connected to VRF and one
+ outside.
+ """
+
+ tcp_l3mdev_accept = 1
+
+ server_netns_name = "tcp_authopt_test_server"
+ client0_netns_name = "tcp_authopt_test_client0"
+ client1_netns_name = "tcp_authopt_test_client1"
+ client2_netns_name = "tcp_authopt_test_client2"
+
+ # 02:* means "locally administered"
+ server_veth0_mac_addr = "02:00:00:01:00:00"
+ server_veth1_mac_addr = "02:00:00:01:00:01"
+ server_veth2_mac_addr = "02:00:00:01:00:02"
+ client0_mac_addr = "02:00:00:02:00:00"
+ client1_mac_addr = "02:00:00:02:01:00"
+ client2_mac_addr = "02:00:00:02:02:00"
+
+ server_ipv4_addr = get_ipv4_addr(1, 1)
+ server_ipv6_addr = get_ipv6_addr(1, 1)
+ client_ipv4_addr = get_ipv4_addr(2, 1)
+ client_ipv6_addr = get_ipv6_addr(2, 1)
+
+ def __init__(self, **kw):
+ import os
+
+ import pytest
+
+ from .conftest import raise_skip_no_netns
+
+ raise_skip_no_netns()
+ if not os.path.exists("/proc/sys/net/ipv4/tcp_l3mdev_accept"):
+ pytest.skip(
+ "missing tcp_l3mdev_accept, is CONFIG_NET_L3_MASTER_DEV enabled?)"
+ )
+ for k, v in kw.items():
+ setattr(self, k, v)
+
+ def get_server_ifindex(self, dev):
+ return ip_link_get_ifindex(dev, f"ip netns exec {self.server_netns_name} ")
+
+ def __enter__(self):
+ self._del_netns()
+ script = f"""
+set -e
+ip netns add {self.server_netns_name}
+ip netns add {self.client0_netns_name}
+ip netns add {self.client1_netns_name}
+ip netns add {self.client2_netns_name}
+# Enable tcp_l3mdev unconditionally
+ip netns exec {self.server_netns_name} sysctl -q net.ipv4.tcp_l3mdev_accept={int(self.tcp_l3mdev_accept)}
+ip link add veth0 netns {self.server_netns_name} type veth peer name veth0 netns {self.client0_netns_name}
+ip link add veth1 netns {self.server_netns_name} type veth peer name veth0 netns {self.client1_netns_name}
+ip link add veth2 netns {self.server_netns_name} type veth peer name veth0 netns {self.client2_netns_name}
+ip link add vrf1 netns {self.server_netns_name} type vrf table 1000
+ip link add vrf2 netns {self.server_netns_name} type vrf table 2000
+ip -n {self.server_netns_name} link set vrf1 up
+ip -n {self.server_netns_name} link set vrf2 up
+ip -n {self.server_netns_name} link set veth1 vrf vrf1
+ip -n {self.server_netns_name} link set veth2 vrf vrf2
+ip -n {self.server_netns_name} link set veth0 up addr {self.server_veth0_mac_addr}
+ip -n {self.server_netns_name} link set veth1 up addr {self.server_veth1_mac_addr}
+ip -n {self.server_netns_name} link set veth2 up addr {self.server_veth2_mac_addr}
+ip -n {self.server_netns_name} addr add {self.server_ipv4_addr}/16 dev veth0
+ip -n {self.server_netns_name} addr add {self.server_ipv6_addr}/64 dev veth0 nodad
+ip -n {self.server_netns_name} addr add {self.server_ipv4_addr}/16 dev veth1
+ip -n {self.server_netns_name} addr add {self.server_ipv6_addr}/64 dev veth1 nodad
+ip -n {self.server_netns_name} addr add {self.server_ipv4_addr}/16 dev veth2
+ip -n {self.server_netns_name} addr add {self.server_ipv6_addr}/64 dev veth2 nodad
+ip -n {self.client0_netns_name} link set veth0 up addr {self.client0_mac_addr}
+ip -n {self.client0_netns_name} addr add {self.client_ipv4_addr}/16 dev veth0
+ip -n {self.client0_netns_name} addr add {self.client_ipv6_addr}/64 dev veth0 nodad
+ip -n {self.client1_netns_name} link set veth0 up addr {self.client1_mac_addr}
+ip -n {self.client1_netns_name} addr add {self.client_ipv4_addr}/16 dev veth0
+ip -n {self.client1_netns_name} addr add {self.client_ipv6_addr}/64 dev veth0 nodad
+ip -n {self.client2_netns_name} link set veth0 up addr {self.client2_mac_addr}
+ip -n {self.client2_netns_name} addr add {self.client_ipv4_addr}/16 dev veth0
+ip -n {self.client2_netns_name} addr add {self.client_ipv6_addr}/64 dev veth0 nodad
+"""
+ subprocess.run(script, shell=True, check=True)
+ self.server_veth0_ifindex = self.get_server_ifindex("veth0")
+ self.server_veth1_ifindex = self.get_server_ifindex("veth1")
+ self.server_veth2_ifindex = self.get_server_ifindex("veth2")
+ self.server_vrf1_ifindex = self.get_server_ifindex("vrf1")
+ self.server_vrf2_ifindex = self.get_server_ifindex("vrf2")
+ return self
+
+ def _del_netns(self):
+ script = f"""\
+set -e
+for ns in {self.server_netns_name} {self.client0_netns_name} {self.client1_netns_name} {self.client2_netns_name}; do
+ if ip netns list | grep -q "$ns"; then
+ ip netns del "$ns"
+ fi
+done
+"""
+ subprocess.run(script, shell=True, check=True)
+
+ def __exit__(self, *a):
+ self._del_netns()
--
2.25.1
