feat: add teslaproxy to capture oauth credentials

BREAKING CHANGE: Change return of connect and get_tokens to return
expiration time.

Pipfile

@@ -14,14 +14,14 @@ pytest-cov = "*"
 tox = "*"
 tox-pipenv = "*"
 twine = "*"
-python-semantic-release = "*"
 black = "*"
 
 [packages]
 aiohttp = "*"
 backoff = "*"
-bs4 = "*"
+beautifulsoup4 = "*"
 wrapt = "*"
+authcaptureproxy = "0.4.0"
 
 [pipenv]
 allow_prereleases = true

setup.py

@@ -24,12 +24,12 @@
 URL = "https://github.com/zabuldon/teslajsonpy"
 EMAIL = "sergey.isachenkol@bool.by"
 AUTHOR = "Sergey Isachenko"
-REQUIRES_PYTHON = ">=3.6"
+REQUIRES_PYTHON = ">=3.6.1"
 LICENSE = "Apache-2.0"
 VERSION = None
 
 # What packages are required for this module to be executed?
-REQUIRED = ["aiohttp", "backoff", "beautifulsoup4", "wrapt"]
+REQUIRED = ["aiohttp", "authcaptureproxy", "backoff", "beautifulsoup4", "wrapt"]
 
 # What packages are optional?
 EXTRAS = {

teslajsonpy/connection.py

@@ -37,7 +37,7 @@ def __init__(
         password: Text = None,
         access_token: Text = None,
         refresh_token: Text = None,
-        authorization_token=None,
+        authorization_token: Text = None,
         expiration: int = 0,
     ) -> None:
         """Initialize connection object."""
@@ -80,6 +80,10 @@ async def get(self, command):
     async def post(self, command, method="post", data=None):
         """Post data to API."""
         now = calendar.timegm(datetime.datetime.now().timetuple())
+        _LOGGER.debug(
+            "Token expiration in %s",
+            str(datetime.timedelta(seconds=self.expiration - now)),
+        )
         if now > self.expiration:
             self.token_refreshed = False
             auth = {}
@@ -91,11 +95,13 @@ async def post(self, command, method="post", data=None):
                     and not self.sso_oauth.get("refresh_token")
                 )
             ):
-                _LOGGER.debug("Getting sso auth code using credentials")
                 if self.email and self.password:
+                    _LOGGER.debug("Getting sso auth code using credentials")
                     self.code = await self.get_authorization_code(
                         self.email, self.password
                     )
+                else:
+                    _LOGGER.debug("Using existing authorization code")
                 auth = await self.get_sso_auth_token(self.code)
             elif self.sso_oauth.get("refresh_token") and now > self.sso_oauth.get(
                 "expires_in", 0
@@ -104,23 +110,39 @@ async def post(self, command, method="post", data=None):
                 auth = await self.refresh_access_token(
                     refresh_token=self.sso_oauth.get("refresh_token")
                 )
-            if auth:
+            if auth and all(
+                [
+                    auth.get(item)
+                    for item in ["access_token", "refresh_token", "expires_in"]
+                ]
+            ):
                 self.sso_oauth = {
                     "access_token": auth["access_token"],
                     "refresh_token": auth["refresh_token"],
                     "expires_in": auth["expires_in"] + now,
                 }
-                _LOGGER.debug("Saving new auth info %s", self.sso_oauth)
+                _LOGGER.debug("Saved new auth info %s", self.sso_oauth)
             else:
                 _LOGGER.debug("Unable to refresh sso oauth token")
+                if auth:
+                    _LOGGER.debug("Auth returned %s", auth)
+                self.code = None
                 self.sso_oauth = {}
                 raise IncompleteCredentials("Need oauth credentials")
             auth = await self.get_bearer_token(
                 access_token=self.sso_oauth.get("access_token")
             )
-            self.__sethead(
-                access_token=auth["access_token"], expires_in=auth["expires_in"]
-            )
+            _LOGGER.debug("Received bearer token %s", auth)
+            if auth.get("created_at"):
+                # use server time if available
+                self.__sethead(
+                    access_token=auth["access_token"],
+                    expiration=auth["expires_in"] + auth["created_at"],
+                )
+            else:
+                self.__sethead(
+                    access_token=auth["access_token"], expires_in=auth["expires_in"]
+                )
             self.refresh_token = auth["refresh_token"]
             self.token_refreshed = True
             _LOGGER.debug("Successfully refreshed oauth")
@@ -347,7 +369,29 @@ async def get_authorization_code(self, email, password) -> Text:
         if not (email and password):
             _LOGGER.debug("No email or password for login; unable to login.")
             return
+        url = self.get_authorization_code_link(new=True)
+        resp = await self.websession.get(url)
+        html = await resp.text()
+        soup: BeautifulSoup = BeautifulSoup(html, "html.parser")
+        data = get_inputs(soup)
+        data["identity"] = self.email
+        data["credential"] = self.password
+        resp = await self.websession.post(url, data=data)
+        _process_resp(resp)
+        code_url = URL(resp.history[-1].url)
+        return code_url.query.get("code")
 
+    def get_authorization_code_link(self, new=False) -> yarl.URL:
+        """Get authorization code url for the oauth3 login method."""
+        # https://tesla-api.timdorr.com/api-basics/authentication#step-2-obtain-an-authorization-code
+        if new:
+            self.code_verifier: Text = secrets.token_urlsafe(64)
+            self.code_challenge = str(
+                base64.urlsafe_b64encode(
+                    hashlib.sha256(self.code_verifier.encode()).hexdigest().encode()
+                ),
+                "utf-8",
+            )
         state = secrets.token_urlsafe(64)
         query = {
             "client_id": "ownerapi",
@@ -360,17 +404,7 @@ async def get_authorization_code(self, email, password) -> Text:
         }
         url = yarl.URL("https://auth.tesla.com/oauth2/v3/authorize")
         url = url.update_query(query)
-        _LOGGER.debug("Getting sso auth token from %s", url)
-        resp = await self.websession.get(url)
-        html = await resp.text()
-        soup: BeautifulSoup = BeautifulSoup(html, "html.parser")
-        data = get_inputs(soup)
-        data["identity"] = self.email
-        data["credential"] = self.password
-        resp = await self.websession.post(url, data=data)
-        _process_resp(resp)
-        code_url = URL(resp.history[-1].url)
-        return code_url.query.get("code")
+        return url
 
     async def get_sso_auth_token(self, code):
         """Get sso auth token."""

teslajsonpy/controller.py

@@ -8,11 +8,12 @@
 import asyncio
 import logging
 import time
-from typing import Callable, Optional, Text, Tuple
+from typing import Callable, Dict, Optional, Text
 
 from aiohttp import ClientConnectorError
 import backoff
 import wrapt
+from yarl import URL
 
 from teslajsonpy.connection import Connection
 from teslajsonpy.const import (
@@ -222,7 +223,12 @@ def __init__(
 
         """
         self.__connection = Connection(
-            websession, email, password, access_token, refresh_token, expiration
+            websession=websession,
+            email=email,
+            password=password,
+            access_token=access_token,
+            refresh_token=refresh_token,
+            expiration=expiration,
         )
         self.__components = []
         self._update_interval: int = update_interval
@@ -252,7 +258,7 @@ def __init__(
 
     async def connect(
         self, test_login=False, wake_if_asleep=False, filtered_vins=None
-    ) -> Tuple[Text, Text]:
+    ) -> Dict[Text, Text]:
         """Connect controller to Tesla.
 
         Args
@@ -261,7 +267,7 @@ async def connect(
             filtered_vins (list, optional): If not empty, filters the cars by the provided VINs.
 
         Returns
-            Tuple[Text, Text]: Returns the refresh_token and access_token
+            Dict[Text, Text]: Returns the refresh_token, access_token, and expires_in time
 
         """
 
@@ -306,7 +312,11 @@ async def connect(
                 await asyncio.gather(*tasks)
             except (TeslaException, RetryLimitError):
                 pass
-        return (self.__connection.refresh_token, self.__connection.access_token)
+        return {
+            "refresh_token": self.__connection.refresh_token,
+            "access_token": self.__connection.access_token,
+            "expiration": self.__connection.expiration,
+        }
 
     def is_token_refreshed(self) -> bool:
         """Return whether token has been changed and not retrieved.
@@ -317,17 +327,21 @@ def is_token_refreshed(self) -> bool:
         """
         return self.__connection.token_refreshed
 
-    def get_tokens(self) -> Tuple[Text, Text]:
-        """Return refresh and access tokens.
+    def get_tokens(self) -> Dict[Text, Text]:
+        """Return oauth data including refresh and access tokens, and expires time.
 
         This will set the the self.__connection token_refreshed to False.
 
         Returns
-            Tuple[Text, Text]: Returns a tuple of refresh and access tokens
+            Dict[Text, Text]: Returns the refresh_token, access_token, and expires time
 
         """
         self.__connection.token_refreshed = False
-        return (self.__connection.refresh_token, self.__connection.access_token)
+        return {
+            "refresh_token": self.__connection.refresh_token,
+            "access_token": self.__connection.access_token,
+            "expiration": self.__connection.expiration,
+        }
 
     def get_expiration(self) -> int:
         """Return expiration for oauth.
@@ -338,6 +352,14 @@ def get_expiration(self) -> int:
         """
         return self.__connection.expiration
 
+    def get_oauth_url(self) -> URL:
+        """Return oauth url."""
+        return self.__connection.get_authorization_code_link(new=True)
+
+    def set_authorization_code(self, code: Text) -> None:
+        """Set authorization code in Connection."""
+        self.__connection.code = code
+
     def register_websocket_callback(self, callback) -> int:
         """Register callback for websocket messages.