From 3ba4278d301bde8788e1a9c93bb2bda604db007f Mon Sep 17 00:00:00 2001 From: eunwoo song Date: Sun, 6 Sep 2026 23:38:15 +0900 Subject: [PATCH] fix: configure tunnels with iproute2 --- meshtastic/tests/test_tunnel.py | 83 +++++++++++++++++++++++++++++++++ meshtastic/tunnel.py | 34 ++++++++++++-- 2 files changed, 114 insertions(+), 3 deletions(-) diff --git a/meshtastic/tests/test_tunnel.py b/meshtastic/tests/test_tunnel.py index e2e917e5b..3fdf9e7cb 100644 --- a/meshtastic/tests/test_tunnel.py +++ b/meshtastic/tests/test_tunnel.py @@ -1,7 +1,9 @@ """Meshtastic unit tests for tunnel.py""" import logging import re +import subprocess import sys +from unittest.mock import call from unittest.mock import MagicMock, patch import pytest @@ -62,6 +64,87 @@ def test_Tunnel_with_interface(mock_platform_system, caplog, iface_with_nodes): assert re.search(r"Not sending packet", caplog.text, re.MULTILINE) +@pytest.mark.unit +@patch("platform.system", return_value="Linux") +@patch("meshtastic.tunnel.threading.Thread") +@patch("meshtastic.tunnel.subprocess.run") +@patch("meshtastic.tunnel.TapDevice") +def test_Tunnel_configures_device_with_iproute2( + mock_tap_device, + mock_run, + _mock_thread, + _mock_platform_system, + iface_with_nodes, +): + """The tunnel must not depend on the obsolete ifconfig command.""" + iface_with_nodes.noProto = False + iface_with_nodes.myInfo.my_node_num = 0x12345678 + tap = mock_tap_device.return_value + tap.name = "mesh0" + + Tunnel(iface_with_nodes) + + mock_tap_device.assert_called_once_with(name="mesh", mtu=200) + assert mock_run.call_args_list == [ + call( + ["ip", "link", "set", "dev", "mesh0", "mtu", "200", "up"], + check=True, + ), + call( + ["ip", "address", "replace", "10.115.86.120/16", "dev", "mesh0"], + check=True, + ), + ] + tap.up.assert_not_called() + tap.ifconfig.assert_not_called() + + +@pytest.mark.unit +@patch("platform.system", return_value="Linux") +@patch("meshtastic.tunnel.threading.Thread") +@patch("meshtastic.tunnel.subprocess.run") +@patch("meshtastic.tunnel.TapDevice") +def test_Tunnel_closes_device_when_iproute2_configuration_fails( + mock_tap_device, + mock_run, + _mock_thread, + _mock_platform_system, + iface_with_nodes, +): + """A failed iproute2 command must not leak the open TUN descriptor.""" + iface_with_nodes.noProto = False + iface_with_nodes.myInfo.my_node_num = 0x12345678 + mock_run.side_effect = subprocess.CalledProcessError(1, ["ip"]) + + with pytest.raises(Tunnel.TunnelError, match="iproute2"): + Tunnel(iface_with_nodes) + + mock_tap_device.return_value.close.assert_called_once_with() + + +@pytest.mark.unit +@patch("platform.system", return_value="Linux") +@patch("meshtastic.tunnel.threading.Thread") +@patch("meshtastic.tunnel.subprocess.run") +@patch("meshtastic.tunnel.TapDevice") +def test_Tunnel_closes_device_when_netmask_is_invalid( + mock_tap_device, + mock_run, + _mock_thread, + _mock_platform_system, + iface_with_nodes, +): + """Invalid network configuration must not leak the open TUN descriptor.""" + iface_with_nodes.noProto = False + iface_with_nodes.myInfo.my_node_num = 0x12345678 + + with pytest.raises(Tunnel.TunnelError, match="iproute2"): + Tunnel(iface_with_nodes, netmask="invalid") + + mock_run.assert_not_called() + mock_tap_device.return_value.close.assert_called_once_with() + + @pytest.mark.unitslow @patch("platform.system") def test_onTunnelReceive_from_ourselves(mock_platform_system, caplog, iface_with_nodes): diff --git a/meshtastic/tunnel.py b/meshtastic/tunnel.py index 46e3a2e0d..3256fab50 100644 --- a/meshtastic/tunnel.py +++ b/meshtastic/tunnel.py @@ -3,6 +3,7 @@ # Note python-pytuntap was too buggy # using pip3 install pytap2 # make sure to "sudo setcap cap_net_admin+eip /usr/bin/python3.8" so python can access tun device without being root +# The Linux tunnel setup uses iproute2 rather than the obsolete ifconfig tool. # sudo ip tuntap del mode tun tun0 # sudo bin/run.sh --port /dev/ttyUSB0 --setch-shortfast # sudo bin/run.sh --port /dev/ttyUSB0 --tunnel --debug @@ -15,8 +16,10 @@ # FIXME: use a more optimal MTU """ +import ipaddress import logging import platform +import subprocess import threading from pubsub import pub # type: ignore[import-untyped] @@ -115,9 +118,14 @@ def __init__(self, iface, subnet: str="10.115", netmask: str="255.255.0.0") -> N f"Not creating a TapDevice() because it is disabled by noProto" ) else: - self.tun = TapDevice(name="mesh") - self.tun.up() - self.tun.ifconfig(address=myAddr, netmask=netmask, mtu=200) + self.tun = TapDevice(name="mesh", mtu=200) + try: + self._configure_tun_device(self.tun, myAddr, netmask, 200) + except (OSError, ValueError, subprocess.CalledProcessError) as error: + self.tun.close() + raise Tunnel.TunnelError( + "Unable to configure the TUN device with iproute2." + ) from error self._rxThread = None if self.iface.noProto: @@ -131,6 +139,26 @@ def __init__(self, iface, subnet: str="10.115", netmask: str="255.255.0.0") -> N ) self._rxThread.start() + @staticmethod + def _configure_tun_device(tun, address: str, netmask: str, mtu: int) -> None: + """Configure a Linux TUN device using the standard iproute2 utility.""" + prefix_length = ipaddress.IPv4Network(f"0.0.0.0/{netmask}").prefixlen + subprocess.run( + ["ip", "link", "set", "dev", tun.name, "mtu", str(mtu), "up"], + check=True, + ) + subprocess.run( + [ + "ip", + "address", + "replace", + f"{address}/{prefix_length}", + "dev", + tun.name, + ], + check=True, + ) + def onReceive(self, packet): """onReceive""" p = packet["decoded"]["payload"]