feat: Support modern vehicles using HTTP proxy (#853)

See https://github.com/alandtse/tesla/blob/dev/README.md#tesla-fleet-api-proxy

.devcontainer.json

@@ -9,15 +9,14 @@
         "ms-python.python",
         "github.vscode-pull-request-github",
         "ryanluker.vscode-coverage-gutters",
-        "ms-python.vscode-pylance"
+        "ms-python.vscode-pylance",
+        "ms-python.pylint"
       ],
       "settings": {
         "files.eol": "\n",
         "editor.tabSize": 4,
         "python.pythonPath": "/usr/bin/python3",
         "python.analysis.autoSearchPaths": false,
-        "python.linting.pylintEnabled": true,
-        "python.linting.enabled": true,
         "python.formatting.provider": "black",
         "python.formatting.blackPath": "/usr/local/py-utils/bin/black",
         "editor.formatOnPaste": false,

README.md

@@ -36,6 +36,14 @@ To use the component, you will need an application to generate a Tesla refresh t
 
 Note: This integration will wake up your vehicle(s) during installation.
 
+## Tesla Fleet API Proxy
+
+Tesla has [deprecated](https://developer.tesla.com/docs/fleet-api) the Owner API for _most_ vehicles in favor of a new Fleet API with end-to-end encryption. You'll know you're affected if you see this error in the log:
+
+> [teslajsonpy.connection] 403: {"response":null,"error":"Tesla Vehicle Command Protocol required, please refer to the documentation here: https://developer.tesla.com/docs/fleet-api#2023-10-09-rest-api-vehicle-commands-endpoint-deprecation-warning","error_description":""}
+
+If your vehicle is affected by this, you'll need to install the [Tesla HTTP Proxy](https://github.com/llamafilm/tesla-http-proxy-addon) add-on and configure this component to use it. This requires a complex setup; see [here](https://github.com/llamafilm/tesla-http-proxy-addon/blob/main/tesla_http_proxy/DOCS.md) for details. After configuring the add-on, tick the box for "Fleet API Proxy" in this component, and the config flow will autofill your Client ID, Proxy URL, and SSL certificate.
+
 <!---->
 
 ## Usage

custom_components/tesla_custom/__init__.py

@@ -5,11 +5,13 @@
 from functools import partial
 from http import HTTPStatus
 import logging
+import ssl
 
 import async_timeout
 from homeassistant.config_entries import SOURCE_IMPORT
 from homeassistant.const import (
     CONF_ACCESS_TOKEN,
+    CONF_CLIENT_ID,
     CONF_DOMAIN,
     CONF_SCAN_INTERVAL,
     CONF_TOKEN,
@@ -29,6 +31,8 @@
 
 from .config_flow import CannotConnect, InvalidAuth, validate_input
 from .const import (
+    CONF_API_PROXY_CERT,
+    CONF_API_PROXY_URL,
     CONF_ENABLE_TESLAMATE,
     CONF_EXPIRATION,
     CONF_INCLUDE_ENERGYSITES,
@@ -131,11 +135,22 @@ def _update_entry(email, data=None, options=None):
 
 async def async_setup_entry(hass, config_entry):
     """Set up Tesla as config entry."""
-    # pylint: disable=too-many-locals,too-many-statements
+    # pylint: disable=too-many-locals,too-many-statements,too-many-branches
     hass.data.setdefault(DOMAIN, {})
     config = config_entry.data
     # Because users can have multiple accounts, we always
     # create a new session so they have separate cookies
+
+    if config[CONF_API_PROXY_CERT]:
+        try:
+            SSL_CONTEXT.load_verify_locations(config[CONF_API_PROXY_CERT])
+            _LOGGER.debug(SSL_CONTEXT)
+        except (FileNotFoundError, ssl.SSLError):
+            _LOGGER.warning(
+                "Unable to load custom SSL certificate from %s",
+                config[CONF_API_PROXY_CERT],
+            )
+
     async_client = httpx.AsyncClient(
         headers={USER_AGENT: SERVER_SOFTWARE}, timeout=60, verify=SSL_CONTEXT
     )
@@ -165,6 +180,9 @@ async def async_setup_entry(hass, config_entry):
             polling_policy=config_entry.options.get(
                 CONF_POLLING_POLICY, DEFAULT_POLLING_POLICY
             ),
+            api_proxy_cert=config.get(CONF_API_PROXY_CERT),
+            api_proxy_url=config.get(CONF_API_PROXY_URL),
+            client_id=config.get(CONF_CLIENT_ID),
         )
         result = await controller.connect(
             include_vehicles=config.get(CONF_INCLUDE_VEHICLES),

custom_components/tesla_custom/config_flow.py

@@ -2,10 +2,12 @@
 
 from http import HTTPStatus
 import logging
+import os
 
 from homeassistant import config_entries, core, exceptions
 from homeassistant.const import (
     CONF_ACCESS_TOKEN,
+    CONF_CLIENT_ID,
     CONF_DOMAIN,
     CONF_SCAN_INTERVAL,
     CONF_TOKEN,
@@ -24,6 +26,9 @@
     ATTR_POLLING_POLICY_ALWAYS,
     ATTR_POLLING_POLICY_CONNECTED,
     ATTR_POLLING_POLICY_NORMAL,
+    CONF_API_PROXY_CERT,
+    CONF_API_PROXY_ENABLE,
+    CONF_API_PROXY_URL,
     CONF_ENABLE_TESLAMATE,
     CONF_EXPIRATION,
     CONF_INCLUDE_ENERGYSITES,
@@ -51,13 +56,35 @@ def __init__(self) -> None:
         """Initialize the tesla flow."""
         self.username = None
         self.reauth = False
+        self.use_proxy = False
 
     async def async_step_import(self, import_config):
         """Import a config entry from configuration.yaml."""
         return await self.async_step_user(import_config)
 
     async def async_step_user(self, user_input=None):
         """Handle the start of the config flow."""
+        data_schema = vol.Schema(
+            {
+                vol.Required(CONF_API_PROXY_ENABLE, default=False): bool,
+            }
+        )
+
+        if user_input is None:
+            return self.async_show_form(
+                step_id="user",
+                data_schema=data_schema,
+            )
+
+        # in case we import a config entry from configuration.yaml
+        if CONF_API_PROXY_CERT in user_input:
+            return await self.async_step_credentials(user_input)
+
+        self.use_proxy = user_input[CONF_API_PROXY_ENABLE]
+        return await self.async_step_credentials()
+
+    async def async_step_credentials(self, user_input=None):
+        """Handle the second step of the config flow."""
         errors = {}
 
         if user_input is not None:
@@ -87,8 +114,8 @@ async def async_step_user(self, user_input=None):
                 )
 
         return self.async_show_form(
-            step_id="user",
-            data_schema=self._async_schema(),
+            step_id="credentials",
+            data_schema=self._async_schema(api_proxy_enable=self.use_proxy),
             errors=errors,
             description_placeholders={},
         )
@@ -106,9 +133,10 @@ def async_get_options_flow(config_entry):
         return OptionsFlowHandler(config_entry)
 
     @callback
-    def _async_schema(self):
+    def _async_schema(self, api_proxy_enable: bool):
         """Fetch schema with defaults."""
-        return vol.Schema(
+
+        schema = vol.Schema(
             {
                 vol.Required(CONF_USERNAME, default=self.username): str,
                 vol.Required(CONF_TOKEN): str,
@@ -118,6 +146,47 @@ def _async_schema(self):
             }
         )
 
+        if api_proxy_enable:
+            # autofill fields if HTTP Proxy is running as addon
+            if "SUPERVISOR_TOKEN" in os.environ:
+                api_proxy_cert = "/share/tesla/selfsigned.pem"
+
+                # find out if addon is running from normal repo or local
+                req = httpx.get(
+                    "http://supervisor/addons",
+                    headers={
+                        "Authorization": f"Bearer {os.environ['SUPERVISOR_TOKEN']}"
+                    },
+                )
+                for addon in req.json()["data"]["addons"]:
+                    if addon["name"] == "Tesla HTTP Proxy":
+                        addon_slug = addon["slug"]
+                        break
+                if not addon_slug:
+                    _LOGGER.warning("Unable to communicate with Tesla HTTP Proxy addon")
+
+                # read Client ID from addon
+                req = httpx.get(
+                    f"http://supervisor/addons/{addon_slug}/info",
+                    headers={
+                        "Authorization": f"Bearer {os.environ['SUPERVISOR_TOKEN']}"
+                    },
+                )
+                client_id = req.json()["data"]["options"]["client_id"]
+                api_proxy_url = "https://" + req.json()["data"]["hostname"]
+
+            else:
+                api_proxy_url = client_id = api_proxy_cert = None
+
+            schema = schema.extend(
+                {
+                    vol.Required(CONF_API_PROXY_URL, default=api_proxy_url): str,
+                    vol.Required(CONF_API_PROXY_CERT, default=api_proxy_cert): str,
+                    vol.Required(CONF_CLIENT_ID, default=client_id): str,
+                }
+            )
+        return schema
+
     @callback
     def _async_entry_for_username(self, username):
         """Find an existing entry for a username."""
@@ -196,6 +265,9 @@ async def validate_input(hass: core.HomeAssistant, data) -> dict:
             expiration=data.get(CONF_EXPIRATION, 0),
             auth_domain=data.get(CONF_DOMAIN, AUTH_DOMAIN),
             polling_policy=data.get(CONF_POLLING_POLICY, DEFAULT_POLLING_POLICY),
+            api_proxy_cert=data.get(CONF_API_PROXY_CERT),
+            api_proxy_url=data.get(CONF_API_PROXY_URL),
+            client_id=data.get(CONF_CLIENT_ID),
         )
         result = await controller.connect(test_login=True)
         config[CONF_TOKEN] = result["refresh_token"]
@@ -205,6 +277,9 @@ async def validate_input(hass: core.HomeAssistant, data) -> dict:
         config[CONF_DOMAIN] = data.get(CONF_DOMAIN, AUTH_DOMAIN)
         config[CONF_INCLUDE_VEHICLES] = data[CONF_INCLUDE_VEHICLES]
         config[CONF_INCLUDE_ENERGYSITES] = data[CONF_INCLUDE_ENERGYSITES]
+        config[CONF_API_PROXY_URL] = data.get(CONF_API_PROXY_URL)
+        config[CONF_API_PROXY_CERT] = data.get(CONF_API_PROXY_CERT)
+        config[CONF_CLIENT_ID] = data.get(CONF_CLIENT_ID)
 
     except IncompleteCredentials as ex:
         _LOGGER.error("Authentication error: %s %s", ex.message, ex)

custom_components/tesla_custom/const.py

@@ -7,6 +7,9 @@
 CONF_POLLING_POLICY = "polling_policy"
 CONF_WAKE_ON_START = "enable_wake_on_start"
 CONF_ENABLE_TESLAMATE = "enable_teslamate"
+CONF_API_PROXY_ENABLE = "api_proxy_enable"
+CONF_API_PROXY_URL = "api_proxy_url"
+CONF_API_PROXY_CERT = "api_proxy_cert"
 DOMAIN = "tesla_custom"
 ATTRIBUTION = "Data provided by Tesla"
 DATA_LISTENER = "listener"

custom_components/tesla_custom/manifest.json

@@ -24,6 +24,6 @@
   "iot_class": "cloud_polling",
   "issue_tracker": "https://github.com/alandtse/tesla/issues",
   "loggers": ["teslajsonpy"],
-  "requirements": ["teslajsonpy==3.9.11"],
+  "requirements": ["teslajsonpy==3.10.1"],
   "version": "3.19.11"
 }

custom_components/tesla_custom/strings.json

@@ -11,15 +11,25 @@
     },
     "step": {
       "user": {
+        "data": {
+          "api_proxy_enable": "Use Fleet API Proxy"
+        },
+        "description": "Newer vehicles may require an API Proxy.  If this applies to\r\n you, install the Tesla HTTP Proxy first.",
+        "title": "Tesla - Configuration"
+      },
+      "credentials": {
         "data": {
           "mfa": "MFA Code (optional)",
           "password": "Password",
           "username": "Email",
           "token": "Refresh Token",
           "include_vehicles": "Include Vehicles",
-          "include_energysites": "Include Energy Sites"
+          "include_energysites": "Include Energy Sites",
+          "api_proxy_url": "Proxy URL",
+          "api_proxy_cert": "Proxy SSL certificate",
+          "client_id": "Tesla developer Client ID"
         },
-        "description": "Use 'Auth App for Tesla' on iOS, 'Tesla Tokens' on Android\r\n or 'teslafi.com' to create a refresh token and enter it below.\r\n Vehicle(s) are forced awake for setup.",
+        "description": "Use 'Auth App for Tesla' on iOS or 'Tesla Tokens' on Android\r\n to create a refresh token and enter it below.\r\n Vehicle(s) are forced awake for setup.",
         "title": "Tesla - Configuration"
       }
     }

custom_components/tesla_custom/translations/en.json

@@ -11,15 +11,25 @@
     },
     "step": {
       "user": {
+        "data": {
+          "api_proxy_enable": "Use Fleet API Proxy"
+        },
+        "description": "Newer vehicles may require an API Proxy.  If this applies to\r\n you, install the Tesla HTTP Proxy first.",
+        "title": "Tesla - Configuration"
+      },
+      "credentials": {
         "data": {
           "mfa": "MFA Code (optional)",
           "password": "Password",
           "username": "Email",
           "token": "Refresh Token",
           "include_vehicles": "Include Vehicles",
-          "include_energysites": "Include Energy Sites"
+          "include_energysites": "Include Energy Sites",
+          "api_proxy_url": "Proxy URL",
+          "api_proxy_cert": "Proxy SSL certificate",
+          "client_id": "Tesla developer Client ID"
         },
-        "description": "Use 'Auth App for Tesla' on iOS, 'Tesla Tokens' on Android\r\n or 'teslafi.com' to create a refresh token and enter it below.\r\n Vehicle(s) are forced awake for setup.",
+        "description": "Use 'Auth App for Tesla' on iOS or 'Tesla Tokens' on Android\r\n to create a refresh token and enter it below.\r\n Vehicle(s) are forced awake for setup.",
         "title": "Tesla - Configuration"
       }
     }

pyproject.toml

@@ -7,7 +7,7 @@ license = "Apache-2.0"
 
 [tool.poetry.dependencies]
 python = "^3.10"
-teslajsonpy = "3.9.11"
+teslajsonpy = "3.10.1"
 
 
 [tool.poetry.group.dev.dependencies]
@@ -21,6 +21,7 @@ pydocstyle = ">=6.0.0"
 prospector = { extras = ["with_all"], version = ">=1.3.1" }
 aiohttp_cors = ">=0.7.0"
 pytest-asyncio = ">=0.20.3"
+pytest-httpx = ">=0.24.0"
 
 [build-system]
 requires = ["poetry-core>=1.0.0"]