Working Bluetooth

Author: Bre77

tesla_fleet_api/tesla/bluetooth.py

@@ -1,5 +1,6 @@
 """Bluetooth only interface."""
 
+import re
 from tesla_fleet_api.tesla.tesla import Tesla
 from tesla_fleet_api.tesla.vehicle.bluetooth import VehicleBluetooth
 
@@ -13,6 +14,10 @@ def __init__(
 
         self.vehicles = Vehicles(self)
 
+    def valid_name(self, name: str) -> bool:
+        """Check if a BLE device name is a valid Tesla vehicle."""
+        return bool(re.match("^S[a-f0-9]{16}[A-F]$", name))
+
 class Vehicles(dict[str, VehicleBluetooth]):
     """Class containing and creating vehicles."""
 

tesla_fleet_api/tesla/vehicle/bluetooth.py

@@ -1,11 +1,8 @@
 from __future__ import annotations
 
 import hashlib
-import hmac
-import struct
-from asyncio import Future, sleep, get_running_loop
-from random import randbytes
-from typing import TYPE_CHECKING, Any, List
+from asyncio import Future, get_running_loop
+from typing import TYPE_CHECKING
 from google.protobuf.message import DecodeError
 
 from bleak import BleakClient, BleakScanner
@@ -22,50 +19,29 @@
 )
 from ...exceptions import (
     MESSAGE_FAULTS,
-    SIGNED_MESSAGE_INFORMATION_FAULTS,
     WHITELIST_OPERATION_STATUS,
-    TeslaFleetMessageFaultIncorrectEpoch,
-    TeslaFleetMessageFaultInvalidTokenOrCounter,
 )
 
 # Protocol
-from .proto.errors_pb2 import GenericError_E
 from .proto.car_server_pb2 import (
     Response,
 )
 from .proto.signatures_pb2 import (
-    SIGNATURE_TYPE_HMAC_PERSONALIZED,
-    TAG_COUNTER,
-    TAG_DOMAIN,
-    TAG_END,
-    TAG_EPOCH,
-    TAG_EXPIRES_AT,
-    TAG_PERSONALIZATION,
-    TAG_SIGNATURE_TYPE,
     SessionInfo,
 )
 from .proto.universal_message_pb2 import (
-    DOMAIN_INFOTAINMENT,
-    DOMAIN_VEHICLE_SECURITY,
-    OPERATIONSTATUS_ERROR,
-    OPERATIONSTATUS_WAIT,
     Destination,
     Domain,
     RoutableMessage,
 )
 from .proto.vcsec_pb2 import (
-    OPERATIONSTATUS_OK,
     FromVCSECMessage,
     KeyFormFactor,
     KeyMetadata,
     PermissionChange,
     PublicKey,
-    SignatureType,
-    SignedMessage,
-    ToVCSECMessage,
     UnsignedMessage,
     WhitelistOperation,
-    WhitelistOperation_information_E,
 
 )
 
@@ -84,9 +60,9 @@ def prependLength(message: bytes) -> bytearray:
 class VehicleBluetooth(Commands):
     """Class describing the Tesla Fleet API vehicle endpoints and commands for a specific vehicle with command signing."""
 
-    _ble_name: str
+    ble_name: str
+    client: BleakClient
     _device: BLEDevice
-    _client: BleakClient
     _futures: dict[Domain, Future]
     _ekey: ec.EllipticCurvePublicKey
     _recv: bytearray = bytearray()
@@ -97,24 +73,30 @@ def __init__(
         self, parent: Tesla, vin: str, key: ec.EllipticCurvePrivateKey | None = None
     ):
         super().__init__(parent, vin, key)
-        self._ble_name = "S" + hashlib.sha1(vin.encode('utf-8')).hexdigest()[:16] + "C"
+        self.ble_name = "S" + hashlib.sha1(vin.encode('utf-8')).hexdigest()[:16] + "C"
         self._futures = {}
 
     async def discover(self, scanner: BleakScanner = BleakScanner()) -> BleakClient:
         """Find the Tesla BLE device."""
 
-        device = await scanner.find_device_by_name(self._ble_name)
+        device = await scanner.find_device_by_name(self.ble_name)
         if not device:
-            raise ValueError(f"Device {self._ble_name} not found")
+            raise ValueError(f"Device {self.ble_name} not found")
         self._device = device
-        self._client = BleakClient(self._device)
+        self.client = BleakClient(self._device, services=[SERVICE_UUID])
         LOGGER.info(f"Discovered device {self._device.name} {self._device.address}")
-        return self._client
+        return self.client
 
-    async def connect(self):
+    async def connect(self, mac:str | None = None) -> None:
         """Connect to the Tesla BLE device."""
-        await self._client.connect()
-        await self._client.start_notify(READ_UUID, self._on_notify)
+        if mac is not None:
+            self.client = BleakClient(mac, services=[SERVICE_UUID])
+        await self.client.connect()
+        await self.client.start_notify(READ_UUID, self._on_notify)
+
+    async def disconnect(self) -> bool:
+        """Disconnect from the Tesla BLE device."""
+        return await self.client.disconnect()
 
     def _on_notify(self,sender: BleakGATTCharacteristic,data : bytearray):
         """Receive data from the Tesla BLE device."""
@@ -126,19 +108,19 @@ def _on_notify(self,sender: BleakGATTCharacteristic,data : bytearray):
         LOGGER.debug(f"Received {len(self._recv)} of {self._recv_len} bytes")
         while len(self._recv) > self._recv_len:
             LOGGER.warn(f"Received more data than expected: {len(self._recv)} > {self._recv_len}")
-            self._on_message(self._recv[:self._recv_len])
+            self._on_message(bytes(self._recv[:self._recv_len]))
             self._recv_len = int.from_bytes(self._recv[self._recv_len:self._recv_len+2], 'big')
             self._recv = self._recv[self._recv_len+2:]
             continue
         if len(self._recv) == self._recv_len:
-            self._on_message(self._recv)
+            self._on_message(bytes(self._recv))
             self._recv = bytearray()
             self._recv_len = 0
 
-    def _on_message(self, data:bytearray):
+    def _on_message(self, data:bytes):
         """Receive messages from the Tesla BLE data."""
         try:
-            msg = RoutableMessage.FromString(bytes(data))
+            msg = RoutableMessage.FromString(data)
         except DecodeError as e:
             LOGGER.error(f"Error parsing message: {e}")
             return
@@ -165,12 +147,6 @@ def _on_message(self, data:bytearray):
             self._futures[msg.from_destination.domain].set_result(msg)
             return
 
-
-    async def disconnect(self):
-        """Disconnect from the Tesla BLE device."""
-        await self._client.stop_notify(READ_UUID)
-        await self._client.disconnect()
-
     async def _create_future(self, domain: Domain) -> Future:
         if(not self._sessions[domain].lock.locked):
             raise ValueError("Session is not locked")
@@ -185,7 +161,9 @@ async def _send(self, msg: RoutableMessage) -> RoutableMessage:
             future = await self._create_future(domain)
             payload = prependLength(msg.SerializeToString())
             LOGGER.info(f"Payload: {payload}")
-            await self._client.write_gatt_char(WRITE_UUID, payload, False)
+
+            await self.client.write_gatt_char(WRITE_UUID, payload, True)
+
             resp = await future
             LOGGER.info(f"Received message {resp}")
 
@@ -194,7 +172,7 @@ async def _send(self, msg: RoutableMessage) -> RoutableMessage:
 
             return resp
 
-    async def pair(self, role: Role = Role.ROLE_DRIVER, form: KeyFormFactor = KeyFormFactor.KEY_FORM_FACTOR_ANDROID_DEVICE):
+    async def pair(self, role: Role = Role.ROLE_OWNER, form: KeyFormFactor = KeyFormFactor.KEY_FORM_FACTOR_CLOUD_KEY):
         """Pair the key."""
 
         request = UnsignedMessage(

tesla_fleet_api/tesla/vehicle/commands.py

@@ -14,7 +14,6 @@
 from asyncio import Lock, sleep
 
 from ...exceptions import (
-    MESSAGE_FAULTS,
     SIGNED_MESSAGE_INFORMATION_FAULTS,
     TeslaFleetMessageFaultIncorrectEpoch,
     TeslaFleetMessageFaultInvalidTokenOrCounter,
@@ -38,7 +37,6 @@
     Response,
 )
 from .proto.signatures_pb2 import (
-    SIGNATURE_TYPE_AES_GCM,
     SIGNATURE_TYPE_AES_GCM_PERSONALIZED,
     SIGNATURE_TYPE_HMAC_PERSONALIZED,
     TAG_COUNTER,
@@ -49,24 +47,24 @@
     TAG_PERSONALIZATION,
     TAG_SIGNATURE_TYPE,
     AES_GCM_Personalized_Signature_Data,
+    KeyIdentity,
     SessionInfo,
+    SignatureData,
 )
 from .proto.universal_message_pb2 import (
     DOMAIN_INFOTAINMENT,
     DOMAIN_VEHICLE_SECURITY,
     OPERATIONSTATUS_ERROR,
     OPERATIONSTATUS_WAIT,
+    Destination,
     Domain,
     RoutableMessage,
+    SessionInfoRequest,
 )
 from .proto.vcsec_pb2 import (
     OPERATIONSTATUS_OK,
     FromVCSECMessage,
 )
-from .proto.universal_message_pb2 import (
-    Domain,
-    RoutableMessage,
-)
 from .proto.car_server_pb2 import (
     Action,
     MediaPlayAction,
@@ -98,7 +96,7 @@
     VehicleControlWindowAction,
     HvacBioweaponModeAction,
     AutoSeatClimateAction,
-    # Ping,
+    Ping,
     ScheduledChargingAction,
     ScheduledDepartureAction,
     HvacClimateKeeperAction,
@@ -122,7 +120,6 @@
     ClosureMoveType_E,
 )
 from .proto.signatures_pb2 import (
-    SessionInfo,
     HMAC_Personalized_Signature_Data,
 )
 from .proto.common_pb2 import (
@@ -200,7 +197,7 @@ def update(self, sessionInfo: SessionInfo):
 
     def hmac_personalized(self) -> HMAC_Personalized_Signature_Data:
         """Sign a command and return session metadata"""
-        self.counter = self.counter+1
+        self.counter += 1
         return HMAC_Personalized_Signature_Data(
             epoch=self.epoch,
             counter=self.counter,
@@ -209,7 +206,7 @@ def hmac_personalized(self) -> HMAC_Personalized_Signature_Data:
 
     def aes_gcm_personalized(self) -> AES_GCM_Personalized_Signature_Data:
         """Sign a command and return session metadata"""
-        self.counter = self.counter+1
+        self.counter += 1
         return AES_GCM_Personalized_Signature_Data(
             epoch=self.epoch,
             nonce=randbytes(12),
@@ -381,15 +378,22 @@ async def _commandHmac(self, session: Session, command: bytes, attempt: int = 1)
             session.hmac, metadata + command, hashlib.sha256
         ).digest()
 
-        msg = RoutableMessage()
-        msg.to_destination.domain = session.domain
-        msg.from_destination.routing_address = self._from_destination
-        msg.protobuf_message_as_bytes = command
-        msg.uuid = randbytes(16)
-        msg.signature_data.HMAC_Personalized_data.CopyFrom(hmac_personalized)
-        msg.signature_data.signer_identity.public_key = self._public_key
-
-        return msg
+        return RoutableMessage(
+            to_destination=Destination(
+                domain=session.domain,
+            ),
+            from_destination=Destination(
+                routing_address=self._from_destination
+            ),
+            protobuf_message_as_bytes=command,
+            uuid=randbytes(16),
+            signature_data=SignatureData(
+                signer_identity=KeyIdentity(
+                    public_key=self._public_key
+                ),
+                HMAC_Personalized_data=hmac_personalized,
+            )
+        )
 
     async def _commandAes(self, session: Session, command: bytes, attempt: int = 1) -> RoutableMessage:
         """Create an encrypted message."""
@@ -428,15 +432,22 @@ async def _commandAes(self, session: Session, command: bytes, attempt: int = 1)
         aes_personalized.tag = ct[-16:]
 
         # I think this whole section could be improved
-        msg = RoutableMessage()
-        msg.to_destination.domain = session.domain
-        msg.from_destination.routing_address = self._from_destination
-        msg.protobuf_message_as_bytes = ct[:-16]
-        msg.uuid = randbytes(16)
-        msg.signature_data.AES_GCM_Personalized_data.CopyFrom(aes_personalized)
-        msg.signature_data.signer_identity.public_key = self._public_key
-
-        return msg
+        return RoutableMessage(
+            to_destination=Destination(
+                domain=session.domain,
+            ),
+            from_destination=Destination(
+                routing_address=self._from_destination
+            ),
+            protobuf_message_as_bytes=ct[:-16],
+            uuid=randbytes(16),
+            signature_data=SignatureData(
+                signer_identity=KeyIdentity(
+                    public_key=self._public_key
+                ),
+                AES_GCM_Personalized_data=aes_personalized,
+            )
+        )
 
 
     async def _sendVehicleSecurity(self, command: UnsignedMessage) -> dict[str, Any]:
@@ -448,33 +459,31 @@ async def _sendInfotainment(self, command: Action) -> dict[str, Any]:
         return await self._command(Domain.DOMAIN_INFOTAINMENT, command.SerializeToString())
 
 
-    async def _handle_command(self, domain: Domain, command: str, attempt: int, retry) -> dict[str, Any]:
-        """Handle a command."""
-        if attempt > MAX_RETRIES:
-            return {"response": {"result": False, "reason": "Too many retries"}}
-        async with session.lock:
-            await sleep(2)
-        return await self._commandAes(domain, command, attempt)
-
     async def _handshake(self, domain: Domain) -> None:
         """Perform a handshake with the vehicle."""
 
         LOGGER.debug(f"Handshake with domain {Domain.Name(domain)}")
-        msg = RoutableMessage()
-        msg.to_destination.domain = domain
-        msg.from_destination.routing_address = self._from_destination
-        msg.session_info_request.public_key = self._public_key
-        msg.uuid = randbytes(16)
+        msg = RoutableMessage(
+            to_destination=Destination(
+                domain=domain,
+            ),
+            from_destination=Destination(
+                routing_address=self._from_destination
+            ),
+            session_info_request=SessionInfoRequest(
+                public_key=self._public_key
+            ),
+            uuid=randbytes(16)
+        )
 
-        # Send handshake message
         await self._send(msg)
 
     async def ping(self) -> dict[str, Any]:
         """Ping the vehicle."""
         return await self._sendInfotainment(
             Action(
                 vehicleAction=VehicleAction(
-                    ping=None
+                    ping=Ping(ping_id=0)
                 )
             )
         )