fix(vehicle): harden signed-command session authentication

AGENTS.md

@@ -147,7 +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.
-
 ## Maintaining this file
 
 Keep this file for knowledge useful to almost every future agent session in this project.

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"""
 

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:

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,
@@ -26,6 +27,8 @@
     MESSAGE_FAULTS,
     SIGNED_MESSAGE_INFORMATION_FAULTS,
     NotOnWhitelistFault,
+    SessionInfoAuthenticationFault,
+    SignedCommandResponseReplayed,
     TeslaFleetError,
     # TeslaFleetMessageFaultInvalidSignature,
     TeslaFleetMessageFaultIncorrectEpoch,
@@ -306,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
@@ -315,6 +320,11 @@ 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.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()
 
     @property
@@ -324,18 +334,67 @@ 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.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"""
@@ -429,22 +488,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 +661,8 @@ async def _command(
                 else:
                     raise ValueError("Invalid request signature data")
 
+                response_counter = resp.signature_data.AES_GCM_Response_data.counter
+
                 metadata = bytes(
                     [
                         Tag.TAG_SIGNATURE_TYPE,
@@ -579,6 +705,8 @@ async def _command(
                     + resp.signature_data.AES_GCM_Response_data.tag,
                     aad.finalize(),
                 )
+                if not session.record_response_counter(request_hash, response_counter):
+                    raise SignedCommandResponseReplayed
 
             if resp.from_destination.domain == Domain.DOMAIN_VEHICLE_SECURITY:
                 try:

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