diff --git a/packages/modules/vehicles/tesla/api.py b/packages/modules/vehicles/tesla/api.py index aec7abbdbf..dec5e3a4fb 100644 --- a/packages/modules/vehicles/tesla/api.py +++ b/packages/modules/vehicles/tesla/api.py @@ -1,10 +1,14 @@ #!/usr/bin/python3 +import ssl import logging import time import json from typing import Tuple +from requests.adapters import HTTPAdapter +from urllib3 import PoolManager + from modules.common import req from modules.vehicles.tesla.config import TeslaSocToken @@ -25,18 +29,71 @@ X_TESLA_USER_AGENT = "" -def post_wake_up_command(vehicle: int, token: TeslaSocToken) -> str: - vehicle_id = __get_vehicle_id(vehicle, token) - command = f"vehicles/{vehicle_id}/wake_up" - log.debug(f"Sending command: '{command}'") - headers = { +class TLS13HttpAdapter(HTTPAdapter): + """HTTP adapter that only negotiates TLS 1.3.""" + + def __init__(self, *args, **kwargs): + ctx = ssl.create_default_context() + try: + ctx.minimum_version = ssl.TLSVersion.TLSv1_3 + ctx.maximum_version = ssl.TLSVersion.TLSv1_3 + except Exception as exc: + raise RuntimeError( + "TLS 1.3 is required for the Tesla API but is not supported by this Python/OpenSSL build. " + "Please upgrade Python/OpenSSL on this system." + ) from exc + self._ssl_context = ctx + super().__init__(*args, **kwargs) + + def init_poolmanager(self, connections, maxsize, block=False, **pool_kwargs): + pool_kwargs["ssl_context"] = self._ssl_context + self.poolmanager = PoolManager( + num_pools=connections, + maxsize=maxsize, + block=block, + **pool_kwargs, + ) + + def proxy_manager_for(self, proxy, **proxy_kwargs): + proxy_kwargs["ssl_context"] = self._ssl_context + return super().proxy_manager_for(proxy, **proxy_kwargs) + + +def __auth_headers(token: TeslaSocToken) -> dict: + if token.access_token is None: + raise Exception("Konfiguration des Tesla SoC unvollständig! Keine Access Token vorhanden.") + return { "user-agent": UA, "x-tesla-user-agent": X_TESLA_USER_AGENT, "authorization": "bearer " + token.access_token } + + +def __no_auth_headers() -> dict: + return { + "user-agent": UA, + "x-tesla-user-agent": X_TESLA_USER_AGENT, + } + + +def __get_tesla_http_session(): session = req.get_http_session() - response = session.post(f"https://owner-api.teslamotors.com/api/1/{command}", headers=headers, timeout=50).json() - return response["response"]["state"] + tls13_adapter = TLS13HttpAdapter() + session.mount("https://owner-api.teslamotors.com/", tls13_adapter) + session.mount("https://auth.tesla.com/", tls13_adapter) + return session + + +def post_wake_up_command(vehicle: int, token: TeslaSocToken) -> str: + vehicle_id = __get_vehicle_id(vehicle, token) + command = f"vehicles/{vehicle_id}/wake_up" + log.debug(f"Sending command: '{command}'") + headers = __auth_headers(token) + with __get_tesla_http_session() as session: + response = session.post( + f"https://owner-api.teslamotors.com/api/1/{command}", headers=headers, timeout=50 + ).json() + return response["response"]["state"] def request_soc_range(vehicle: int, token: TeslaSocToken) -> Tuple[float, float, float]: @@ -52,28 +109,37 @@ def request_soc_range(vehicle: int, token: TeslaSocToken) -> Tuple[float, float, def validate_token(token: TeslaSocToken) -> TeslaSocToken: - if token.access_token is None and token.refresh_token is None: - raise Exception("Konfiguration des Tesla SoC unvollständig! Keine Token vorhanden.") + if token is None: + raise Exception("Konfiguration des Tesla SoC fehlt vollständig.") + if token.refresh_token is None: + raise Exception("Konfiguration des Tesla SoC unvollständig! Keine Refresh Token vorhanden.") + if token.access_token is None: + log.debug("No access token present. Refreshing token") + return __refresh_token(token) expiration = token.created_at + token.expires_in - log.debug("No need to authenticate. Valid token already present.") if time.time() > expiration: - log.debug("Access token expired. Refreshing token.") + log.debug("Access Token expired. Refreshing token") token = __refresh_token(token) + else: + log.debug("No need to authenticate. Valid token already present.") return token def __refresh_token(token: TeslaSocToken) -> TeslaSocToken: - headers = {"user-agent": UA, "x-tesla-user-agent": X_TESLA_USER_AGENT} + headers = __no_auth_headers() payload = { "grant_type": "refresh_token", "client_id": "ownerapi", "refresh_token": token.refresh_token, "scope": "openid email offline_access", } - resp = req.get_http_session().post("https://auth.tesla.com/oauth2/v3/token", - headers=headers, - json=payload, - timeout=50) + with __get_tesla_http_session() as session: + resp = session.post( + "https://auth.tesla.com/oauth2/v3/token", + headers=headers, + json=payload, + timeout=50, + ) log.debug("received refresh token") resp_json = resp.json() token.refresh_token = resp_json["refresh_token"] @@ -96,12 +162,11 @@ def __get_vehicle_id(index: int, token: TeslaSocToken) -> str: def __request_data(data_part: str, token: TeslaSocToken) -> str: log.debug(f"Requesting data: '{data_part}'") - headers = { - "user-agent": UA, - "x-tesla-user-agent": X_TESLA_USER_AGENT, - "authorization": "bearer " + token.access_token - } - response = req.get_http_session().get(f"https://owner-api.teslamotors.com/api/1/{data_part}", - headers=headers, - timeout=50) - return response.text + headers = __auth_headers(token) + with __get_tesla_http_session() as session: + response = session.get( + f"https://owner-api.teslamotors.com/api/1/{data_part}", + headers=headers, + timeout=50, + ) + return response.text