From d399b718533b8b05bb002d8c5e84f10b2dbf6630 Mon Sep 17 00:00:00 2001 From: firstmate crewmate Date: Sun, 12 Jul 2026 20:23:01 +1000 Subject: [PATCH 1/7] fix(tesla): create private key PEM files atomically at mode 0600 get_private_key and get_rsa_private_key wrote the PEM with the process default mode then chmod'd to 0o600 afterward, leaving a brief world-readable window on every fresh key and silently leaving the key 0644 forever if the chmod failed. Both now open with O_EXCL at 0o600 via a custom aiofiles opener, so the file is born owner-only with no window and no swallowed failure. O_EXCL also closes the TOCTOU race between the exists() check and the write - a caller that loses the create race falls back to reading the winner's file instead of raising. --- tesla_fleet_api/tesla/tesla.py | 90 ++++++++++++++++-------------- tests/test_tesla_private_key.py | 98 ++++++++++++++++++++++++++++++--- 2 files changed, 140 insertions(+), 48 deletions(-) diff --git a/tesla_fleet_api/tesla/tesla.py b/tesla_fleet_api/tesla/tesla.py index ca1a53c..3787ffe 100644 --- a/tesla_fleet_api/tesla/tesla.py +++ b/tesla_fleet_api/tesla/tesla.py @@ -1,6 +1,7 @@ """Tesla Fleet API for Python.""" import base64 +import os from os.path import exists import aiofiles @@ -16,6 +17,11 @@ from cryptography.hazmat.backends import default_backend +def _owner_only_opener(file: str, flags: int) -> int: + """Open a new file exclusively, born at mode 0o600 with no chmod window.""" + return os.open(file, os.O_WRONLY | os.O_CREAT | os.O_EXCL, 0o600) + + class Tesla: """Base class describing interactions with Tesla products.""" @@ -33,42 +39,42 @@ async def get_private_key( ) -> ec.EllipticCurvePrivateKey: """Get or create the private key. - The private key is stored as an unencrypted PEM file with permissions - 0o600 when created. + A newly created key file is opened with O_EXCL so it is born at mode + 0o600 with no world-readable window. If another process wins the + create race, its file is read instead of raising. """ if not exists(path): self.private_key = ec.generate_private_key( ec.SECP256R1(), default_backend() ) - # save the key pem = self.private_key.private_bytes( encoding=serialization.Encoding.PEM, format=serialization.PrivateFormat.TraditionalOpenSSL, encryption_algorithm=serialization.NoEncryption(), ) - async with aiofiles.open(path, "wb") as key_file: - await key_file.write(pem) try: - from os import chmod - - chmod(path, 0o600) - except OSError: + async with aiofiles.open( + path, "wb", opener=_owner_only_opener + ) as key_file: + await key_file.write(pem) + return self.private_key + except FileExistsError: pass - else: - try: - 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: - raise PermissionError(f"Permission denied when trying to read {path}") - - if not isinstance(value, ec.EllipticCurvePrivateKey): - raise AssertionError("Loaded key is not an EllipticCurvePrivateKey") - self.private_key = value + + try: + 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: + raise PermissionError(f"Permission denied when trying to read {path}") + + if not isinstance(value, ec.EllipticCurvePrivateKey): + raise AssertionError("Loaded key is not an EllipticCurvePrivateKey") + self.private_key = value return self.private_key @property @@ -110,8 +116,10 @@ async def get_rsa_private_key( """Get or create an RSA private key for energy gateway client registration. The default 4096-bit key matches the format expected by the Powerwall - TEDapi v1r LAN protocol. The private key is stored as an unencrypted - PEM file with permissions 0o600 when created. + TEDapi v1r LAN protocol. A newly created key file is opened with + O_EXCL so it is born at mode 0o600 with no world-readable window. If + another process wins the create race, its file is read instead of + raising. """ if not exists(path): self.rsa_private_key = rsa.generate_private_key( @@ -124,23 +132,23 @@ async def get_rsa_private_key( format=serialization.PrivateFormat.TraditionalOpenSSL, encryption_algorithm=serialization.NoEncryption(), ) - async with aiofiles.open(path, "wb") as key_file: - await key_file.write(pem) try: - from os import chmod - - chmod(path, 0o600) - except OSError: + async with aiofiles.open( + path, "wb", opener=_owner_only_opener + ) as key_file: + await key_file.write(pem) + return self.rsa_private_key + except FileExistsError: pass - else: - 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() - ) - if not isinstance(value, rsa.RSAPrivateKey): - raise AssertionError("Loaded key is not an RSAPrivateKey") - self.rsa_private_key = value + + 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() + ) + if not isinstance(value, rsa.RSAPrivateKey): + raise AssertionError("Loaded key is not an RSAPrivateKey") + self.rsa_private_key = value return self.rsa_private_key @property diff --git a/tests/test_tesla_private_key.py b/tests/test_tesla_private_key.py index 487ada5..80d185e 100644 --- a/tests/test_tesla_private_key.py +++ b/tests/test_tesla_private_key.py @@ -1,10 +1,10 @@ -"""``get_private_key`` must write its PEM file with owner-only permissions. +"""``get_private_key``/``get_rsa_private_key`` must create their PEM files atomically. -The key it creates signs Tesla vehicle commands (Fleet API and BLE) - written -with the process default mode (typically group/world-readable), it leaks -command-signing material to any other local user. ``get_rsa_private_key`` -(the TEDapi/Powerwall key created later in the same module) already chmods to -0o600 after writing; this locks ``get_private_key`` into the same contract. +The keys sign Tesla vehicle commands (Fleet API and BLE) and register with the +Powerwall TEDapi gateway. Both are created via an O_EXCL-opened file at mode +0o600 so there is never a write-then-chmod window where the key is +world-readable, and a losing concurrent creator falls back to reading the +winner's file instead of raising. """ from __future__ import annotations @@ -12,11 +12,30 @@ import stat import tempfile from pathlib import Path -from unittest import IsolatedAsyncioTestCase +from unittest import IsolatedAsyncioTestCase, mock + +from cryptography.hazmat.primitives import serialization +from cryptography.hazmat.primitives.asymmetric import ec, rsa from tesla_fleet_api.tesla.tesla import Tesla +def _ec_pem(key: ec.EllipticCurvePrivateKey) -> bytes: + return key.private_bytes( + encoding=serialization.Encoding.PEM, + format=serialization.PrivateFormat.TraditionalOpenSSL, + encryption_algorithm=serialization.NoEncryption(), + ) + + +def _rsa_pem(key: rsa.RSAPrivateKey) -> bytes: + return key.private_bytes( + encoding=serialization.Encoding.PEM, + format=serialization.PrivateFormat.TraditionalOpenSSL, + encryption_algorithm=serialization.NoEncryption(), + ) + + class GetPrivateKeyPermissionsTests(IsolatedAsyncioTestCase): async def test_new_key_file_is_owner_only(self) -> None: with tempfile.TemporaryDirectory() as tmp_dir: @@ -27,3 +46,68 @@ async def test_new_key_file_is_owner_only(self) -> None: mode = stat.S_IMODE(Path(path).stat().st_mode) self.assertEqual(mode, 0o600) + + async def test_existing_key_read_path_unchanged(self) -> None: + with tempfile.TemporaryDirectory() as tmp_dir: + path = str(Path(tmp_dir) / "private_key.pem") + creator = Tesla() + created = await creator.get_private_key(path) + + reader = Tesla() + read_back = await reader.get_private_key(path) + + self.assertEqual(_ec_pem(read_back), _ec_pem(created)) + mode = stat.S_IMODE(Path(path).stat().st_mode) + self.assertEqual(mode, 0o600) + + async def test_concurrent_create_falls_back_to_read(self) -> None: + with tempfile.TemporaryDirectory() as tmp_dir: + path = str(Path(tmp_dir) / "private_key.pem") + winner = Tesla() + winner_key = await winner.get_private_key(path) + + # Simulate a caller whose exists() check ran before the winner's + # create - it must fall back to reading the winner's file rather + # than raising FileExistsError. + loser = Tesla() + with mock.patch("tesla_fleet_api.tesla.tesla.exists", return_value=False): + key = await loser.get_private_key(path) + + self.assertEqual(_ec_pem(key), _ec_pem(winner_key)) + + +class GetRsaPrivateKeyPermissionsTests(IsolatedAsyncioTestCase): + async def test_new_key_file_is_owner_only(self) -> None: + with tempfile.TemporaryDirectory() as tmp_dir: + path = str(Path(tmp_dir) / "tedapi_rsa_private.pem") + tesla = Tesla() + + await tesla.get_rsa_private_key(path, key_size=1024) + + mode = stat.S_IMODE(Path(path).stat().st_mode) + self.assertEqual(mode, 0o600) + + async def test_existing_key_read_path_unchanged(self) -> None: + with tempfile.TemporaryDirectory() as tmp_dir: + path = str(Path(tmp_dir) / "tedapi_rsa_private.pem") + creator = Tesla() + created = await creator.get_rsa_private_key(path, key_size=1024) + + reader = Tesla() + read_back = await reader.get_rsa_private_key(path, key_size=1024) + + self.assertEqual(_rsa_pem(read_back), _rsa_pem(created)) + mode = stat.S_IMODE(Path(path).stat().st_mode) + self.assertEqual(mode, 0o600) + + async def test_concurrent_create_falls_back_to_read(self) -> None: + with tempfile.TemporaryDirectory() as tmp_dir: + path = str(Path(tmp_dir) / "tedapi_rsa_private.pem") + winner = Tesla() + winner_key = await winner.get_rsa_private_key(path, key_size=1024) + + loser = Tesla() + with mock.patch("tesla_fleet_api.tesla.tesla.exists", return_value=False): + key = await loser.get_rsa_private_key(path, key_size=1024) + + self.assertEqual(_rsa_pem(key), _rsa_pem(winner_key)) From 514ca9967819c50f93bde6a5dd0d025303753777 Mon Sep 17 00:00:00 2001 From: firstmate crewmate Date: Sun, 12 Jul 2026 20:27:44 +1000 Subject: [PATCH 2/7] no-mistakes(review): Force key permissions after atomic create --- tesla_fleet_api.egg-info/PKG-INFO | 24 +++++++++++++----------- tesla_fleet_api.egg-info/SOURCES.txt | 5 +++++ tesla_fleet_api/tesla/tesla.py | 8 +++++++- tests/test_tesla_private_key.py | 27 +++++++++++++++++++++++++++ 4 files changed, 52 insertions(+), 12 deletions(-) diff --git a/tesla_fleet_api.egg-info/PKG-INFO b/tesla_fleet_api.egg-info/PKG-INFO index 679b352..8e97d9c 100644 --- a/tesla_fleet_api.egg-info/PKG-INFO +++ b/tesla_fleet_api.egg-info/PKG-INFO @@ -1,6 +1,6 @@ Metadata-Version: 2.4 Name: tesla_fleet_api -Version: 1.6.4 +Version: 1.7.1 Summary: Tesla Fleet API library for Python Author-email: Brett Adams License-Expression: Apache-2.0 @@ -195,15 +195,17 @@ default with a passive GATT read about every 20 seconds. Pass leaving it enabled can keep an already-awake car awake longer, so disconnect or disable keepalive when vehicle sleep is preferred. -BLE connect/notify and GATT write failures from `VehicleBluetooth` raise +BLE connect/notify failures, and GATT writes rejected before backend I/O, raise `BluetoothTransportError`, a `TeslaFleetError` subclass, with the original -transport exception chained as `__cause__`. By default, a response-wait timeout -from a mutating BLE command raises `BluetoothUnconfirmedCommand`, a -`BluetoothTimeout` subclass, because the vehicle may have executed it despite a -lost acknowledgement. The BLE-only `optimistic` and `raise_unconfirmed` -factory options can instead resolve some ambiguous timeouts as best-effort -successes; see [Bluetooth for Vehicles](docs/bluetooth_vehicles.md) for the -full confirmation ladder. Catch `TeslaFleetError` to handle Bluetooth transport +transport exception chained as `__cause__` when available. A GATT write that +entered backend I/O and then failed or timed out is delivery-ambiguous and +raises `BluetoothTimeout`/`BluetoothUnconfirmedCommand` instead. Mutating BLE +commands use a confirmation ladder controlled by `confirmation` (`"ack"` by +default) and `raise_unconfirmed` (`False` by default): an inconclusive lost +acknowledgement resolves as best-effort success unless you opt in to +`BluetoothUnconfirmedCommand`, while a command proven not to have applied raises +`BluetoothCommandFailed`. See [Bluetooth for Vehicles](docs/bluetooth_vehicles.md) +for the full ladder. Catch `TeslaFleetError` to handle Bluetooth transport failures (including `bleak.exc.BleakError` and builtin `TimeoutError` from ESPHome proxies) and response-wait timeouts through the same library error hierarchy. @@ -224,7 +226,7 @@ async def main(): # Primary: local Bluetooth tesla_bluetooth = TeslaBluetooth() await tesla_bluetooth.get_private_key("path/to/private_key.pem") - primary = tesla_bluetooth.vehicles.create("", verify_commands=True) + primary = tesla_bluetooth.vehicles.create("", confirmation="verify") # Secondary (fallback): Teslemetry cloud teslemetry = Teslemetry(access_token="", session=session) @@ -257,7 +259,7 @@ await router.set_operation(...) # local first, cloud on failure Enable `DEBUG` logging for `tesla_fleet_api` to see which backend served a routed call and why failover happened. -> **Warning:** Because a failed call is replayed on the next backend, a non-idempotent command (e.g. `honk_horn`, `actuate_trunk`, `door_unlock`, `charge_start`) that fails _mid-flight_ — after a backend may have already partially applied it — can be **double-executed** (or executed more than once across a longer chain) when it is retried on the next backend. This is a deliberate tradeoff of per-command failover. `BluetoothUnconfirmedCommand` is the exception: it propagates without failover because the BLE command may already have executed. When the primary is `VehicleBluetooth`, pass `verify_commands=True` to resolve supported mutating command timeouts by state before they reach the router, or `raise_unconfirmed=False` only when a best-effort success is acceptable; callers needing exactly-once semantics for other commands should gate dispatch with an explicit `health` check or call the underlying backends directly. +> **Warning:** Because a failed call is replayed on the next backend, a non-idempotent command (e.g. `honk_horn`, `actuate_trunk`, `door_unlock`, `charge_start`) that fails _mid-flight_ — after a backend may have already partially applied it — can be **double-executed** (or executed more than once across a longer chain) when it is retried on the next backend. This is a deliberate tradeoff of per-command failover. `BluetoothUnconfirmedCommand` is the exception: it propagates without failover because the BLE command may already have executed. When the primary is `VehicleBluetooth`, pass `confirmation="verify"` to resolve supported mutating command timeouts by state before they reach the router, and set `raise_unconfirmed=True` when callers must see still-ambiguous outcomes instead of the default best-effort success; callers needing exactly-once semantics for other commands should gate dispatch with an explicit `health` check or call the underlying backends directly. > > Dispatch is implemented via `__getattr__`, which does not proxy dunder methods, so `async with Router(...)` does **not** manage a backend's BLE connection lifecycle (`__aenter__`/`__aexit__`). Commands still auto-connect on send; for explicit connect/disconnect reach through `router.primary` (or `router.backends`). diff --git a/tesla_fleet_api.egg-info/SOURCES.txt b/tesla_fleet_api.egg-info/SOURCES.txt index 7c95146..112db0f 100644 --- a/tesla_fleet_api.egg-info/SOURCES.txt +++ b/tesla_fleet_api.egg-info/SOURCES.txt @@ -61,9 +61,11 @@ tesla_fleet_api/tessie/__init__.py tesla_fleet_api/tessie/tessie.py tesla_fleet_api/tessie/vehicle.py tests/test_auto_seat_climate.py +tests/test_ble_broadcast_confirmation.py tests/test_ble_charging_commands.py tests/test_ble_climate_commands.py tests/test_ble_command_verification.py +tests/test_ble_confirmation_mode.py tests/test_ble_expects_data.py tests/test_ble_keepalive.py tests/test_ble_message_routing.py @@ -72,10 +74,12 @@ tests/test_ble_mocked_commands.py tests/test_ble_mocked_media_commands.py tests/test_ble_mocked_state_readers.py tests/test_ble_nav_misc_commands.py +tests/test_ble_optimistic_and_best_effort.py tests/test_ble_pair.py tests/test_ble_reassembling_buffer.py tests/test_ble_send_transport.py tests/test_ble_unconfirmed_command.py +tests/test_ble_write_timeout_router.py tests/test_command_counter_lock.py tests/test_command_logging.py tests/test_cross_transport_parity.py @@ -84,4 +88,5 @@ tests/test_firmware_at_least.py tests/test_fleet_auth_refresh.py tests/test_power_mode_commands.py tests/test_router.py +tests/test_tesla_private_key.py tests/test_tessie_vehicle_params.py \ No newline at end of file diff --git a/tesla_fleet_api/tesla/tesla.py b/tesla_fleet_api/tesla/tesla.py index 3787ffe..f47e005 100644 --- a/tesla_fleet_api/tesla/tesla.py +++ b/tesla_fleet_api/tesla/tesla.py @@ -19,7 +19,13 @@ def _owner_only_opener(file: str, flags: int) -> int: """Open a new file exclusively, born at mode 0o600 with no chmod window.""" - return os.open(file, os.O_WRONLY | os.O_CREAT | os.O_EXCL, 0o600) + fd = os.open(file, os.O_WRONLY | os.O_CREAT | os.O_EXCL, 0o600) + try: + os.fchmod(fd, 0o600) + except OSError: + os.close(fd) + raise + return fd class Tesla: diff --git a/tests/test_tesla_private_key.py b/tests/test_tesla_private_key.py index 80d185e..9238515 100644 --- a/tests/test_tesla_private_key.py +++ b/tests/test_tesla_private_key.py @@ -10,6 +10,7 @@ from __future__ import annotations import stat +import os import tempfile from pathlib import Path from unittest import IsolatedAsyncioTestCase, mock @@ -47,6 +48,19 @@ async def test_new_key_file_is_owner_only(self) -> None: mode = stat.S_IMODE(Path(path).stat().st_mode) self.assertEqual(mode, 0o600) + async def test_new_key_file_is_owner_only_with_restrictive_umask(self) -> None: + with tempfile.TemporaryDirectory() as tmp_dir: + path = str(Path(tmp_dir) / "private_key.pem") + tesla = Tesla() + old_umask = os.umask(0o777) + try: + await tesla.get_private_key(path) + finally: + os.umask(old_umask) + + mode = stat.S_IMODE(Path(path).stat().st_mode) + self.assertEqual(mode, 0o600) + async def test_existing_key_read_path_unchanged(self) -> None: with tempfile.TemporaryDirectory() as tmp_dir: path = str(Path(tmp_dir) / "private_key.pem") @@ -87,6 +101,19 @@ async def test_new_key_file_is_owner_only(self) -> None: mode = stat.S_IMODE(Path(path).stat().st_mode) self.assertEqual(mode, 0o600) + async def test_new_key_file_is_owner_only_with_restrictive_umask(self) -> None: + with tempfile.TemporaryDirectory() as tmp_dir: + path = str(Path(tmp_dir) / "tedapi_rsa_private.pem") + tesla = Tesla() + old_umask = os.umask(0o777) + try: + await tesla.get_rsa_private_key(path, key_size=1024) + finally: + os.umask(old_umask) + + mode = stat.S_IMODE(Path(path).stat().st_mode) + self.assertEqual(mode, 0o600) + async def test_existing_key_read_path_unchanged(self) -> None: with tempfile.TemporaryDirectory() as tmp_dir: path = str(Path(tmp_dir) / "tedapi_rsa_private.pem") From fad6440d5e60de4dcdaeea0ea63c30db96465dae Mon Sep 17 00:00:00 2001 From: firstmate crewmate Date: Sun, 12 Jul 2026 20:31:10 +1000 Subject: [PATCH 3/7] no-mistakes(review): Make key opener portable without fchmod --- tesla_fleet_api/tesla/tesla.py | 6 +++++- tests/test_tesla_private_key.py | 33 +++++++++++++++++++++++++++++++++ 2 files changed, 38 insertions(+), 1 deletion(-) diff --git a/tesla_fleet_api/tesla/tesla.py b/tesla_fleet_api/tesla/tesla.py index f47e005..1866fe9 100644 --- a/tesla_fleet_api/tesla/tesla.py +++ b/tesla_fleet_api/tesla/tesla.py @@ -21,7 +21,11 @@ def _owner_only_opener(file: str, flags: int) -> int: """Open a new file exclusively, born at mode 0o600 with no chmod window.""" fd = os.open(file, os.O_WRONLY | os.O_CREAT | os.O_EXCL, 0o600) try: - os.fchmod(fd, 0o600) + fchmod = getattr(os, "fchmod", None) + if fchmod is not None: + fchmod(fd, 0o600) + else: + os.chmod(file, 0o600) except OSError: os.close(fd) raise diff --git a/tests/test_tesla_private_key.py b/tests/test_tesla_private_key.py index 9238515..9910e93 100644 --- a/tests/test_tesla_private_key.py +++ b/tests/test_tesla_private_key.py @@ -38,6 +38,39 @@ def _rsa_pem(key: rsa.RSAPrivateKey) -> bytes: class GetPrivateKeyPermissionsTests(IsolatedAsyncioTestCase): + async def test_new_key_file_uses_chmod_without_fchmod(self) -> None: + with tempfile.TemporaryDirectory() as tmp_dir: + path = str(Path(tmp_dir) / "private_key.pem") + + with ( + mock.patch("tesla_fleet_api.tesla.tesla.os.fchmod", new=None), + mock.patch( + "tesla_fleet_api.tesla.tesla.os.chmod", wraps=os.chmod + ) as chmod, + ): + await Tesla().get_private_key(path) + + chmod.assert_called_once_with(path, 0o600) + + async def test_new_key_file_closes_fd_when_chmod_fails(self) -> None: + with tempfile.TemporaryDirectory() as tmp_dir: + path = str(Path(tmp_dir) / "private_key.pem") + + with ( + mock.patch("tesla_fleet_api.tesla.tesla.os.fchmod", new=None), + mock.patch( + "tesla_fleet_api.tesla.tesla.os.chmod", + side_effect=OSError("chmod failed"), + ), + mock.patch( + "tesla_fleet_api.tesla.tesla.os.close", wraps=os.close + ) as close, + ): + with self.assertRaises(OSError): + await Tesla().get_private_key(path) + + close.assert_called_once() + async def test_new_key_file_is_owner_only(self) -> None: with tempfile.TemporaryDirectory() as tmp_dir: path = str(Path(tmp_dir) / "private_key.pem") From 09a14d42535f8cb70de63e50978383955afa3002 Mon Sep 17 00:00:00 2001 From: firstmate crewmate Date: Sun, 12 Jul 2026 20:34:31 +1000 Subject: [PATCH 4/7] no-mistakes(review): Retry raced private-key reads --- tesla_fleet_api/tesla/tesla.py | 46 ++++++++++++++++++++++--------- tests/test_tesla_private_key.py | 48 +++++++++++++++++++++++++++++++++ 2 files changed, 82 insertions(+), 12 deletions(-) diff --git a/tesla_fleet_api/tesla/tesla.py b/tesla_fleet_api/tesla/tesla.py index 1866fe9..dfeb28b 100644 --- a/tesla_fleet_api/tesla/tesla.py +++ b/tesla_fleet_api/tesla/tesla.py @@ -1,7 +1,9 @@ """Tesla Fleet API for Python.""" import base64 +import asyncio import os +import time from os.path import exists import aiofiles @@ -16,6 +18,9 @@ from cryptography.hazmat.primitives import serialization from cryptography.hazmat.backends import default_backend +_KEY_READ_RETRY_TIMEOUT = 1.0 +_KEY_READ_RETRY_INTERVAL = 0.05 + def _owner_only_opener(file: str, flags: int) -> int: """Open a new file exclusively, born at mode 0o600 with no chmod window.""" @@ -32,6 +37,21 @@ def _owner_only_opener(file: str, flags: int) -> int: return fd +async def _load_pem_private_key(path: str, retry_invalid: bool = False) -> object: + deadline = time.monotonic() + _KEY_READ_RETRY_TIMEOUT + while True: + async with aiofiles.open(path, "rb") as key_file: + key_data = await key_file.read() + try: + return serialization.load_pem_private_key( + key_data, password=None, backend=default_backend() + ) + except ValueError: + if not retry_invalid or time.monotonic() >= deadline: + raise + await asyncio.sleep(_KEY_READ_RETRY_INTERVAL) + + class Tesla: """Base class describing interactions with Tesla products.""" @@ -69,14 +89,16 @@ async def get_private_key( await key_file.write(pem) return self.private_key except FileExistsError: - pass + value = await _load_pem_private_key(path, retry_invalid=True) + if not isinstance(value, ec.EllipticCurvePrivateKey): + raise AssertionError( + "Loaded key is not an EllipticCurvePrivateKey" + ) + self.private_key = value + return self.private_key try: - 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() - ) + value = await _load_pem_private_key(path) except FileNotFoundError: raise FileNotFoundError(f"Private key file not found at {path}") except PermissionError: @@ -149,13 +171,13 @@ async def get_rsa_private_key( await key_file.write(pem) return self.rsa_private_key except FileExistsError: - pass + value = await _load_pem_private_key(path, retry_invalid=True) + if not isinstance(value, rsa.RSAPrivateKey): + raise AssertionError("Loaded key is not an RSAPrivateKey") + self.rsa_private_key = value + return self.rsa_private_key - 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() - ) + value = await _load_pem_private_key(path) if not isinstance(value, rsa.RSAPrivateKey): raise AssertionError("Loaded key is not an RSAPrivateKey") self.rsa_private_key = value diff --git a/tests/test_tesla_private_key.py b/tests/test_tesla_private_key.py index 9910e93..2935668 100644 --- a/tests/test_tesla_private_key.py +++ b/tests/test_tesla_private_key.py @@ -9,6 +9,7 @@ from __future__ import annotations +import asyncio import stat import os import tempfile @@ -122,6 +123,28 @@ async def test_concurrent_create_falls_back_to_read(self) -> None: self.assertEqual(_ec_pem(key), _ec_pem(winner_key)) + async def test_concurrent_create_waits_for_winner_to_finish_writing(self) -> None: + with tempfile.TemporaryDirectory() as tmp_dir: + path = Path(tmp_dir) / "private_key.pem" + winner_key = ec.generate_private_key(ec.SECP256R1()) + path.touch(mode=0o600) + + async def finish_write() -> None: + await asyncio.sleep(0.1) + path.write_bytes(_ec_pem(winner_key)) + + writer = asyncio.create_task(finish_write()) + try: + loser = Tesla() + with mock.patch( + "tesla_fleet_api.tesla.tesla.exists", return_value=False + ): + key = await loser.get_private_key(str(path)) + finally: + await writer + + self.assertEqual(_ec_pem(key), _ec_pem(winner_key)) + class GetRsaPrivateKeyPermissionsTests(IsolatedAsyncioTestCase): async def test_new_key_file_is_owner_only(self) -> None: @@ -171,3 +194,28 @@ async def test_concurrent_create_falls_back_to_read(self) -> None: key = await loser.get_rsa_private_key(path, key_size=1024) self.assertEqual(_rsa_pem(key), _rsa_pem(winner_key)) + + async def test_concurrent_create_waits_for_winner_to_finish_writing(self) -> None: + with tempfile.TemporaryDirectory() as tmp_dir: + path = Path(tmp_dir) / "tedapi_rsa_private.pem" + winner_key = rsa.generate_private_key( + public_exponent=65537, + key_size=1024, + ) + path.touch(mode=0o600) + + async def finish_write() -> None: + await asyncio.sleep(0.1) + path.write_bytes(_rsa_pem(winner_key)) + + writer = asyncio.create_task(finish_write()) + try: + loser = Tesla() + with mock.patch( + "tesla_fleet_api.tesla.tesla.exists", return_value=False + ): + key = await loser.get_rsa_private_key(str(path), key_size=1024) + finally: + await writer + + self.assertEqual(_rsa_pem(key), _rsa_pem(winner_key)) From c8252540a0f7516fcd54f3cd0f86a9be3ef977c1 Mon Sep 17 00:00:00 2001 From: firstmate crewmate Date: Sun, 12 Jul 2026 20:37:58 +1000 Subject: [PATCH 5/7] no-mistakes(review): Retry existing key reads during creation race --- tesla_fleet_api/tesla/tesla.py | 4 ++-- tests/test_tesla_private_key.py | 39 +++++++++++++++++++++++++++++++++ 2 files changed, 41 insertions(+), 2 deletions(-) diff --git a/tesla_fleet_api/tesla/tesla.py b/tesla_fleet_api/tesla/tesla.py index dfeb28b..7256b7b 100644 --- a/tesla_fleet_api/tesla/tesla.py +++ b/tesla_fleet_api/tesla/tesla.py @@ -98,7 +98,7 @@ async def get_private_key( return self.private_key try: - value = await _load_pem_private_key(path) + value = await _load_pem_private_key(path, retry_invalid=True) except FileNotFoundError: raise FileNotFoundError(f"Private key file not found at {path}") except PermissionError: @@ -177,7 +177,7 @@ async def get_rsa_private_key( self.rsa_private_key = value return self.rsa_private_key - value = await _load_pem_private_key(path) + value = await _load_pem_private_key(path, retry_invalid=True) if not isinstance(value, rsa.RSAPrivateKey): raise AssertionError("Loaded key is not an RSAPrivateKey") self.rsa_private_key = value diff --git a/tests/test_tesla_private_key.py b/tests/test_tesla_private_key.py index 2935668..d9b25de 100644 --- a/tests/test_tesla_private_key.py +++ b/tests/test_tesla_private_key.py @@ -145,6 +145,24 @@ async def finish_write() -> None: self.assertEqual(_ec_pem(key), _ec_pem(winner_key)) + async def test_existing_path_waits_for_winner_to_finish_writing(self) -> None: + with tempfile.TemporaryDirectory() as tmp_dir: + path = Path(tmp_dir) / "private_key.pem" + winner_key = ec.generate_private_key(ec.SECP256R1()) + path.touch(mode=0o600) + + async def finish_write() -> None: + await asyncio.sleep(0.1) + path.write_bytes(_ec_pem(winner_key)) + + writer = asyncio.create_task(finish_write()) + try: + key = await Tesla().get_private_key(str(path)) + finally: + await writer + + self.assertEqual(_ec_pem(key), _ec_pem(winner_key)) + class GetRsaPrivateKeyPermissionsTests(IsolatedAsyncioTestCase): async def test_new_key_file_is_owner_only(self) -> None: @@ -219,3 +237,24 @@ async def finish_write() -> None: await writer self.assertEqual(_rsa_pem(key), _rsa_pem(winner_key)) + + async def test_existing_path_waits_for_winner_to_finish_writing(self) -> None: + with tempfile.TemporaryDirectory() as tmp_dir: + path = Path(tmp_dir) / "tedapi_rsa_private.pem" + winner_key = rsa.generate_private_key( + public_exponent=65537, + key_size=1024, + ) + path.touch(mode=0o600) + + async def finish_write() -> None: + await asyncio.sleep(0.1) + path.write_bytes(_rsa_pem(winner_key)) + + writer = asyncio.create_task(finish_write()) + try: + key = await Tesla().get_rsa_private_key(str(path), key_size=1024) + finally: + await writer + + self.assertEqual(_rsa_pem(key), _rsa_pem(winner_key)) From 5ecb1e66fc8a3c8b1c97faaa396973acca3a535b Mon Sep 17 00:00:00 2001 From: firstmate crewmate Date: Sun, 12 Jul 2026 20:43:33 +1000 Subject: [PATCH 6/7] no-mistakes(document): Document atomic key file creation --- README.md | 6 ++++++ docs/bluetooth_vehicles.md | 6 ++++++ docs/fleet_api_energy_sites.md | 6 ++++++ tesla_fleet_api.egg-info/PKG-INFO | 6 ++++++ tesla_fleet_api/tesla/tesla.py | 15 ++++++++------- 5 files changed, 32 insertions(+), 7 deletions(-) diff --git a/README.md b/README.md index defbe67..3e6d2cf 100644 --- a/README.md +++ b/README.md @@ -166,6 +166,12 @@ asyncio.run(main()) For more detailed examples, see [Bluetooth for Vehicles](docs/bluetooth_vehicles.md). +`get_private_key(path)` loads an existing EC private key or creates a new +unencrypted PEM key file. Newly created key files are created owner-readable +and owner-writable only (`0600`) from the start, with no write-then-chmod +window, and concurrent creators fall back to reading the file that won the +create race. + `VehicleBluetooth` keeps a held BLE connection alive during idle periods by default with a passive GATT read about every 20 seconds. Pass `keepalive_interval=None` (or `0`) when creating the vehicle to disable it; diff --git a/docs/bluetooth_vehicles.md b/docs/bluetooth_vehicles.md index c18df66..db7be59 100644 --- a/docs/bluetooth_vehicles.md +++ b/docs/bluetooth_vehicles.md @@ -39,6 +39,12 @@ async def main(): asyncio.run(main()) ``` +`get_private_key(path)` loads an existing EC private key or creates a new +unencrypted PEM key file. Newly created key files are created owner-readable +and owner-writable only (`0600`) from the start, with no write-then-chmod +window, and concurrent creators fall back to reading the file that won the +create race. + ## Keeping the Connection Alive (`keepalive_interval`) An idle held BLE link to the vehicle drops on its own after roughly 42 seconds diff --git a/docs/fleet_api_energy_sites.md b/docs/fleet_api_energy_sites.md index 593f207..3a4e1af 100644 --- a/docs/fleet_api_energy_sites.md +++ b/docs/fleet_api_energy_sites.md @@ -354,6 +354,12 @@ asyncio.run(main()) Energy gateways (Powerwalls, etc.) support gRPC commands sent via `POST /api/1/energy_sites/{id}/command`. These are undocumented Tesla API endpoints that communicate directly with the gateway hardware. All device command methods require the `energy_cmds` scope. +`get_rsa_private_key(path)` loads an existing RSA private key for gateway +client registration or creates a new unencrypted PEM key file. Newly created +key files are created owner-readable and owner-writable only (`0600`) from the +start, with no write-then-chmod window, and concurrent creators fall back to +reading the file that won the create race. + ### Available Commands | Method | Category | Description | diff --git a/tesla_fleet_api.egg-info/PKG-INFO b/tesla_fleet_api.egg-info/PKG-INFO index 8e97d9c..5c7e5df 100644 --- a/tesla_fleet_api.egg-info/PKG-INFO +++ b/tesla_fleet_api.egg-info/PKG-INFO @@ -189,6 +189,12 @@ asyncio.run(main()) For more detailed examples, see [Bluetooth for Vehicles](docs/bluetooth_vehicles.md). +`get_private_key(path)` loads an existing EC private key or creates a new +unencrypted PEM key file. Newly created key files are created owner-readable +and owner-writable only (`0600`) from the start, with no write-then-chmod +window, and concurrent creators fall back to reading the file that won the +create race. + `VehicleBluetooth` keeps a held BLE connection alive during idle periods by default with a passive GATT read about every 20 seconds. Pass `keepalive_interval=None` (or `0`) when creating the vehicle to disable it; diff --git a/tesla_fleet_api/tesla/tesla.py b/tesla_fleet_api/tesla/tesla.py index 7256b7b..2942f78 100644 --- a/tesla_fleet_api/tesla/tesla.py +++ b/tesla_fleet_api/tesla/tesla.py @@ -69,9 +69,10 @@ async def get_private_key( ) -> ec.EllipticCurvePrivateKey: """Get or create the private key. - A newly created key file is opened with O_EXCL so it is born at mode - 0o600 with no world-readable window. If another process wins the - create race, its file is read instead of raising. + The private key is stored as an unencrypted PEM file. A newly created + key file is opened with O_EXCL so it is born at mode 0o600 with no + world-readable window. If another process wins the create race, its + file is read instead of raising. """ if not exists(path): self.private_key = ec.generate_private_key( @@ -148,10 +149,10 @@ async def get_rsa_private_key( """Get or create an RSA private key for energy gateway client registration. The default 4096-bit key matches the format expected by the Powerwall - TEDapi v1r LAN protocol. A newly created key file is opened with - O_EXCL so it is born at mode 0o600 with no world-readable window. If - another process wins the create race, its file is read instead of - raising. + TEDapi v1r LAN protocol. The private key is stored as an unencrypted + PEM file. A newly created key file is opened with O_EXCL so it is born + at mode 0o600 with no world-readable window. If another process wins + the create race, its file is read instead of raising. """ if not exists(path): self.rsa_private_key = rsa.generate_private_key( From 596415a0f7966f624336771fa2886486e3fd1a92 Mon Sep 17 00:00:00 2001 From: firstmate crewmate Date: Sun, 12 Jul 2026 20:44:26 +1000 Subject: [PATCH 7/7] no-mistakes(lint): Apply formatting fixes --- tesla_fleet_api/tesla/tesla.py | 4 +--- 1 file changed, 1 insertion(+), 3 deletions(-) diff --git a/tesla_fleet_api/tesla/tesla.py b/tesla_fleet_api/tesla/tesla.py index 2942f78..1c2f8c3 100644 --- a/tesla_fleet_api/tesla/tesla.py +++ b/tesla_fleet_api/tesla/tesla.py @@ -92,9 +92,7 @@ async def get_private_key( except FileExistsError: value = await _load_pem_private_key(path, retry_invalid=True) if not isinstance(value, ec.EllipticCurvePrivateKey): - raise AssertionError( - "Loaded key is not an EllipticCurvePrivateKey" - ) + raise AssertionError("Loaded key is not an EllipticCurvePrivateKey") self.private_key = value return self.private_key