diff --git a/AGENTS.md b/AGENTS.md index ba475a3..7da6775 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -78,7 +78,7 @@ Commands (vehicle/commands.py) - protobuf-based signed command implementation (A `Vehicles` (vehicle/vehicles.py) is a `dict[str, Vehicle]` with factory methods: - `createFleet(vin)` → `VehicleFleet` - `createSigned(vin)` → `VehicleSigned` -- `createBluetooth(vin, confirmation="ack", keepalive_interval=..., raise_unconfirmed=False, *, verify_commands=None, optimistic=None)` → `VehicleBluetooth` +- `createBluetooth(vin, confirmation="ack", keepalive_interval=..., raise_unconfirmed=False, *, verify_commands=None, optimistic=None, key=None)` → `VehicleBluetooth` Teslemetry/Tessie override `Vehicles` with their own vehicle classes (`TeslemetryVehicle`, `TessieVehicle`) extending `VehicleFleet` with service-specific commands (e.g., `closure()`, `seat_heater()` for Teslemetry; `wake()`, `lock()` for Tessie). @@ -153,6 +153,7 @@ Keep the `tesla-protocol` floor at `>=0.5.0`; earlier releases have generated `. - **`VehicleAction`/`GetVehicleData` proto coverage is locked by test, not just by convention**: `tests/test_proto_coverage_lock.py` walks both descriptors and fails if any field has no wrapper (`commands.py`) or reader (`bluetooth.py`) and isn't on one of its two small, reasoned allowlists — keep that test in sync with any future `tesla-protocol` bump rather than special-casing new fields elsewhere. The only fields deliberately left unwrapped today are the 7-field push-style subscription/streaming family (`createStreamSession`/`streamMessage`/`vehicleDataSubscription`/`vehicleDataAck`/`vitalsSubscription`/`vitalsAck`/`cancelVehicleDataSubscription`, which need a public lifecycle/iterator API atop the private `_stream_sinks` routing above) and `getVehicleImageState` (needs chunked binary-transfer paging). CarServer's `GetVehicleState` sub-state is exposed as `legacy_vehicle_state()` (`bluetooth.py`), matching the `VehicleData.legacy_vehicle_state` reply field name, to avoid confusion with `vehicle_state()` (VCSEC `VehicleStatus`, a different message/domain). `set_rate_tariff`/`add_managed_charging_site` (`commands.py`) take `tesla_protocol` message types directly for their deeply-nested arguments rather than a parallel flattened dataclass API. - **Energy-gateway authorized-client pairing has security- and protocol-specific constraints**: use RSA for LAN TEDapi v1r, treat `PENDING_VERIFICATION_TIMEOUT` as terminal, and account for presence-free key removal. The authoritative pairing, retry, encoding, and removal guidance is in `docs/energy_local_control.md`; enum values and API contracts live in `const.py` and the relevant method docstrings. - **`register_client()` (`teslemetry/teslemetry.py`) is Teslemetry-only OAuth Dynamic Client Registration (RFC 7591)**: a module-level function, not a `Teslemetry` instance method, since registration precedes having a `client_id` or access token — callers pass a bare `aiohttp.ClientSession`. It always registers a new client (no dedup/caching) and raises `TeslemetryRegistrationError` (`exceptions.py`) on transport failure, a non-2xx response, a non-JSON body, or a response missing a usable `client_id`; a non-dict-but-valid-JSON body (list/scalar) is treated as the same malformed-response error rather than raising an uncaught `AttributeError`. Fleet API and Tessie have no equivalent — don't add one speculatively. See `docs/teslemetry.md`'s "OAuth Dynamic Client Registration" section and `tests/test_teslemetry_register_client.py`. +- **`False`, not `None`, is the "signing is disabled" value for `Commands.__init__`'s `private_key` (and `VehicleBluetooth.__init__`/`Vehicles.createBluetooth`/`VehiclesBluetooth.create`/`createBluetooth`'s `key`)**: `None` — the default and an explicit `None` — keeps its long-standing meaning of falling back to the parent's key, raising `ValueError("No private key.")` if it has none; `False` disables signing for a passive BLE listener that only observes broadcasts. `None` is deliberately *not* the opt-out: a caller already passing `private_key=None` to mean "I haven't got one" must keep getting that `ValueError`, not a silently unsignable vehicle. Because `False` and `None` are both falsy, every branch on this argument must test **identity** (`is False`/`is not None`) — a truthiness check (`if private_key:`) collapses the two states and reintroduces the bug. `self.private_key` is `EllipticCurvePrivateKey | None`, its `None` meaning signing-disabled — `_handshake` (reached by `_command`, i.e. every signed command, and by `_ensure_handshake`, used by signed reads) raises `SigningDisabled` (`exceptions.py`) up front rather than failing deep in the signing/crypto path. `pair()`'s fast path never calls `_handshake` (it builds and sends its own whitelist request directly), so it carries its own identical guard at the top instead — `_handshake` is not a single choke point every signed-session entry point routes through; each entry point that doesn't call it needs its own `self.private_key is None` check. Tests: `tests/test_ble_null_key.py`. ## Maintaining this file diff --git a/docs/bluetooth_vehicles.md b/docs/bluetooth_vehicles.md index b163edf..412989c 100644 --- a/docs/bluetooth_vehicles.md +++ b/docs/bluetooth_vehicles.md @@ -574,6 +574,26 @@ the `unsubscribe()` closure returned at registration. Callback exceptions are logged and do not stop later listeners or normal message routing; `KeyboardInterrupt` and `SystemExit` still propagate. +### Passive listening without a private key + +A vehicle that only decodes broadcasts and never sends a command has no +reason to hold a signing key. Pass `key=False` to `vehicles.create` (or +`VehicleBluetooth` directly) to construct without one: + +```python +vehicle = tesla_bluetooth.vehicles.create("", key=False) +``` + +`key=None` - the default, and an explicit `None` - still falls back to the +parent's private key exactly as before, raising `ValueError("No private +key.")` if the parent has none. Only `False` disables signing, so a caller +already passing `key=None` to mean "I haven't got one" keeps getting that +error rather than silently ending up with a vehicle that cannot sign. + +Reads and listeners that don't need a signed session still work; any command +that does raises `SigningDisabled` naming that signing was explicitly disabled +for this vehicle. + ### Connection-status events Use `listen_connection_status(callback)` to receive BLE session transitions diff --git a/tesla_fleet_api/exceptions.py b/tesla_fleet_api/exceptions.py index d83ccc8..d9e1a2a 100644 --- a/tesla_fleet_api/exceptions.py +++ b/tesla_fleet_api/exceptions.py @@ -408,6 +408,23 @@ class LibraryError(Exception): """Errors related to this library.""" +class SigningDisabled(LibraryError): + """A signed operation was attempted on a vehicle constructed with signing explicitly disabled. + + Pass ``private_key=False`` only for a passive listener that never sends a + command; construct with a real key (or leave the argument at ``None`` to + inherit the parent's) to issue signed commands. + """ + + def __init__(self) -> None: + super().__init__( + "This vehicle was constructed with private_key=False, explicitly " + "disabling command signing. It can only observe unsolicited " + "broadcasts (the listen_* methods); any signed command or read " + "needs a real private_key." + ) + + class SignedCommandRequired(TeslaFleetError): """The requested action requires a signed command; the unsigned cloud API cannot actuate it. diff --git a/tesla_fleet_api/tesla/vehicle/bluetooth.py b/tesla_fleet_api/tesla/vehicle/bluetooth.py index 5ec045d..6921915 100644 --- a/tesla_fleet_api/tesla/vehicle/bluetooth.py +++ b/tesla_fleet_api/tesla/vehicle/bluetooth.py @@ -7,7 +7,7 @@ import warnings from collections import deque from random import randbytes -from typing import TYPE_CHECKING, Any, Callable, Generic, TypeVar +from typing import TYPE_CHECKING, Any, Callable, Generic, Literal, TypeVar import bleak from bleak.backends.characteristic import BleakGATTCharacteristic @@ -24,6 +24,7 @@ BluetoothTimeout, BluetoothTransportError, BluetoothUnconfirmedCommand, + SigningDisabled, TeslaFleetError, WhitelistOperationStatus, ) @@ -512,7 +513,7 @@ def __init__( self, parent: BluetoothParentT, vin: str, - key: ec.EllipticCurvePrivateKey | None = None, + key: ec.EllipticCurvePrivateKey | Literal[False] | None = None, device: BLEDevice | None = None, confirmation: BluetoothConfirmation | bool = "ack", keepalive_interval: float | None = DEFAULT_KEEPALIVE_INTERVAL, @@ -529,6 +530,11 @@ def __init__( ``confirmation``, overriding any value passed there (a ``True`` ``optimistic`` wins over a ``True`` ``verify_commands`` if both are somehow passed, matching the old dominance order). + + Passing ``key=False`` explicitly disables command signing, for a + passive listener that only observes broadcasts via the ``listen_*`` + methods and never sends a command. ``key=None`` (the default, and an + explicit ``None``) keeps the usual fallback to the parent's key. """ super().__init__(parent, vin, key) if isinstance(confirmation, bool): @@ -1519,6 +1525,9 @@ async def pair( if poll_interval <= 0: raise ValueError("poll_interval must be greater than 0") + if self.private_key is None: + raise SigningDisabled() + request = UnsignedMessage( WhitelistOperation=WhitelistOperation( addKeyToWhitelistAndAddPermissions=PermissionChange( diff --git a/tesla_fleet_api/tesla/vehicle/commands.py b/tesla_fleet_api/tesla/vehicle/commands.py index c82dd72..75d29c0 100644 --- a/tesla_fleet_api/tesla/vehicle/commands.py +++ b/tesla_fleet_api/tesla/vehicle/commands.py @@ -29,6 +29,7 @@ NotOnWhitelistFault, SessionInfoAuthenticationFault, SignedCommandResponseReplayed, + SigningDisabled, TeslaFleetError, # TeslaFleetMessageFaultInvalidSignature, TeslaFleetMessageFaultIncorrectEpoch, @@ -428,7 +429,7 @@ def aes_gcm_personalized(self) -> AES_GCM_Personalized_Signature_Data: class Commands(ABC, Vehicle[CommandParentT], Generic[CommandParentT]): """Class describing the Tesla Fleet API vehicle endpoints and commands for a specific vehicle with command signing.""" - private_key: ec.EllipticCurvePrivateKey + private_key: ec.EllipticCurvePrivateKey | None _public_key: bytes _from_destination: bytes _sessions: dict[int, Session[CommandParentT]] @@ -440,9 +441,19 @@ def __init__( self, parent: CommandParentT, vin: str, - private_key: ec.EllipticCurvePrivateKey | None = None, + private_key: ec.EllipticCurvePrivateKey | Literal[False] | None = None, public_key: bytes | None = None, ): + """Initialize with a signing key, or ``private_key=False`` to disable signing. + + ``None`` (the default, and an explicit ``None``) keeps the long-standing + behaviour: fall back to the parent's key, raising ``ValueError`` if it + has none. Passing ``private_key=False`` explicitly disables signing for + this vehicle - for a passive BLE listener that only observes broadcasts + and never sends a command. ``False`` is used rather than ``None`` so + that no caller who already passes ``private_key=None`` meaning "I + haven't got one" silently gets a vehicle that cannot sign. + """ super().__init__(parent, vin) self._from_destination = randbytes(16) @@ -453,19 +464,30 @@ def __init__( Domain.DOMAIN_INFOTAINMENT: Session(self, Domain.DOMAIN_INFOTAINMENT), } - if private_key: + # Identity checks, not truthiness: ``False`` and ``None`` are both + # falsy, and collapsing them would make an explicit ``None`` silently + # disable signing instead of falling back to the parent's key. + if private_key is False: + self.private_key = None + elif private_key is not None: self.private_key = private_key - elif parent.private_key: + elif parent.private_key is not None: self.private_key = parent.private_key else: raise ValueError("No private key.") - self._public_key = public_key or self.private_key.public_key().public_bytes( - encoding=Encoding.X962, format=PublicFormat.UncompressedPoint + self._public_key = public_key or ( + self.private_key.public_key().public_bytes( + encoding=Encoding.X962, format=PublicFormat.UncompressedPoint + ) + if self.private_key is not None + else b"" ) def shared_key(self, vehicleKey: bytes) -> bytes: """Derive the 16-byte shared key used for signed-command session encryption.""" + if self.private_key is None: + raise SigningDisabled() exchange = self.private_key.exchange( ec.ECDH(), ec.EllipticCurvePublicKey.from_encoded_point(ec.SECP256R1(), vehicleKey), @@ -1088,6 +1110,8 @@ async def handshakeInfotainment(self) -> None: async def _handshake(self, domain: Domain) -> bool: """Perform a handshake with the vehicle.""" + if self.private_key is None: + raise SigningDisabled() LOGGER.debug(f"Handshake with domain {Domain.Name(domain)}") msg = RoutableMessage( diff --git a/tesla_fleet_api/tesla/vehicle/vehicles.py b/tesla_fleet_api/tesla/vehicle/vehicles.py index 822acf8..b8c03e6 100644 --- a/tesla_fleet_api/tesla/vehicle/vehicles.py +++ b/tesla_fleet_api/tesla/vehicle/vehicles.py @@ -1,5 +1,5 @@ from __future__ import annotations -from typing import TYPE_CHECKING, Any, Generic, TypeVar +from typing import TYPE_CHECKING, Any, Generic, Literal, TypeVar from bleak.backends.device import BLEDevice from cryptography.hazmat.primitives.asymmetric import ec @@ -53,6 +53,7 @@ def createBluetooth( raise_unconfirmed: bool = False, *, verify_commands: bool | None = None, + key: ec.EllipticCurvePrivateKey | Literal[False] | None = None, ) -> VehicleBluetooth[FleetParentT]: """Creates a bluetooth vehicle that uses command protocol. @@ -67,10 +68,13 @@ def createBluetooth( success. ``verify_commands``/``optimistic`` are deprecated aliases for ``confirmation="verify"``/``confirmation="optimistic"``. See ``VehicleBluetooth``'s docstring for the full ladder. + ``key=False`` explicitly disables signing, for a passive listener; + ``key=None`` (the default) keeps the usual parent-key fallback. """ vehicle = self.Bluetooth( self._parent, vin, + key, confirmation=confirmation, keepalive_interval=keepalive_interval, optimistic=optimistic, @@ -101,7 +105,7 @@ def __init__(self, parent: BluetoothClientT): def create( self, vin: str, - key: ec.EllipticCurvePrivateKey | None = None, + key: ec.EllipticCurvePrivateKey | Literal[False] | None = None, device: BLEDevice | None = None, confirmation: BluetoothConfirmation | bool = "ack", keepalive_interval: float | None = DEFAULT_KEEPALIVE_INTERVAL, @@ -123,6 +127,8 @@ def create( success. ``verify_commands``/``optimistic`` are deprecated aliases for ``confirmation="verify"``/``confirmation="optimistic"``. See ``VehicleBluetooth``'s docstring for the full ladder. + ``key=False`` explicitly disables signing, for a passive listener; + ``key=None`` (the default) keeps the usual parent-key fallback. """ return self.createBluetooth( vin, @@ -138,7 +144,7 @@ def create( def createBluetooth( self, vin: str, - key: ec.EllipticCurvePrivateKey | None = None, + key: ec.EllipticCurvePrivateKey | Literal[False] | None = None, device: BLEDevice | None = None, confirmation: BluetoothConfirmation | bool = "ack", keepalive_interval: float | None = DEFAULT_KEEPALIVE_INTERVAL, @@ -160,6 +166,8 @@ def createBluetooth( success. ``verify_commands``/``optimistic`` are deprecated aliases for ``confirmation="verify"``/``confirmation="optimistic"``. See ``VehicleBluetooth``'s docstring for the full ladder. + ``key=False`` explicitly disables signing, for a passive listener; + ``key=None`` (the default) keeps the usual parent-key fallback. """ vehicle = self.Bluetooth( self._parent, diff --git a/tesla_fleet_api/teslemetry/vehicle.py b/tesla_fleet_api/teslemetry/vehicle.py index 5250e08..b07a058 100644 --- a/tesla_fleet_api/teslemetry/vehicle.py +++ b/tesla_fleet_api/teslemetry/vehicle.py @@ -1,6 +1,8 @@ from __future__ import annotations -from typing import TYPE_CHECKING, Any +from typing import TYPE_CHECKING, Any, Literal + +from cryptography.hazmat.primitives.asymmetric import ec from tesla_fleet_api.const import ( BluetoothConfirmation, @@ -673,6 +675,7 @@ def createBluetooth( raise_unconfirmed: bool = False, *, verify_commands: bool | None = None, + key: ec.EllipticCurvePrivateKey | Literal[False] | None = None, ) -> Any: """Not supported; parameters match the Fleet API Bluetooth factory.""" raise NotImplementedError("Teslemetry cannot use local Bluetooth") diff --git a/tesla_fleet_api/tessie/vehicle.py b/tesla_fleet_api/tessie/vehicle.py index 02139f3..6ca359b 100644 --- a/tesla_fleet_api/tessie/vehicle.py +++ b/tesla_fleet_api/tessie/vehicle.py @@ -1,5 +1,7 @@ from __future__ import annotations -from typing import TYPE_CHECKING, Any +from typing import TYPE_CHECKING, Any, Literal + +from cryptography.hazmat.primitives.asymmetric import ec from tesla_fleet_api.const import BluetoothConfirmation, Method from tesla_fleet_api.tesla.vehicle.vehicles import Vehicles @@ -1222,6 +1224,7 @@ def createBluetooth( raise_unconfirmed: bool = False, *, verify_commands: bool | None = None, + key: ec.EllipticCurvePrivateKey | Literal[False] | None = None, ) -> Any: """Not supported; parameters match the Fleet API Bluetooth factory.""" raise NotImplementedError("Tessie cannot use local Bluetooth") diff --git a/tests/test_ble_null_key.py b/tests/test_ble_null_key.py new file mode 100644 index 0000000..ba99da5 --- /dev/null +++ b/tests/test_ble_null_key.py @@ -0,0 +1,170 @@ +"""Tests for constructing a ``VehicleBluetooth`` with signing explicitly disabled. + +``key``/``private_key`` keeps ``None`` as its long-standing meaning: whether +omitted or passed explicitly, it falls back to the parent's key and raises if +the parent has none. ``False`` is the distinct, additive opt-out: it constructs +successfully with signing disabled, still receives broadcasts via the +``listen_*`` methods, and raises a clear ``SigningDisabled`` (not a generic +attribute/type error from deep in the signing path) on any operation that +actually needs to sign. + +``False`` rather than ``None`` is the opt-out precisely so that a caller who +already writes ``private_key=None`` meaning "I haven't got one" keeps getting +today's ``ValueError`` instead of silently ending up with a vehicle that +cannot sign - see ``ExplicitNoneIsUnchangedTests``. +""" + +from __future__ import annotations + +from typing import Any +from unittest import IsolatedAsyncioTestCase +from unittest.mock import AsyncMock, MagicMock + +from cryptography.hazmat.primitives.asymmetric import ec + +from tesla_fleet_api.exceptions import SigningDisabled +from tesla_fleet_api.tesla.vehicle.bluetooth import VehicleBluetooth +from tesla_protocol.command.universal_message_pb2 import ( + Destination, + Domain, + RoutableMessage, +) +from tesla_protocol.command.vcsec_pb2 import ( + FromVCSECMessage, + VehicleLockState_E, + VehicleStatus, +) + +VIN = "5YJXCAE43LF123456" + + +def _status_broadcast(status: VehicleStatus) -> RoutableMessage: + body = FromVCSECMessage(vehicleStatus=status) + return RoutableMessage( + from_destination=Destination(domain=Domain.DOMAIN_VEHICLE_SECURITY), + protobuf_message_as_bytes=body.SerializeToString(), + ) + + +class OmittedKeyStillRaisesTests(IsolatedAsyncioTestCase): + async def test_omitted_key_with_no_parent_key_raises_value_error(self) -> None: + """Not passing a key at all keeps today's behaviour unchanged.""" + parent = MagicMock() + parent.private_key = None + + with self.assertRaisesRegex(ValueError, "No private key."): + VehicleBluetooth(parent, VIN) + + async def test_omitted_key_falls_back_to_parent_key(self) -> None: + """Not passing a key still inherits the parent's, as before.""" + parent = MagicMock() + parent.private_key = ec.generate_private_key(ec.SECP256R1()) + + vehicle = VehicleBluetooth(parent, VIN) + + self.assertIs(vehicle.private_key, parent.private_key) + + +class ExplicitNoneIsUnchangedTests(IsolatedAsyncioTestCase): + """An explicit ``key=None`` must behave exactly as it always has. + + This is the regression the ``False`` opt-out exists to prevent: a caller + writing ``key=None`` to mean "I haven't got one" must keep getting the + ``ValueError`` that tells them so, not a silently unsignable vehicle. + """ + + async def test_explicit_none_with_no_parent_key_raises_value_error(self) -> None: + parent = MagicMock() + parent.private_key = None + + with self.assertRaisesRegex(ValueError, "No private key."): + VehicleBluetooth(parent, VIN, key=None) + + async def test_explicit_none_falls_back_to_parent_key(self) -> None: + parent = MagicMock() + parent.private_key = ec.generate_private_key(ec.SECP256R1()) + + vehicle = VehicleBluetooth(parent, VIN, key=None) + + self.assertIs(vehicle.private_key, parent.private_key) + + +class ExplicitFalseKeyTests(IsolatedAsyncioTestCase): + async def test_explicit_false_key_constructs_even_with_no_parent_key(self) -> None: + """An explicit ``key=False`` disables signing rather than raising.""" + parent = MagicMock() + parent.private_key = None + + vehicle = VehicleBluetooth(parent, VIN, key=False) + + self.assertIsNone(vehicle.private_key) + + async def test_explicit_false_key_overrides_an_available_parent_key(self) -> None: + """``key=False`` disables signing even when the parent does have a key.""" + parent = MagicMock() + parent.private_key = ec.generate_private_key(ec.SECP256R1()) + + vehicle = VehicleBluetooth(parent, VIN, key=False) + + self.assertIsNone(vehicle.private_key) + + async def test_can_still_receive_broadcasts(self) -> None: + """A key-less vehicle still fans out unsolicited status broadcasts.""" + parent = MagicMock() + parent.private_key = None + vehicle = VehicleBluetooth(parent, VIN, key=False) + + seen: list[Any] = [] + vehicle.listen_vehicle_lock_state(seen.append) + + vehicle._on_message( + _status_broadcast( + VehicleStatus( + vehicleLockState=VehicleLockState_E.VEHICLELOCKSTATE_LOCKED + ) + ) + ) + + self.assertEqual(seen, [VehicleLockState_E.VEHICLELOCKSTATE_LOCKED]) + + +class SignedOperationOnDisabledKeyTests(IsolatedAsyncioTestCase): + async def test_signed_command_raises_signing_disabled(self) -> None: + """A signed operation on a key-less vehicle fails clearly, not deep in the signing path.""" + parent = MagicMock() + parent.private_key = None + vehicle = VehicleBluetooth(parent, VIN, key=False) + vehicle.connect_if_needed = AsyncMock() # type: ignore[method-assign] + vehicle._send = AsyncMock() # type: ignore[method-assign] + + with self.assertRaises(SigningDisabled): + await vehicle.door_lock() + + async def test_handshake_raises_signing_disabled(self) -> None: + parent = MagicMock() + parent.private_key = None + vehicle = VehicleBluetooth(parent, VIN, key=False) + + with self.assertRaises(SigningDisabled): + await vehicle.handshakeVehicleSecurity() + + async def test_pair_raises_signing_disabled_without_touching_transport( + self, + ) -> None: + """pair() must fail before it builds/sends a whitelist request. + + A key-less vehicle has an empty ``_public_key``; without this guard + pair() would proceed to connect and send a malformed whitelist + request to real hardware instead of failing clearly up front. + """ + parent = MagicMock() + parent.private_key = None + vehicle = VehicleBluetooth(parent, VIN, key=False) + vehicle.connect_if_needed = AsyncMock() # type: ignore[method-assign] + vehicle._send = AsyncMock() # type: ignore[method-assign] + + with self.assertRaises(SigningDisabled): + await vehicle.pair() + + vehicle.connect_if_needed.assert_not_awaited() + vehicle._send.assert_not_awaited()