Merge pull request dracoventions#3 from mvaneijken/main

Pull into ems_p1monitor_api

.github/workflows/docker+pypi.yml

@@ -130,7 +130,7 @@ jobs:
       uses: docker/build-push-action@v2
       with:
         context: contrib/docker
-        platforms: linux/amd64,linux/arm64,linux/arm/v7
+        platforms: linux/amd64,linux/arm64
         push: true
         tags: twcmanager/twcmanager:latest
         build-args: |
@@ -176,7 +176,7 @@ jobs:
       uses: docker/build-push-action@v2
       with:
         context: contrib/docker
-        platforms: linux/amd64,linux/arm64,linux/arm/v7
+        platforms: linux/amd64,linux/arm64
         push: true
         tags: twcmanager/twcmanager:${{ steps.branch_name.outputs.SOURCE_TAG }}
         build-args: |

lib/TWCManager/Control/HTTPControl.py

@@ -669,23 +669,27 @@ def do_GET(self):
                 return
 
             webroutes = [
-              { "route": "/debug",    "tmpl": "debug.html.j2" },
-              { "route": "/schedule", "tmpl": "schedule.html.j2" },
-              { "route": "/settings", "tmpl": "settings.html.j2" },
-              { "rstart": "/vehicleDetail", "tmpl": "vehicleDetail.html.j2" },
-              { "route": "/vehicles", "tmpl": "vehicles.html.j2" }
+              { "route": "/debug",                      "tmpl": "debug.html.j2" },
+              { "route": "/schedule",                   "tmpl": "schedule.html.j2" },
+              { "route": "/settings",                   "tmpl": "settings.html.j2" },
+              { "route": "/teslaAccount/login",         "error": "insecure" },
+              { "route": "/teslaAccount/mfaCode",       "error": "insecure" },
+              { "route": "/teslaAccount/submitCaptcha", "error": "insecure" },
+              { "rstart": "/teslaAccount",              "tmpl": "main.html.j2" },
+              { "rstart": "/vehicleDetail",             "tmpl": "vehicleDetail.html.j2" },
+              { "route": "/vehicles",                   "tmpl": "vehicles.html.j2" }
             ]
 
-            if (self.url.path == "/teslaAccount/login" or
-                self.url.path == "/teslaAccount/mfaCode"):
-                # For security, these details should be submitted via a POST request
-                # Send a 405 Method Not Allowed in response.
-                self.send_response(405)
-                page = "This function may only be requested via the POST HTTP method."
-                self.wfile.write(page.encode("utf-8"))
+            if self.url.path == "/teslaAccount/getCaptchaImage":
+                self.send_response(200)
+                self.send_header("Content-type", "image/svg+xml")
+                self.end_headers()
+                self.wfile.write(master.getModuleByName(
+                    "TeslaAPI"
+                ).getCaptchaImage())
                 return
 
-            if self.url.path == "/" or self.url.path.startswith("/teslaAccount"):
+            if self.url.path == "/":
                 self.send_response(200)
                 self.send_header("Content-type", "text/html")
                 self.end_headers()
@@ -711,10 +715,27 @@ def do_GET(self):
             for webroute in webroutes:
                 if self.url.path == webroute.get("route", "INVALID"):
                     route = webroute
+                    break
                 elif self.url.path.startswith(webroute.get("rstart", "INVALID")):
                     route = webroute
+                    break
+
+            if route and route.get("error", None):
+
+                if route["error"] == "insecure":
+                    # For security, these details should be submitted via a POST request
+                    # Send a 405 Method Not Allowed in response.
+                    self.send_response(405)
+                    page = "This function may only be requested via the POST HTTP method."
+                    self.wfile.write(page.encode("utf-8"))
+                    return
+
+                else:
+                    self.send_response(500)
+                    self.wfile.write("".encode("utf-8"))
+                    return
 
-            if route:
+            elif route:
                 self.send_response(200)
                 self.send_header("Content-type", "text/html")
                 self.end_headers()
@@ -836,6 +857,19 @@ def do_POST(self):
                 self.wfile.write("".encode("utf-8"))
                 return
 
+            if self.url.path == "/teslaAccount/submitCaptcha":
+                captchaCode = self.getFieldValue("captchaCode")
+
+                resp = master.getModuleByName(
+                    "TeslaAPI"
+                ).submitCaptchaCode(captchaCode)
+
+                self.send_response(302)
+                self.send_header("Location", "/teslaAccount/" + str(resp))
+                self.end_headers()
+                self.wfile.write("".encode("utf-8"))
+                return
+
             if self.url.path == "/graphs/dates":
                 # User has submitted dates to graph this period.
                 objIni = self.getFieldValue("dateIni")

lib/TWCManager/Control/themes/Default/handle_teslalogin.html.j2

@@ -38,6 +38,18 @@
   <font color='red'>
     <b>Error encountered during Phase 1 (GET) of the Tesla Authentication process.</b>
   </font>
+{% elif url.path == "/teslaAccount/Phase1Captcha" %}
+  <font color='black'>
+    <b>Tesla Login has requested a Captcha code for this login session. Please type the characters shown on the Captcha image into the textbox below.</b>
+
+    <br><img src="/teslaAccount/getCaptchaImage" style="width:400px;height:200px;" />
+
+    <form action="/teslaAccount/submitCaptcha" method=POST>
+      Enter Captcha Code:
+      <input type=text size=6 name=captchaCode />
+      <input type=submit value="Send Captcha" />
+    </form>
+  </font>
 {% elif url.path == "/teslaAccount/Phase2Error" or url.path == "/teslaAccount/Phase2ErrorTip" %}
   <font color='red'>
     <b>Error encountered during Phase 2 (POST) of the Tesla Authentication process.</b>
@@ -58,7 +70,8 @@
 
 {% if not master.teslaLoginAskLater
       and url.path != "/teslaAccount/True" 
-      and not url.path.startswith("/teslaAccount/MFA") %}
+      and not url.path.startswith("/teslaAccount/MFA")
+      and not url.path == "/teslaAccount/Phase1Captcha" %}
       <!-- Check if we have already stored the Tesla credentials
            If we can access the Tesla API okay, don't prompt -->
       {% if not apiAvailable %}

lib/TWCManager/Vehicle/TeslaAPI.py

@@ -14,8 +14,11 @@
 
 class TeslaAPI:
 
+    __apiCaptcha = None
+    __apiCaptchaCode = None
     authURL = "https://auth.tesla.com/oauth2/v3/authorize"
     callbackURL = "https://auth.tesla.com/void/callback"
+    captchaURL = "https://auth.tesla.com/captcha"
     carApiLastErrorTime = 0
     carApiBearerToken = ""
     carApiRefreshToken = ""
@@ -33,11 +36,14 @@ class TeslaAPI:
     carApiVehicles = []
     config = None
     master = None
+    __email = None
     errorCount = 0
     maxLoginRetries = 10
     minChargeLevel = -1
     params = None
+    __password = None
     refreshURL = "https://owner-api.teslamotors.com/oauth/token"
+    __resp = None
     session = None
     verifier = ""
 
@@ -77,6 +83,10 @@ def addVehicle(self, json):
 
     def apiLogin(self, email, password):
 
+        # Populate auth details for Phase 1
+        self.__email = email
+        self.__password = password
+
         for attempt in range(self.maxLoginRetries):
 
             self.verifier = base64.urlsafe_b64encode(os.urandom(86)).rstrip(b"=")
@@ -98,14 +108,22 @@ def apiLogin(self, email, password):
             )
 
             self.session = requests.Session()
-            resp = self.session.get(self.authURL, params=self.params)
+            self.__resp = self.session.get(self.authURL, params=self.params)
 
-            if resp.ok and "<title>" in resp.text:
+            if self.__resp.ok and "<title>" in self.__resp.text:
                 logger.log(
                     logging.INFO6,
                     "Tesla Auth form fetch success, attempt: " + str(attempt),
                 )
-                break
+
+                if 'img data-id="captcha"' in self.__resp.text:
+                    logger.log(
+                        logging.INFO6,
+                        "Tesla Auth form challenged us for Captcha. Redirecting.")
+                    self.getApiCaptcha()
+                    return "Phase1Captcha"
+                else:
+                    return self.apiLoginPhaseOne()
             else:
                 logger.log(
                     logging.INFO6,
@@ -122,9 +140,15 @@ def apiLogin(self, email, password):
             )
             return "Phase1Error"
 
-        csrf = re.search(r'name="_csrf".+value="([^"]+)"', resp.text).group(1)
+    def apiLoginPhaseOne(self):
+
+        # Picks up on the first phase of authentication, after redirecting to
+        # handle Captcha if this was requested, or directly if we were lucky
+        # enough not to be challenged.
+
+        csrf = re.search(r'name="_csrf".+value="([^"]+)"', self.__resp.text).group(1)
         transaction_id = re.search(
-            r'name="transaction_id".+value="([^"]+)"', resp.text
+            r'name="transaction_id".+value="([^"]+)"', self.__resp.text
         ).group(1)
 
         if not csrf or not transaction_id:
@@ -138,10 +162,22 @@ def apiLogin(self, email, password):
             "_process": "1",
             "transaction_id": transaction_id,
             "cancel": "",
-            "identity": email,
-            "credential": password,
+            "identity": self.__email,
+            "credential": self.__password,
         }
 
+        # If a captcha code is stored, inject it into the data parameter
+        if self.__apiCaptchaCode:
+            data["captcha"] = self.__apiCaptchaCode
+
+            # Clear captcha data
+            self.__apiCaptcha = None
+
+        # Clear stored credentials
+        self.__email = None
+        self.__password = None
+
+        # Call login Phase 2
         return self.apiLoginPhaseTwo(data)
 
     def apiLoginPhaseTwo(self, data):
@@ -1057,6 +1093,24 @@ def applyChargeLimit(self, limit, checkArrival=False, checkDeparture=False):
         if checkArrival:
             self.updateChargeAtHome()
 
+    def getApiCaptcha(self):
+        # This will fetch the current Captcha image displayed by Tesla's auth
+        # website, and store it in memory
+
+        self.__apiCaptcha = self.session.get(self.captchaURL)
+
+    def getCaptchaImage(self):
+        # This will serve the Tesla Captcha image
+
+        if self.__apiCaptcha:
+            return(self.__apiCaptcha.content)
+        else:
+            logger.log(
+                logging.INFO2,
+                "ERROR: Captcha image requested, but we have none buffered. This is likely due to a stale login session, but if you see it regularly, please report it."
+            )
+            return ""
+
     def getCarApiBearerToken(self):
         return self.carApiBearerToken
 
@@ -1181,6 +1235,10 @@ def setCarApiTokenExpireTime(self, value):
         self.carApiTokenExpireTime = value
         return True
 
+    def submitCaptchaCode(self, code):
+        self.__apiCaptchaCode = code
+        return self.apiLoginPhaseOne()
+
     def updateCarApiLastErrorTime(self, vehicle=None):
         timestamp = time.time()
         logger.log(