Merge pull request #156 from alandtse/sphinx

Signed-off-by: Alan Tse <alandtse@gmail.com>

Pipfile

@@ -28,7 +28,7 @@ aiohttp = "*"
 backoff = "*"
 beautifulsoup4 = "*"
 wrapt = "*"
-authcaptureproxy = "0.4.0"
+authcaptureproxy = "~=0.4.2"
 
 [pipenv]
 allow_prereleases = true

setup.cfg

@@ -38,8 +38,6 @@ force_grid_wrap=0
 use_parentheses=True
 line_length=88
 indent = "    "
-# by default isort don't check module indexes
-not_skip = __init__.py
 # will group `import x` and `from x import` of the same module.
 force_sort_within_sections = true
 sections = FUTURE,STDLIB,THIRDPARTY,FIRSTPARTY,LOCALFOLDER

teslajsonpy/connection.py

@@ -1,7 +1,8 @@
-#  SPDX-License-Identifier: Apache-2.0
 """
 Python Package for controlling Tesla API.
 
+SPDX-License-Identifier: Apache-2.0
+
 Underlying connection logic.
 
 For more details about this api, please refer to the documentation at
@@ -366,9 +367,17 @@ async def _process_messages() -> None:
     #         }
     #     )
 
-    async def get_authorization_code(self, email, password) -> Text:
+    async def get_authorization_code(
+        self,
+        email: Text,
+        password: Text,
+        mfa_code: Text = "",
+        mfa_device: int = 0,
+        retry_limit: int = 3,
+    ) -> Text:
         """Get authorization code from the oauth3 login method."""
         # https://tesla-api.timdorr.com/api-basics/authentication#step-2-obtain-an-authorization-code
+        # pylint: disable=too-many-locals
         if not (email and password):
             _LOGGER.debug("No email or password for login; unable to login.")
             return
@@ -377,10 +386,68 @@ async def get_authorization_code(self, email, password) -> Text:
         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)
+        data["identity"] = email
+        data["credential"] = password
+        transaction_id: Text = data.get("transaction_id")
+        for attempt in range(retry_limit):
+            _LOGGER.debug("Attempt #%s", attempt)
+            resp = await self.websession.post(url, data=data)
+            _process_resp(resp)
+            if not resp.history:
+                html = await resp.text()
+                if "/mfa/verify" in html:
+                    mfa_resp = await self.websession.get(
+                        "https://auth.tesla.com/oauth2/v3/authorize/mfa/factors",
+                        params={"transaction_id": transaction_id},
+                    )
+                    _process_resp(mfa_resp)
+                    # {
+                    #     "data": [
+                    #         {
+                    #         "dispatchRequired": false,
+                    #         "id": "X-4Y-44e4-b9a4-54e114a13c40",
+                    #         "name": "Pixel",
+                    #         "factorType": "token:software",
+                    #         "factorProvider": "TESLA",
+                    #         "securityLevel": 1,
+                    #         "activatedAt": "2021-02-10T23:53:40.000Z",
+                    #         "updatedAt": "2021-02-10T23:54:20.000Z"
+                    #         }
+                    #     ]
+                    # }
+                    mfa_json = await mfa_resp.json()
+                    if len(mfa_json.get("data", [])) > 1:
+                        factor_id = mfa_json["data"][mfa_device]["id"]
+                    if not mfa_code:
+                        _LOGGER.debug("No MFA provided")
+                        _LOGGER.debug("MFA Devices: %s", mfa_json["data"])
+                        raise IncompleteCredentials(
+                            "MFA Code missing", devices=mfa_json["data"]
+                        )
+                    mfa_resp = await self.websession.post(
+                        "https://auth.tesla.com/oauth2/v3/authorize/mfa/verify",
+                        json={
+                            "transaction_id": transaction_id,
+                            "factor_id": factor_id,
+                            "passcode": mfa_code,
+                        },
+                    )
+                    _process_resp(mfa_resp)
+                    mfa_json = await mfa_resp.json()
+                    if not (
+                        mfa_json["data"].get("approved")
+                        and mfa_json["data"].get("valid")
+                    ):
+                        _LOGGER.debug("MFA Code invalid")
+                        raise IncompleteCredentials(
+                            "MFA Code invalid", devices=mfa_json["data"]
+                        )
+                    resp = await self.websession.post(url, data=data)
+                    _process_resp(resp)
+            await asyncio.sleep(3)
+        if not (resp.history):
+            _LOGGER.debug("Failed to authenticate")
+            raise IncompleteCredentials("Unable to login with credentials")
         code_url = URL(resp.history[-1].url)
         return code_url.query.get("code")
 

teslajsonpy/controller.py

@@ -130,7 +130,7 @@ def valid_result(result):
             result = await wrapped(*args, **kwargs)
         except TeslaException as ex:
             _LOGGER.debug(
-                "Exception: %s\n%s(%s %s)", ex.message, wrapped.__name__, args, kwargs
+                "Exception: %s\n%s(%s %s)", str(ex), wrapped.__name__, args, kwargs
             )
             raise
     if valid_result(result) or is_wake_command:
@@ -187,7 +187,7 @@ def valid_result(result):
         )
     except TeslaException as ex:
         _LOGGER.debug(
-            "Exception: %s\n%s(%s %s)", ex.message, wrapped.__name__, args, kwargs
+            "Exception: %s\n%s(%s %s)", str(ex), wrapped.__name__, args, kwargs
         )
         raise
     if valid_result(result):

teslajsonpy/exceptions.py

@@ -6,14 +6,15 @@
 https://github.com/zabuldon/teslajsonpy
 """
 import logging
+from typing import Any, Dict, Text
 
 _LOGGER = logging.getLogger(__name__)
 
 
 class TeslaException(Exception):
     """Class of Tesla API exceptions."""
 
-    def __init__(self, code, *args, **kwargs):
+    def __init__(self, code: Text, *args, **kwargs):
         """Initialize exceptions for the Tesla API."""
         self.message = ""
         super().__init__(*args, **kwargs)
@@ -52,7 +53,10 @@ class RetryLimitError(TeslaException):
 class IncompleteCredentials(TeslaException):
     """Class of exceptions for incomplete credentials."""
 
-    pass
+    def __init__(self, code: Text, *args, devices: Dict[Any, Any] = None, **kwargs):
+        """Initialize exception to include list of devices."""
+        super().__init__(code, *args, **kwargs)
+        self.devices = devices or {}
 
 
 class UnknownPresetMode(TeslaException):

teslajsonpy/teslaproxy.py

@@ -47,7 +47,16 @@ def __init__(self, proxy_url: URL, host_url: URL) -> None:
         }
 
         self.modifiers.update(
-            {"prepend_url_ajax": partial(self.prepend_relative_urls, self.access_url())}
+            {
+                "prepend_url_ajax": partial(
+                    self.prepend_relative_urls, self.access_url()
+                ),
+                "application/javascript": {
+                    "prepend_url_i18n": partial(
+                        self.prepend_i18n_path, URL(self.access_url().path)
+                    )
+                },
+            }
         )
 
     async def test_url(
@@ -121,6 +130,31 @@ async def prepend_relative_urls(self, base_url: URL, html: Text) -> Text:
             html=html,
         )
 
+    async def prepend_i18n_path(self, base_url: URL, html: Text) -> Text:
+        """Prepend path for i18n loadPath so it'll reach the proxy.
+
+        This is intended to be used for to place the proxy_url path in front of relative urls for loadPath in i18next.
+
+        Args:
+            base_url (URL): Base URL to prepend
+            html (Text): Text to replace
+
+        Returns
+            Text: Replaced text
+
+        """
+        if not base_url:
+            _LOGGER.debug("No base_url specified")
+            return html
+
+        return await find_regex_urls(
+            partial(prepend_url, base_url, encoded=True),
+            {
+                "method_func": r"""(?:loadPath:)\s*?["']([^"']*)[\"\']""",
+            },
+            html=html,
+        )
+
     async def modify_headers(
         self, site: URL, request: web.Request
     ) -> multidict.MultiDict: