Add natlab test for start telio with tun

nat-lab/tests/telio.py

@@ -514,6 +514,18 @@ async def simple_start(self):
         if isinstance(self.get_router(), LinuxRouter):
             await self.get_proxy().set_fwmark(int(LINUX_FWMARK_VALUE))
 
+    async def create_tun(self, tun_name: bytes) -> int:
+        return await self.get_proxy().create_tun(tun_name)
+
+    async def start_with_tun(self, tun: int):
+        await self.get_proxy().start_with_tun(
+            private_key=self._node.private_key,
+            adapter=self._adapter_type,
+            tun=tun,
+        )
+        if isinstance(self.get_router(), LinuxRouter):
+            await self.get_proxy().set_fwmark(int(LINUX_FWMARK_VALUE))
+
     async def wait_for_state_peer(
         self,
         public_key,

nat-lab/tests/test_start_with_tun.py

@@ -0,0 +1,40 @@
+from typing import List
+from contextlib import AsyncExitStack
+from helpers import SetupParameters, ping_between_all_nodes, setup_mesh_nodes
+from utils.bindings import (
+    default_features,
+    TelioAdapterType,
+)
+from utils.connection_util import ConnectionTag
+
+
+def _generate_setup_parameters(
+    conn_tags: List[ConnectionTag],
+) -> List[SetupParameters]:
+    return [
+        SetupParameters(
+            connection_tag=conn_tag,
+            adapter_type_override=TelioAdapterType.BORING_TUN,
+            features=default_features(),
+            fingerprint=f"{conn_tag}",
+        )
+        for conn_tag in conn_tags
+    ]
+
+
+async def test_start_with_tun() -> None:
+    setup_params = _generate_setup_parameters(
+        [
+            ConnectionTag.DOCKER_CONE_CLIENT_1,
+            ConnectionTag.DOCKER_CONE_CLIENT_2,
+        ]
+    )
+
+    async with AsyncExitStack() as exit_stack:
+        env = await setup_mesh_nodes(exit_stack, setup_params)
+        alpha_client, _ = env.clients
+
+        await alpha_client.stop_device()
+        tun = await alpha_client.create_tun(b"tun11")
+        await alpha_client.start_with_tun(tun)
+        await ping_between_all_nodes(env)

nat-lab/tests/uniffi/libtelio_proxy.py

@@ -99,6 +99,16 @@ def start_named(self, private_key, adapter, name: str):
             lambda r: r.start_named(private_key, adapter.value, name)
         )
 
+    @move_to_async_thread
+    def create_tun(self, tun_name: bytes) -> int:
+        return self._handle_remote_error(lambda r: r.create_tun(tun_name))
+
+    @move_to_async_thread
+    def start_with_tun(self, private_key, adapter, tun: int):
+        self._handle_remote_error(
+            lambda r: r.start_with_tun(private_key, adapter.value, tun)
+        )
+
     @move_to_async_thread
     def set_fwmark(self, fwmark: int):
         self._handle_remote_error(lambda r: r.set_fwmark(fwmark))

nat-lab/tests/uniffi/libtelio_remote.py

@@ -1,5 +1,7 @@
 import datetime
+import fcntl
 import os
+import struct
 import Pyro5.api  # type: ignore
 import Pyro5.server  # type: ignore
 import shutil
@@ -109,6 +111,26 @@ def start_named(self, private_key, adapter, name: str):
             private_key, libtelio.TelioAdapterType(adapter), name
         )
 
+    @serialize_error
+    def create_tun(self, tun_name: bytes) -> int:
+        # Constants for TUN/TAP interface creation (from Linux's if_tun.h)
+        TUNSETIFF = 0x400454CA
+        IFF_TUN = 0x0001
+        IFF_NO_PI = 0x1000
+
+        tun_fd = os.open("/dev/net/tun", os.O_RDWR)
+        # '16sH' means we need to pass 16-byte string (interface name) and 2-byte short (flags)
+        ifr = struct.pack("16sH", tun_name, IFF_TUN | IFF_NO_PI)
+        fcntl.ioctl(tun_fd, TUNSETIFF, ifr)
+
+        return tun_fd
+
+    @serialize_error
+    def start_with_tun(self, private_key, adapter, tun: int):
+        self._libtelio.start_with_tun(
+            private_key, libtelio.TelioAdapterType(adapter), tun
+        )
+
     @serialize_error
     def set_fwmark(self, fwmark: int):
         self._libtelio.set_fwmark(fwmark)