Ensure WAN NAT for Wi-Fi tethering hotspot

AGNOS kernels (4.9, CONFIG_NF_TABLES not set — verified in upstream AGNOS boot image) cannot run NetworkManager's shared-mode firewall rules, so tethered clients get DHCP but no WAN access. Idempotently apply masquerade/forward rules via iptables-legacy on every hotspot activation path (UI toggle, autoconnect fallback, boot restore), replacing a manually re-installed systemd service after each AGNOS update.
2026-09-18 05:23:57 +08:00 · 2026-09-14 00:12:54 -04:00
parent 9d0ab7a849
commit b640bbc20e
3 changed files with 274 additions and 0 deletions
@@ -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()
@@ -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
@@ -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")