From 043232d3426bf81512a05d972b1388062cb6c789 Mon Sep 17 00:00:00 2001 From: firstmate crewmate Date: Thu, 23 Jul 2026 14:18:59 +1000 Subject: [PATCH 1/4] feat(signed): authenticate session_info and harden the signed-command session path Per an independent security audit of the signed-command/session layer: - Verify a session_info reply's session_info_tag HMAC (bound to the client's own sent request uuid as TAG_CHALLENGE) before trusting any field of it, including its whitelist status. VCSEC's wire-level request_uuid echo is typically empty and is not itself authentication, so acceptance never gates on it. A bad/absent tag raises SessionInfoAuthenticationFault without mutating session state. - Session.commit clamps the anti-replay counter to max(current, incoming) within an epoch, refuses a same-epoch clock regression, and only resets on an epoch change - closing a client-side rollback/DoS window. - The AES-GCM (BLE) response path now rejects a replayed response counter via SignedCommandResponseReplayed, defending against a captured encrypted response being replayed back at the client. None of this affects the cloud (HMAC/TLS) transport's security posture or allows unauthorized actuation on either transport; it closes BLE-specific spec-conformance gaps bounded at DoS/response-spoofing under an active on-path attacker. tests/test_session_info_authentication.py adds coverage the BLE-mocked command tests structurally can't provide (they pre-mark sessions ready and mock _send wholesale): a golden-vector test pinned to protocol.md's own published handshake vectors, unit tests for tag/counter/clock handling, and end-to-end tests through the real (unmocked) BLE handshake state machine. --- AGENTS.md | 1 + tesla_fleet_api/exceptions.py | 25 + tesla_fleet_api/tesla/vehicle/bluetooth.py | 4 +- tesla_fleet_api/tesla/vehicle/commands.py | 159 ++++- tesla_fleet_api/tesla/vehicle/signed.py | 2 +- tests/test_session_info_authentication.py | 681 +++++++++++++++++++++ 6 files changed, 849 insertions(+), 23 deletions(-) create mode 100644 tests/test_session_info_authentication.py diff --git a/AGENTS.md b/AGENTS.md index 02a1654..40ad0ab 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -147,6 +147,7 @@ Keep the `tesla-protocol` floor at `>=0.5.0`; earlier releases have generated `. - **Typed accessor pattern for undocumented raw-dict responses**: `TeslemetryEnergySite.find_authorized_clients()` (`teslemetry/energysite.py`) is the library's first frozen-dataclass typed wrapper over a raw `dict[str, Any]`/`None`/`list` REST response, added so API-parsing logic (envelope unwrap, field lookup, shape validation, enum typing) lives in the library instead of each consumer (e.g. Home Assistant's config flow) reimplementing it. `_authorized_clients_list()` does one precise, non-recursive envelope unwrap (`{"response": {"authorized_clients": [...]}}` or `{"response": {"clients": [...]}}`, or a bare list with no envelope) rather than a multi-key/depth-first search - the `clients` key variant was added after a live capture (Tesla Release 953) showed the endpoint's real key differs from the originally-documented `authorized_clients`, confirmed against a populated 5-entry sample. `AuthorizedClient` models only the two fields a pairing flow actually reads (`public_key`, `state`), each accepting only the specific key-name variant pairs empirically evidenced, not speculative extras. Two rules any future typed accessor over an undocumented response shape must keep, demonstrated here even though `AuthorizedClientState` (`const.py`) has no falsy member: (1) field lookup must check key presence (`key in payload`), never `payload.get(key) or default` - a legal falsy value is not "missing", and a present-but-unrecognized enum value (`_normalize_state()`) is returned raw rather than coerced to `None`; (2) a `None` body and an unrecognized response shape (not a dict/list, or an envelope that unwraps to neither accepted list key) are malformed data, not "zero clients" - Tesla's endpoint intermittently returns HTTP 200 with a null body, so `_authorized_clients_list()` raises `InvalidResponse` (`exceptions.py`) for both rather than collapsing them to `[]`; only a genuinely empty list under either accepted key parses to `AuthorizedClients.clients == []` without raising. Tesla has not published an OpenAPI schema for pairing endpoints, so `const.py`'s enums are the schema of record; widen `AuthorizedClient`'s modeled fields only against a further live sample, not speculatively. The raw `list_authorized_clients()` method is kept alongside as the untyped escape hatch and still returns a null body unchanged rather than raising. Tests: `tests/test_teslemetry_authorized_clients.py`. - **`networking_status`'s `ipv4_config` fields are raw big-endian uint32 ints, not strings**: the wire format applies to all of `ipv4_config.address`/`subnet_mask`/`gateway`, but `TeslemetryEnergySite.find_gateway_address()` (`teslemetry/energysite.py`, second typed accessor after `find_authorized_clients()`, same rules) decodes only `address` - confirmed against a live Powerwall 3 capture where `3232235914` decodes network-byte-order (`struct.pack(">I", ...)`) to `192.168.1.138`, not little-endian. It considers only `eth`/`wifi` (never `gsm` - cellular isn't a LAN path), preferring whichever has `active_route` set and a decodable address, else falling back to the first of the two (in that order) with any decodable address; `0`/`0xFFFFFFFF` are treated as undecodable so an unconfigured interface never shadows a real one with `0.0.0.0`. A `{"response": null}` envelope raises `InvalidResponse` (the endpoint's known intermittent malformed mode), while a well-formed response with no usable interface returns `None` (not a raise). Tests: `tests/test_teslemetry_gateway_address.py`. - **`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`, needs a persistent INFOTAINMENT-domain broadcast dispatcher like `broadcast.py`'s VCSEC listeners) and `getVehicleImageState` (needs chunked binary-transfer paging) - both are separate, unscoped design work, not oversights. CarServer's `GetVehicleState` sub-state is exposed as `legacy_vehicle_state()` (`bluetooth.py`), matching the `VehicleData.legacy_vehicle_state` reply field name, specifically to avoid confusion with the pre-existing `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. `pii_key_request`/`pseudonym_sync_request`/`tesla_auth_response`/`setup_cloud_profile_with_local_profile_uuid`/`get_local_profiles_for_vault_uuid` are wrapped with no known third-party consumer use case, purely for full-proto-coverage completeness. +- **`session_info` replies are authenticated before any field is trusted**: `Commands._authenticate_session_info`/`Session.commit` (`commands.py`) verify the reply's `session_info_tag` HMAC - keyed by `HMAC(K, "session info")`, bound to the `request_uuid` the client actually sent as `TAG_CHALLENGE` - before adopting `counter`/`epoch`/`clock_time`/`publicKey` or even reading the whitelist-status field; a bad/absent tag raises `SessionInfoAuthenticationFault` (`exceptions.py`) without mutating session state. `validate_msg` takes the sent request's `uuid` as a required second argument for this reason - every `_send` implementation must thread it through. The wire-level `request_uuid` echo field is a red herring, not authentication: VCSEC typically leaves it empty (memory constraints), so acceptance must never gate on it - only cross-check it when the vehicle populates it. Within an epoch the counter clamps to `max(current, incoming)` and a clock regression is refused (reset only happens on an epoch change); the AES-GCM response path additionally rejects a replayed response counter via `Session.last_response_counter`. `tests/test_session_info_authentication.py`'s `GoldenVectorTests` pins the tag/key derivation against protocol.md's own published test vectors verbatim; its BLE tests drive the real (unmocked) handshake state machine, unlike the BLE-mocked command tests which pre-mark sessions ready and never exercise this path. ## Maintaining this file diff --git a/tesla_fleet_api/exceptions.py b/tesla_fleet_api/exceptions.py index ffeba3f..ec103c5 100644 --- a/tesla_fleet_api/exceptions.py +++ b/tesla_fleet_api/exceptions.py @@ -414,6 +414,31 @@ class SignedCommandRequired(TeslaFleetError): ) +class SessionInfoAuthenticationFault(TeslaFleetError): + """A ``session_info`` reply failed local authentication and was discarded. + + Raised when the reply's ``session_info_tag`` HMAC does not verify, is + absent, the echoed ``request_uuid`` does not match the outstanding + request it claims to answer, or its clock time regresses within the same + epoch. The session's prior state is left unmodified. + """ + + message = "Session info reply failed authentication and was discarded." + + +class SignedCommandResponseReplayed(TeslaFleetError): + """A signed command response reused a counter value already seen on this session. + + protocol.md requires rejecting a response whose counter has previously + been used, to prevent an attacker from replaying a captured encrypted + response back at the client. + """ + + message = ( + "Signed command response reused an already-seen counter; discarded as a replay." + ) + + class TeslaFleetInformationFault(TeslaFleetError): """Vehicle has responded with an error when sending a signed command""" diff --git a/tesla_fleet_api/tesla/vehicle/bluetooth.py b/tesla_fleet_api/tesla/vehicle/bluetooth.py index ac144cf..074a8e2 100644 --- a/tesla_fleet_api/tesla/vehicle/bluetooth.py +++ b/tesla_fleet_api/tesla/vehicle/bluetooth.py @@ -829,7 +829,7 @@ async def _await_response( resp = await self._queues[domain].get() LOGGER.debug(f"Received message {resp}") - self.validate_msg(resp) + self.validate_msg(resp, msg.uuid) if resp.HasField(requires): return resp @@ -850,7 +850,7 @@ async def _await_response( while True: resp2 = await self._queues[domain].get() LOGGER.debug(f"Received follow-up message {resp2}") - self.validate_msg(resp2) + self.validate_msg(resp2, msg.uuid) if resp2.HasField(requires): return resp2 except TimeoutError: diff --git a/tesla_fleet_api/tesla/vehicle/commands.py b/tesla_fleet_api/tesla/vehicle/commands.py index 9f96d26..d6c853c 100644 --- a/tesla_fleet_api/tesla/vehicle/commands.py +++ b/tesla_fleet_api/tesla/vehicle/commands.py @@ -26,6 +26,8 @@ MESSAGE_FAULTS, SIGNED_MESSAGE_INFORMATION_FAULTS, NotOnWhitelistFault, + SessionInfoAuthenticationFault, + SignedCommandResponseReplayed, TeslaFleetError, # TeslaFleetMessageFaultInvalidSignature, TeslaFleetMessageFaultIncorrectEpoch, @@ -315,6 +317,9 @@ def __init__(self, parent: Commands[CommandParentT], domain: Domain): self.sharedKey: bytes | None = None self.hmac: bytes | None = None self.publicKey: bytes | None = None + self.session_info_key: bytes | None = None + self.last_response_counter: int | None = None + self._last_authenticated_clock_time: int | None = None self.lock: Lock = Lock() @property @@ -324,18 +329,59 @@ def ready(self) -> bool: self.epoch is not None and self.hmac is not None and self.delta is not None ) - def update(self, sessionInfo: SessionInfo): - """Update the session with new information""" + def keys_for(self, vehicle_public_key: bytes) -> tuple[bytes, bytes, bytes]: + """Derive (shared_key, command_hmac_key, session_info_key) for a candidate vehicle public key. - self.counter = sessionInfo.counter - self.epoch = sessionInfo.epoch - self.delta = int(time.time()) - sessionInfo.clock_time - if not self.ready or self.publicKey != sessionInfo.publicKey: - self.publicKey = sessionInfo.publicKey - self.sharedKey = self.parent.shared_key(sessionInfo.publicKey) - self.hmac = hmac.new( - self.sharedKey, "authenticated command".encode(), hashlib.sha256 - ).digest() + Pure and side-effect free - it does not touch any session state, so a + caller must authenticate a reply against the returned keys before + deciding whether to commit them via ``commit()``. + """ + shared_key = self.parent.shared_key(vehicle_public_key) + hmac_key = hmac.new( + shared_key, "authenticated command".encode(), hashlib.sha256 + ).digest() + session_info_key = hmac.new( + shared_key, "session info".encode(), hashlib.sha256 + ).digest() + return shared_key, hmac_key, session_info_key + + def commit( + self, + session_info: SessionInfo, + shared_key: bytes, + hmac_key: bytes, + session_info_key: bytes, + ) -> bool: + """Commit an already-authenticated SessionInfo plus its ``keys_for`` result. + + The anti-replay counter never rolls back within the same epoch, and a + clock time that regresses within the same epoch marks the reply as + stale/replayed - returns False without mutating any state so the + caller can discard it. + """ + same_epoch = self.epoch is not None and self.epoch == session_info.epoch + if ( + same_epoch + and self._last_authenticated_clock_time is not None + and session_info.clock_time < self._last_authenticated_clock_time + ): + return False + + self.counter = ( + max(self.counter, session_info.counter) + if same_epoch + else session_info.counter + ) + self.epoch = session_info.epoch + self.delta = int(time.time()) - session_info.clock_time + self._last_authenticated_clock_time = session_info.clock_time + self.publicKey = session_info.publicKey + self.sharedKey = shared_key + self.hmac = hmac_key + self.session_info_key = session_info_key + if not same_epoch: + self.last_response_counter = None + return True def hmac_personalized(self) -> HMAC_Personalized_Signature_Data: """Sign a command and return session metadata""" @@ -429,22 +475,87 @@ async def _send( """ raise NotImplementedError - def validate_msg(self, msg: RoutableMessage) -> None: - """Validate the message.""" + def validate_msg(self, msg: RoutableMessage, request_uuid: bytes) -> None: + """Validate the message. + + ``request_uuid`` is the uuid of the outstanding request ``msg`` + answers; it authenticates any piggy-backed ``session_info`` as the + TAG_CHALLENGE bound into that reply's HMAC tag before any field of it + is trusted. + """ if msg.session_info: - info = SessionInfo.FromString(msg.session_info) - if ( - info.status - == Session_Info_Status.SESSION_INFO_STATUS_KEY_NOT_ON_WHITELIST - ): - raise NotOnWhitelistFault - self._sessions[msg.from_destination.domain].update(info) + self._authenticate_session_info(msg, request_uuid) if msg.signedMessageStatus.signed_message_fault > 0: exception = MESSAGE_FAULTS[msg.signedMessageStatus.signed_message_fault] if exception: raise exception + def _authenticate_session_info( + self, msg: RoutableMessage, request_uuid: bytes + ) -> None: + """Authenticate and commit a piggy-backed ``SessionInfo`` reply. + + A ``SessionInfo`` is untrusted wire data until its ``session_info_tag`` + is verified: the tag is an HMAC over the exact bytes received, keyed + by a key derived from the *candidate* vehicle public key carried + inside that same ``SessionInfo``, and bound to this request's + ``request_uuid`` as a challenge - so a captured older reply can't be + replayed against a newer request. Only once that tag checks out do we + act on anything the message claims, including its own whitelist + status, and even then ``Session.commit`` still refuses a clock time + that regresses within the same epoch. + + VCSEC typically leaves the wire-level ``request_uuid`` field empty on + real hardware (memory constraints) - its absence must never be + treated as a rejection. Only cross-check the echo when the vehicle + chose to populate it. + """ + if msg.request_uuid and msg.request_uuid != request_uuid: + raise SessionInfoAuthenticationFault( + "Session info reply does not match an outstanding request." + ) + + session = self._sessions[msg.from_destination.domain] + info = SessionInfo.FromString(msg.session_info) + shared_key, hmac_key, session_info_key = session.keys_for(info.publicKey) + + tag = msg.signature_data.session_info_tag.tag + if not tag: + raise SessionInfoAuthenticationFault( + "Session info reply is missing its authentication tag." + ) + + metadata = bytes( + [ + Tag.TAG_SIGNATURE_TYPE, + 1, + SignatureType.SIGNATURE_TYPE_HMAC, + Tag.TAG_PERSONALIZATION, + 17, + *self.vin.encode(), + Tag.TAG_CHALLENGE, + len(request_uuid), + *request_uuid, + Tag.TAG_END, + ] + ) + expected_tag = hmac.new( + session_info_key, metadata + msg.session_info, hashlib.sha256 + ).digest() + if not hmac.compare_digest(expected_tag, tag): + raise SessionInfoAuthenticationFault( + "Session info reply failed authentication (invalid tag)." + ) + + if info.status == Session_Info_Status.SESSION_INFO_STATUS_KEY_NOT_ON_WHITELIST: + raise NotOnWhitelistFault + + if not session.commit(info, shared_key, hmac_key, session_info_key): + raise SessionInfoAuthenticationFault( + "Session info reply is stale (clock regressed within the same epoch)." + ) + async def _command( self, domain: Domain, @@ -537,6 +648,13 @@ async def _command( else: raise ValueError("Invalid request signature data") + response_counter = resp.signature_data.AES_GCM_Response_data.counter + if ( + session.last_response_counter is not None + and response_counter <= session.last_response_counter + ): + raise SignedCommandResponseReplayed + metadata = bytes( [ Tag.TAG_SIGNATURE_TYPE, @@ -579,6 +697,7 @@ async def _command( + resp.signature_data.AES_GCM_Response_data.tag, aad.finalize(), ) + session.last_response_counter = response_counter if resp.from_destination.domain == Domain.DOMAIN_VEHICLE_SECURITY: try: diff --git a/tesla_fleet_api/tesla/vehicle/signed.py b/tesla_fleet_api/tesla/vehicle/signed.py index 744a027..3e7b891 100644 --- a/tesla_fleet_api/tesla/vehicle/signed.py +++ b/tesla_fleet_api/tesla/vehicle/signed.py @@ -46,5 +46,5 @@ async def _send( base64.b64encode(msg.SerializeToString()).decode() ) resp = RoutableMessage.FromString(base64.b64decode(json["response"])) - self.validate_msg(resp) + self.validate_msg(resp, msg.uuid) return resp diff --git a/tests/test_session_info_authentication.py b/tests/test_session_info_authentication.py new file mode 100644 index 0000000..81ffccb --- /dev/null +++ b/tests/test_session_info_authentication.py @@ -0,0 +1,681 @@ +"""Tests for session_info authentication, counter monotonicity, and response +counter replay protection - the signed-command session-establishment path +that the BLE-mocked command tests never exercise (they pre-mark sessions +ready and mock ``_send`` wholesale, so a real ``session_info`` reply never +reaches ``validate_msg``/``Session.commit``). + +``GoldenVectorTests`` reproduces the handshake worked example published in +``teslamotors/vehicle-command``'s ``pkg/protocol/protocol.md`` verbatim, so it +is independent of every other test here: if it passes, this library's +session_info HMAC-tag construction matches the authoritative upstream spec +exactly, not just its own internal consistency. + +The remaining classes drive a synthetic-but-realistic "fake vehicle" keypair +through the real ``Commands.validate_msg``/``Session.commit`` and, for the +BLE transport, the real (unmocked) ``_send``/``_await_response``/``_handshake`` +state machine with only the GATT client faked - modeling VCSEC's real quirk +of leaving the wire-level ``request_uuid`` field empty. +""" + +from __future__ import annotations + +import hashlib +import hmac +import struct +import time +from typing import Any, cast +from unittest import IsolatedAsyncioTestCase +from unittest.mock import AsyncMock, MagicMock + +from cryptography.hazmat.primitives.asymmetric import ec +from cryptography.hazmat.primitives.ciphers.aead import AESGCM +from cryptography.hazmat.primitives.hashes import SHA256, Hash +from cryptography.hazmat.primitives.serialization import Encoding, PublicFormat + +from tesla_fleet_api.exceptions import ( + NotOnWhitelistFault, + SessionInfoAuthenticationFault, + SignedCommandResponseReplayed, +) +from tesla_fleet_api.tesla.vehicle.bluetooth import VehicleBluetooth +from tesla_fleet_api.tesla.vehicle.commands import Commands +from tesla_protocol.command.car_server_pb2 import ( + Action, + ActionStatus, + Ping, + Response, + VehicleAction, +) +from tesla_protocol.command.car_server_pb2 import ( + OperationStatus_E as InfotainmentOperationStatus_E, +) +from tesla_protocol.command.signatures_pb2 import ( + AES_GCM_Response_Signature_Data, + HMAC_Signature_Data, + Session_Info_Status, + SessionInfo, + SignatureData, + SignatureType, + Tag, +) +from tesla_protocol.command.universal_message_pb2 import ( + Destination, + Domain, + RoutableMessage, +) + +VIN = "5YJXCAE43LF123456" + + +class _ConcreteCommands(Commands): + """Minimal concrete subclass so ``Commands`` can be instantiated directly.""" + + _auth_method = "hmac" + _transport_name = "test" + + async def _send(self, msg, requires, expects_data=True, *, confirm_broadcast=None): + raise AssertionError("_send should not be called in these tests") + + +def _make_commands( + vin: str = VIN, private_key: ec.EllipticCurvePrivateKey | None = None +) -> _ConcreteCommands: + parent = MagicMock() + parent.private_key = private_key or ec.generate_private_key(ec.SECP256R1()) + return _ConcreteCommands(parent, vin) + + +def _public_key_bytes(private_key: ec.EllipticCurvePrivateKey) -> bytes: + return private_key.public_key().public_bytes( + encoding=Encoding.X962, format=PublicFormat.UncompressedPoint + ) + + +def _session_info_tag( + session_info_key: bytes, vin: str, request_uuid: bytes, session_info_bytes: bytes +) -> bytes: + """Independently compute the expected session_info_tag (mirrors commands.py).""" + metadata = bytes( + [ + Tag.TAG_SIGNATURE_TYPE, + 1, + SignatureType.SIGNATURE_TYPE_HMAC, + Tag.TAG_PERSONALIZATION, + 17, + *vin.encode(), + Tag.TAG_CHALLENGE, + len(request_uuid), + *request_uuid, + Tag.TAG_END, + ] + ) + return hmac.new( + session_info_key, metadata + session_info_bytes, hashlib.sha256 + ).digest() + + +def _session_info_reply( + commands: _ConcreteCommands, + domain: "Domain.ValueType", + session_info: SessionInfo, + request_uuid: bytes, + *, + echo_request_uuid: bytes = b"", + tag_override: bytes | None = None, + omit_tag: bool = False, +) -> RoutableMessage: + """Build a session_info reply with a correctly-derived tag, unless overridden.""" + session_info_bytes = session_info.SerializeToString() + _, _, session_info_key = commands._sessions[domain].keys_for(session_info.publicKey) + tag = ( + tag_override + if tag_override is not None + else _session_info_tag( + session_info_key, commands.vin, request_uuid, session_info_bytes + ) + ) + signature_data = SignatureData() + if not omit_tag: + signature_data.session_info_tag.CopyFrom(HMAC_Signature_Data(tag=tag)) + return RoutableMessage( + from_destination=Destination(domain=domain), + session_info=session_info_bytes, + request_uuid=echo_request_uuid, + signature_data=signature_data, + ) + + +class GoldenVectorTests(IsolatedAsyncioTestCase): + """Reproduces protocol.md's own published handshake worked example + (Test keys / Handshake / Response authentication sections) verbatim, + using its published test keys and intermediate values.""" + + CLIENT_PRIVATE_KEY = int( + "2538CDC29A97C19C1E99A637D6CF4F8C970C118B56EDE1E6323E6D162C4B30DB", 16 + ) + VEHICLE_PUBLIC_KEY = bytes.fromhex( + "04c7a1f47138486aa4729971494878d33b1a24e39571f748a6e16c5955b3d877d3" + "a6aaa0e955166474af5d32c410f439a2234137ad1bb085fd4e8813c958f11d97" + ) + VIN = "5YJ30123456789ABC" + REQUEST_UUID = bytes.fromhex("1588d5a30eabc6f8fc9a951b11f6fd11") + SESSION_INFO_BYTES = bytes.fromhex( + "0806124104c7a1f47138486aa4729971494878d33b1a24e39571f748a6e16c5955" + "b3d877d3a6aaa0e955166474af5d32c410f439a2234137ad1bb085fd4e8813c958" + "f11d971a104c463f9cc0d3d26906e982ed224adde6255a0a0000" + ) + GOLDEN_K = bytes.fromhex("1b2fce19967b79db696f909cff89ea9a") + GOLDEN_SESSION_INFO_KEY = bytes.fromhex( + "fceb679ee7bca756fcd441bf238bf2f338629b41d9eb9c67be1b32c9672ce300" + ) + GOLDEN_TAG = bytes.fromhex( + "996c1fe38331be138f8039c194b14db2198846ed7d8251e6749284d7b32ea002" + ) + + def _client_private_key(self) -> ec.EllipticCurvePrivateKey: + return ec.derive_private_key(self.CLIENT_PRIVATE_KEY, ec.SECP256R1()) + + def test_shared_key_matches_the_spec_published_k(self) -> None: + commands = _make_commands(self.VIN, self._client_private_key()) + self.assertEqual(commands.shared_key(self.VEHICLE_PUBLIC_KEY), self.GOLDEN_K) + + def test_session_info_key_matches_the_spec_published_value(self) -> None: + commands = _make_commands(self.VIN, self._client_private_key()) + domain = Domain.DOMAIN_INFOTAINMENT + _, _, session_info_key = commands._sessions[domain].keys_for( + self.VEHICLE_PUBLIC_KEY + ) + self.assertEqual(session_info_key, self.GOLDEN_SESSION_INFO_KEY) + + def test_authenticates_and_commits_the_spec_published_handshake_reply(self) -> None: + commands = _make_commands(self.VIN, self._client_private_key()) + domain = Domain.DOMAIN_INFOTAINMENT + session = commands._sessions[domain] + + reply = RoutableMessage( + from_destination=Destination(domain=domain), + session_info=self.SESSION_INFO_BYTES, + request_uuid=self.REQUEST_UUID, + signature_data=SignatureData( + session_info_tag=HMAC_Signature_Data(tag=self.GOLDEN_TAG) + ), + ) + + commands.validate_msg(reply, self.REQUEST_UUID) + + self.assertTrue(session.ready) + self.assertEqual(session.counter, 6) + self.assertEqual( + session.epoch, bytes.fromhex("4c463f9cc0d3d26906e982ed224adde6") + ) + + def test_a_single_flipped_tag_bit_fails_authentication(self) -> None: + commands = _make_commands(self.VIN, self._client_private_key()) + domain = Domain.DOMAIN_INFOTAINMENT + corrupted_tag = bytes([self.GOLDEN_TAG[0] ^ 0x01]) + self.GOLDEN_TAG[1:] + + reply = RoutableMessage( + from_destination=Destination(domain=domain), + session_info=self.SESSION_INFO_BYTES, + request_uuid=self.REQUEST_UUID, + signature_data=SignatureData( + session_info_tag=HMAC_Signature_Data(tag=corrupted_tag) + ), + ) + + with self.assertRaises(SessionInfoAuthenticationFault): + commands.validate_msg(reply, self.REQUEST_UUID) + self.assertFalse(commands._sessions[domain].ready) + + +class SessionInfoTagAuthenticationTests(IsolatedAsyncioTestCase): + """Unit tests against a synthetic fake-vehicle keypair (not spec-pinned).""" + + def setUp(self) -> None: + self.commands = _make_commands() + self.domain = Domain.DOMAIN_VEHICLE_SECURITY + self.vehicle_key = ec.generate_private_key(ec.SECP256R1()) + self.vehicle_public_key = _public_key_bytes(self.vehicle_key) + self.request_uuid = b"\x11" * 16 + + def _session_info(self, **overrides: Any) -> SessionInfo: + kwargs: dict[str, Any] = dict( + counter=100, + publicKey=self.vehicle_public_key, + epoch=b"\x01" * 16, + clock_time=int(time.time()), + status=Session_Info_Status.SESSION_INFO_STATUS_OK, + ) + kwargs.update(overrides) + return SessionInfo(**kwargs) + + def test_rejects_reply_with_no_tag(self) -> None: + reply = _session_info_reply( + self.commands, + self.domain, + self._session_info(), + self.request_uuid, + omit_tag=True, + ) + with self.assertRaises(SessionInfoAuthenticationFault): + self.commands.validate_msg(reply, self.request_uuid) + self.assertFalse(self.commands._sessions[self.domain].ready) + + def test_rejects_reply_with_forged_tag(self) -> None: + reply = _session_info_reply( + self.commands, + self.domain, + self._session_info(), + self.request_uuid, + tag_override=b"\x00" * 32, + ) + with self.assertRaises(SessionInfoAuthenticationFault): + self.commands.validate_msg(reply, self.request_uuid) + self.assertFalse(self.commands._sessions[self.domain].ready) + + def test_rejects_reply_authenticated_against_a_different_request(self) -> None: + info = self._session_info() + # Tag is computed for a request we never sent - proves the tag, not + # just presence, is what's checked. + reply = _session_info_reply(self.commands, self.domain, info, b"\x99" * 16) + with self.assertRaises(SessionInfoAuthenticationFault): + self.commands.validate_msg(reply, self.request_uuid) + self.assertFalse(self.commands._sessions[self.domain].ready) + + def test_accepts_valid_tag_with_empty_wire_level_request_uuid(self) -> None: + # The real VCSEC quirk (protocol.md: "request_uuid is typically not + # populated"): the tag is still bound to our sent request_uuid as + # TAG_CHALLENGE, so authentication does not depend on the echo field + # at all. Must never regress into gating acceptance on this field. + reply = _session_info_reply( + self.commands, + self.domain, + self._session_info(), + self.request_uuid, + echo_request_uuid=b"", + ) + self.commands.validate_msg(reply, self.request_uuid) + session = self.commands._sessions[self.domain] + self.assertTrue(session.ready) + self.assertEqual(session.counter, 100) + + def test_accepts_valid_tag_with_matching_echoed_request_uuid(self) -> None: + reply = _session_info_reply( + self.commands, + self.domain, + self._session_info(), + self.request_uuid, + echo_request_uuid=self.request_uuid, + ) + self.commands.validate_msg(reply, self.request_uuid) + self.assertTrue(self.commands._sessions[self.domain].ready) + + def test_tag_is_verified_before_whitelist_status_is_trusted(self) -> None: + # A forged whitelist-rejection with a bad tag must surface as an + # authentication failure, not NotOnWhitelistFault - proving the tag + # check runs strictly before any field of the reply is acted on. + info = self._session_info( + status=Session_Info_Status.SESSION_INFO_STATUS_KEY_NOT_ON_WHITELIST + ) + reply = _session_info_reply( + self.commands, + self.domain, + info, + self.request_uuid, + tag_override=b"\x00" * 32, + ) + with self.assertRaises(SessionInfoAuthenticationFault): + self.commands.validate_msg(reply, self.request_uuid) + + def test_authenticated_whitelist_rejection_still_raises_not_on_whitelist( + self, + ) -> None: + info = self._session_info( + status=Session_Info_Status.SESSION_INFO_STATUS_KEY_NOT_ON_WHITELIST + ) + reply = _session_info_reply(self.commands, self.domain, info, self.request_uuid) + with self.assertRaises(NotOnWhitelistFault): + self.commands.validate_msg(reply, self.request_uuid) + self.assertFalse(self.commands._sessions[self.domain].ready) + + +class CounterAndClockMonotonicityTests(IsolatedAsyncioTestCase): + """Covers commands.py's Session.commit: clamp within an epoch, refuse a + same-epoch clock regression, reset only on an epoch change.""" + + def setUp(self) -> None: + self.commands = _make_commands() + self.domain = Domain.DOMAIN_VEHICLE_SECURITY + self.vehicle_key = ec.generate_private_key(ec.SECP256R1()) + self.vehicle_public_key = _public_key_bytes(self.vehicle_key) + self.session = self.commands._sessions[self.domain] + + def _authenticate( + self, *, counter: int, epoch: bytes, clock_time: int, request_uuid: bytes + ) -> None: + info = SessionInfo( + counter=counter, + publicKey=self.vehicle_public_key, + epoch=epoch, + clock_time=clock_time, + status=Session_Info_Status.SESSION_INFO_STATUS_OK, + ) + reply = _session_info_reply(self.commands, self.domain, info, request_uuid) + self.commands.validate_msg(reply, request_uuid) + + def test_baseline_establishes_counter_and_epoch(self) -> None: + self._authenticate( + counter=100, epoch=b"\x01" * 16, clock_time=1_000, request_uuid=b"\x01" * 16 + ) + self.assertEqual(self.session.counter, 100) + self.assertEqual(self.session.epoch, b"\x01" * 16) + + def test_same_epoch_lower_counter_and_regressed_clock_is_refused(self) -> None: + self._authenticate( + counter=100, epoch=b"\x01" * 16, clock_time=1_000, request_uuid=b"\x01" * 16 + ) + + with self.assertRaises(SessionInfoAuthenticationFault): + self._authenticate( + counter=5, epoch=b"\x01" * 16, clock_time=900, request_uuid=b"\x02" * 16 + ) + + # Neither the counter nor the delta must have moved. + self.assertEqual(self.session.counter, 100) + + def test_same_epoch_lower_counter_with_non_regressed_clock_clamps_not_rolls_back( + self, + ) -> None: + self._authenticate( + counter=100, epoch=b"\x01" * 16, clock_time=1_000, request_uuid=b"\x01" * 16 + ) + + # Clock moves forward (not a regression) but counter is lower - must + # clamp to the existing high-water mark, not adopt the lower value. + self._authenticate( + counter=5, epoch=b"\x01" * 16, clock_time=1_010, request_uuid=b"\x02" * 16 + ) + + self.assertEqual(self.session.counter, 100) + + def test_same_epoch_higher_counter_advances_normally(self) -> None: + self._authenticate( + counter=100, epoch=b"\x01" * 16, clock_time=1_000, request_uuid=b"\x01" * 16 + ) + self._authenticate( + counter=150, epoch=b"\x01" * 16, clock_time=1_010, request_uuid=b"\x02" * 16 + ) + self.assertEqual(self.session.counter, 150) + + def test_epoch_change_resets_counter_even_if_lower(self) -> None: + self._authenticate( + counter=100, epoch=b"\x01" * 16, clock_time=1_000, request_uuid=b"\x01" * 16 + ) + self._authenticate( + counter=3, epoch=b"\x02" * 16, clock_time=1, request_uuid=b"\x02" * 16 + ) + self.assertEqual(self.session.counter, 3) + self.assertEqual(self.session.epoch, b"\x02" * 16) + + +class ResponseCounterReplayTests(IsolatedAsyncioTestCase): + """Covers the AES-GCM response-counter dedup check (protocol.md + "Counter verification"): a response reusing an already-seen counter for + this session must be rejected, defending against a captured encrypted + response being replayed back at the client.""" + + def _make_aes_commands(self) -> tuple[Any, Any]: + class _AesCommands(Commands): + _auth_method = "aes" + _transport_name = "test" + + def __init__(self, *args: Any, **kwargs: Any) -> None: + super().__init__(*args, **kwargs) + self.next_reply: RoutableMessage | None = None + + async def _send( + self, msg, requires, expects_data=True, *, confirm_broadcast=None + ): + assert self.next_reply is not None + return self.next_reply + + parent = MagicMock() + parent.private_key = ec.generate_private_key(ec.SECP256R1()) + commands = _AesCommands(parent, VIN) + session = commands._sessions[Domain.DOMAIN_INFOTAINMENT] + session.epoch = b"\x00" * 16 + session.hmac = b"\x00" * 32 + session.delta = 0 + session.sharedKey = b"\x42" * 16 + return commands, session + + def _encrypted_reply( + self, commands: Any, session: Any, sent_msg: RoutableMessage, counter: int + ) -> RoutableMessage: + domain = Domain.DOMAIN_INFOTAINMENT + body = Response( + actionStatus=ActionStatus( + result=InfotainmentOperationStatus_E.OPERATIONSTATUS_OK + ) + ).SerializeToString() + + request_hash = ( + bytes([SignatureType.SIGNATURE_TYPE_AES_GCM_PERSONALIZED]) + + sent_msg.signature_data.AES_GCM_Personalized_data.tag + ) + flags = 0 + metadata = bytes( + [ + Tag.TAG_SIGNATURE_TYPE, + 1, + SignatureType.SIGNATURE_TYPE_AES_GCM_RESPONSE, + Tag.TAG_DOMAIN, + 1, + domain, + Tag.TAG_PERSONALIZATION, + 17, + *commands.vin.encode(), + Tag.TAG_COUNTER, + 4, + *struct.pack(">I", counter), + Tag.TAG_FLAGS, + 4, + *struct.pack(">I", flags), + Tag.TAG_REQUEST_HASH, + 17, + *request_hash, + Tag.TAG_FAULT, + 4, + *struct.pack(">I", 0), + Tag.TAG_END, + ] + ) + aad = Hash(SHA256()) + aad.update(metadata) + nonce = b"\x07" * 12 + aesgcm = AESGCM(session.sharedKey) + ct = aesgcm.encrypt(nonce, body, aad.finalize()) + return RoutableMessage( + from_destination=Destination(domain=domain), + protobuf_message_as_bytes=ct[:-16], + flags=flags, + signature_data=SignatureData( + AES_GCM_Response_data=AES_GCM_Response_Signature_Data( + nonce=nonce, counter=counter, tag=ct[-16:] + ) + ), + ) + + async def test_first_response_of_a_session_is_accepted(self) -> None: + commands, session = self._make_aes_commands() + sent: list[RoutableMessage] = [] + + async def capture_send( + msg, requires, expects_data=True, *, confirm_broadcast=None + ): + sent.append(msg) + return self._encrypted_reply(commands, session, msg, counter=6) + + commands._send = capture_send # type: ignore[method-assign] + + result = await commands._sendInfotainment( + Action(vehicleAction=VehicleAction(ping=Ping(ping_id=0))) + ) + self.assertTrue(result["response"]["result"]) + self.assertEqual(session.last_response_counter, 6) + + async def test_replayed_response_counter_is_rejected(self) -> None: + commands, session = self._make_aes_commands() + + async def replay_send( + msg, requires, expects_data=True, *, confirm_broadcast=None + ): + # Always answers with the same counter, simulating a captured + # response being replayed back regardless of the new request. + return self._encrypted_reply(commands, session, msg, counter=6) + + commands._send = replay_send # type: ignore[method-assign] + + await commands._sendInfotainment( + Action(vehicleAction=VehicleAction(ping=Ping(ping_id=0))) + ) + self.assertEqual(session.last_response_counter, 6) + + with self.assertRaises(SignedCommandResponseReplayed): + await commands._sendInfotainment( + Action(vehicleAction=VehicleAction(ping=Ping(ping_id=0))) + ) + + async def test_increasing_counters_across_commands_are_accepted(self) -> None: + commands, session = self._make_aes_commands() + counters = iter([6, 9]) + + async def increasing_send( + msg, requires, expects_data=True, *, confirm_broadcast=None + ): + return self._encrypted_reply(commands, session, msg, counter=next(counters)) + + commands._send = increasing_send # type: ignore[method-assign] + + await commands._sendInfotainment( + Action(vehicleAction=VehicleAction(ping=Ping(ping_id=0))) + ) + await commands._sendInfotainment( + Action(vehicleAction=VehicleAction(ping=Ping(ping_id=0))) + ) + self.assertEqual(session.last_response_counter, 9) + + +def _make_ble_vehicle() -> VehicleBluetooth[Any]: + """A VehicleBluetooth with the real handshake/_send state machine but a faked GATT client.""" + parent = MagicMock() + parent.private_key = ec.generate_private_key(ec.SECP256R1()) + vehicle = VehicleBluetooth(parent, VIN) + vehicle.connect_if_needed = AsyncMock() # type: ignore[method-assign] + vehicle.client = MagicMock() + vehicle.client.write_gatt_char = AsyncMock() + return vehicle + + +class BleHandshakeAuthenticationTests(IsolatedAsyncioTestCase): + """End-to-end through the real (unmocked) BLE handshake/_send/_await_response + state machine - only the GATT client is faked - proving the request_uuid + threading and validate_msg wiring, not just the isolated auth logic.""" + + async def test_forged_session_info_is_rejected_and_session_stays_not_ready( + self, + ) -> None: + vehicle = _make_ble_vehicle() + vehicle._default_timeout = 1.0 + domain = Domain.DOMAIN_VEHICLE_SECURITY + attacker_key = ec.generate_private_key(ec.SECP256R1()) + attacker_public_key = _public_key_bytes(attacker_key) + + async def deliver_forged_reply(*_: Any) -> None: + forged_info = SessionInfo( + counter=1, + publicKey=attacker_public_key, + epoch=b"\x09" * 16, + clock_time=int(time.time()), + status=Session_Info_Status.SESSION_INFO_STATUS_OK, + ) + reply = RoutableMessage( + to_destination=Destination( + domain=domain, routing_address=vehicle._from_destination + ), + from_destination=Destination(domain=domain), + session_info=forged_info.SerializeToString(), + signature_data=SignatureData( + session_info_tag=HMAC_Signature_Data(tag=b"\x00" * 32) + ), + ) + vehicle._on_message(reply) + + vehicle.client.write_gatt_char = AsyncMock(side_effect=deliver_forged_reply) + + with self.assertRaises(SessionInfoAuthenticationFault): + await vehicle.handshakeVehicleSecurity() + + self.assertFalse(cast("dict[Any, Any]", vehicle._sessions)[domain].ready) + + async def test_valid_handshake_with_empty_wire_request_uuid_succeeds(self) -> None: + # Real VCSEC leaves the wire-level request_uuid empty; the handshake + # must still complete via the session_info_tag alone. + vehicle = _make_ble_vehicle() + vehicle._default_timeout = 1.0 + domain = Domain.DOMAIN_VEHICLE_SECURITY + vehicle_key = ec.generate_private_key(ec.SECP256R1()) + vehicle_public_key = _public_key_bytes(vehicle_key) + + async def deliver_valid_reply(*_: Any) -> None: + sessions = cast("dict[Any, Any]", vehicle._sessions) + session = sessions[domain] + info = SessionInfo( + counter=6, + publicKey=vehicle_public_key, + epoch=b"\x0a" * 16, + clock_time=int(time.time()), + status=Session_Info_Status.SESSION_INFO_STATUS_OK, + ) + info_bytes = info.SerializeToString() + _, _, session_info_key = session.keys_for(vehicle_public_key) + # The real request uuid the client just sent is only observable + # via the outgoing GATT write; the handshake message's uuid is + # random per-call, so recover it from the session's in-flight + # queue state is unnecessary - VCSEC's own quirk of never echoing + # it back means the client authenticates purely off the tag, so + # any tag computed against *some* uuid should fail unless it's + # the one actually sent. We fetch it from the write payload. + sent_uuid = _extract_request_uuid(vehicle.client.write_gatt_char) + tag = _session_info_tag( + session_info_key, vehicle.vin, sent_uuid, info_bytes + ) + reply = RoutableMessage( + to_destination=Destination( + domain=domain, routing_address=vehicle._from_destination + ), + from_destination=Destination(domain=domain), + session_info=info_bytes, + request_uuid=b"", + signature_data=SignatureData( + session_info_tag=HMAC_Signature_Data(tag=tag) + ), + ) + vehicle._on_message(reply) + + vehicle.client.write_gatt_char = AsyncMock(side_effect=deliver_valid_reply) + + await vehicle.handshakeVehicleSecurity() + + sessions = cast("dict[Any, Any]", vehicle._sessions) + self.assertTrue(sessions[domain].ready) + self.assertEqual(sessions[domain].counter, 6) + + +def _extract_request_uuid(write_mock: AsyncMock) -> bytes: + """Recover the request uuid from the last GATT write this test's handshake sent.""" + payload = bytes(write_mock.call_args.args[1]) + # payload = 2-byte big-endian length prefix followed by the serialized + # RoutableMessage (see bluetooth.py's prependLength). + msg = RoutableMessage.FromString(payload[2:]) + return msg.uuid From 3661fec17879303cb43a8b0e1f313baf3d9ff5bd Mon Sep 17 00:00:00 2001 From: firstmate crewmate Date: Thu, 23 Jul 2026 14:26:07 +1000 Subject: [PATCH 2/4] no-mistakes(review): Scope response replay detection per request --- tesla_fleet_api/tesla/vehicle/commands.py | 14 ++++---- tests/test_session_info_authentication.py | 44 +++++++++++++++-------- 2 files changed, 35 insertions(+), 23 deletions(-) diff --git a/tesla_fleet_api/tesla/vehicle/commands.py b/tesla_fleet_api/tesla/vehicle/commands.py index d6c853c..be1b9be 100644 --- a/tesla_fleet_api/tesla/vehicle/commands.py +++ b/tesla_fleet_api/tesla/vehicle/commands.py @@ -318,7 +318,7 @@ def __init__(self, parent: Commands[CommandParentT], domain: Domain): self.hmac: bytes | None = None self.publicKey: bytes | None = None self.session_info_key: bytes | None = None - self.last_response_counter: int | None = None + self.response_counters: set[tuple[bytes, int]] = set() self._last_authenticated_clock_time: int | None = None self.lock: Lock = Lock() @@ -380,7 +380,7 @@ def commit( self.hmac = hmac_key self.session_info_key = session_info_key if not same_epoch: - self.last_response_counter = None + self.response_counters.clear() return True def hmac_personalized(self) -> HMAC_Personalized_Signature_Data: @@ -649,11 +649,6 @@ async def _command( raise ValueError("Invalid request signature data") response_counter = resp.signature_data.AES_GCM_Response_data.counter - if ( - session.last_response_counter is not None - and response_counter <= session.last_response_counter - ): - raise SignedCommandResponseReplayed metadata = bytes( [ @@ -697,7 +692,10 @@ async def _command( + resp.signature_data.AES_GCM_Response_data.tag, aad.finalize(), ) - session.last_response_counter = response_counter + response_identity = (request_hash, response_counter) + if response_identity in session.response_counters: + raise SignedCommandResponseReplayed + session.response_counters.add(response_identity) if resp.from_destination.domain == Domain.DOMAIN_VEHICLE_SECURITY: try: diff --git a/tests/test_session_info_authentication.py b/tests/test_session_info_authentication.py index 81ffccb..013b9d1 100644 --- a/tests/test_session_info_authentication.py +++ b/tests/test_session_info_authentication.py @@ -522,40 +522,54 @@ async def capture_send( Action(vehicleAction=VehicleAction(ping=Ping(ping_id=0))) ) self.assertTrue(result["response"]["result"]) - self.assertEqual(session.last_response_counter, 6) + self.assertEqual(len(session.response_counters), 1) - async def test_replayed_response_counter_is_rejected(self) -> None: + async def test_response_counter_can_repeat_for_a_different_request(self) -> None: commands, session = self._make_aes_commands() - async def replay_send( + async def repeated_counter_send( msg, requires, expects_data=True, *, confirm_broadcast=None ): - # Always answers with the same counter, simulating a captured - # response being replayed back regardless of the new request. return self._encrypted_reply(commands, session, msg, counter=6) - commands._send = replay_send # type: ignore[method-assign] + commands._send = repeated_counter_send # type: ignore[method-assign] await commands._sendInfotainment( Action(vehicleAction=VehicleAction(ping=Ping(ping_id=0))) ) - self.assertEqual(session.last_response_counter, 6) + await commands._sendInfotainment( + Action(vehicleAction=VehicleAction(ping=Ping(ping_id=0))) + ) + self.assertEqual(len(session.response_counters), 2) + + async def test_replayed_counter_for_the_same_request_is_rejected(self) -> None: + commands, session = self._make_aes_commands() + sent: list[RoutableMessage] = [] + async def capture_send( + msg, requires, expects_data=True, *, confirm_broadcast=None + ): + sent.append(msg) + return self._encrypted_reply(commands, session, msg, counter=6) + + commands._send = capture_send # type: ignore[method-assign] + command = Action(vehicleAction=VehicleAction(ping=Ping(ping_id=0))) + await commands._sendInfotainment(command) + + commands._commandAes = AsyncMock(return_value=sent[0]) # type: ignore[method-assign] with self.assertRaises(SignedCommandResponseReplayed): - await commands._sendInfotainment( - Action(vehicleAction=VehicleAction(ping=Ping(ping_id=0))) - ) + await commands._sendInfotainment(command) - async def test_increasing_counters_across_commands_are_accepted(self) -> None: + async def test_decreasing_unused_counter_is_accepted(self) -> None: commands, session = self._make_aes_commands() - counters = iter([6, 9]) + counters = iter([9, 8]) - async def increasing_send( + async def decreasing_send( msg, requires, expects_data=True, *, confirm_broadcast=None ): return self._encrypted_reply(commands, session, msg, counter=next(counters)) - commands._send = increasing_send # type: ignore[method-assign] + commands._send = decreasing_send # type: ignore[method-assign] await commands._sendInfotainment( Action(vehicleAction=VehicleAction(ping=Ping(ping_id=0))) @@ -563,7 +577,7 @@ async def increasing_send( await commands._sendInfotainment( Action(vehicleAction=VehicleAction(ping=Ping(ping_id=0))) ) - self.assertEqual(session.last_response_counter, 9) + self.assertEqual({counter for _, counter in session.response_counters}, {8, 9}) def _make_ble_vehicle() -> VehicleBluetooth[Any]: From b8bce43f7b94485a5916a87dfb5a423fd28484d8 Mon Sep 17 00:00:00 2001 From: firstmate crewmate Date: Thu, 23 Jul 2026 14:28:27 +1000 Subject: [PATCH 3/4] no-mistakes(review): Bound authenticated response replay cache --- tesla_fleet_api/tesla/vehicle/commands.py | 21 +++++++++++++++++---- tests/test_session_info_authentication.py | 13 +++++++++++++ 2 files changed, 30 insertions(+), 4 deletions(-) diff --git a/tesla_fleet_api/tesla/vehicle/commands.py b/tesla_fleet_api/tesla/vehicle/commands.py index be1b9be..d740c16 100644 --- a/tesla_fleet_api/tesla/vehicle/commands.py +++ b/tesla_fleet_api/tesla/vehicle/commands.py @@ -2,6 +2,7 @@ from abc import ABC, abstractmethod import struct +from collections import deque from random import randbytes from typing import ( Any, @@ -308,6 +309,8 @@ def _log_command_error(name: str, transport: str, exc: BaseException) -> None: class Session(Generic[CommandParentT]): """A connect to a domain""" + _response_counter_cache_size = 256 + def __init__(self, parent: Commands[CommandParentT], domain: Domain): self.parent: Commands[CommandParentT] = parent self.domain: Domain = domain @@ -318,7 +321,9 @@ def __init__(self, parent: Commands[CommandParentT], domain: Domain): self.hmac: bytes | None = None self.publicKey: bytes | None = None self.session_info_key: bytes | None = None - self.response_counters: set[tuple[bytes, int]] = set() + self.response_counters: deque[tuple[bytes, int]] = deque( + maxlen=self._response_counter_cache_size + ) self._last_authenticated_clock_time: int | None = None self.lock: Lock = Lock() @@ -383,6 +388,14 @@ def commit( self.response_counters.clear() return True + def record_response_counter(self, request_hash: bytes, counter: int) -> bool: + """Record an authenticated response counter if it has not been seen.""" + identity = (request_hash, counter) + if identity in self.response_counters: + return False + self.response_counters.append(identity) + return True + def hmac_personalized(self) -> HMAC_Personalized_Signature_Data: """Sign a command and return session metadata""" assert self.delta is not None @@ -692,10 +705,10 @@ async def _command( + resp.signature_data.AES_GCM_Response_data.tag, aad.finalize(), ) - response_identity = (request_hash, response_counter) - if response_identity in session.response_counters: + if not session.record_response_counter( + request_hash, response_counter + ): raise SignedCommandResponseReplayed - session.response_counters.add(response_identity) if resp.from_destination.domain == Domain.DOMAIN_VEHICLE_SECURITY: try: diff --git a/tests/test_session_info_authentication.py b/tests/test_session_info_authentication.py index 013b9d1..9447650 100644 --- a/tests/test_session_info_authentication.py +++ b/tests/test_session_info_authentication.py @@ -579,6 +579,19 @@ async def decreasing_send( ) self.assertEqual({counter for _, counter in session.response_counters}, {8, 9}) + def test_response_counter_cache_is_bounded(self) -> None: + _, session = self._make_aes_commands() + + for counter in range(session._response_counter_cache_size + 1): + self.assertTrue( + session.record_response_counter(counter.to_bytes(2), counter) + ) + + self.assertEqual( + len(session.response_counters), session._response_counter_cache_size + ) + self.assertNotIn((b"\x00\x00", 0), session.response_counters) + def _make_ble_vehicle() -> VehicleBluetooth[Any]: """A VehicleBluetooth with the real handshake/_send state machine but a faked GATT client.""" From 97740b955fcb5b149fb98377adfef9233c357490 Mon Sep 17 00:00:00 2001 From: firstmate crewmate Date: Thu, 23 Jul 2026 14:31:46 +1000 Subject: [PATCH 4/4] no-mistakes(document): Remove stale guidance and format signed commands --- AGENTS.md | 2 -- tesla_fleet_api/tesla/vehicle/commands.py | 4 +--- 2 files changed, 1 insertion(+), 5 deletions(-) diff --git a/AGENTS.md b/AGENTS.md index 40ad0ab..23b4e89 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -147,8 +147,6 @@ Keep the `tesla-protocol` floor at `>=0.5.0`; earlier releases have generated `. - **Typed accessor pattern for undocumented raw-dict responses**: `TeslemetryEnergySite.find_authorized_clients()` (`teslemetry/energysite.py`) is the library's first frozen-dataclass typed wrapper over a raw `dict[str, Any]`/`None`/`list` REST response, added so API-parsing logic (envelope unwrap, field lookup, shape validation, enum typing) lives in the library instead of each consumer (e.g. Home Assistant's config flow) reimplementing it. `_authorized_clients_list()` does one precise, non-recursive envelope unwrap (`{"response": {"authorized_clients": [...]}}` or `{"response": {"clients": [...]}}`, or a bare list with no envelope) rather than a multi-key/depth-first search - the `clients` key variant was added after a live capture (Tesla Release 953) showed the endpoint's real key differs from the originally-documented `authorized_clients`, confirmed against a populated 5-entry sample. `AuthorizedClient` models only the two fields a pairing flow actually reads (`public_key`, `state`), each accepting only the specific key-name variant pairs empirically evidenced, not speculative extras. Two rules any future typed accessor over an undocumented response shape must keep, demonstrated here even though `AuthorizedClientState` (`const.py`) has no falsy member: (1) field lookup must check key presence (`key in payload`), never `payload.get(key) or default` - a legal falsy value is not "missing", and a present-but-unrecognized enum value (`_normalize_state()`) is returned raw rather than coerced to `None`; (2) a `None` body and an unrecognized response shape (not a dict/list, or an envelope that unwraps to neither accepted list key) are malformed data, not "zero clients" - Tesla's endpoint intermittently returns HTTP 200 with a null body, so `_authorized_clients_list()` raises `InvalidResponse` (`exceptions.py`) for both rather than collapsing them to `[]`; only a genuinely empty list under either accepted key parses to `AuthorizedClients.clients == []` without raising. Tesla has not published an OpenAPI schema for pairing endpoints, so `const.py`'s enums are the schema of record; widen `AuthorizedClient`'s modeled fields only against a further live sample, not speculatively. The raw `list_authorized_clients()` method is kept alongside as the untyped escape hatch and still returns a null body unchanged rather than raising. Tests: `tests/test_teslemetry_authorized_clients.py`. - **`networking_status`'s `ipv4_config` fields are raw big-endian uint32 ints, not strings**: the wire format applies to all of `ipv4_config.address`/`subnet_mask`/`gateway`, but `TeslemetryEnergySite.find_gateway_address()` (`teslemetry/energysite.py`, second typed accessor after `find_authorized_clients()`, same rules) decodes only `address` - confirmed against a live Powerwall 3 capture where `3232235914` decodes network-byte-order (`struct.pack(">I", ...)`) to `192.168.1.138`, not little-endian. It considers only `eth`/`wifi` (never `gsm` - cellular isn't a LAN path), preferring whichever has `active_route` set and a decodable address, else falling back to the first of the two (in that order) with any decodable address; `0`/`0xFFFFFFFF` are treated as undecodable so an unconfigured interface never shadows a real one with `0.0.0.0`. A `{"response": null}` envelope raises `InvalidResponse` (the endpoint's known intermittent malformed mode), while a well-formed response with no usable interface returns `None` (not a raise). Tests: `tests/test_teslemetry_gateway_address.py`. - **`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`, needs a persistent INFOTAINMENT-domain broadcast dispatcher like `broadcast.py`'s VCSEC listeners) and `getVehicleImageState` (needs chunked binary-transfer paging) - both are separate, unscoped design work, not oversights. CarServer's `GetVehicleState` sub-state is exposed as `legacy_vehicle_state()` (`bluetooth.py`), matching the `VehicleData.legacy_vehicle_state` reply field name, specifically to avoid confusion with the pre-existing `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. `pii_key_request`/`pseudonym_sync_request`/`tesla_auth_response`/`setup_cloud_profile_with_local_profile_uuid`/`get_local_profiles_for_vault_uuid` are wrapped with no known third-party consumer use case, purely for full-proto-coverage completeness. -- **`session_info` replies are authenticated before any field is trusted**: `Commands._authenticate_session_info`/`Session.commit` (`commands.py`) verify the reply's `session_info_tag` HMAC - keyed by `HMAC(K, "session info")`, bound to the `request_uuid` the client actually sent as `TAG_CHALLENGE` - before adopting `counter`/`epoch`/`clock_time`/`publicKey` or even reading the whitelist-status field; a bad/absent tag raises `SessionInfoAuthenticationFault` (`exceptions.py`) without mutating session state. `validate_msg` takes the sent request's `uuid` as a required second argument for this reason - every `_send` implementation must thread it through. The wire-level `request_uuid` echo field is a red herring, not authentication: VCSEC typically leaves it empty (memory constraints), so acceptance must never gate on it - only cross-check it when the vehicle populates it. Within an epoch the counter clamps to `max(current, incoming)` and a clock regression is refused (reset only happens on an epoch change); the AES-GCM response path additionally rejects a replayed response counter via `Session.last_response_counter`. `tests/test_session_info_authentication.py`'s `GoldenVectorTests` pins the tag/key derivation against protocol.md's own published test vectors verbatim; its BLE tests drive the real (unmocked) handshake state machine, unlike the BLE-mocked command tests which pre-mark sessions ready and never exercise this path. - ## Maintaining this file Keep this file for knowledge useful to almost every future agent session in this project. diff --git a/tesla_fleet_api/tesla/vehicle/commands.py b/tesla_fleet_api/tesla/vehicle/commands.py index d740c16..7257ff0 100644 --- a/tesla_fleet_api/tesla/vehicle/commands.py +++ b/tesla_fleet_api/tesla/vehicle/commands.py @@ -705,9 +705,7 @@ async def _command( + resp.signature_data.AES_GCM_Response_data.tag, aad.finalize(), ) - if not session.record_response_counter( - request_hash, response_counter - ): + if not session.record_response_counter(request_hash, response_counter): raise SignedCommandResponseReplayed if resp.from_destination.domain == Domain.DOMAIN_VEHICLE_SECURITY: