diff --git a/setup.py b/setup.py index dca0c13..f20dc4d 100644 --- a/setup.py +++ b/setup.py @@ -5,7 +5,7 @@ setuptools.setup( name="tesla_fleet_api", - version="0.8.1", + version="0.8.2", author="Brett Adams", author_email="hello@teslemetry.com", description="Tesla Fleet API library for Python", diff --git a/tesla_fleet_api/const.py b/tesla_fleet_api/const.py index 957d8b7..b7b652b 100644 --- a/tesla_fleet_api/const.py +++ b/tesla_fleet_api/const.py @@ -3,7 +3,7 @@ from enum import Enum import logging -VERSION = "0.8.1" +VERSION = "0.8.2" LOGGER = logging.getLogger(__package__) SERVERS = { "na": "https://fleet-api.prd.na.vn.cloud.tesla.com", diff --git a/tesla_fleet_api/teslafleetapi.py b/tesla_fleet_api/teslafleetapi.py index ad3c269..b877dc8 100644 --- a/tesla_fleet_api/teslafleetapi.py +++ b/tesla_fleet_api/teslafleetapi.py @@ -4,11 +4,11 @@ from typing import Any, Awaitable from os.path import exists import aiohttp +import aiofiles # cryptography -from cryptography.hazmat.primitives.ciphers.aead import AESGCM from cryptography.hazmat.primitives.asymmetric import ec -from cryptography.hazmat.primitives import hashes, serialization +from cryptography.hazmat.primitives import serialization from cryptography.hazmat.backends import default_backend from .exceptions import raise_for_status, InvalidRegion, LibraryError, ResponseError @@ -30,7 +30,7 @@ class TeslaFleetApi: session: aiohttp.ClientSession headers: dict[str, str] refresh_hook: Awaitable | None = None - _private_key: ec.EllipticCurvePrivateKey + _private_key: ec.EllipticCurvePrivateKey | None = None def __init__( self, @@ -161,8 +161,8 @@ async def products(self) -> dict[str, Any]: "api/1/products", ) - def private_key(self, path: str = "private_key.pem") -> ec.EllipticCurvePrivateKey: - """Create or load the private key.""" + async def get_private_key(self, path: str = "private_key.pem") -> ec.EllipticCurvePrivateKey: + """Get or create the private key.""" if not exists(path): self._private_key = ec.generate_private_key( ec.SECP256R1(), default_backend() @@ -173,16 +173,15 @@ def private_key(self, path: str = "private_key.pem") -> ec.EllipticCurvePrivateK format=serialization.PrivateFormat.TraditionalOpenSSL, encryption_algorithm=serialization.NoEncryption(), ) - with open(path, "wb") as key_file: - key_file.write(pem) - return self._private_key + async with aiofiles.open(path, "wb") as key_file: + await key_file.write(pem) else: try: - with open(path, "rb") as key_file: - key_data = key_file.read() - value = serialization.load_pem_private_key( - key_data, password=None, backend=default_backend() - ) + async with aiofiles.open(path, "rb") as key_file: + key_data = await key_file.read() + value = serialization.load_pem_private_key( + key_data, password=None, backend=default_backend() + ) except FileNotFoundError: raise FileNotFoundError(f"Private key file not found at {path}") except PermissionError: @@ -191,4 +190,4 @@ def private_key(self, path: str = "private_key.pem") -> ec.EllipticCurvePrivateK if not isinstance(value, ec.EllipticCurvePrivateKey): raise AssertionError("Loaded key is not an EllipticCurvePrivateKey") self._private_key = value - return self._private_key + return self._private_key diff --git a/tesla_fleet_api/vehiclesigned.py b/tesla_fleet_api/vehiclesigned.py index f423101..ec3fcde 100644 --- a/tesla_fleet_api/vehiclesigned.py +++ b/tesla_fleet_api/vehiclesigned.py @@ -1,6 +1,5 @@ from __future__ import annotations import base64 -from dataclasses import dataclass from random import randbytes from typing import Any, TYPE_CHECKING import time @@ -155,7 +154,7 @@ def tag( class VehicleSigned(VehicleSpecific): """Class describing the Tesla Fleet API vehicle endpoints and commands for a specific vehicle with command signing.""" - _key: ec.EllipticCurvePrivateKey + _private_key: ec.EllipticCurvePrivateKey _public_key: bytes _from_destination: bytes _sessions: dict[int, Session] @@ -165,13 +164,13 @@ def __init__( ): super().__init__(parent, vin) if key: - self._key = key + self._private_key = key elif parent._parent._private_key: - self._key = parent._parent._private_key + self._private_key = parent._parent._private_key else: raise ValueError("No private key.") - self._public_key = self._key.public_key().public_bytes( + self._public_key = self._private_key.public_key().public_bytes( encoding=Encoding.X962, format=PublicFormat.UncompressedPoint ) self._from_destination = randbytes(16) @@ -201,7 +200,7 @@ async def _handshake(self, domain: int) -> None: vehicle_public_key = info.publicKey # Derive shared key from private key _key and vehicle public key - shared = self._key.exchange( + shared = self._private_key.exchange( ec.ECDH(), ec.EllipticCurvePublicKey.from_encoded_point( ec.SECP256R1(), vehicle_public_key @@ -215,8 +214,6 @@ async def _handshake(self, domain: int) -> None: delta=int(time.time()) - info.clock_time, ) - print(self._sessions[domain]) - async def _sendVehicleSecurity(self, command: UnsignedMessage) -> dict[str, Any]: """Sign and send a message to Infotainment computer.""" if DOMAIN_VEHICLE_SECURITY not in self._sessions: