diff --git a/system/ui/lib/tests/test_tethering_nat.py b/system/ui/lib/tests/test_tethering_nat.py index b833fe395a..d8186e6da7 100644 --- a/system/ui/lib/tests/test_tethering_nat.py +++ b/system/ui/lib/tests/test_tethering_nat.py @@ -30,6 +30,9 @@ class TestTetheringNat(unittest.TestCase): 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)) + self.assertIsNone(tethering_nat._subnet_cidr("-1.2.3.4", 24)) + self.assertIsNone(tethering_nat._subnet_cidr("1.2.three.4", 24)) + self.assertIsNone(tethering_nat._subnet_cidr("1.2.3.4", True)) def test_interface_subnet_parses_ip_output(self): result = subprocess.CompletedProcess( @@ -42,6 +45,11 @@ class TestTetheringNat(unittest.TestCase): with mock.patch.object(subprocess, "run", return_value=result): self.assertIsNone(tethering_nat._interface_subnet("wlan0")) + def test_interface_subnet_ignores_malformed_address_lines(self): + result = subprocess.CompletedProcess([], 0, "11: wlan0 inet\n11: wlan0 inet not-an-address\n", "") + 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(), diff --git a/system/ui/lib/tethering_nat.py b/system/ui/lib/tethering_nat.py index 70f3f8c6c5..3f4ae8b658 100644 --- a/system/ui/lib/tethering_nat.py +++ b/system/ui/lib/tethering_nat.py @@ -35,8 +35,9 @@ def _interface_subnet(interface: str) -> str | 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) + inet_index = parts.index("inet") if "inet" in parts else -1 + if inet_index >= 0 and len(parts) > inet_index + 1 and "/" in parts[inet_index + 1]: + addr, prefix = parts[inet_index + 1].split("/", 1) try: prefix = int(prefix) except ValueError: @@ -47,10 +48,13 @@ def _interface_subnet(interface: str) -> str | None: def _subnet_cidr(addr: str, prefix: int) -> str | None: - if prefix < 8 or prefix > 32 or len(addr.split(".")) != 4: + if not isinstance(prefix, int) or isinstance(prefix, bool) or prefix < 8 or prefix > 32: return None - octets = [int(o) for o in addr.split(".")] - if any(o > 255 for o in octets): + try: + octets = [int(o) for o in addr.split(".")] + except (AttributeError, TypeError, ValueError): + return None + if len(octets) != 4 or any(o < 0 or 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] diff --git a/system/ui/lib/wifi_manager.py b/system/ui/lib/wifi_manager.py index 7c1f131ebf..2f5e184e88 100644 --- a/system/ui/lib/wifi_manager.py +++ b/system/ui/lib/wifi_manager.py @@ -320,7 +320,8 @@ class WifiManager: self._wifi_state = WifiState(ssid=ssid, status=status) # Hotspot may already be active (boot restore / autoconnect fallback) - if ssid == self._tethering_ssid: + tethering_ssid = getattr(self, "_tethering_ssid", None) + if tethering_ssid is not None and ssid == tethering_ssid: self._ensure_tethering_nat() if block: @@ -597,7 +598,8 @@ class WifiManager: # 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: + tethering_ssid = getattr(self, "_tethering_ssid", None) + if tethering_ssid is not None and wifi_state.ssid == tethering_ssid: self._ensure_tethering_nat() # Persist volatile connections (created by AddAndActivateConnection2) to disk