diff --git a/system/ui/lib/tests/test_tethering_nat.py b/system/ui/lib/tests/test_tethering_nat.py new file mode 100644 index 0000000000..b833fe395a --- /dev/null +++ b/system/ui/lib/tests/test_tethering_nat.py @@ -0,0 +1,116 @@ +import subprocess +import sys +import types +import unittest +from unittest import mock + +# swaglog pulls in compiled msgq, unavailable in some checkouts; stub it +if "openpilot.common.swaglog" not in sys.modules: + try: + import openpilot.common.swaglog # noqa: F401 + except ImportError: + stub = types.ModuleType("openpilot.common.swaglog") + stub.cloudlog = mock.MagicMock() + sys.modules["openpilot.common.swaglog"] = stub + +from openpilot.system.ui.lib import tethering_nat + + +def _run_ok(cmd, **kwargs): + return subprocess.CompletedProcess(cmd, 0, "", "") + + +class TestTetheringNat(unittest.TestCase): + + def test_subnet_cidr(self): + self.assertEqual(tethering_nat._subnet_cidr("10.42.0.1", 24), "10.42.0.0/24") + self.assertEqual(tethering_nat._subnet_cidr("192.168.43.1", 24), "192.168.43.0/24") + self.assertEqual(tethering_nat._subnet_cidr("10.99.62.240", 24), "10.99.62.0/24") + self.assertEqual(tethering_nat._subnet_cidr("172.16.5.9", 12), "172.16.0.0/12") + self.assertEqual(tethering_nat._subnet_cidr("8.8.8.8", 32), "8.8.8.8/32") + self.assertIsNone(tethering_nat._subnet_cidr("300.1.1.1", 24)) + self.assertIsNone(tethering_nat._subnet_cidr("1.2.3", 24)) + + def test_interface_subnet_parses_ip_output(self): + result = subprocess.CompletedProcess( + [], 0, "11: wlan0 inet 10.42.0.1/24 brd 10.42.0.255 scope global wlan0\n", "") + with mock.patch.object(subprocess, "run", return_value=result): + self.assertEqual(tethering_nat._interface_subnet("wlan0"), "10.42.0.0/24") + + def test_interface_subnet_no_address(self): + result = subprocess.CompletedProcess([], 0, "", "") + with mock.patch.object(subprocess, "run", return_value=result): + self.assertIsNone(tethering_nat._interface_subnet("wlan0")) + + def test_hotspot_subnets_live_first_then_candidates(self): + with mock.patch.object(tethering_nat, "_interface_subnet", return_value="10.42.0.0/24"): + self.assertEqual(tethering_nat.hotspot_subnets(), + ("10.42.0.0/24", "192.168.43.0/24")) + with mock.patch.object(tethering_nat, "_interface_subnet", return_value=None): + self.assertEqual(tethering_nat.hotspot_subnets(), + ("10.42.0.0/24", "192.168.43.0/24")) + with mock.patch.object(tethering_nat, "_interface_subnet", return_value="192.168.0.0/24"): + self.assertEqual(tethering_nat.hotspot_subnets(), + ("192.168.0.0/24", "10.42.0.0/24", "192.168.43.0/24")) + + def test_ensure_noops_without_iptables(self): + with mock.patch.object(tethering_nat.shutil, "which", return_value=None): + self.assertFalse(tethering_nat.ensure_tethering_nat()) + + def test_ensure_applies_rules_idempotently(self): + calls = [] + + def fake_run(cmd, **kwargs): + calls.append(cmd) + return subprocess.CompletedProcess(cmd, 0, "", "") + + with mock.patch.object(tethering_nat.shutil, "which", return_value="/usr/sbin/iptables-legacy"), \ + mock.patch.object(tethering_nat.subprocess, "run", side_effect=fake_run), \ + mock.patch.object(tethering_nat, "_interface_subnet", return_value=None): + self.assertTrue(tethering_nat.ensure_tethering_nat()) + + sysctls = [c for c in calls if "sysctl" in c] + self.assertEqual(len(sysctls), 1) + self.assertIn("net.ipv4.ip_forward=1", sysctls[0]) + + # check-only passes: nothing re-added; 3 checks per subnet, 2 candidate subnets + ipt = [c for c in calls if "iptables-legacy" in c] + self.assertEqual(len(ipt), 6) + self.assertTrue(all("-C" in c for c in ipt)) + + def test_ensure_adds_when_check_fails(self): + calls = [] + + def fake_run(cmd, **kwargs): + calls.append(cmd) + # sudo checks fail (rule missing), sudo adds succeed + returncode = 1 if "-C" in cmd else 0 + return subprocess.CompletedProcess(cmd, returncode, "", "") + + with mock.patch.object(tethering_nat.shutil, "which", return_value="/usr/sbin/iptables-legacy"), \ + mock.patch.object(tethering_nat.subprocess, "run", side_effect=fake_run), \ + mock.patch.object(tethering_nat, "_interface_subnet", return_value=None): + self.assertTrue(tethering_nat.ensure_tethering_nat()) + + ipt = [c for c in calls if "iptables-legacy" in c] + self.assertEqual(len(ipt), 12) # check+add per rule, 3 rules x 2 subnets + self.assertEqual(len([c for c in ipt if "-A" in c]), 6) + + def test_ensure_excludes_live_subnet_when_not_hotspot(self): + calls = [] + + def fake_run(cmd, **kwargs): + calls.append(cmd) + return subprocess.CompletedProcess(cmd, 0, "", "") + + with mock.patch.object(tethering_nat.shutil, "which", return_value="/usr/sbin/iptables-legacy"), \ + mock.patch.object(tethering_nat.subprocess, "run", side_effect=fake_run), \ + mock.patch.object(tethering_nat, "_interface_subnet", return_value="10.99.62.0/24"): + self.assertTrue(tethering_nat.ensure_tethering_nat(include_live_subnet=False)) + + ipt = [c for c in calls if "iptables-legacy" in c] + self.assertFalse(any("10.99.62.0/24" in " ".join(c) for c in ipt)) + + +if __name__ == "__main__": + unittest.main() diff --git a/system/ui/lib/tethering_nat.py b/system/ui/lib/tethering_nat.py new file mode 100644 index 0000000000..70f3f8c6c5 --- /dev/null +++ b/system/ui/lib/tethering_nat.py @@ -0,0 +1,139 @@ +"""WAN NAT for the Wi-Fi tethering hotspot. + +AGNOS kernels (4.9, CONFIG_NF_TABLES not set — verified in upstream AGNOS) +can't run NetworkManager's shared-mode firewall rules, so tethered clients +get DHCP but no WAN access. `ensure_tethering_nat()` idempotently applies +masquerade/forward rules via iptables-legacy and enables IPv4 forwarding. +Safe to call repeatedly, from any hotspot activation path. +""" + +import shutil +import subprocess + +from openpilot.common.swaglog import cloudlog + +IPTABLES = "iptables-legacy" +IPV4_FORWARD_SYSCTL = "net.ipv4.ip_forward=1" + +# NetworkManager's default shared range (profile without pinned address-data) +NM_SHARED_SUBNET = "10.42.0.0/24" +# WifiManager's pinned tethering address (TETHERING_IP_ADDRESS/24) +WIFI_MANAGER_SUBNET = "192.168.43.0/24" + + +def _interface_subnet(interface: str) -> str | None: + """Return the interface's current IPv4 subnet as a CIDR, or None.""" + try: + result = subprocess.run( + ["ip", "-4", "-o", "addr", "show", "dev", interface], + capture_output=True, text=True, timeout=5, check=True, + ) + except (OSError, subprocess.SubprocessError) as exc: + cloudlog.warning(f"Failed to read {interface} addresses for tethering NAT: {exc}") + return None + + for line in result.stdout.splitlines(): + # Example: "11: wlan0 inet 10.42.0.1/24 brd ..." + parts = line.split() + if "inet" in parts and "/" in parts[parts.index("inet") + 1]: + addr, prefix = parts[parts.index("inet") + 1].split("/", 1) + try: + prefix = int(prefix) + except ValueError: + continue + if prefix > 0 and not addr.startswith("127."): + return _subnet_cidr(addr, prefix) + return None + + +def _subnet_cidr(addr: str, prefix: int) -> str | None: + if prefix < 8 or prefix > 32 or len(addr.split(".")) != 4: + return None + octets = [int(o) for o in addr.split(".")] + if any(o > 255 for o in octets): + return None + mask = ((0xFFFFFFFF << (32 - prefix)) & 0xFFFFFFFF) + network = (octets[0] << 24) | (octets[1] << 16) | (octets[2] << 8) | octets[3] + network &= mask + return f"{(network >> 24) & 0xFF}.{(network >> 16) & 0xFF}.{(network >> 8) & 0xFF}.{network & 0xFF}/{prefix}" + + +def hotspot_subnets(interface: str = "wlan0") -> tuple[str, ...]: + """Candidate hotspot subnets to NAT, live subnet first when present.""" + subnets = [] + live = _interface_subnet(interface) + if live is not None: + subnets.append(live) + for candidate in (NM_SHARED_SUBNET, WIFI_MANAGER_SUBNET): + if candidate not in subnets: + subnets.append(candidate) + return tuple(subnets) + + +def _ensure_rule(check_args: tuple[str, ...], add_args: tuple[str, ...]) -> bool: + """Add a rule if missing. Both the check and the add require root.""" + try: + result = subprocess.run(["sudo", "-n", IPTABLES, *check_args], + capture_output=True, timeout=5) + if result.returncode == 0: + return True + result = subprocess.run(["sudo", "-n", IPTABLES, *add_args], + capture_output=True, text=True, timeout=5) + if result.returncode != 0: + cloudlog.warning(f"Failed to apply tethering NAT rule ({' '.join(add_args)}): {result.stderr.strip()}") + return False + return True + except (OSError, subprocess.SubprocessError) as exc: + cloudlog.warning(f"Error applying tethering NAT rule ({' '.join(add_args)}): {exc}") + return False + + +def ensure_tethering_nat(interface: str = "wlan0", include_live_subnet: bool = True) -> bool: + """Idempotently ensure WAN NAT for the hotspot subnet(s). Never raises. + + `include_live_subnet` also NATs the interface's current subnet; only pass + True while the hotspot is active so a client connection's LAN subnet never + gets masqueraded. Rules for subnets that aren't routed are inert. + + Returns False (without raising) where unsupported, e.g. PCs without + iptables-legacy. + """ + if shutil.which(IPTABLES) is None: + cloudlog.debug(f"{IPTABLES} not available; skipping tethering NAT") + return False + + ok = True + try: + result = subprocess.run(["sudo", "-n", "sysctl", "-w", IPV4_FORWARD_SYSCTL], + capture_output=True, text=True, timeout=5) + if result.returncode != 0: + cloudlog.warning(f"Failed to enable IPv4 forwarding for tethering: {result.stderr.strip()}") + ok = False + + subnets: list[str] = [] + if include_live_subnet: + live = _interface_subnet(interface) + if live is not None: + subnets.append(live) + for candidate in (NM_SHARED_SUBNET, WIFI_MANAGER_SUBNET): + if candidate not in subnets: + subnets.append(candidate) + + for subnet in subnets: + ok &= _ensure_rule( + ("-t", "nat", "-C", "POSTROUTING", "-s", subnet, "!", "-d", subnet, "-j", "MASQUERADE"), + ("-t", "nat", "-A", "POSTROUTING", "-s", subnet, "!", "-d", subnet, "-j", "MASQUERADE"), + ) + ok &= _ensure_rule( + ("-C", "FORWARD", "-s", subnet, "-j", "ACCEPT"), + ("-A", "FORWARD", "-s", subnet, "-j", "ACCEPT"), + ) + ok &= _ensure_rule( + ("-C", "FORWARD", "-d", subnet, "-m", "state", "--state", "RELATED,ESTABLISHED", "-j", "ACCEPT"), + ("-A", "FORWARD", "-d", subnet, "-m", "state", "--state", "RELATED,ESTABLISHED", "-j", "ACCEPT"), + ) + except (OSError, subprocess.SubprocessError) as exc: + cloudlog.warning(f"Error applying tethering NAT: {exc}") + return False + + return ok diff --git a/system/ui/lib/wifi_manager.py b/system/ui/lib/wifi_manager.py index 49bb0dcb6c..7c1f131ebf 100644 --- a/system/ui/lib/wifi_manager.py +++ b/system/ui/lib/wifi_manager.py @@ -49,6 +49,8 @@ try: except Exception: Params = None +from openpilot.system.ui.lib.tethering_nat import ensure_tethering_nat + TETHERING_IP_ADDRESS = "192.168.43.1" DEFAULT_TETHERING_PASSWORD = "swagswagcomma" SIGNAL_QUEUE_SIZE = 10 @@ -317,6 +319,10 @@ class WifiManager: self._wifi_state = WifiState(ssid=ssid, status=status) + # Hotspot may already be active (boot restore / autoconnect fallback) + if ssid == self._tethering_ssid: + self._ensure_tethering_nat() + if block: worker() else: @@ -589,6 +595,11 @@ class WifiManager: self._enqueue_callbacks(self._activated) self._update_active_connection_info() + # AGNOS (no nf_tables — verified upstream) never installs shared-mode + # NAT rules; ensure them on every hotspot activation path + if wifi_state.ssid == self._tethering_ssid: + self._ensure_tethering_nat() + # Persist volatile connections (created by AddAndActivateConnection2) to disk if conn_path is not None: conn_addr = DBusAddress(conn_path, bus_name=NM, interface=NM_CONNECTION_IFACE) @@ -1057,6 +1068,14 @@ class WifiManager: def set_ipv4_forward(self, enabled: bool): self._ipv4_forward = enabled + def _ensure_tethering_nat(self): + def worker(): + try: + ensure_tethering_nat() + except Exception: + cloudlog.exception("Failed to ensure tethering NAT") + threading.Thread(target=worker, daemon=True).start() + def set_tethering_active(self, active: bool): if self._backend_unavailable: cloudlog.warning(f"Ignoring set_tethering_active({active}); Wi-Fi backend unavailable")