mirror of
https://github.com/firestar5683/StarPilot.git
synced 2026-09-18 05:23:57 +08:00
Harden tethering NAT activation
This commit is contained in:
@@ -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(),
|
||||
|
||||
@@ -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]
|
||||
|
||||
@@ -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
|
||||
|
||||
Reference in New Issue
Block a user