Merge pull request #223 from dirwin31/RemoveVerifyHeadUnit

Remove Verify Head Unit

openpilot/starpilot/system/android_auto/identity.py

@@ -83,13 +83,16 @@ def load_identity(directory: Path | None = None, now: datetime | None = None) ->
                   expires.isoformat() if expires is not None else "unknown", days_left)
 
 
-CONFIG_VERSION = 4
+CONFIG_VERSION = 5
 # Before config versions, every save wrote these defaults, which pinned projection
 # at 12 fps / 4000 kbps. Unversioned files holding exactly them get today's defaults.
 LEGACY_DEFAULTS = {"fps": 12, "bitrate_kbps": 4000}
 # Before version 3 the car view could only publish RGBA, and every save wrote these off.
 # They were never a choice, so older files get today's defaults for them.
 PIPELINE_RESET = ("gpu_nv12", "async_readback")
+# Before version 5 every save wrote verify_head_unit on, and some cars send certificates the root cannot parse.
+# Older files get today's default (off).
+VERIFY_RESET = ("verify_head_unit",)
 
 DEFAULT_CONFIG = {
   "config_version": CONFIG_VERSION,
@@ -98,7 +101,7 @@ DEFAULT_CONFIG = {
   "companion_address": "",    # explicitly selected bonded car used before a separate wireless adapter
   "companion_name": "",
   "rfcomm_channel": 0,         # 0 = discover through SDP (normal); set only to work around a broken SDP record
-  "verify_head_unit": True,    # verify the car's certificate against root-cert.pem when present
+  "verify_head_unit": False,   # verify the car's certificate against root-cert.pem when present
   "connection": "wireless",    # "wireless": Bluetooth + the car's Wi-Fi; "wired": USB cable from the car to the comma's USB-C port
   "usb_mode": "auto",          # wired: "auto" waits for the car's accessory handshake, then presents as an accessory itself;
                                # "handshake" waits for it only; "direct" presents as an accessory at once
@@ -134,6 +137,8 @@ def load_config(path: Path | None = None) -> dict:
         stored = {key: value for key, value in stored.items() if LEGACY_DEFAULTS.get(key, object()) != value}
       if not isinstance(version, int) or version < 3:
         stored = {key: value for key, value in stored.items() if key not in PIPELINE_RESET}
+      if not isinstance(version, int) or version < 5:
+        stored = {key: value for key, value in stored.items() if key not in VERIFY_RESET}
       config.update({key: value for key, value in stored.items() if key in DEFAULT_CONFIG and isinstance(value, type(DEFAULT_CONFIG[key]))})
   except (OSError, ValueError):
     pass

openpilot/starpilot/system/android_auto/supervisor.py

@@ -1142,7 +1142,7 @@ class Supervisor:
   def _project(self, result, lease, ident, connect=None) -> None:
     config = self.config
     sock = connect() if connect is not None else self._connect_tcp(result, lease)
-    ca = ident.root if config.get("verify_head_unit", True) else None
+    ca = ident.root if config.get("verify_head_unit", False) else None
     session = ProjectionSession(sock, ident.cert, ident.key, self.log, ca)
     self._stage("authenticating")
     session.authenticate()

openpilot/starpilot/system/android_auto/tests/test_auto_connect.py

@@ -259,7 +259,50 @@ def test_config_v3_preserves_explicit_pipeline_preferences(tmp_path):
   path.write_text(json.dumps({"config_version": 3, "gpu_nv12": False, "async_readback": False, "bitrate_kbps": 5000}))
   config = identity_store.load_config(path)
   assert config["gpu_nv12"] is False and config["async_readback"] is False and config["bitrate_kbps"] == 5000
-  assert config["config_version"] == 4
+  assert config["config_version"] == 5
+
+
+def test_config_v4_resets_head_unit_verification(tmp_path):
+  from openpilot.starpilot.system.android_auto import identity as identity_store
+  path = tmp_path / "config.json"
+  # Every v4 save wrote verification on; it was never chosen.
+  path.write_text(json.dumps({"config_version": 4, "verify_head_unit": True}))
+  config = identity_store.load_config(path)
+  assert config["verify_head_unit"] is False
+  identity_store.save_config(config, path)
+  # Saving and loading again keeps it off.
+  saved = json.loads(path.read_text())
+  assert saved["config_version"] == 5 and saved["verify_head_unit"] is False
+  assert identity_store.load_config(path)["verify_head_unit"] is False
+  # From v5 on it is a real choice and stays as saved.
+  path.write_text(json.dumps({**json.loads(path.read_text()), "verify_head_unit": True}))
+  assert identity_store.load_config(path)["verify_head_unit"] is True
+
+
+@pytest.mark.parametrize("stored, verified", [({}, False), ({"verify_head_unit": False}, False), ({"verify_head_unit": True}, True)])
+def test_projection_verifies_head_unit_only_when_chosen(identity, tmp_path, monkeypatch, stored, verified):
+  from openpilot.starpilot.system.android_auto import identity as identity_store, supervisor
+  sup, _ = make_supervisor(identity, tmp_path, monkeypatch, lambda *a, **k: socket.socketpair()[0])
+  identity_store.save_config({**identity_store.load_config(), **stored})
+  sup.config = identity_store.load_config()
+  ident = identity_store.load_identity()
+  assert ident.root is not None  # the root is installed, so only the setting decides
+  seen = []
+
+  class Stop(Exception):
+    pass
+
+  class RecordingSession:
+    def __init__(self, sock, cert, key, log, ca):
+      seen.append(ca)
+
+    def authenticate(self):
+      raise Stop
+
+  monkeypatch.setattr(supervisor, "ProjectionSession", RecordingSession)
+  with pytest.raises(Stop):
+    sup._project(None, None, ident, connect=lambda: socket.socketpair()[0])
+  assert seen == [ident.root if verified else None]
 
 
 def _wireless_backoff_harness(sup, monkeypatch, failure):