From b3042480f161c5d5d1298d738ba102801ad221a1 Mon Sep 17 00:00:00 2001 From: JingHao-Leon <102573344+JingHao-Leon@users.noreply.github.com> Date: Tue, 20 Oct 2026 10:50:58 +0800 Subject: [PATCH 1/2] test: reproduce CN OAuth token endpoint bug The OAuth token endpoint is hardcoded to the global host, so a China client redeems its authorization code (and refreshes its access token) against fleet-auth.prd.vn.cloud.tesla.com instead of auth.tesla.cn, while the authorize URL already switches host for China. Covers TeslaFleetOAuth.get_refresh_token/refresh_access_token and TeslaFleetApi.partner_login for both regions. --- tests/test_fleet_auth_refresh.py | 112 ++++++++++++++++++++++++++++++- 1 file changed, 111 insertions(+), 1 deletion(-) diff --git a/tests/test_fleet_auth_refresh.py b/tests/test_fleet_auth_refresh.py index fb0adcd..f2bfd41 100644 --- a/tests/test_fleet_auth_refresh.py +++ b/tests/test_fleet_auth_refresh.py @@ -3,15 +3,20 @@ These tests assert what ``_request`` does today; they do not assume any auth-refresh / retry feature exists. If that feature is added later, extend these tests to cover the retry path rather than deleting them. + +``OAuthTokenEndpointTests`` covers the OAuth *token* endpoint, which has to +follow the client's region: China is served by its own auth host +(``auth.tesla.cn``), every other region by the shared global one. """ from contextlib import asynccontextmanager from unittest import IsolatedAsyncioTestCase from unittest.mock import AsyncMock, MagicMock -from tesla_fleet_api.const import Method +from tesla_fleet_api.const import Method, Scope from tesla_fleet_api.exceptions import MissingToken, NotFound, ResponseError from tesla_fleet_api.tesla.fleet import TeslaFleetApi +from tesla_fleet_api.tesla.oauth import TeslaFleetOAuth class _RequestTestApi(TeslaFleetApi): @@ -114,3 +119,108 @@ async def test_error_status_routes_through_raise_for_status(self) -> None: api = _make_api(response=resp) with self.assertRaises(NotFound): await api.request(Method.GET, "api/1/products") + + +def _fake_token_response() -> MagicMock: + """A 200 response carrying an OAuth token payload.""" + resp = MagicMock() + resp.ok = True + resp.status = 200 + resp.json = AsyncMock( + return_value={ + "access_token": "access-token", + "refresh_token": "refresh-token", + "expires_in": 3600, + } + ) + return resp + + +def _make_post_session() -> MagicMock: + """A session whose ``post`` returns an async context manager yielding a token response.""" + session = MagicMock() + + @asynccontextmanager + async def _ctx(*args, **kwargs): + yield _fake_token_response() + + session.post = MagicMock(side_effect=lambda *a, **k: _ctx(*a, **k)) + return session + + +class OAuthTokenEndpointTests(IsolatedAsyncioTestCase): + """The OAuth token endpoint must follow the client's region. + + China is served by its own auth host (``auth.tesla.cn``); every other + region shares the global one (``fleet-auth.prd.vn.cloud.tesla.com``). + The *authorize* URL already switched host for China - the token URL did + not, so a China client redeemed its authorization code against the + global host and got its refresh token from the wrong region. + """ + + GLOBAL_TOKEN_URL = "https://fleet-auth.prd.vn.cloud.tesla.com/oauth2/v3/token" + CN_TOKEN_URL = "https://auth.tesla.cn/oauth2/v3/token" + + def _make_oauth(self, region: str) -> tuple[TeslaFleetOAuth, MagicMock]: + session = _make_post_session() + api = TeslaFleetOAuth( + session=session, + region=region, + client_id="client-id", + client_secret="client-secret", + redirect_uri="https://example.com/callback", + refresh_token="refresh-token", + ) + return api, session + + async def test_cn_exchanges_the_authorization_code_on_the_cn_host(self) -> None: + api, session = self._make_oauth("cn") + + await api.get_refresh_token("cn_abc123") + + self.assertEqual(session.post.call_args.args[0], self.CN_TOKEN_URL) + + async def test_cn_refreshes_the_access_token_on_the_cn_host(self) -> None: + api, session = self._make_oauth("cn") + + await api.refresh_access_token() + + self.assertEqual(session.post.call_args.args[0], self.CN_TOKEN_URL) + + async def test_na_exchanges_the_authorization_code_on_the_global_host(self) -> None: + api, session = self._make_oauth("na") + + await api.get_refresh_token("na_abc123") + + self.assertEqual(session.post.call_args.args[0], self.GLOBAL_TOKEN_URL) + + async def test_na_refreshes_the_access_token_on_the_global_host(self) -> None: + api, session = self._make_oauth("na") + + await api.refresh_access_token() + + self.assertEqual(session.post.call_args.args[0], self.GLOBAL_TOKEN_URL) + + async def test_cn_partner_login_uses_the_cn_host(self) -> None: + session = _make_post_session() + api = TeslaFleetApi(session=session, access_token="access-token", region="cn") + + await api.partner_login("client-id", "client-secret") + + self.assertEqual(session.post.call_args.args[0], self.CN_TOKEN_URL) + + async def test_na_partner_login_uses_the_global_host(self) -> None: + session = _make_post_session() + api = TeslaFleetApi(session=session, access_token="access-token", region="na") + + await api.partner_login("client-id", "client-secret") + + self.assertEqual(session.post.call_args.args[0], self.GLOBAL_TOKEN_URL) + + async def test_cn_authorize_url_already_uses_the_cn_host(self) -> None: + """Characterization: the authorize URL switched host for China, the token URL did not.""" + api, _ = self._make_oauth("cn") + + url = api.get_login_url([Scope.OPENID]) + + self.assertTrue(url.startswith("https://auth.tesla.cn/oauth2/v3/authorize")) From 0675c711dcd11d11717703d34cc705653604acea Mon Sep 17 00:00:00 2001 From: JingHao-Leon <102573344+JingHao-Leon@users.noreply.github.com> Date: Tue, 20 Oct 2026 10:54:59 +0800 Subject: [PATCH 2/2] fix(oauth): send the OAuth token request to the region's auth host China is served by its own auth host (auth.tesla.cn); every other region shares the global one. The authorize URL already switched host for China, but the token endpoint stayed hardcoded to the global host, so a China client redeemed its authorization code and refreshed its access token against the wrong region. Add const.auth_server() as the single source of truth and use it for TeslaFleetOAuth.get_refresh_token, TeslaFleetOAuth.refresh_access_token and TeslaFleetApi.partner_login. --- tesla_fleet_api/const.py | 12 ++++++++++++ tesla_fleet_api/tesla/fleet.py | 11 +++++++++-- tesla_fleet_api/tesla/oauth.py | 13 ++++++++++--- 3 files changed, 31 insertions(+), 5 deletions(-) diff --git a/tesla_fleet_api/const.py b/tesla_fleet_api/const.py index 09b3b2c..107e112 100644 --- a/tesla_fleet_api/const.py +++ b/tesla_fleet_api/const.py @@ -25,6 +25,18 @@ def is_valid_region(value: str) -> TypeGuard[Region]: return value in SERVERS +def auth_server(region: Region | None) -> str: + """The OAuth token endpoint for ``region``. + + China is served by its own auth host; every other region - and an unset + region - shares the global one. The authorize endpoint follows the same + split (``auth.tesla.cn`` for China, ``auth.tesla.com`` elsewhere). + """ + if region == "cn": + return "https://auth.tesla.cn/oauth2/v3/token" + return "https://fleet-auth.prd.vn.cloud.tesla.com/oauth2/v3/token" + + class IntEnum(int, Enum): """Integer Enum.""" diff --git a/tesla_fleet_api/tesla/fleet.py b/tesla_fleet_api/tesla/fleet.py index 6ea8be7..81d9e1a 100644 --- a/tesla_fleet_api/tesla/fleet.py +++ b/tesla_fleet_api/tesla/fleet.py @@ -7,7 +7,14 @@ import aiohttp from tesla_fleet_api import __version__ as VERSION -from tesla_fleet_api.const import LOGGER, SERVERS, Method, Region, Scope +from tesla_fleet_api.const import ( + LOGGER, + SERVERS, + Method, + Region, + Scope, + auth_server, +) from tesla_fleet_api.exceptions import ( InvalidRegion, LibraryError, @@ -267,7 +274,7 @@ async def partner_login( } async with self.session.post( - "https://fleet-auth.prd.vn.cloud.tesla.com/oauth2/v3/token", + auth_server(self.region), headers={"Content-Type": "application/x-www-form-urlencoded"}, data=data, ) as resp: diff --git a/tesla_fleet_api/tesla/oauth.py b/tesla_fleet_api/tesla/oauth.py index 48d3c69..13f71ec 100644 --- a/tesla_fleet_api/tesla/oauth.py +++ b/tesla_fleet_api/tesla/oauth.py @@ -4,7 +4,14 @@ import aiohttp -from tesla_fleet_api.const import SERVERS, Method, Region, Scope, is_valid_region +from tesla_fleet_api.const import ( + SERVERS, + Method, + Region, + Scope, + auth_server, + is_valid_region, +) from tesla_fleet_api.tesla import TeslaFleetApi @@ -68,7 +75,7 @@ async def get_refresh_token(self, code: str) -> None: self.server = SERVERS.get(self.region) async with self.session.post( - "https://fleet-auth.prd.vn.cloud.tesla.com/oauth2/v3/token", + auth_server(self.region), data={ "grant_type": "authorization_code", "client_id": self.client_id, @@ -106,7 +113,7 @@ async def refresh_access_token(self) -> dict[str, Any]: if not self.refresh_token: raise ValueError("Refresh token is missing") async with self.session.post( - "https://fleet-auth.prd.vn.cloud.tesla.com/oauth2/v3/token", + auth_server(self.region), data={ "grant_type": "refresh_token", "client_id": self.client_id,