From b7c90e530b1914fd5d28d9064e5ccade4e61a872 Mon Sep 17 00:00:00 2001 From: Brett Adams Date: Fri, 2 Oct 2026 08:36:41 +1000 Subject: [PATCH] fix(signed): re-request session info when a reply fails authentication A session_info reply whose session_info_tag fails authentication made the handshake fail on the first bad reply. Tesla's vehicle-command instead discards it and re-sends the unsigned, idempotent SessionInfoRequest after RetryInterval (1s on BLE). Match that: Commands._handshake now retries up to 3 attempts 1s apart on SessionInfoAuthenticationFault and still raises after the last. The unauthenticated reply is never trusted. Observed live: right after a key-card-approved whitelist add, VCSEC's first handshake reply carried session_info_tag with a zero-length tag (signature_data = 6a 02 32 00) alongside an otherwise sane SessionInfo (status OK, new key handle), so pairing appeared to fail although the key was enrolled. The captured 141-byte frame is the regression fixture. --- AGENTS.md | 7 ++ tesla_fleet_api/tesla/vehicle/commands.py | 18 ++++- tests/test_session_info_authentication.py | 98 +++++++++++++++++++++++ 3 files changed, 122 insertions(+), 1 deletion(-) diff --git a/AGENTS.md b/AGENTS.md index cd5c4d4..e08fc79 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -416,6 +416,13 @@ break. so polling survives reconnects) until `timeout`. **Never re-send the whitelist op** — it re-prompts the user. Deadline with neither path confirming raises `BluetoothTimeout`. +- **A session_info reply that fails authentication is re-requested, never + trusted**: `Commands._handshake` re-sends the unsigned session-info request up + to `_handshake_attempts` (3) times `_handshake_retry_interval` (1s) apart on + `SessionInfoAuthenticationFault`, matching vehicle-command's `tryStartSession`, + and raises after the last. VCSEC was observed sending a present-but-empty + `session_info_tag` on the first handshake right after a key-card-approved + whitelist add (fixture in `tests/test_session_info_authentication.py`). - **Idle keepalive**: an idle held link to the vehicle drops at ~42s mean; a trivial passive GATT read on an idle cadence extends the session ~10x, so `keepalive_interval` (default `DEFAULT_KEEPALIVE_INTERVAL`, `None`/`0` diff --git a/tesla_fleet_api/tesla/vehicle/commands.py b/tesla_fleet_api/tesla/vehicle/commands.py index 016ab1d..f653f22 100644 --- a/tesla_fleet_api/tesla/vehicle/commands.py +++ b/tesla_fleet_api/tesla/vehicle/commands.py @@ -477,6 +477,8 @@ class Commands(ABC, Vehicle[CommandParentT], Generic[CommandParentT]): _auth_method: ClassVar[Literal["hmac", "aes"]] # Transport identity for debug logging; set per concrete subclass. _transport_name: ClassVar[str] + _handshake_attempts: ClassVar[int] = 3 + _handshake_retry_interval: ClassVar[float] = 1.0 def __init__( self, @@ -1217,7 +1219,21 @@ async def _handshake(self, domain: Domain) -> bool: raise SigningDisabled() LOGGER.debug(f"Handshake with domain {Domain.Name(domain)}") - await self._send(self._session_info_request(domain), "session_info") + # Like vehicle-command's tryStartSession: a session_info reply that + # fails authentication is discarded and the (unsigned, idempotent) + # request re-sent, instead of failing the handshake on one bad reply. + for attempt in range(self._handshake_attempts): + try: + await self._send(self._session_info_request(domain), "session_info") + break + except SessionInfoAuthenticationFault as e: + LOGGER.debug( + f"Discarding unauthenticated session info from " + f"{Domain.Name(domain)} ({e.data}), attempt {attempt + 1}" + ) + if attempt + 1 >= self._handshake_attempts: + raise + await sleep(self._handshake_retry_interval) return self._sessions[domain].ready def _session_info_request(self, domain: Domain) -> RoutableMessage: diff --git a/tests/test_session_info_authentication.py b/tests/test_session_info_authentication.py index f64c068..986707b 100644 --- a/tests/test_session_info_authentication.py +++ b/tests/test_session_info_authentication.py @@ -653,6 +653,7 @@ async def test_forged_session_info_is_rejected_and_session_stays_not_ready( ) -> None: vehicle = _make_ble_vehicle() vehicle._default_timeout = 1.0 + vehicle._handshake_retry_interval = 0.01 domain = Domain.DOMAIN_VEHICLE_SECURITY attacker_key = ec.generate_private_key(ec.SECP256R1()) attacker_public_key = _public_key_bytes(attacker_key) @@ -745,3 +746,100 @@ def _extract_request_uuid(write_mock: AsyncMock) -> bytes: # RoutableMessage (see bluetooth.py's prependLength). msg = RoutableMessage.FromString(payload[2:]) return msg.uuid + + +# Captured live 2026-10-01 22:21:06Z: VCSEC's first session_info reply after a +# key-card-approved whitelist add carried session_info_tag with a zero-length +# tag (signature_data = 6a 02 32 00), so it can never authenticate. +CAPTURED_EMPTY_TAG_REPLY = bytes.fromhex( + "32121210b6897a37fea9067b25d7c387b9bcd58b3a0208026a0232007a5c1241" + "04ca71e4c8d4ee4bc18621cc8bbe33ec86d371a098a500c469510c02ac98530e" + "02951b4a83c80ffff1f139e5fa869a0367478d28c33e247d373d629f8ce65a8a" + "fe1a10e80d05f79f7277b1da527db0ce40dd3e25312e0000300e92031044e116" + "4ca45e6b61322b965995ca8073" +) + + +class EmptyTagHandshakeRetryTests(IsolatedAsyncioTestCase): + """A session_info reply that fails authentication is discarded and the + session-info request re-sent (vehicle-command's tryStartSession parity), + never trusted - and the handshake still fails after the last attempt.""" + + def test_captured_reply_has_present_but_empty_tag(self) -> None: + msg = RoutableMessage.FromString(CAPTURED_EMPTY_TAG_REPLY) + self.assertTrue(msg.signature_data.HasField("session_info_tag")) + self.assertEqual(msg.signature_data.session_info_tag.tag, b"") + self.assertEqual(msg.SerializeToString(), CAPTURED_EMPTY_TAG_REPLY) + commands = _make_commands(vin="LRW3F7EK4NC716336") + with self.assertRaises(SessionInfoAuthenticationFault) as ctx: + commands.validate_msg(msg, msg.request_uuid) + self.assertIn("missing its authentication tag", str(ctx.exception.data)) + self.assertFalse(commands._sessions[Domain.DOMAIN_VEHICLE_SECURITY].ready) + + def _empty_tag_reply(self, vehicle: VehicleBluetooth[Any]) -> RoutableMessage: + msg = RoutableMessage.FromString(CAPTURED_EMPTY_TAG_REPLY) + msg.to_destination.routing_address = vehicle._from_destination + msg.request_uuid = _extract_request_uuid(vehicle.client.write_gatt_char) + return msg + + async def test_handshake_retries_past_one_empty_tag_reply(self) -> None: + vehicle = _make_ble_vehicle() + vehicle._default_timeout = 1.0 + vehicle._handshake_retry_interval = 0.01 + domain = Domain.DOMAIN_VEHICLE_SECURITY + vehicle_public_key = _public_key_bytes(ec.generate_private_key(ec.SECP256R1())) + writes = 0 + + async def deliver(*_: Any) -> None: + nonlocal writes + writes += 1 + if writes == 1: + vehicle._on_message(self._empty_tag_reply(vehicle)) + return + session = cast("dict[Any, Any]", vehicle._sessions)[domain] + info = SessionInfo( + publicKey=vehicle_public_key, + epoch=b"\x0b" * 16, + clock_time=11825, + handle=14, + ).SerializeToString() + _, _, key = session.keys_for(vehicle_public_key) + sent = _extract_request_uuid(vehicle.client.write_gatt_char) + vehicle._on_message( + RoutableMessage( + to_destination=Destination( + routing_address=vehicle._from_destination + ), + from_destination=Destination(domain=domain), + session_info=info, + request_uuid=sent, + signature_data=SignatureData( + session_info_tag=HMAC_Signature_Data( + tag=_session_info_tag(key, vehicle.vin, sent, info) + ) + ), + ) + ) + + vehicle.client.write_gatt_char = AsyncMock(side_effect=deliver) + await vehicle.handshakeVehicleSecurity() + self.assertTrue(cast("dict[Any, Any]", vehicle._sessions)[domain].ready) + self.assertEqual(writes, 2) + + async def test_persistent_empty_tag_still_raises(self) -> None: + vehicle = _make_ble_vehicle() + vehicle._default_timeout = 1.0 + vehicle._handshake_retry_interval = 0.01 + + async def deliver(*_: Any) -> None: + vehicle._on_message(self._empty_tag_reply(vehicle)) + + vehicle.client.write_gatt_char = AsyncMock(side_effect=deliver) + with self.assertRaises(SessionInfoAuthenticationFault): + await vehicle.handshakeVehicleSecurity() + self.assertEqual(vehicle.client.write_gatt_char.await_count, 3) + self.assertFalse( + cast("dict[Any, Any]", vehicle._sessions)[ + Domain.DOMAIN_VEHICLE_SECURITY + ].ready + )