wifi: replace NetworkManager with wpa_supplicant

openpilot/common/hardware/comma/hardware.py

@@ -4,10 +4,13 @@
 import socket
 import subprocess
 import time
+import uuid
 from functools import cached_property, lru_cache
 from pathlib import Path
 
 from openpilot.cereal import log
+from openpilot.common.nm_keyfile import decode_nm_keyfile_ssid
+from openpilot.common.wifi import WPA_CTRL_PATH, decode_wpa_ssid
 from openpilot.common.utils import sudo_read, sudo_write
 from openpilot.common.gpio import gpio_set, gpio_init, get_irqs_for_action
 from openpilot.common.esim.base import LPABase
@@ -41,7 +44,7 @@ def wpa_supplicant_cmd(cmd: str, timeout: float = 0.2) -> dict[str, str]:
   with socket.socket(socket.AF_UNIX, socket.SOCK_DGRAM) as sock:
     sock.settimeout(timeout)
     sock.bind(f"\0openpilot-wpa-{os.getpid()}-{time.monotonic_ns()}")
-    sock.connect("/run/wpa_supplicant/wlan0")
+    sock.connect(WPA_CTRL_PATH)
     sock.send(cmd.encode())
 
     while True:
@@ -57,6 +60,12 @@ def get_default_route_iface():
     routes = [(int(route[6]), route[0]) for line in f.readlines()[1:] if (route := line.split())[1] == "00000000" and int(route[3], 16) & 0x1]
   return min(routes)[1] if routes else None
 
+def _normalize_uuid(value: str) -> str | None:
+  try:
+    return str(uuid.UUID(value))
+  except ValueError:
+    return None
+
 class HardwareComma(HardwareBase):
   """
     This platform covers the Snapdragon 845-based comma devices:
@@ -211,23 +220,28 @@ def get_network_metered(self, network_type) -> bool:
       return Params().get_bool("GsmMetered")
     try:
       if network_type == NetworkType.wifi:
-        ssid = wpa_supplicant_cmd("STATUS").get("ssid", "")
-        if ssid:
-          # wpa_supplicant escapes non-printable bytes as \xNN; NM keyfile stores ASCII SSIDs as a literal and others as a byte;byte; list
-          ssid_bytes = ssid.encode().decode('unicode_escape').encode('latin-1')
-          ssid_keyfile_list = ';'.join(str(b) for b in ssid_bytes) + ';'
-
-          nm_dirs = ("/run/NetworkManager/system-connections", "/data/etc/NetworkManager/system-connections")
-          for fpath in (p for d in nm_dirs for p in Path(d).glob("*.nmconnection")):
+        status = wpa_supplicant_cmd("STATUS")
+        profile_uuid = status.get("id_str", "").strip('"')
+        normalized_profile_uuid = _normalize_uuid(profile_uuid) if profile_uuid else None
+        if profile_uuid and normalized_profile_uuid is None:
+          return super().get_network_metered(network_type)
+        ssid = decode_wpa_ssid(status.get("ssid", ""))
+        if profile_uuid or ssid:
+          nm_dirs = ("/data/etc/NetworkManager/system-connections", "/run/NetworkManager/system-connections")
+          for fpath in (path for directory in nm_dirs for path in Path(directory).glob("*.nmconnection")):
             raw = sudo_read(str(fpath))
             if not raw:
               continue
             cp = configparser.ConfigParser(interpolation=None)
             try:
               cp.read_string(raw)
-              keyfile_ssid = cp.get("wifi", "ssid", fallback="")
-              if keyfile_ssid != ssid and keyfile_ssid != ssid_keyfile_list:
-                continue
+              if profile_uuid:
+                if _normalize_uuid(cp.get("connection", "uuid", fallback="")) != normalized_profile_uuid:
+                  continue
+              else:
+                wifi_section = "wifi" if cp.has_section("wifi") else "802-11-wireless"
+                if decode_nm_keyfile_ssid(cp.get(wifi_section, "ssid", fallback="")) != ssid:
+                  continue
               metered = cp.getint("connection", "metered", fallback=0)
             except (configparser.Error, ValueError):
               continue

openpilot/common/hardware/comma/tests/test_hardware.py

@@ -0,0 +1,124 @@
+from pathlib import Path
+from unittest import TestCase
+from unittest.mock import MagicMock, patch
+
+from openpilot.common.hardware.comma import hardware as hardware_module
+from openpilot.common.hardware.comma.hardware import HardwareComma, NetworkType
+
+
+class TestHardwareComma(TestCase):
+  def test_canonical_wifi_profile_metered(self):
+    profile = """\
+[connection]
+metered=1
+
+[802-11-wireless]
+ssid=TestNet
+"""
+
+    with (
+      patch.object(hardware_module, "wpa_supplicant_cmd", return_value={"ssid": "TestNet"}),
+      patch.object(Path, "glob", return_value=[Path("profile.nmconnection")]),
+      patch.object(hardware_module, "sudo_read", return_value=profile),
+    ):
+      assert HardwareComma().get_network_metered(NetworkType.wifi)
+
+  def test_escaped_wifi_profile_metered(self):
+    profile = """\
+[connection]
+metered=1
+
+[wifi]
+ssid=\\sGuest\\s
+"""
+
+    with (
+      patch.object(hardware_module, "wpa_supplicant_cmd", return_value={"ssid": " Guest "}),
+      patch.object(Path, "glob", return_value=[Path("profile.nmconnection")]),
+      patch.object(hardware_module, "sudo_read", return_value=profile),
+    ):
+      assert HardwareComma().get_network_metered(NetworkType.wifi)
+  def test_selected_profile_uuid_controls_metering(self):
+    first_uuid = "11111111-1111-1111-1111-111111111111"
+    second_uuid = "22222222-2222-2222-2222-222222222222"
+    profiles = {
+      "first.nmconnection": f"""\\
+[connection]
+uuid={first_uuid}
+metered=1
+
+[wifi]
+ssid=Duplicate
+""",
+      "second.nmconnection": f"""\\
+[connection]
+uuid={second_uuid}
+metered=2
+
+[wifi]
+ssid=Duplicate
+""",
+    }
+
+    with (
+      patch.object(hardware_module, "wpa_supplicant_cmd", return_value={"ssid": "Duplicate", "id_str": second_uuid}),
+      patch.object(Path, "glob", return_value=[Path(name) for name in profiles]),
+      patch.object(hardware_module, "sudo_read", side_effect=lambda path: profiles[path]),
+    ):
+      assert not HardwareComma().get_network_metered(NetworkType.wifi)
+
+  def test_selected_profile_uuid_is_normalized_for_metering(self):
+    active_uuid = "22222222-2222-4abc-8def-222222222222"
+    profiles = {
+      "first.nmconnection": """
+[connection]
+uuid=11111111-1111-4111-8111-111111111111
+metered=2
+
+[wifi]
+ssid=Duplicate
+""",
+      "second.nmconnection": "",
+    }
+    for raw_uuid in (active_uuid.upper(), f"{{{active_uuid.upper()}}}"):
+      with self.subTest(raw_uuid=raw_uuid):
+        profiles["second.nmconnection"] = f"""
+[connection]
+uuid={raw_uuid}
+metered=1
+
+[wifi]
+ssid=Duplicate
+"""
+        with (
+          patch.object(hardware_module, "wpa_supplicant_cmd", return_value={"ssid": "Duplicate", "id_str": active_uuid}),
+          patch.object(Path, "glob", return_value=[Path(name) for name in profiles]),
+          patch.object(hardware_module, "sudo_read", side_effect=lambda path: profiles[path]),
+        ):
+          assert HardwareComma().get_network_metered(NetworkType.wifi)
+
+  def test_malformed_active_profile_uuid_does_not_fall_back_to_ssid(self):
+    profile = """
+[connection]
+uuid=22222222-2222-4abc-8def-222222222222
+metered=1
+
+[wifi]
+ssid=Duplicate
+"""
+    with (
+      patch.object(hardware_module, "wpa_supplicant_cmd", return_value={"ssid": "Duplicate", "id_str": "not-a-uuid"}),
+      patch.object(Path, "glob", return_value=[Path("profile.nmconnection")]),
+      patch.object(hardware_module, "sudo_read", return_value=profile),
+    ):
+      assert not HardwareComma().get_network_metered(NetworkType.wifi)
+
+  def test_hardware_uses_owned_control_socket(self):
+    sock = MagicMock()
+    sock.__enter__.return_value = sock
+    sock.recv.return_value = b"FAIL\\n"
+
+    with patch.object(hardware_module.socket, "socket", return_value=sock):
+      assert hardware_module.wpa_supplicant_cmd("STATUS") == {}
+
+    sock.connect.assert_called_once_with("/run/openpilot-wpa/wlan0")

openpilot/common/nm_keyfile.py

@@ -0,0 +1,47 @@
+NM_KEYFILE_ESCAPES = {
+  "\\": "\\",
+  "n": "\n",
+  "r": "\r",
+  "s": " ",
+  "t": "\t",
+}
+
+
+def _decode_nm_keyfile_string(value: str, *, semicolon_escape: bool) -> str:
+  decoded = []
+  i = 0
+  while i < len(value):
+    if value[i] == "\\" and i + 1 < len(value):
+      escaped = NM_KEYFILE_ESCAPES.get(value[i + 1])
+      if escaped is None and semicolon_escape and value[i + 1] == ";":
+        escaped = ";"
+      if escaped is not None:
+        decoded.append(escaped)
+        i += 2
+        continue
+    decoded.append(value[i])
+    i += 1
+  return "".join(decoded)
+
+
+def decode_nm_keyfile_string(value: str) -> str:
+  return _decode_nm_keyfile_string(value, semicolon_escape=False)
+
+
+def decode_nm_keyfile_ssid(ssid: str) -> str:
+  # Netplan uses a semicolon-terminated decimal byte list for SSIDs that do not
+  # round-trip as a plain keyfile string. Detect that representation before
+  # interpreting any keyfile escapes.
+  if ssid.endswith(";"):
+    parts = ssid[:-1].split(";")
+    if parts and all(part.isascii() and part.isdigit() for part in parts):
+      try:
+        ssid_bytes = bytes(int(part) for part in parts)
+      except ValueError:
+        pass
+      else:
+        if all(byte == 0 for byte in ssid_bytes):
+          return ""
+        return ssid_bytes.decode("utf-8", errors="surrogateescape")
+
+  return _decode_nm_keyfile_string(ssid, semicolon_escape=True)

openpilot/common/tests/test_nm_keyfile.py

@@ -0,0 +1,33 @@
+import unittest
+
+from openpilot.common.nm_keyfile import decode_nm_keyfile_ssid
+
+
+class TestNmKeyfile(unittest.TestCase):
+  def test_decodes_escaped_semicolon(self):
+    self.assertEqual(decode_nm_keyfile_ssid(r"Cafe\;Guest"), "Cafe;Guest")
+
+  def test_decodes_escaped_backslash(self):
+    self.assertEqual(decode_nm_keyfile_ssid(r"Cafe\\Guest"), r"Cafe\Guest")
+
+  def test_preserves_literal_backslash_before_semicolon(self):
+    self.assertEqual(decode_nm_keyfile_ssid(r"Cafe\\;Guest"), r"Cafe\;Guest")
+
+  def test_decodes_boundary_spaces(self):
+    self.assertEqual(decode_nm_keyfile_ssid(r"\sCafe\s"), " Cafe ")
+
+  def test_decodes_decimal_byte_list(self):
+    self.assertEqual(decode_nm_keyfile_ssid("65;66;67;"), "ABC")
+
+  def test_preserves_non_utf8_decimal_byte_list(self):
+    self.assertEqual(
+      decode_nm_keyfile_ssid("255;65;"),
+      b"\xffA".decode("utf-8", errors="surrogateescape"),
+    )
+
+  def test_preserves_unknown_escape(self):
+    self.assertEqual(decode_nm_keyfile_ssid(r"Cafe\q"), r"Cafe\q")
+
+
+if __name__ == "__main__":
+  unittest.main()

openpilot/common/wifi.py

@@ -0,0 +1,67 @@
+import os
+
+
+WPA_CTRL_DIR = "/run/openpilot-wpa"
+WPA_CTRL_PATH = os.path.join(WPA_CTRL_DIR, "wlan0")
+WPA_PID_FILE = os.path.join(WPA_CTRL_DIR, "wpa_supplicant.pid")
+
+_HEX = "0123456789abcdefABCDEF"
+
+
+def decode_wpa_ssid(encoded: str) -> str:
+  """Decode a wpa_supplicant printf_encode'd SSID without losing byte identity."""
+  out = bytearray()
+  i = 0
+  while i < len(encoded):
+    char = encoded[i]
+    if char != "\\":
+      out.append(ord(char) & 0xff)
+      i += 1
+      continue
+
+    i += 1
+    if i >= len(encoded):
+      break
+
+    escaped = encoded[i]
+    if escaped == "\\":
+      out.append(ord("\\"))
+      i += 1
+    elif escaped == '"':
+      out.append(ord('"'))
+      i += 1
+    elif escaped == "n":
+      out.append(ord("\n"))
+      i += 1
+    elif escaped == "r":
+      out.append(ord("\r"))
+      i += 1
+    elif escaped == "t":
+      out.append(ord("\t"))
+      i += 1
+    elif escaped == "e":
+      out.append(0x1b)
+      i += 1
+    elif escaped == "x":
+      i += 1
+      if i + 1 < len(encoded) and encoded[i] in _HEX and encoded[i + 1] in _HEX:
+        out.append(int(encoded[i:i + 2], 16))
+        i += 2
+      elif i < len(encoded) and encoded[i] in _HEX:
+        out.append(int(encoded[i], 16))
+        i += 1
+    elif "0" <= escaped <= "7":
+      value = ord(escaped) - ord("0")
+      i += 1
+      if i < len(encoded) and "0" <= encoded[i] <= "7":
+        value = value * 8 + ord(encoded[i]) - ord("0")
+        i += 1
+        if i < len(encoded) and "0" <= encoded[i] <= "7":
+          value = value * 8 + ord(encoded[i]) - ord("0")
+          i += 1
+      out.append(value & 0xff)
+    # Unknown escapes consume only the backslash.
+
+  if not out or all(byte == 0 for byte in out):
+    return ""
+  return out.decode("utf-8", errors="surrogateescape")

openpilot/selfdrive/ui/layouts/settings/settings.py

@@ -53,7 +53,6 @@ def __init__(self):
 
     # Panel configuration
     wifi_manager = WifiManager()
-    wifi_manager.set_active(False)
 
     self._panels = {
       PanelType.DEVICE: PanelInfo(tr_noop("Device"), DeviceLayout()),

openpilot/selfdrive/ui/mici/layouts/settings/network/__init__.py

@@ -3,7 +3,8 @@
 from openpilot.selfdrive.ui.mici.layouts.settings.network.wifi_ui import WifiIcon
 from openpilot.selfdrive.ui.mici.widgets.button import BigButton
 from openpilot.system.ui.lib.application import gui_app
-from openpilot.system.ui.lib.wifi_manager import WifiManager, ConnectStatus, SecurityType, normalize_ssid
+from openpilot.system.ui.lib.wifi_manager import WifiManager, ConnectStatus, SecurityType
+from openpilot.system.ui.lib.wpa_ctrl import normalize_ssid
 
 
 class WifiNetworkButton(BigButton):