fix(tesla): create private keys atomically with owner-only permissions

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;

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

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 |

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 <hello@teslemetry.com>
 License-Expression: Apache-2.0
@@ -189,21 +189,29 @@ 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;
 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 +232,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("<vin>", verify_commands=True)
+        primary = tesla_bluetooth.vehicles.create("<vin>", confirmation="verify")
 
         # Secondary (fallback): Teslemetry cloud
         teslemetry = Teslemetry(access_token="<access_token>", session=session)
@@ -257,7 +265,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`).
 

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

tesla_fleet_api/tesla/tesla.py

@@ -1,6 +1,9 @@
 """Tesla Fleet API for Python."""
 
 import base64
+import asyncio
+import os
+import time
 from os.path import exists
 import aiofiles
 
@@ -15,6 +18,39 @@
 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."""
+    fd = os.open(file, os.O_WRONLY | os.O_CREAT | os.O_EXCL, 0o600)
+    try:
+        fchmod = getattr(os, "fchmod", None)
+        if fchmod is not None:
+            fchmod(fd, 0o600)
+        else:
+            os.chmod(file, 0o600)
+    except OSError:
+        os.close(fd)
+        raise
+    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."""
@@ -33,42 +69,43 @@ 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.
+        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(
                 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:
-                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
+                async with aiofiles.open(
+                    path, "wb", opener=_owner_only_opener
+                ) as key_file:
+                    await key_file.write(pem)
+                return self.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")
+                self.private_key = value
+                return self.private_key
+
+        try:
+            value = await _load_pem_private_key(path, retry_invalid=True)
+        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
@@ -111,7 +148,9 @@ async def get_rsa_private_key(
 
         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.
+        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(
@@ -124,23 +163,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:
-                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, "wb", opener=_owner_only_opener
+                ) as key_file:
+                    await key_file.write(pem)
+                return self.rsa_private_key
+            except FileExistsError:
+                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
+
+        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
 
     @property