Glad to help! @estebanp did all the heavy lifting and deserves most of the credit 
Is anyone else finding this is no longer working again? Tesla seem to keep moving the goalposts in terms of what they allow through the WAF in front of their auth service. I have several different tools I modified in different ways and slowly over the last few days each has stopped working, either getting a ânot authorizedâ response, or worse, just blocked and times out with no response at all.
Just tested mine and it still is working. 20.49.0.
Itâs seemingly entirely random. I canât get this integration to work on two different HA installs at the moment, but one of those installs is able to use Node Red to renew a refresh token and the other is not.
Hello. I have spent some time pulling the information in this thread out into a custom component repository, and I have changed the configuration to use config flow rather than yaml, as we want to add it to some âlocked downâ home assistant instances my organisation administrates.
With a small bit more work I could release it on github and make it available to install through HACS.
I wanted to check here and with @estebanp that this hasnât already been done somewhere - I donât want to duplicate anything or take credit for other peopleâs work. I think it would be easier to manage and keep up to date in a repository somewhere.
I donât have a tesla or powerwall myself so Iâm not actually an end user.
While testing I am getting auth errors that include <title>Tesla SSO â Page Not Found</title>
@racingsnake and @peteretep mine is also still working (20.49.0). I got one failure to switch in the last month, looks like the gateway needed a restart (version update). If you enable logging and post the log (filtering sensitive information), I could provide some hints on what could be happening. Try restarting the PW.
@peteretep it has not been added to HACS and is not shared in a public github. Can you consider adding it to the official âTesla Powerwallâ component: https://www.home-assistant.io/integrations/powerwall/ instead of HACS?
If that is too much work, I am fine if you go for HACS.
There are many different version of the code on this thread - and Iâm not 100% sure I am running the most up to date/working version.
Could you once more share your working code - then I will add it to a repository as a custom component, make changes for config flow. Once I have done that I could look into the tesla integration, but I am not at all sure I will have time to do so
Thanks!
"""
Monitors and controls the Tesla gateway.
"""
import logging
import aiohttp
import asyncio
import async_timeout
import base64
import hashlib
import json
import os
import re
import time
from urllib.parse import parse_qs
import voluptuous as vol
from homeassistant.const import (
CONF_USERNAME,
CONF_PASSWORD,
CONF_ACCESS_TOKEN
)
from homeassistant.helpers.aiohttp_client import async_get_clientsession
import homeassistant.helpers.config_validation as cv
DOMAIN = 'tesla_gateway'
CONF_REFRESH_TOKEN = "refresh_token"
_LOGGER = logging.getLogger(__name__)
DEFAULT_TIMEOUT = 100
CONFIG_SCHEMA = vol.Schema({
DOMAIN: vol.Schema({
vol.Required(CONF_USERNAME): cv.string,
vol.Required(CONF_PASSWORD): cv.string,
vol.Optional(CONF_ACCESS_TOKEN, default=''): cv.string,
vol.Optional(CONF_REFRESH_TOKEN, default=''): cv.string,
}),
}, extra=vol.ALLOW_EXTRA)
tesla_base_url = 'https://owner-api.teslamotors.com'
tesla_auth_url = 'https://auth.tesla.com'
step_max_attempts = 7
step_attempt_sleep = 3
TESLA_CLIENT_ID = '81527cff06843c8634fdc09e8ac0abefb46ac849f38fe1e431c2ef2106796384'
@asyncio.coroutine
def async_setup(hass, config):
# Tesla gateway is SSL but has no valid certificates
websession = async_get_clientsession(hass, verify_ssl=False)
domain_config = config[DOMAIN]
conf_user = domain_config[CONF_USERNAME]
conf_password = domain_config[CONF_PASSWORD]
@asyncio.coroutine
def SSO_login():
# Code extracted from https://github.com/enode-engineering/tesla-oauth2/blob/2414d74a50f38ab7b3ad5424de4e867ac2709dcf/tesla.py
# Login process explained at https://tesla-api.timdorr.com/api-basics/authentication
authorize_url = tesla_auth_url + '/oauth2/v3/authorize'
callback_url = tesla_auth_url + '/void/callback'
headers = {
"User-Agent": "curl",
"x-tesla-user-agent": "TeslaApp/3.10.9-433/adff2e065/android/10",
"X-Requested-With": "com.teslamotors.tesla",
}
verifier_bytes = os.urandom(86)
code_verifier = base64.urlsafe_b64encode(verifier_bytes).rstrip(b"=")
code_challenge = base64.urlsafe_b64encode(hashlib.sha256(code_verifier).digest()).rstrip(b"=").decode("utf-8")
state = base64.urlsafe_b64encode(os.urandom(16)).rstrip(b"=").decode("utf-8")
params = (
("client_id", "ownerapi"),
("code_challenge", code_challenge),
("code_challenge_method", "S256"),
("redirect_uri", callback_url),
("response_type", "code"),
("scope", "openid email offline_access"),
("state", state),
)
try:
# Step 1: Obtain the login page
_LOGGER.debug('Step 1: GET %s\nparams %s', authorize_url, params)
for attempt in range(step_max_attempts):
with async_timeout.timeout(DEFAULT_TIMEOUT, loop=hass.loop):
response = yield from websession.get(authorize_url,
headers=headers,
params=params,
raise_for_status=False)
returned_text = yield from response.text()
if response.status == 200 and "<title>" in returned_text:
crsf_regex_result = re.search(r'name="_csrf".+value="([^"]+)"', returned_text)
if crsf_regex_result:
_LOGGER.debug('Step 1: Success on attempt %d', attempt)
break
_LOGGER.warning('Step 1: Error %d on attempt %d, call %s:\n%s', response.status, attempt, response.url, returned_text)
time.sleep(step_attempt_sleep)
else:
raise ValueError('Step 1: failed after %d attempts, last response %s:\n%s', step_max_attempts, response.status, returned_text)
# Step 2: Obtain an authorization code
csrf = crsf_regex_result.group(1)
transaction_id = re.search(r'name="transaction_id".+value="([^"]+)"', returned_text).group(1)
body = {
"_csrf": csrf,
"_phase": "authenticate",
"_process": "1",
"transaction_id": transaction_id,
"cancel": "",
"identity": conf_user,
"credential": conf_password,
}
_LOGGER.debug('Step 2: POST %s\nparams: %s\nbody: %s', authorize_url, params, body)
for attempt in range(step_max_attempts):
with async_timeout.timeout(DEFAULT_TIMEOUT, loop=hass.loop):
response = yield from websession.post(authorize_url,
headers=headers,
params=params,
data=body,
raise_for_status=False,
allow_redirects=False)
returned_text = yield from response.text()
if "We could not sign you in" in returned_text and response.status == 401:
raise ValueError('Step 2: Invalid credentials. Error %d on call %s:\n%s', response.status, response.url, returned_text)
if response.status == 302 or "<title>" in returned_text:
_LOGGER.debug('Step 2: Success on attempt %d', attempt)
break
_LOGGER.warning('Step 2: Error %d on call %s:\n%s', response.status, response.url, returned_text)
time.sleep(step_attempt_sleep)
else:
raise ValueError('Step 2: failed after %d attempts, last response %s:\n%s', step_max_attempts, response.status, returned_text)
is_mfa = True if response.status == 200 and "/mfa/verify" in returned_text else False
if is_mfa:
raise ValueError('Multi-factor authentication enabled for the account and not supported')
# Step 3: Exchange authorization code for bearer token
code = parse_qs(response.headers["location"])[callback_url + '?code']
token_url = tesla_auth_url + '/oauth2/v3/token'
body = {
"grant_type": "authorization_code",
"client_id": "ownerapi",
"code_verifier": code_verifier.decode("utf-8"),
"code": code,
"redirect_uri": callback_url
}
_LOGGER.debug('Step 3: POST %s', token_url)
with async_timeout.timeout(DEFAULT_TIMEOUT, loop=hass.loop):
response = yield from websession.post(token_url,
headers=headers,
data=body,
raise_for_status=False)
returned_json = yield from response.json()
access_token = returned_json['access_token']
domain_config[CONF_ACCESS_TOKEN] = access_token
domain_config[CONF_REFRESH_TOKEN] = returned_json['refresh_token']
return access_token
except asyncio.TimeoutError:
_LOGGER.warning('Timeout call %s.', response.url)
except aiohttp.ClientError:
_LOGGER.error('Client error %s.', response.url)
return None
@asyncio.coroutine
def SSO_refresh_token():
token_oauth2_url = tesla_auth_url + '/oauth2/v3/token'
headers = {
"User-Agent": "curl",
"x-tesla-user-agent": "TeslaApp/3.10.9-433/adff2e065/android/10",
"X-Requested-With": "com.teslamotors.tesla",
}
body = {
"grant_type": "refresh_token",
"refresh_token": domain_config[CONF_REFRESH_TOKEN],
"client_id": "ownerapi",
"scope": "openid email offline_access",
}
with async_timeout.timeout(DEFAULT_TIMEOUT, loop=hass.loop):
response = yield from websession.post(token_oauth2_url,
headers=headers,
data=body,
raise_for_status=False)
returned_json = yield from response.json()
access_token = returned_json['access_token']
domain_config[CONF_ACCESS_TOKEN] = access_token
domain_config[CONF_REFRESH_TOKEN] = returned_json['refresh_token']
return access_token
@asyncio.coroutine
def OWNER_get_token(access_token):
try:
token_oauth_url = tesla_base_url + '/oauth/token'
headers = {
"User-Agent": "curl",
"authorization": "bearer " + access_token,
}
body = {
"grant_type": "urn:ietf:params:oauth:grant-type:jwt-bearer",
"client_id": TESLA_CLIENT_ID,
}
with async_timeout.timeout(DEFAULT_TIMEOUT, loop=hass.loop):
response = yield from websession.post(token_oauth_url,
headers=headers,
data=body,
raise_for_status=False)
returned_json = yield from response.json()
owner_access_token = returned_json["access_token"]
return owner_access_token
except asyncio.TimeoutError:
_LOGGER.warning('Timeout call %s.', response.url)
except aiohttp.ClientError:
_LOGGER.error('Client error %s.', response.url)
return None
@asyncio.coroutine
def OWNER_revoke(owner_token):
revoke_url = tesla_base_url + '/oauth/revoke'
headers = {'Content-type': 'application/json'}
body = {
'token': owner_token
}
try:
with async_timeout.timeout(DEFAULT_TIMEOUT, loop=hass.loop):
response = yield from websession.post(revoke_url,
headers=headers,
json=body,
raise_for_status=False)
if response.status != 200:
returned_text = yield from response.text()
_LOGGER.warning('Error %d on call %s:\n%s', response.status, response.url, returned_text)
else:
_LOGGER.debug('revoke completed')
return True
except asyncio.TimeoutError:
_LOGGER.warning('Timeout call %s.', response.url)
except aiohttp.ClientError:
_LOGGER.error('Client error %s.', response.url)
return False
@asyncio.coroutine
def get_energy_site_id(owner_token):
list_url = tesla_base_url + '/api/1/products'
headers = {
'Authorization': 'Bearer ' + owner_token
}
body = {}
try:
with async_timeout.timeout(DEFAULT_TIMEOUT, loop=hass.loop):
response = yield from websession.get(list_url,
headers=headers,
json=body,
raise_for_status=False)
if response.status != 200:
returned_text = yield from response.text()
_LOGGER.warning('Error %d on call %s:\n%s', response.status, response.url, returned_text)
else:
returned_json = yield from response.json()
for r in returned_json['response']:
if 'energy_site_id' in r:
return r['energy_site_id']
return None
except asyncio.TimeoutError:
_LOGGER.warning('Timeout call %s.', response.url)
except aiohttp.ClientError:
_LOGGER.error('Client error %s.', response.url)
return None
@asyncio.coroutine
def set_operation(owner_token,energy_site_id,service_data):
operation_url = tesla_base_url + '/api/1/energy_sites/{}/operation'.format(energy_site_id)
headers = {
'Content-type': 'application/json',
'Authorization': 'Bearer ' + owner_token
}
body = {
'default_real_mode': service_data['real_mode'],
'backup_reserve_percent':int(service_data['backup_reserve_percent'])
}
try:
with async_timeout.timeout(DEFAULT_TIMEOUT, loop=hass.loop):
response = yield from websession.post(operation_url,
json=body,
headers=headers,
raise_for_status=False)
if response.status != 200:
returned_text = yield from response.text()
_LOGGER.warning('Error %d on call %s:\n%s', response.status, response.url, returned_text)
else:
returned_json = yield from response.json()
_LOGGER.debug('set operation successful, request: %s response: %s', body, returned_json)
except asyncio.TimeoutError:
_LOGGER.warning('Timeout call %s.', response.url)
except aiohttp.ClientError:
_LOGGER.error('Client error %s.', response.url)
@asyncio.coroutine
def get_owner_api_token():
access_token = domain_config[CONF_ACCESS_TOKEN]
if not access_token:
access_token = yield from SSO_login()
else:
access_token = yield from SSO_refresh_token()
if not access_token:
return None
owner_token = yield from OWNER_get_token(access_token)
return owner_token
@asyncio.coroutine
def async_set_operation(service):
owner_token = yield from get_owner_api_token()
if owner_token:
energy_site_id = yield from get_energy_site_id(owner_token)
if energy_site_id:
yield from set_operation(owner_token, energy_site_id, service.data)
yield from OWNER_revoke(owner_token)
hass.services.async_register(DOMAIN, 'set_operation', async_set_operation)
@asyncio.coroutine
def set_reserve(owner_token,energy_site_id,service_data):
operation_url = tesla_base_url + '/api/1/energy_sites/{}/backup'.format(energy_site_id)
headers = {
'Content-type': 'application/json',
'Authorization': 'Bearer ' + owner_token
}
body = {
'backup_reserve_percent': int(service_data['reserve_percent'])
}
_LOGGER.debug(body)
try:
with async_timeout.timeout(DEFAULT_TIMEOUT, loop=hass.loop):
response = yield from websession.post(operation_url,
json=body,
headers=headers,
raise_for_status=False)
if response.status != 200:
returned_text = yield from response.text()
_LOGGER.warning('Error %d on call %s:\n%s', response.status, response.url, returned_text)
else:
returned_json = yield from response.json()
_LOGGER.debug('set reserve successful, response: %s', returned_json)
except asyncio.TimeoutError:
_LOGGER.warning('Timeout call %s.', response.url)
except aiohttp.ClientError:
_LOGGER.error('Client error %s.', response.url)
@asyncio.coroutine
def async_set_reserve(service):
owner_token = yield from get_owner_api_token()
if owner_token:
energy_site_id = yield from get_energy_site_id(owner_token)
if energy_site_id:
yield from set_reserve(owner_token, energy_site_id, service.data)
yield from OWNER_revoke(owner_token)
hass.services.async_register(DOMAIN, 'set_reserve', async_set_reserve)
return True
@estebanp hi, I am trying to follow all that you have done but struggling. I can see various different explanations but they are too complex for a n00b like me. is it possible that you put it in simple terms - what code do I need to put where?
I have the powerwall integration that comes as standard in HA working, reading my gatewayâs numbers with the local physical IP, but I would like to be able to set the mode / reserve percent in order to charge optimally, which I understand is what I need your code to be able to do.
many thanks in advance
Ian
Correct, the powerall integration from HA does not have the functionality provided in this custom component (change operation and/or reserve %). There is some history behind that, it can be found here: https://github.com/vloschiavo/powerwall2/issues/26
Yeah, the instructions are not simple for everyone.
Option A, learn how to setup a custom component (even a âhello worldâ) and then take the pieces from this thread, look at Add support for Tesla Powerwall Be patient when going through the thread, it was explained how to set it up.
Option B, wait for @peteretep to set it up with config flow and HACS so it will be easier for you to setup.
thanks. I already got solcast custom component working following the instructions provided elsewhere for that, so I think I am ok on that . I already looked at your post Add support for Tesla Powerwall in depth , it seems very clear, and was getting ready to give it a go (I didnât try it yet) but then working my down the thread to your post at Add support for Tesla Powerwall has an order of magnitude more code, and no references to the separate files and where they go, thats why I became confused about what code is needed (and where).
thanks
Ian
@peteretep Iâd happily give it a go if you need a beta tester who has a powerwall to go through your HACS / custom component installation path for this function. I do not have anything setup for changing my powerwall mode at present, so I would not be losing anything by giving it a go.
I see the confusion, most of my posts are for the __init__.py file. If you see some python code, it is for that file.
1 Like
I have created a repository here, adding the latest code:
I havenât tested installation or anything else yet, and I havenât pushed any of my config flow work.
Please feel free to try it out and contribute any improvements you can.
Please note - I donât have a powerwall or tesla myself, I am working on this mainly to get experience with Home Assistant custom component development. @estebanp if you wish I can change the ownership of the repo to you.
2 Likes
installation worked like a charm using HACS. Config was trivial - just adding my tesla.com user and pwd to the file. nothing more needed.
A first try with
service: tesla_gateway.set_reserve
data:
reserve_percent: 20
worked as expected, I saw the change to the reserve percent in the tesla app a couple of minutes later. However, I did a second change to 10 percent a few minutes after that: at this exact same time, the powerwall lost its connection to tesla, and the app lost connection to the powerwall.
I got onto the powerwall locally and re-initiated its connection to tesla, but the app still couldnât see the powerwall. I then tried signing out of the app and now the app cannot sign back in (same credentials of course). Iâm going to have to contact tesla support, I wonder if my account has locked 
- its it possible that the authentication from this tool, although with valid credentials, is causing something in tesla-land to throw an alert that an account breach is taking place?
edit - whilst I was typing this the app recovered. have just done another set to 0% and its worked as expected. wonder if there is something watching out for frequency of logins. or could have been a one-off glitch, lets see. now to try to automate the âif not sunny tomorrow then set reserve percent high during off-peak periodâ logic!
Thanks for testing. This morning I updated it to work with configuration with the user interface, so if you update it you will have to reinstall that way I think.
I canât comment on any of the problems youâre having Iâm afraid - I donât have anything to test with myself, and am just repackaging @estebanpâs code. I hope it doesnât break your stuff!
Very glad the installation went smoothly.
Regarding the authentication, the component authenticates one time and then renews authentication token every time its called. The token doesnt need to be renewed every time, but we would need to do a time-based approach and trigger renewal in the background if it is not called.
I have seen some weird cases, usually around a firmware update. I call it ~2 times a day and has behaved fairly stable (maybe around 5% instability).
Honestly, I dunno what Tesla could be detecting there⌠the authentication process was practically reversed engineered and âguessedâ. If you look at the code, it does things that are weird like redirecting the login to an inexistent page. It also guesses some values to pass on some fields⌠I highly doubt the app is doing that. Until Tesla releases an official public API, this is hacky territory.
I suggest you keep an eye on it for a couple of days and let us know when you see the problem again. Know this is somehow expected to eventually fail and stop working at some pointâŚ
I am fine if you stay as the repository owner. Thanks for putting it into HACS!
Itâs not quite in HACS, but it is compatible with that installation method.
1 Like