diff --git a/README.md b/README.md index de51404..ccc8d12 100644 --- a/README.md +++ b/README.md @@ -9,7 +9,9 @@ The Owner API will [stop working](https://developer.tesla.com/docs/fleet-api#202 ## Overview -This module depends on Python [requests](https://pypi.org/project/requests/), [requests_oauthlib](https://pypi.org/project/requests-oauthlib/) and [websocket-client](https://pypi.org/project/websocket-client/). It requires Python 3.10+ when using urllib3 2.0, which comes with requests 2.30.0+, or you can pin urllib3 to 1.26.x by installing `urllib3<2`. +This module depends on Python [requests](https://pypi.org/project/requests/), [requests_oauthlib](https://pypi.org/project/requests-oauthlib/), [httpx](https://pypi.org/project/httpx/) and [websocket-client](https://pypi.org/project/websocket-client/). It requires Python 3.10+ when using urllib3 2.0, which comes with requests 2.30.0+, or you can pin urllib3 to 1.26.x by installing `urllib3<2`. + +As of June 2026, Tesla requires HTTP/2 for its SSO (auth.tesla.com) and Owner API endpoints; requests, which only speaks HTTP/1.1, started receiving `403 Client Error: forbidden` responses on token refresh. The `Tesla` class therefore uses `httpx` (installed with the `[http2]` extra so the `h2` package is present) to perform these requests over HTTP/2, automatically falling back to requests over HTTP/1.1 when httpx is unavailable. On Python 2, where httpx cannot be installed, the requests fallback is always used. The `Tesla` class extends `requests_oauthlib.OAuth2Session` which extends `requests.Session` and therefore inherits methods like `get()` and `post()` that can be used to perform API calls. Module characteristics: @@ -814,7 +816,7 @@ TeslaPy is available on PyPI: Make sure you have [Python](https://www.python.org/) 2.7+ or 3.5+ installed on your system. Alternatively, clone the repository to your machine and run demo application [cli.py](https://github.com/tdorssers/TeslaPy/blob/master/cli.py), [menu.py](https://github.com/tdorssers/TeslaPy/blob/master/menu.py) or [gui.py](https://github.com/tdorssers/TeslaPy/blob/master/gui.py) to get started, after installing [requests_oauthlib](https://pypi.org/project/requests-oauthlib/) 0.8.0+, [geopy](https://pypi.org/project/geopy/) 1.14.0+, [pywebview](https://pypi.org/project/pywebview/) 3.0+ (optional), [selenium](https://pypi.org/project/selenium/) 3.13.0+ (optional) and [websocket-client](https://pypi.org/project/websocket-client/) 0.59+ using [PIP](https://pypi.org/project/pip/) as follows: -`python -m pip install requests_oauthlib geopy pywebview selenium websocket-client` +`python -m pip install requests_oauthlib 'httpx[http2]' geopy pywebview selenium websocket-client` and install [ChromeDriver](https://sites.google.com/chromium.org/driver/) to use Selenium or on Ubuntu as follows: diff --git a/requirements.txt b/requirements.txt index e00c40e..195edd5 100644 --- a/requirements.txt +++ b/requirements.txt @@ -1,5 +1,6 @@ geopy>=1.20.0 requests-oauthlib>=1.3.0 websocket_client>=0.59.0 +httpx[http2]>=0.27.0 selenium>=3.141.0 pywebview>=3.2 \ No newline at end of file diff --git a/setup.cfg b/setup.cfg index eccc1af..6619ed4 100644 --- a/setup.cfg +++ b/setup.cfg @@ -33,3 +33,4 @@ install_requires = requests >= 2.4.2 requests_oauthlib websocket-client >= 0.59.0 + httpx[http2] >= 0.27.0 diff --git a/teslapy/__init__.py b/teslapy/__init__.py index 329f3b6..02b12ab 100644 --- a/teslapy/__init__.py +++ b/teslapy/__init__.py @@ -32,8 +32,18 @@ from requests.packages.urllib3.util.retry import Retry from urllib3.poolmanager import PoolManager from oauthlib.oauth2.rfc6749.errors import * +from oauthlib.common import urldecode import websocket # websocket-client v0.49.0 up to v0.58.0 is not supported +# Optional HTTP/2 transport. Since June 2026 Tesla's SSO and Owner API servers +# require HTTP/2, which `requests` cannot speak, so token refresh fails with a +# 403. `httpx` is used when available and we fall back to `requests` otherwise. +try: + import httpx + HAS_HTTPX = True +except ImportError: + HAS_HTTPX = False + requests.packages.urllib3.disable_warnings() BASE_URL = 'https://owner-api.teslamotors.com/' @@ -163,7 +173,12 @@ def request(self, method, url, serialize=True, **kwargs): kwargs.setdefault('timeout', self.timeout) if serialize and 'data' in kwargs: kwargs['json'] = kwargs.pop('data') - response = super(Tesla, self).request(method, url, **kwargs) + # Use HTTP/2 for Owner API calls when httpx is available, otherwise + # fall back to requests (HTTP/1.1 over TLS 1.3 via TLSAdapter) + if HAS_HTTPX: + response = self._request_http2(method, url, **kwargs) + else: + response = super(Tesla, self).request(method, url, **kwargs) # Error message handling if serialize and 400 <= response.status_code < 600: try: @@ -177,6 +192,64 @@ def request(self, method, url, serialize=True, **kwargs): return response.json(object_hook=JsonDict) return response.text + @staticmethod + def _httpx_auth_verify(verify=True): + """ Returns an httpx `verify` value pinned to TLS 1.3 that advertises + HTTP/2 over ALPN. Tesla's servers reject connections that do not + negotiate TLS 1.3, mirroring the requests `TLSAdapter`. A custom SSL + context, CA bundle path or disabled verification is passed through. """ + if verify is False or isinstance(verify, (str, bytes, ssl.SSLContext)): + return verify + try: + context = ssl.create_default_context() + context.minimum_version = ssl.TLSVersion.TLSv1_3 + # httpx only sets ALPN protocols when it builds the SSL context + # itself, so HTTP/2 must be advertised on our custom context + context.set_alpn_protocols(['h2', 'http/1.1']) + return context + except Exception: + return verify + + def _httpx_client_kwargs(self, verify): + """ Builds `httpx.Client` keyword arguments from the session settings, + forwarding the proxy and trust_env configuration. """ + client_kwargs = {'http2': True, 'trust_env': self.trust_env, + 'verify': self._httpx_auth_verify(verify)} + proxy = self.proxies.get('https') + if proxy: + client_kwargs['proxy'] = proxy + return client_kwargs + + def _request_http2(self, method, url, **kwargs): + """ Sends an Owner API request over HTTP/2 using httpx and returns a + requests compatible response. Refreshes an expired access token first, + as `OAuth2Session` would for the requests transport, and falls back to + requests (HTTP/1.1) when the HTTP/2 request fails. """ + # Auto-refresh an expired token (OAuth2Session does this for requests) + if (not kwargs.get('withhold_token') and self.authorized + and 0 < (self.expires_at or 0) < time.time()): + self.refresh_token() + # Start from the session headers so httpx sends the same fingerprint + # (Content-Type, X-Tesla-User-Agent, User-Agent) as the requests session + headers = dict(self.headers) + if not kwargs.get('withhold_token') and self.authorized: + token = self.token.get('access_token') + if token: + headers['Authorization'] = 'Bearer ' + token + request_kwargs = {'headers': headers, 'follow_redirects': True, + 'timeout': kwargs.get('timeout', self.timeout)} + for key in ('params', 'json', 'data'): + if key in kwargs: + request_kwargs[key] = kwargs[key] + try: + with httpx.Client(**self._httpx_client_kwargs(self.verify)) as client: + response = client.request(method, url, **request_kwargs) + return _HTTP2Response(response) + except Exception as exc: + logger.warning('HTTP/2 request failed, falling back to HTTP/1.1: %s', + exc) + return super(Tesla, self).request(method, url, **kwargs) + @staticmethod def new_code_verifier(): """ Generate code verifier for PKCE as per RFC 7636 section 4.1 """ @@ -241,6 +314,15 @@ def fetch_token(self, token_url='oauth2/v3/token', **kwargs): kwargs['authorization_response'] = self.authenticator(url) # Use authorization code in redirected location to get token token_url = urljoin(self.sso_base_url, token_url) + # Try HTTP/2 first as Tesla requires it for auth.tesla.com since 2026 + if HAS_HTTPX: + try: + self._fetch_token_http2(token_url, **kwargs) + self._token_updater() # Save new token + return self.token + except Exception as exc: + logger.warning('HTTP/2 token fetch failed, falling back to ' + 'HTTP/1.1: %s', exc) kwargs['include_client_id'] = True kwargs.setdefault('verify', self.verify) kwargs.setdefault('code_verifier', self.code_verifier) @@ -248,6 +330,28 @@ def fetch_token(self, token_url='oauth2/v3/token', **kwargs): self._token_updater() # Save new token return self.token + def _fetch_token_http2(self, token_url, **kwargs): + """ Exchanges the authorization code for a token over HTTP/2 using the + oauthlib client to build the request body and parse the response. """ + # Extract the authorization code from the redirected location + self._client.parse_request_uri_response( + kwargs['authorization_response'], state=self._state) + body = self._client.prepare_request_body( + code=self._client.code, redirect_uri=self.redirect_uri, + include_client_id=True, + code_verifier=kwargs.get('code_verifier', self.code_verifier)) + headers = {'Accept': 'application/json', + 'Content-Type': 'application/x-www-form-urlencoded'} + timeout = kwargs.get('timeout', self.timeout) + verify = kwargs.get('verify', self.verify) + with httpx.Client(**self._httpx_client_kwargs(verify)) as client: + response = client.post(token_url, data=dict(urldecode(body)), + headers=headers, timeout=timeout) + response.raise_for_status() + self.token = self._client.parse_request_body_response( + response.text, scope=self.scope) + return self.token + def refresh_token(self, token_url='oauth2/v3/token', **kwargs): """ Overriddes base method to refresh Tesla's SSO token. Raises ValueError and ServerError. @@ -262,11 +366,45 @@ def refresh_token(self, token_url='oauth2/v3/token', **kwargs): if not self.authorized and not kwargs.get('refresh_token'): raise ValueError('`refresh_token` is not set') token_url = urljoin(self.sso_base_url, token_url) + # Try HTTP/2 first as Tesla requires it for auth.tesla.com since 2026 + if HAS_HTTPX: + try: + self._refresh_token_http2(token_url, **kwargs) + self._token_updater() # Save new token + return self.token + except Exception as exc: + logger.warning('HTTP/2 token refresh failed, falling back to ' + 'HTTP/1.1: %s', exc) kwargs.setdefault('verify', self.verify) super(Tesla, self).refresh_token(token_url, **kwargs) self._token_updater() # Save new token return self.token + def _refresh_token_http2(self, token_url, **kwargs): + """ Refreshes Tesla's SSO token over HTTP/2 using the oauthlib client + to build the request body and parse the response. """ + refresh_token = (kwargs.get('refresh_token') + or self.token.get('refresh_token')) + if not refresh_token: + raise ValueError('`refresh_token` is not set') + body = self._client.prepare_refresh_body( + refresh_token=refresh_token, scope=self.scope, + **self.auto_refresh_kwargs) + headers = {'Accept': 'application/json', + 'Content-Type': 'application/x-www-form-urlencoded'} + timeout = kwargs.get('timeout', self.timeout) + verify = kwargs.get('verify', self.verify) + with httpx.Client(**self._httpx_client_kwargs(verify)) as client: + response = client.post(token_url, data=dict(urldecode(body)), + headers=headers, timeout=timeout) + response.raise_for_status() + self.token = self._client.parse_request_body_response( + response.text, scope=self.scope) + # Tesla may omit the refresh token in the response, so preserve it + if 'refresh_token' not in self.token: + self.token['refresh_token'] = refresh_token + return self.token + def close(self): """ Overriddes base method to remove all adapters on close """ super(Tesla, self).close() @@ -409,6 +547,32 @@ def wall_connector_list(self): if p.get('resource_type') == 'wall_connector'] +class _HTTP2Response(object): + """ Adapts an `httpx.Response` to the subset of the `requests.Response` + interface used by `Tesla.request`, so the HTTP/2 path is transparent. """ + + def __init__(self, response): + self._content = response.content + self.status_code = response.status_code + self.reason = response.reason_phrase + self.headers = response.headers + + @property + def text(self): + """ Returns the response body decoded as text """ + return self._content.decode('utf-8', 'replace') + + def json(self, **kwargs): + """ Deserializes the response body as JSON """ + return json.loads(self._content, **kwargs) + + def raise_for_status(self): + """ Raises HTTPError, if one occurred """ + if 400 <= self.status_code < 600: + raise HTTPError('%s Client Error: %s' % (self.status_code, + self.reason), response=self) + + class VehicleError(Exception): """ Vehicle exception class """ pass