Skip to content

Commit

Permalink
Added #65, fixed #68 and added #71
Browse files Browse the repository at this point in the history
  • Loading branch information
@tdorssers
tdorssers committed Apr 18, 2022
1 parent 08e8cd3 commit e88ec08
Show file tree
Hide file tree
Showing 2 changed files with 61 additions and 18 deletions.
30 changes: 29 additions & 1 deletion README.md
View file
Original file line number Diff line number Diff line change
Expand Up @@ -33,6 +33,9 @@ TeslaPy 2.0.0+ no longer implements headless authentication. The constructor dif
| `cache_loader` | (optional) function that returns the cache dict |
| `cache_dumper` | (optional) function with one argument, the cache dict |
| `sso_base_url` | (optional) URL of SSO service, set to `https://auth.tesla.cn/` if your email is registered in another region |
| `state` | (optional) state string for CSRF protection |
| `code_verifier` | (optional) PKCE code verifier string |
| `app_user_agent` | (optional) X-Tesla-User-Agent string |

TeslaPy 2.1.0+ no longer implements [RFC 7523](https://tools.ietf.org/html/rfc7523) and uses the SSO token for all API requests.

Expand All @@ -43,9 +46,11 @@ The convenience method `api()` uses named endpoints listed in *endpoints.json* t
| Call | Description |
| --- | --- |
| `request()` | performs API call using relative or absolute URL, serialization and error message handling |
| `authorization_url()` | forms authorization URL with [PKCE](https://oauth.net/2/pkce/) extension |
| `new_code_verifier()` | generates code verifier for [PKCE](https://oauth.net/2/pkce/) |
| `authorization_url()` | forms authorization URL with [PKCE](https://oauth.net/2/pkce/) extension and tries to detect the accounts registered region |
| `fetch_token()` | requests an SSO token using Authorization Code grant with [PKCE](https://oauth.net/2/pkce/) extension |
| `refresh_token()` | requests an SSO token using [Refresh Token](https://oauth.net/2/grant-types/refresh-token/) grant |
| `close()` | remove all requests adapter instances |
| `logout()` | removes token from cache, returns logout URL and optionally signs out using system's default web browser |
| `vehicle_list()` | returns a list of Vehicle objects |
| `battery_list()` | returns a list of Battery objects |
Expand Down Expand Up @@ -189,6 +194,29 @@ print(vehicles[0])
tesla.close()
```

#### Alternative staged

Support for staged authorization has been added to TeslaPy 2.5.0. The keyword arguments `state` and `code_verifier` are accepted by the `Tesla` class constructor, the `authorization_url()` method and the `fetch_token()` method.

```python
import teslapy
# First stage
tesla = teslapy.Tesla('[email protected]')
if not tesla.authorized:
state = tesla.new_state()
code_verifier = tesla.new_code_verifier()
print('Use browser to login. Page Not Found will be shown at success.')
print('Open: ' + tesla.authorization_url(state=state, code_verifier=code_verifier))
tesla.close()
# Second stage
tesla = teslapy.Tesla('[email protected]', state=state, code_verifier=code_verifier)
if not tesla.authorized:
tesla.fetch_token(authorization_response=input('Enter URL after authentication: '))
vehicles = tesla.vehicle_list()
print(vehicles[0])
tesla.close()
```

#### 3rd party authentication apps

TeslaPy 2.4.0+ supports usage of a refresh token obtained by 3rd party [authentication apps](https://teslascope.com/help/generating-tokens). The refresh token is used to obtain an access token and both are cached for persistence, so you only need to supply the refresh token only once.
Expand Down
49 changes: 32 additions & 17 deletions teslapy/__init__.py
View file
Original file line number Diff line number Diff line change
Expand Up @@ -34,6 +34,7 @@
SSO_BASE_URL = 'https://auth.tesla.com/'
SSO_CLIENT_ID = 'ownerapi'
STREAMING_BASE_URL = 'wss://streaming.vn.teslamotors.com/'
APP_USER_AGENT = 'TeslaApp/4.7.0'

# Setup module logging
logger = logging.getLogger(__name__)
Expand Down Expand Up @@ -62,13 +63,18 @@ class Tesla(OAuth2Session):
cache_dumper: (optional) Function with one argument, the cache dict.
sso_base_url: (optional) URL of SSO service, set to `https://auth.tesla.cn/`
if your email is registered in another region.
kwargs: (optional) Extra arguments for the Session constructor.
code_verifier (optional): PKCE code verifier string.
app_user_agent (optional): X-Tesla-User-Agent string.
Extra keyword arguments to pass to OAuth2Session constructor using `kwargs`:
state (optional): A state string for CSRF protection.
"""

def __init__(self, email, verify=True, proxy=None, retry=0, timeout=10,
user_agent=__name__ + '/' + __version__, authenticator=None,
cache_file='cache.json', cache_loader=None, cache_dumper=None,
sso_base_url=None, **kwargs):
sso_base_url=None, code_verifier=None,
app_user_agent=APP_USER_AGENT, **kwargs):
super(Tesla, self).__init__(client_id=SSO_CLIENT_ID, **kwargs)
if not email:
raise ValueError('`email` is not set')
Expand All @@ -81,7 +87,7 @@ def __init__(self, email, verify=True, proxy=None, retry=0, timeout=10,
self.endpoints = {}
self.sso_base_url = sso_base_url or SSO_BASE_URL
self._auto_refresh_url = None
self.code_verifier = None
self.code_verifier = code_verifier
# Set OAuth2Session properties
self.scope = ('openid', 'email', 'offline_access')
self.redirect_uri = SSO_BASE_URL + 'void/callback'
Expand All @@ -90,7 +96,7 @@ def __init__(self, email, verify=True, proxy=None, retry=0, timeout=10,
self.token_updater = self._token_updater
self.mount('https://', requests.adapters.HTTPAdapter(max_retries=retry))
self.headers.update({'Content-Type': 'application/json',
'X-Tesla-User-Agent': 'TeslaApp/4.5.0',
'X-Tesla-User-Agent': app_user_agent,
'User-Agent': user_agent})
self.verify = verify
if proxy:
Expand Down Expand Up @@ -152,30 +158,41 @@ def request(self, method, url, serialize=True, **kwargs):
return response.json(object_hook=JsonDict)
return response.text

def authorization_url(self, url='oauth2/v3/authorize', **kwargs):
@staticmethod
def new_code_verifier():
""" Generate code verifier for PKCE as per RFC 7636 section 4.1 """
result = base64.urlsafe_b64encode(os.urandom(32)).rstrip(b'=')
logger.debug('Generated new code verifier %s.',
result.decode() if isinstance(result, bytes) else result)
return result

def authorization_url(self, url='oauth2/v3/authorize',
code_verifier=None, **kwargs):
""" Overriddes base method to form an authorization URL with PKCE
extension for Tesla's SSO service.
url (optional): Authorization endpoint url.
code_verifier (optional): PKCE code verifier string.
Extra keyword arguments to pass to base method using `kwargs`:
state (optional): A state string for CSRF protection.
Return type: String
Return type: String or None
"""
if self.authorized:
return None
# Generate code verifier and challenge for PKCE (RFC 7636)
self.code_verifier = base64.urlsafe_b64encode(os.urandom(32)).rstrip(b'=')
self.code_verifier = code_verifier or self.new_code_verifier()
unencoded_digest = hashlib.sha256(self.code_verifier).digest()
code_challenge = base64.urlsafe_b64encode(unencoded_digest).rstrip(b'=')
# Prepare for OAuth 2 Authorization Code Grant flow
url = urljoin(self.sso_base_url, url)
kwargs['code_challenge'] = code_challenge
kwargs['code_challenge_method'] = 'S256'
without_hint = super(Tesla, self).authorization_url(url, **kwargs)[0]
without_hint, state = super(Tesla, self).authorization_url(url, **kwargs)
# Detect account's registered region
kwargs['login_hint'] = self.email
kwargs['state'] = state
with_hint = super(Tesla, self).authorization_url(url, **kwargs)[0]
response = self.get(with_hint, allow_redirects=False)
if response.is_redirect:
Expand All @@ -192,6 +209,7 @@ def fetch_token(self, token_url='oauth2/v3/token', **kwargs):
Extra keyword arguments to pass to base method using `kwargs`:
authorization_response (optional): Authorization response URL.
code_verifier (optional): Code verifier cryptographic random string.
Return type: dict
"""
Expand Down Expand Up @@ -240,7 +258,7 @@ def logout(self, sign_out=False):
sign_out (optional): sign out using system's default web browser.
Return type: String
Return type: String or None
"""
if not self.authorized:
return None
Expand Down Expand Up @@ -679,7 +697,7 @@ def api(self, name, **kwargs):
return self.tesla.api(name, pathvars, **kwargs)

def get_calendar_history_data(
self, kind='savings', period='day', start_date=None,
self, kind='energy', period='day', start_date=None,
end_date=time.strftime('%Y-%m-%dT%H:%M:%S.000Z'),
installation_timezone=None, timezone=None, tariff=None):
""" Retrieve live status of product
Expand All @@ -697,14 +715,13 @@ def get_calendar_history_data(
"""
return self.api('CALENDAR_HISTORY_DATA', kind=kind, period=period,
start_date=start_date, end_date=end_date,
timezone=timezone,
installation_timezone=installation_timezone,
tariff=tariff)['response']
timezone=timezone, tariff=tariff)['response']

def get_history_data(
self, kind='savings', period='day', start_date=None,
self, kind='energy', period='day', start_date=None,
end_date=time.strftime('%Y-%m-%dT%H:%M:%S.000Z'),
installation_timezone=None, timezone=None, tariff=None):
installation_timezone=None, timezone=None):
""" Retrieve live status of product
kind: A telemetry type of 'backup', 'energy', 'power',
'self_consumption', 'time_of_use_energy', and
Expand All @@ -716,13 +733,11 @@ def get_history_data(
start_date: The state date in the data requested in the json format
'2021-02-27T07:59:59.999Z'
installation_timezone: Timezone of installation location for 'savings'
tariff: Unclear format use in 'savings' only
"""
return self.api('HISTORY_DATA', kind=kind, period=period,
start_date=start_date, end_date=end_date,
timezone=timezone,
installation_timezone=installation_timezone,
tariff=tariff)['response']
timezone=timezone)['response']

def command(self, name, **kwargs):
""" Wrapper method for product command response error handling """
Expand Down

0 comments on commit e88ec08

Please sign in to comment.