From 47fbbf71af06ff9767e15f3e62908f3e716b100f Mon Sep 17 00:00:00 2001 From: Brett Adams Date: Thu, 13 Aug 2026 16:12:41 +1000 Subject: [PATCH] feat(teslemetry): add OAuth dynamic client registration helper Adds register_client()/TeslemetryClientRegistration/TeslemetryRegistrationError so the Home Assistant Teslemetry integration's client-registration transport (RFC 7591 DCR: HTTP POST, response parsing, error handling) can move out of its own oauth.py and into this library. --- AGENTS.md | 1 + docs/teslemetry.md | 38 ++++++ tesla_fleet_api/__init__.py | 8 +- tesla_fleet_api/exceptions.py | 12 ++ tesla_fleet_api/teslemetry/__init__.py | 8 +- tesla_fleet_api/teslemetry/teslemetry.py | 83 ++++++++++++- tests/test_teslemetry_register_client.py | 151 +++++++++++++++++++++++ 7 files changed, 298 insertions(+), 3 deletions(-) create mode 100644 tests/test_teslemetry_register_client.py diff --git a/AGENTS.md b/AGENTS.md index 49ec4fa..38585b9 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -150,6 +150,7 @@ Keep the `tesla-protocol` floor at `>=0.5.0`; earlier releases have generated `. - **`_stream_sinks` peels subscription pushes off the command-reply queue before routing**: a `vehicleDataSubscription`'s pushes arrive addressed to us on the same domain queue (`_queues`) an ordinary command's reply uses, correlated by the subscribe request's own `request_uuid`. `_on_message` (`bluetooth.py`) checks `self._stream_sinks.get(msg.request_uuid)` before touching `_queues` — a match routes into that subscription's own bounded, drop-oldest `_StreamSink` instead, so `_send`'s pre-send drain can never discard a push and `_await_response` can never return one as an unrelated command's reply. `_register_stream_sink`/`_unregister_stream_sink` are the only entry points into the registry; there is no public subscription API yet. Tests: `tests/test_ble_stream_sink.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`, 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`. ## Maintaining this file diff --git a/docs/teslemetry.md b/docs/teslemetry.md index 7a99dc8..48e7d42 100644 --- a/docs/teslemetry.md +++ b/docs/teslemetry.md @@ -645,3 +645,41 @@ async def main(): asyncio.run(main()) ``` + +## OAuth Dynamic Client Registration + +Teslemetry supports OAuth 2.0 Dynamic Client Registration +([RFC 7591](https://www.rfc-editor.org/rfc/rfc7591)) so each installation can +register its own `client_id` instead of every client sharing one hardcoded +value. `register_client` is a standalone helper - it needs only an +`aiohttp.ClientSession`, not a `Teslemetry` instance, since no `client_id` or +access token exists yet at registration time. It always registers a new +client; callers are responsible for persisting the returned `client_id` and +skipping registration on subsequent runs. + +It raises `tesla_fleet_api.exceptions.TeslemetryRegistrationError` if the +registration endpoint can't be reached, the response isn't valid JSON, or the +response doesn't contain a usable `client_id`. + +```python +import aiohttp +from tesla_fleet_api import register_client +from tesla_fleet_api.exceptions import TeslemetryRegistrationError + +async def main(): + async with aiohttp.ClientSession() as session: + try: + registration = await register_client( + session, + client_name="Home Assistant", + software_id="home-assistant", + software_version="2026.1.0", + ) + except TeslemetryRegistrationError as e: + print(e) + return + + print(registration.client_id) + +asyncio.run(main()) +``` diff --git a/tesla_fleet_api/__init__.py b/tesla_fleet_api/__init__.py index 4fffbf3..cc11c5c 100644 --- a/tesla_fleet_api/__init__.py +++ b/tesla_fleet_api/__init__.py @@ -14,7 +14,11 @@ from tesla_fleet_api.tesla.bluetooth import TeslaBluetooth from tesla_fleet_api.tesla.fleet import TeslaFleetApi from tesla_fleet_api.tesla.oauth import TeslaFleetOAuth -from tesla_fleet_api.teslemetry.teslemetry import Teslemetry +from tesla_fleet_api.teslemetry.teslemetry import ( + Teslemetry, + TeslemetryClientRegistration, + register_client, +) from tesla_fleet_api.tessie.tessie import Tessie from tesla_fleet_api.util import firmware_at_least, firmware_compare @@ -27,10 +31,12 @@ "TeslaBluetooth", "TeslaFleetOAuth", "Teslemetry", + "TeslemetryClientRegistration", "Tessie", "firmware_at_least", "firmware_compare", "get_tariff_periods", "is_valid_region", + "register_client", "unwrap_tariff_v2", ] diff --git a/tesla_fleet_api/exceptions.py b/tesla_fleet_api/exceptions.py index 18794be..d83ccc8 100644 --- a/tesla_fleet_api/exceptions.py +++ b/tesla_fleet_api/exceptions.py @@ -205,6 +205,18 @@ class OAuthExpired(TeslaFleetError): key = "token expired (401)" +class TeslemetryRegistrationError(TeslaFleetError): # Teslemetry specific + """Teslemetry OAuth dynamic client registration (RFC 7591) failed. + + Covers a transport/timeout failure reaching the registration endpoint, a + non-2xx response, a response body that isn't valid JSON, and a + well-formed response missing a usable ``client_id``. The specific reason + is carried in ``data``. + """ + + message = "Teslemetry dynamic client registration failed." + + class LoginRequired(TeslaFleetError): # Native and Teslemetry """The user has reset their password and a new auth code is required, or the refresh_token has already been used.""" diff --git a/tesla_fleet_api/teslemetry/__init__.py b/tesla_fleet_api/teslemetry/__init__.py index 35acaa9..afac4b7 100644 --- a/tesla_fleet_api/teslemetry/__init__.py +++ b/tesla_fleet_api/teslemetry/__init__.py @@ -1,12 +1,18 @@ from tesla_fleet_api.tesla.charging import Charging from tesla_fleet_api.tesla.energysite import EnergySite, EnergySites from tesla_fleet_api.tesla.user import User -from tesla_fleet_api.teslemetry.teslemetry import Teslemetry +from tesla_fleet_api.teslemetry.teslemetry import ( + Teslemetry, + TeslemetryClientRegistration, + register_client, +) from tesla_fleet_api.teslemetry.vehicle import TeslemetryVehicle as Vehicle from tesla_fleet_api.teslemetry.vehicle import TeslemetryVehicles as Vehicles __all__ = [ "Teslemetry", + "TeslemetryClientRegistration", + "register_client", "Charging", "EnergySites", "EnergySite", diff --git a/tesla_fleet_api/teslemetry/teslemetry.py b/tesla_fleet_api/teslemetry/teslemetry.py index e72d10a..0ce42f6 100644 --- a/tesla_fleet_api/teslemetry/teslemetry.py +++ b/tesla_fleet_api/teslemetry/teslemetry.py @@ -1,14 +1,95 @@ from collections.abc import Awaitable, Callable +from dataclasses import dataclass from time import time -from typing import Any +from typing import Any, Final, cast import aiohttp from tesla_fleet_api.const import LOGGER, Method, is_valid_region +from tesla_fleet_api.exceptions import TeslemetryRegistrationError from tesla_fleet_api.tesla import TeslaFleetApi from tesla_fleet_api.teslemetry.energysite import TeslemetryEnergySites from tesla_fleet_api.teslemetry.vehicle import TeslemetryVehicles +REGISTER_URL: Final = "https://api.teslemetry.com/oauth/register" + + +@dataclass(frozen=True, slots=True) +class TeslemetryClientRegistration: + """Parsed result of :func:`register_client`. + + ``client_id`` is the OAuth client identifier this installation should + present for every future authorization, including reauthentication. + ``raw`` is the full decoded registration response for anything this + wrapper doesn't model. + """ + + client_id: str + raw: dict[str, Any] + + +async def register_client( + session: aiohttp.ClientSession, + client_name: str, + software_id: str, + software_version: str, +) -> TeslemetryClientRegistration: + """Dynamically register an OAuth client with Teslemetry (RFC 7591). + + Posts to Teslemetry's client-registration endpoint and parses the + result. This is a bare transport call - it always registers a new + client and has no knowledge of whether the caller already has one; + callers are responsible for persisting the returned ``client_id`` and + skipping registration on subsequent runs. + + Args: + session: An aiohttp session to issue the request on. + client_name: Human-readable client name shown during consent. + software_id: Identifier for the calling application. + software_version: Version string for the calling application. + + Returns: + The parsed registration result. + + Raises: + TeslemetryRegistrationError: The endpoint could not be reached, the + response was not valid JSON, or the response didn't contain a + usable ``client_id``. + """ + try: + async with session.post( + REGISTER_URL, + json={ + "client_name": client_name, + "software_id": software_id, + "software_version": software_version, + }, + ) as response: + response.raise_for_status() + registration = await response.json() + except (aiohttp.ClientError, TimeoutError) as err: + raise TeslemetryRegistrationError( + "Could not reach Teslemetry to register a client", + status=getattr(err, "status", None), + ) from err + except ValueError as err: + raise TeslemetryRegistrationError( + "Teslemetry returned a malformed registration response" + ) from err + + registration_dict = ( + cast("dict[str, Any]", registration) if isinstance(registration, dict) else None + ) + client_id = registration_dict.get("client_id") if registration_dict else None + if not isinstance(client_id, str) or not client_id: + raise TeslemetryRegistrationError( + "Teslemetry registration response did not contain a client_id" + ) + + return TeslemetryClientRegistration( + client_id=client_id, raw=registration_dict or {} + ) + class Teslemetry(TeslaFleetApi): vehicles: TeslemetryVehicles diff --git a/tests/test_teslemetry_register_client.py b/tests/test_teslemetry_register_client.py new file mode 100644 index 0000000..1edc372 --- /dev/null +++ b/tests/test_teslemetry_register_client.py @@ -0,0 +1,151 @@ +"""Tests for ``register_client()``, the Teslemetry OAuth dynamic client +registration (RFC 7591) transport helper. + +Mirrors the registration semantics of the Home Assistant Teslemetry +integration's ``oauth.py`` (endpoint, request payload, response parsing, +and failure modes): a transport/timeout failure, a server-rejected +registration, and a malformed/missing-``client_id`` response all raise +:class:`~tesla_fleet_api.exceptions.TeslemetryRegistrationError`. +""" + +from __future__ import annotations + +from contextlib import asynccontextmanager +from typing import Any +from unittest import IsolatedAsyncioTestCase +from unittest.mock import AsyncMock, MagicMock + +import aiohttp + +from tesla_fleet_api.exceptions import TeslemetryRegistrationError +from tesla_fleet_api.teslemetry.teslemetry import ( + REGISTER_URL, + TeslemetryClientRegistration, + register_client, +) + + +def _fake_response( + *, json_body: object = None, raise_for_status_error: Exception | None = None +) -> MagicMock: + resp = MagicMock() + resp.raise_for_status = MagicMock(side_effect=raise_for_status_error) + if isinstance(json_body, Exception): + resp.json = AsyncMock(side_effect=json_body) + else: + resp.json = AsyncMock(return_value=json_body) + return resp + + +def _make_session( + *, + response: object = None, + post_error: Exception | None = None, +) -> MagicMock: + session = MagicMock() + + if post_error is not None: + session.post = MagicMock(side_effect=post_error) + return session + + @asynccontextmanager + async def _ctx(*args: Any, **kwargs: Any): + yield response + + session.post = MagicMock(side_effect=lambda *a, **k: _ctx(*a, **k)) + return session + + +class RegisterClientSuccessTests(IsolatedAsyncioTestCase): + async def test_registers_new_client(self) -> None: + session = _make_session( + response=_fake_response(json_body={"client_id": "new-client-id"}) + ) + + result = await register_client( + session, "Home Assistant", "home-assistant", "2026.1.0" + ) + + self.assertEqual( + result, + TeslemetryClientRegistration( + client_id="new-client-id", raw={"client_id": "new-client-id"} + ), + ) + args, kwargs = session.post.call_args + self.assertEqual(args[0], REGISTER_URL) + self.assertEqual( + kwargs["json"], + { + "client_name": "Home Assistant", + "software_id": "home-assistant", + "software_version": "2026.1.0", + }, + ) + + +class RegisterClientRejectionTests(IsolatedAsyncioTestCase): + async def test_server_rejected_registration_raises_typed_error(self) -> None: + error = aiohttp.ClientResponseError( + MagicMock(), (), status=400, message="Bad Request" + ) + session = _make_session( + response=_fake_response(json_body={}, raise_for_status_error=error) + ) + + with self.assertRaises(TeslemetryRegistrationError): + await register_client(session, "Home Assistant", "home-assistant", "1") + + async def test_connection_error_raises_typed_error(self) -> None: + session = _make_session(post_error=aiohttp.ClientConnectionError("boom")) + + with self.assertRaises(TeslemetryRegistrationError): + await register_client(session, "Home Assistant", "home-assistant", "1") + + async def test_timeout_raises_typed_error(self) -> None: + session = _make_session(post_error=TimeoutError()) + + with self.assertRaises(TeslemetryRegistrationError): + await register_client(session, "Home Assistant", "home-assistant", "1") + + +class RegisterClientMalformedResponseTests(IsolatedAsyncioTestCase): + async def test_non_json_body_raises_typed_error(self) -> None: + session = _make_session( + response=_fake_response(json_body=ValueError("not json")) + ) + + with self.assertRaises(TeslemetryRegistrationError): + await register_client(session, "Home Assistant", "home-assistant", "1") + + async def test_missing_client_id_raises_typed_error(self) -> None: + session = _make_session(response=_fake_response(json_body={})) + + with self.assertRaises(TeslemetryRegistrationError): + await register_client(session, "Home Assistant", "home-assistant", "1") + + async def test_non_string_client_id_raises_typed_error(self) -> None: + session = _make_session(response=_fake_response(json_body={"client_id": 12345})) + + with self.assertRaises(TeslemetryRegistrationError): + await register_client(session, "Home Assistant", "home-assistant", "1") + + async def test_empty_string_client_id_raises_typed_error(self) -> None: + session = _make_session(response=_fake_response(json_body={"client_id": ""})) + + with self.assertRaises(TeslemetryRegistrationError): + await register_client(session, "Home Assistant", "home-assistant", "1") + + async def test_non_dict_body_raises_typed_error(self) -> None: + session = _make_session( + response=_fake_response(json_body=["unexpected", "list", "body"]) + ) + + with self.assertRaises(TeslemetryRegistrationError): + await register_client(session, "Home Assistant", "home-assistant", "1") + + async def test_null_body_raises_typed_error(self) -> None: + session = _make_session(response=_fake_response(json_body=None)) + + with self.assertRaises(TeslemetryRegistrationError): + await register_client(session, "Home Assistant", "home-assistant", "1")