diff --git a/CHANGELOG.md b/CHANGELOG.md new file mode 100644 index 000000000..ef4fc34c1 --- /dev/null +++ b/CHANGELOG.md @@ -0,0 +1,6 @@ +# Changelog + +## [Unreleased] +### Fixed +- Replaced deprecated `ifconfig` command with `ip` (iproute2) in `meshtastic.tunnel` for modern Linux compatibility. +- Implemented robust error handling and rollback logic for tunnel interface configuration. diff --git a/meshtastic/tunnel.py b/meshtastic/tunnel.py index 46e3a2e0d..94694bfbe 100644 --- a/meshtastic/tunnel.py +++ b/meshtastic/tunnel.py @@ -18,6 +18,8 @@ import logging import platform import threading +import subprocess +import socket from pubsub import pub # type: ignore[import-untyped] from pytap2 import TapDevice @@ -117,7 +119,26 @@ def __init__(self, iface, subnet: str="10.115", netmask: str="255.255.0.0") -> N else: self.tun = TapDevice(name="mesh") self.tun.up() - self.tun.ifconfig(address=myAddr, netmask=netmask, mtu=200) + + # Replaced ifconfig with ip commands + # self.tun.ifconfig(address=myAddr, netmask=netmask, mtu=200) + + cidr = self._cidr_from_netmask(netmask) + try: + # 1. Set interface up + subprocess.check_call(["ip", "link", "set", "dev", self.tun.name, "up"]) + # 2. Add address + subprocess.check_call(["ip", "addr", "add", f"{myAddr}/{cidr}", "dev", self.tun.name]) + # 3. Set MTU + subprocess.check_call(["ip", "link", "set", "dev", self.tun.name, "mtu", "200"]) + except Exception as e: + logger.error(f"Failed to configure tunnel interface: {e}") + logger.info("Attempting to rollback (set interface down)...") + try: + subprocess.check_call(["ip", "link", "set", "dev", self.tun.name, "down"]) + except Exception as rollback_e: + logger.error(f"Rollback failed: {rollback_e}") + raise e self._rxThread = None if self.iface.noProto: @@ -131,6 +152,10 @@ def __init__(self, iface, subnet: str="10.115", netmask: str="255.255.0.0") -> N ) self._rxThread.start() + def _cidr_from_netmask(self, netmask): + """Convert a netmask string (e.g. '255.255.0.0') to a CIDR prefix length (e.g. 16)""" + return sum([bin(int(x)).count('1') for x in netmask.split('.')]) + def onReceive(self, packet): """onReceive""" p = packet["decoded"]["payload"] diff --git a/tests/test_tunnel_linux.py b/tests/test_tunnel_linux.py new file mode 100644 index 000000000..b48712f45 --- /dev/null +++ b/tests/test_tunnel_linux.py @@ -0,0 +1,162 @@ + +import unittest +from unittest.mock import MagicMock, patch +import sys +import os +import subprocess + +# MOCK DEPENDENCIES BEFORE IMPORTING TARGET +# This is necessary because the environment lacks pypubsub/pytap2 and we cannot install them. +missing_modules = [ + "pubsub", "pytap2", "tabulate", "serial", "bleak", "requests", "yaml", + "packaging", "dotmap", "pyqrcode", "print_color", "dash", "flask" +] +for mod in missing_modules: + sys.modules[mod] = MagicMock() + +# Explicitly mock submodules that are imported via 'from x.y import z' +sys.modules["packaging.version"] = MagicMock() +sys.modules["serial.tools"] = MagicMock() +sys.modules["serial.tools.list_ports"] = MagicMock() + +# Mock meshtastic.protobuf +mock_pb2 = MagicMock() +mock_pb2.IP_TUNNEL_APP = 123 +sys.modules["meshtastic.protobuf"] = MagicMock() +sys.modules["meshtastic.protobuf.portnums_pb2"] = mock_pb2 + +# Ensure we can import meshtastic +sys.path.append(os.getcwd()) + +from meshtastic.tunnel import Tunnel + +class TestTunnelLinux(unittest.TestCase): + """Test the Linux-specific Tunnel class""" + + def setUp(self): + self.mock_iface = MagicMock() + self.mock_iface.noProto = False + self.mock_iface.myInfo.my_node_num = 1234 + + # Mock nodes + self.mock_iface.nodes = { + "node1": {"num": 1234, "user": {"id": "node1_id"}} + } + + @patch("platform.system", return_value="Linux") + @patch("meshtastic.tunnel.TapDevice") + @patch("subprocess.check_call") + @patch("meshtastic.tunnel.mt_config") + @patch("meshtastic.tunnel.pub") + def test_tunnel_init_ip_commands(self, mock_pub, mock_config, mock_check_call, mock_tap_device, mock_platform): + """Test that ip commands are called correctly instead of ifconfig""" + + # Mock the TapDevice instance + mock_tun_instance = MagicMock() + mock_tun_instance.name = "mesh0" + mock_tap_device.return_value = mock_tun_instance + + # Initialize Tunnel + # subnet="10.115", netmask="255.255.0.0" (default) + tunnel = Tunnel(self.mock_iface) + + # Check that TapDevice was created + mock_tap_device.assert_called() + + # Verify that ifconfig was NOT called on the tun device + # (This is the key part of the fix - replacing tun.ifconfig with subprocess calls) + # Note: In the old code it called tun.ifconfig. In the new code it shouldn't. + # We can't strictly assert "not called" on a method that doesn't exist on our mock + # unless we configured the mock to have it. + # But we CAN verify that subprocess.check_call was called with the right ip commands. + + # Expected calls: + # 1. ip link set dev mesh0 up + # 2. ip addr add 10.115.4.210/16 dev mesh0 (1234 decimal is 0x04D2. 0x04=4, 0xD2=210) + # 3. ip link set dev mesh0 mtu 200 + + # Note: 1234 = 0x04D2. + # _nodeNumToIp(1234) -> 10.115.4.210 + + expected_calls = [ + unittest.mock.call(["ip", "link", "set", "dev", "mesh0", "up"]), + unittest.mock.call(["ip", "addr", "add", "10.115.4.210/16", "dev", "mesh0"]), + unittest.mock.call(["ip", "link", "set", "dev", "mesh0", "mtu", "200"]) + ] + + mock_check_call.assert_has_calls(expected_calls, any_order=True) + + @patch("platform.system", return_value="Linux") + @patch("meshtastic.tunnel.TapDevice") + @patch("subprocess.check_call") + @patch("meshtastic.tunnel.mt_config") + @patch("meshtastic.tunnel.pub") + def test_tunnel_ip_command_fails(self, mock_pub, mock_config, mock_check_call, mock_tap_device, mock_platform): + """Test behavior when ip command fails""" + # Mock failure on the first call + mock_check_call.side_effect = subprocess.CalledProcessError(1, ["ip", "link", "set"]) + + # Mock TapDevice + mock_tun_instance = MagicMock() + mock_tun_instance.name = "mesh0" + mock_tap_device.return_value = mock_tun_instance + + # Attempt to initialize Tunnel, expecting it to raise the CalledProcessError + with self.assertRaises(subprocess.CalledProcessError): + Tunnel(self.mock_iface) + + + @patch("meshtastic.tunnel.TapDevice") + @patch("subprocess.check_call") + def test_cidr_conversion_valid(self, mock_check_call, mock_tap_device): + """Test valid netmask conversions""" + mock_tun = MagicMock() + mock_tun.name = "mesh0" + mock_tap_device.return_value = mock_tun + + tunnel = Tunnel(self.mock_iface) + self.assertEqual(tunnel._cidr_from_netmask("255.255.0.0"), 16) + self.assertEqual(tunnel._cidr_from_netmask("255.255.255.0"), 24) + self.assertEqual(tunnel._cidr_from_netmask("255.0.0.0"), 8) + + @patch("meshtastic.tunnel.TapDevice") + @patch("subprocess.check_call") + def test_cidr_conversion_invalid(self, mock_check_call, mock_tap_device): + """Test invalid netmask raises ValueError or behaves predictably""" + mock_tun = MagicMock() + mock_tun.name = "mesh0" + mock_tap_device.return_value = mock_tun + + tunnel = Tunnel(self.mock_iface) + # Should raise ValueError because 'foo' is not an integer + with self.assertRaises(ValueError): + tunnel._cidr_from_netmask("255.255.foo.0") + + @patch("meshtastic.tunnel.TapDevice") + @patch("subprocess.check_call") + def test_chaos_partial_failure(self, mock_check_call, mock_tap_device): + """CHAOS: Test partial failure where 1st command succeeds but 2nd crashes""" + mock_tun = MagicMock() + mock_tun.name = "mesh0" + mock_tap_device.return_value = mock_tun + + # First call succeeds, Second call raises Error + # Sequence: [UP (success), ADDR (fail), DOWN (rollback, success)] + mock_check_call.side_effect = [ + None, + subprocess.CalledProcessError(1, ["ip", "addr", "add"]), + None + ] + + # The constructor should still fail (re-raise) + with self.assertRaises(subprocess.CalledProcessError): + Tunnel(self.mock_iface) + + # Verify we tried exactly 3 commands: UP, ADDR (fail), DOWN (rollback) + self.assertEqual(mock_check_call.call_count, 3) + + # Verify the last call was the rollback + mock_check_call.assert_called_with(["ip", "link", "set", "dev", "mesh0", "down"]) + +if __name__ == '__main__': + unittest.main()