pandad: check comms protocol version (#28983)

* proto version check

* bump panda

* update bootstub

* test

* bump panda

* fix that

* bump panda

* fix test case

* more fix

---------

Co-authored-by: Comma Device <device@comma.ai>
old-commit-hash: 2720fdf2a1

selfdrive/boardd/pandad.py

@@ -7,7 +7,7 @@ import subprocess
 from typing import List, NoReturn
 from functools import cmp_to_key
 
-from panda import Panda, PandaDFU, FW_PATH
+from panda import Panda, PandaDFU, PandaProtocolMismatch, FW_PATH
 from common.basedir import BASEDIR
 from common.params import Params
 from selfdrive.boardd.set_time import set_time
@@ -25,7 +25,12 @@ def get_expected_signature(panda: Panda) -> bytes:
 
 
 def flash_panda(panda_serial: str) -> Panda:
-  panda = Panda(panda_serial)
+  try:
+    panda = Panda(panda_serial)
+  except PandaProtocolMismatch:
+    cloudlog.warning("detected protocol mismatch, reflashing panda")
+    HARDWARE.recover_internal_panda()
+    raise
 
   fw_signature = get_expected_signature(panda)
   internal_panda = panda.is_internal()
@@ -42,10 +47,11 @@ def flash_panda(panda_serial: str) -> Panda:
   if panda.bootstub:
     bootstub_version = panda.get_version()
     cloudlog.info(f"Flashed firmware not booting, flashing development bootloader. {bootstub_version=}, {internal_panda=}")
+
     if internal_panda:
       HARDWARE.recover_internal_panda()
     panda.recover(reset=(not internal_panda))
-    cloudlog.info("Done flashing bootloader")
+    cloudlog.info("Done flashing bootstub")
 
   if panda.bootstub:
     cloudlog.info("Panda still not booting, exiting")
@@ -152,6 +158,9 @@ def main() -> NoReturn:
       # a panda was disconnected while setting everything up. let's try again
       cloudlog.exception("Panda USB exception while setting up")
       continue
+    except PandaProtocolMismatch:
+      cloudlog.exception("pandad.protocol_mismatch")
+      continue
     except Exception:
       cloudlog.exception("pandad.uncaught_exception")
       continue

selfdrive/boardd/tests/test_pandad.py

@@ -6,7 +6,7 @@ import unittest
 import cereal.messaging as messaging
 from cereal import log
 from common.gpio import gpio_set, gpio_init
-from panda import Panda, PandaDFU
+from panda import Panda, PandaDFU, PandaProtocolMismatch
 from selfdrive.test.helpers import phone_only
 from selfdrive.manager.process_config import managed_processes
 from system.hardware import HARDWARE
@@ -22,8 +22,8 @@ class TestPandad(unittest.TestCase):
 
   def _wait_for_boardd(self, timeout=30):
     sm = messaging.SubMaster(['peripheralState'])
-    for _ in range(timeout):
-      sm.update(1000)
+    for _ in range(timeout*10):
+      sm.update(100)
       if sm['peripheralState'].pandaType != log.PandaState.PandaType.unknown:
         break
 
@@ -34,6 +34,27 @@ class TestPandad(unittest.TestCase):
     HARDWARE.recover_internal_panda()
     assert Panda.wait_for_dfu(None, 10)
 
+  def _flash_and_test(self, fn, expect_mismatch=False):
+    self._go_to_dfu()
+    pd = PandaDFU(None)
+    if fn is None:
+      fn = os.path.join(HERE, pd.get_mcu_type().config.bootstub_fn)
+    with open(fn, "rb") as f:
+      pd.program_bootstub(f.read())
+    pd.reset()
+    HARDWARE.reset_internal_panda()
+
+    assert Panda.wait_for_panda(None, 10)
+    if expect_mismatch:
+      with self.assertRaises(PandaProtocolMismatch):
+        Panda()
+    else:
+      with Panda() as p:
+        assert p.bootstub
+
+    managed_processes['pandad'].start()
+    self._wait_for_boardd(45)
+
   @phone_only
   def test_in_dfu(self):
     HARDWARE.recover_internal_panda()
@@ -71,24 +92,18 @@ class TestPandad(unittest.TestCase):
     managed_processes['pandad'].start()
     self._wait_for_boardd(8)
 
+  @phone_only
+  def test_protocol_version_check(self):
+    if HARDWARE.get_device_type() == 'tici':
+      self.skipTest("")
+
+    # flash old fw
+    fn = os.path.join(HERE, "bootstub.panda_h7_spiv0.bin")
+    self._flash_and_test(fn, expect_mismatch=True)
+
   @phone_only
   def test_release_to_devel_bootstub(self):
-    # flash release bootstub
-    self._go_to_dfu()
-    pd = PandaDFU(None)
-    fn = os.path.join(HERE, pd.get_mcu_type().config.bootstub_fn)
-    with open(fn, "rb") as f:
-      pd.program_bootstub(f.read())
-    pd.reset()
-    HARDWARE.reset_internal_panda()
-
-    assert Panda.wait_for_panda(None, 10)
-    with Panda() as p:
-      assert p.bootstub
-
-    managed_processes['pandad'].start()
-    self._wait_for_boardd(45)
-
+    self._flash_and_test(None)
 
 if __name__ == "__main__":
   unittest.main()