From 86a7feb78b70fe48bd5d5c191660bc26a86eb035 Mon Sep 17 00:00:00 2001 From: firstmate crewmate Date: Thu, 23 Jul 2026 07:23:00 +1000 Subject: [PATCH] feat(vehicle): add niche low-certainty-value signed commands Wraps the remaining five VehicleAction fields with no known third-party consumer use case today: pii_key_request, pseudonym_sync_request, tesla_auth_response, setup_cloud_profile_with_local_profile_uuid, and get_local_profiles_for_vault_uuid. Included per the captain's full-proto-coverage directive; low-certainty value, easy to drop if reconsidered later since each is purely additive. --- tesla_fleet_api/tesla/vehicle/commands.py | 111 ++++++++++++++++++++++ tests/test_ble_niche_commands.py | 106 +++++++++++++++++++++ 2 files changed, 217 insertions(+) create mode 100644 tests/test_ble_niche_commands.py diff --git a/tesla_fleet_api/tesla/vehicle/commands.py b/tesla_fleet_api/tesla/vehicle/commands.py index 6f32aaa..390a0cf 100644 --- a/tesla_fleet_api/tesla/vehicle/commands.py +++ b/tesla_fleet_api/tesla/vehicle/commands.py @@ -174,7 +174,14 @@ # Group 12: Power modes SetLowPowerModeAction, SetKeepAccessoryPowerModeAction, + # Group 18: Niche/low-certainty-value commands (captain full-coverage directive) + PiiKeyRequest, + PseudonymSyncRequest, + TeslaAuthResponseAction, + SetupCloudProfileWithLocalProfileUuidAction, + GetLocalProfilesForVaultUuidAction, ) +from google.protobuf.timestamp_pb2 import Timestamp from tesla_protocol.command.vehicle_pb2 import ( VehicleData, VehicleState, @@ -2404,3 +2411,107 @@ async def auto_secure_vehicle(self) -> dict[str, Any]: return await self._sendVehicleSecurity( UnsignedMessage(RKEAction=RKEAction_E.RKE_ACTION_AUTO_SECURE_VEHICLE) ) + + # Group 18: Niche/low-certainty-value commands + # + # Included per the captain's full-proto-coverage directive; each has no + # known third-party consumer use case today, unlike every other command + # in this file. See AGENTS.md for detail on why each is niche. + + async def pii_key_request( + self, subscriber_public_key: str, pii_key_expiration: int + ) -> dict[str, Any]: + """Requests a PII (privacy/telemetry pseudonymization) key. + + ``pii_key_expiration`` is a Unix timestamp in seconds. + """ + return await self._sendInfotainment( + Action( + vehicleAction=VehicleAction( + piiKeyRequest=PiiKeyRequest( + subscriber_public_key=subscriber_public_key, + pii_key_expiration=Timestamp(seconds=pii_key_expiration), + ) + ) + ) + ) + + async def pseudonym_sync_request( + self, last_known_pseudonym_hashed: bytes + ) -> dict[str, Any]: + """Syncs the vehicle's privacy/telemetry pseudonym with a previously known hashed value.""" + return await self._sendInfotainment( + Action( + vehicleAction=VehicleAction( + pseudonymSyncRequest=PseudonymSyncRequest( + last_known_pseudonym_hashed=last_known_pseudonym_hashed + ) + ) + ) + ) + + async def tesla_auth_response( + self, + client_id: str, + scope: str, + access_token: str, + refresh_token: str, + expiry_timestamp: int, + error: str = "", + scoped_token: str = "", + ) -> dict[str, Any]: + """Passes a completed "Sign in with Tesla" OAuth response to the vehicle screen. + + This is a pass-through: the caller must already have a completed OAuth + response from elsewhere in their own auth flow; the library cannot + originate these fields itself. + """ + return await self._sendInfotainment( + Action( + vehicleAction=VehicleAction( + teslaAuthResponseAction=TeslaAuthResponseAction( + client_id=client_id, + scope=scope, + access_token=access_token, + refresh_token=refresh_token, + expiry_timestamp=expiry_timestamp, + error=error, + scoped_token=scoped_token, + ) + ) + ) + ) + + async def setup_cloud_profile_with_local_profile_uuid( + self, + cloud_vault_uuid: str, + local_profile_uuid: str, + delete_local_profile_after_setup: bool = False, + ) -> dict[str, Any]: + """Links a local Tesla profile to a cloud vault profile UUID.""" + return await self._sendInfotainment( + Action( + vehicleAction=VehicleAction( + setupCloudProfileWithLocalProfileUuidAction=SetupCloudProfileWithLocalProfileUuidAction( + cloud_vault_uuid=cloud_vault_uuid, + local_profile_uuid=local_profile_uuid, + delete_local_profile_after_setup=delete_local_profile_after_setup, + ) + ) + ) + ) + + async def get_local_profiles_for_vault_uuid( + self, vault_uuid: str + ) -> dict[str, Any]: + """Gets local Tesla profiles associated with a cloud vault UUID.""" + return await self._sendInfotainment( + Action( + vehicleAction=VehicleAction( + getLocalProfilesForVaultUuidAction=GetLocalProfilesForVaultUuidAction( + vault_uuid=vault_uuid + ) + ) + ), + mutating=False, + ) diff --git a/tests/test_ble_niche_commands.py b/tests/test_ble_niche_commands.py new file mode 100644 index 0000000..7d0be87 --- /dev/null +++ b/tests/test_ble_niche_commands.py @@ -0,0 +1,106 @@ +"""Niche/low-certainty-value command group over the mocked BLE transport. + +pii_key_request, pseudonym_sync_request, tesla_auth_response, +setup_cloud_profile_with_local_profile_uuid, and +get_local_profiles_for_vault_uuid - included per the captain's +full-proto-coverage directive despite no known third-party consumer use +case (see commands.py's Group 18 docstring). +""" + +from tesla_protocol.command.car_server_pb2 import Action +from tesla_protocol.command.universal_message_pb2 import Domain + +from ble_mocked_transport import ( + MockedBleTransportTestCase, + decrypt_sent_command, + infotainment_action_ok_reply, +) + + +def _decode_vehicle_action(vehicle, sent_msg): + plaintext = decrypt_sent_command(vehicle, sent_msg) + action = Action.FromString(plaintext) + assert sent_msg.to_destination.domain == Domain.DOMAIN_INFOTAINMENT + return action.vehicleAction + + +class PiiKeyRequestTests(MockedBleTransportTestCase): + async def test_sends_public_key_and_expiration(self) -> None: + vehicle, send = self.make_vehicle() + send.return_value = infotainment_action_ok_reply() + + await vehicle.pii_key_request("subscriber-pubkey", 1893456000) + + vehicle_action = _decode_vehicle_action(vehicle, send.await_args.args[0]) + req = vehicle_action.piiKeyRequest + self.assertEqual(req.subscriber_public_key, "subscriber-pubkey") + self.assertEqual(req.pii_key_expiration.seconds, 1893456000) + + +class PseudonymSyncRequestTests(MockedBleTransportTestCase): + async def test_sends_last_known_pseudonym_hashed(self) -> None: + vehicle, send = self.make_vehicle() + send.return_value = infotainment_action_ok_reply() + + await vehicle.pseudonym_sync_request(b"\xaa\xbb\xcc") + + vehicle_action = _decode_vehicle_action(vehicle, send.await_args.args[0]) + self.assertEqual( + vehicle_action.pseudonymSyncRequest.last_known_pseudonym_hashed, + b"\xaa\xbb\xcc", + ) + + +class TeslaAuthResponseTests(MockedBleTransportTestCase): + async def test_sends_full_oauth_response(self) -> None: + vehicle, send = self.make_vehicle() + send.return_value = infotainment_action_ok_reply() + + await vehicle.tesla_auth_response( + client_id="my-client", + scope="openid", + access_token="access-tok", + refresh_token="refresh-tok", + expiry_timestamp=1893456000, + scoped_token="scoped-tok", + ) + + vehicle_action = _decode_vehicle_action(vehicle, send.await_args.args[0]) + action = vehicle_action.teslaAuthResponseAction + self.assertEqual(action.client_id, "my-client") + self.assertEqual(action.scope, "openid") + self.assertEqual(action.access_token, "access-tok") + self.assertEqual(action.refresh_token, "refresh-tok") + self.assertEqual(action.expiry_timestamp, 1893456000) + self.assertEqual(action.error, "") + self.assertEqual(action.scoped_token, "scoped-tok") + + +class SetupCloudProfileWithLocalProfileUuidTests(MockedBleTransportTestCase): + async def test_sends_uuids_and_delete_flag(self) -> None: + vehicle, send = self.make_vehicle() + send.return_value = infotainment_action_ok_reply() + + await vehicle.setup_cloud_profile_with_local_profile_uuid( + "cloud-uuid", "local-uuid", delete_local_profile_after_setup=True + ) + + vehicle_action = _decode_vehicle_action(vehicle, send.await_args.args[0]) + action = vehicle_action.setupCloudProfileWithLocalProfileUuidAction + self.assertEqual(action.cloud_vault_uuid, "cloud-uuid") + self.assertEqual(action.local_profile_uuid, "local-uuid") + self.assertTrue(action.delete_local_profile_after_setup) + + +class GetLocalProfilesForVaultUuidTests(MockedBleTransportTestCase): + async def test_sends_vault_uuid(self) -> None: + vehicle, send = self.make_vehicle() + send.return_value = infotainment_action_ok_reply() + + await vehicle.get_local_profiles_for_vault_uuid("vault-uuid") + + vehicle_action = _decode_vehicle_action(vehicle, send.await_args.args[0]) + self.assertEqual( + vehicle_action.getLocalProfilesForVaultUuidAction.vault_uuid, + "vault-uuid", + )