Replace tunnel ifconfig calls with iproute2

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):

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"]