feat: add UI to provide oauth2 code

psa_car_controller/psa/connected_car_api/configuration.py

@@ -72,7 +72,7 @@ class Configuration(object):
         # Debug file location
         self.logger_file = None
         # Debug switch
-        self.debug = True
+        self.debug = False
 
         # SSL/TLS verification
         # Set this to false to skip verifying SSL certificate when calling API

psa_car_controller/psa/oauth.py

@@ -7,7 +7,7 @@ from typing import Tuple
 from http import HTTPStatus
 from typing import Optional
 
-from oauth2_client.credentials_manager import CredentialManager, ServiceInformation, OAuthError
+from oauth2_client.credentials_manager import CredentialManager, ServiceInformation
 from requests import Response, RequestException
 
 from psa_car_controller.common.utils import rate_limit
@@ -18,45 +18,43 @@ from psa_car_controller.psa.connected_car_api.rest import ApiException
 logger = logging.getLogger(__name__)
 
 
+def generate_sha256_pkce(length: int) -> Tuple[str, str]:
+    if not (43 <= length <= 128):
+        raise ValueError("Invalid length: %d" % length)
+    verifier = secrets.token_urlsafe(length)
+    encoded = base64.urlsafe_b64encode(hashlib.sha256(verifier.encode('ascii')).digest())
+    challenge = encoded.decode('ascii')[:-1]
+    return verifier, challenge
+
+
 class OpenIdCredentialManager(CredentialManager):
+    @staticmethod
+    def create(service_information: ServiceInformation, scheme: str, country_code: str,
+               proxies: Optional[dict] = None):
+        manager = OpenIdCredentialManager(service_information, proxies)
+        manager.redirect_uri = scheme + "://oauth2redirect/" + country_code.lower()
+        return manager
+
     def __init__(self, service_information: ServiceInformation, proxies: Optional[dict] = None):
         super().__init__(service_information, proxies)
         self.refresh_callbacks = []
+        self.code_verifier = None
+        self.redirect_uri = None
 
     def _grant_password_request_realm(self, login: str, password: str, realm: str) -> dict:
         return {"grant_type": 'password', "username": login, "scope": ' '.join(self.service_information.scopes),
                 "password": password, "realm": realm}
 
-    def generate_sha256_pkce(self, length: int) -> Tuple[str, str]:
-        if not (43 <= length <= 128):
-            raise ValueError("Invalid length: %d" % length)
-        verifier = secrets.token_urlsafe(length)
-        encoded = base64.urlsafe_b64encode(hashlib.sha256(verifier.encode('ascii')).digest())
-        challenge = encoded.decode('ascii')[:-1]
-        return verifier, challenge
+    def generate_redirect_url(self):
+        self.code_verifier, code_challenge = generate_sha256_pkce(64)
+        return self.generate_authorize_url(self.redirect_uri, secrets.token_urlsafe(16),
+                                           code_challenge=code_challenge, code_challenge_method="S256")
 
-    def init_with_oauth2_redirect(self, scheme: str, country_code: str):
-        ret = ""
-        while True:
-            redir_uri = scheme + "://oauth2redirect/" + country_code.lower()
-            code_verifier, code_challenge = self.generate_sha256_pkce(64)
-            url = self.generate_authorize_url(redir_uri, secrets.token_urlsafe(16),
-                                              code_challenge=code_challenge, code_challenge_method="S256")
-
-            logger.info("Now login to this URL in a browser: %s", url)
-
-            try:
-                ret = input("\nCopy+paste the resulting mymXX-code (in F12 > Network, "
-                            "when you hit the final OK button, 36 chars, UUID format): ")
-                logger.info("Try getting a token with code %s", ret)
-                assert len(ret) == 36, "Invalid code length"
-                self._token_request({"grant_type": 'authorization_code', "code": ret,
-                                     "redirect_uri": redir_uri, "code_verifier": code_verifier}, False)
-            except (OAuthError, AssertionError):
-                logger.exception("Failed to get a token")
-                if input("Retry ? yes/NO") == "yes":
-                    continue
-            break
+    def connect_with_code(self, code: str):
+        assert len(code) == 36, "Invalid code length"
+        self._token_request({"grant_type": 'authorization_code', "code": code,
+                             "redirect_uri": self.redirect_uri, "code_verifier": self.code_verifier},
+                            False)
 
     @staticmethod
     def _is_token_expired(response: Response) -> bool:

psa_car_controller/psa/setup/apk_parser.py

@@ -1,4 +1,5 @@
 import json
+import logging
 import os
 
 from androguard.core.bytecodes.apk import APK
@@ -8,6 +9,8 @@ from cryptography.hazmat.primitives.serialization import pkcs12
 
 from psa_car_controller.psa.constants import BRAND
 
+logging.getLogger("androguard").setLevel(logging.ERROR)
+
 
 class ApkParser:
     def __init__(self, filename, country_code):

psa_car_controller/psa/setup/app_decoder.py

@@ -8,6 +8,7 @@ import requests
 from psa_car_controller.psa.constants import BRAND
 from psa_car_controller.psa.setup.apk_parser import ApkParser
 from psa_car_controller.psa.setup.github import urlretrieve_from_github
+from psa_car_controller.psacc.application.car_controller import PSACarController
 from psa_car_controller.psacc.application.psa_client import PSAClient
 from psa_car_controller.psacc.application.charge_control import ChargeControl, ChargeControls
 
@@ -17,6 +18,7 @@ APP_VERSION = "1.33.0"
 GITHUB_USER = "flobz"
 GITHUB_REPO = "psa_apk"
 TIMEOUT_IN_S = 10
+app = PSACarController()
 
 
 def get_content_from_apk(filename: str, country_code: str) -> ApkParser:
@@ -26,91 +28,108 @@ def get_content_from_apk(filename: str, country_code: str) -> ApkParser:
     return apk_parser
 
 
-def firstLaunchConfig(package_name, client_email, client_password, country_code,  # pylint: disable=too-many-locals
-                      config_prefix=""):
-    filename = package_name.split(".")[-1] + ".apk"
-    apk_parser = get_content_from_apk(filename, country_code)
-
-    try:
-        res = requests.post(apk_parser.host_brandid_prod + "/GetAccessToken",
-                            headers={
-                                "Connection": "Keep-Alive",
-                                "Content-Type": "application/json",
-                                "User-Agent": "okhttp/2.3.0"
-                            },
-                            params={"jsonRequest": json.dumps(
-                                {"siteCode": apk_parser.site_code, "culture": "fr-FR", "action": "authenticate",
-                                 "fields": {"USR_EMAIL": {"value": client_email},
-                                            "USR_PASSWORD": {"value": client_password}}
-                                 }
-                            )},
-                            timeout=TIMEOUT_IN_S
-                            )
-
-        token = res.json()["accessToken"]
-    except Exception as ex:
-        msg = traceback.format_exc() + f"\nHOST_BRANDID : {apk_parser.host_brandid_prod} " \
-                                       f"sitecode: {apk_parser.site_code}"
+class InitialSetup:
+    def __init__(self, package_name, client_email, client_password, country_code):
+        self.package_name = package_name
+        filename = package_name.split(".")[-1] + ".apk"
+        apk_parser = get_content_from_apk(filename, country_code)
+        self.culture = apk_parser.culture
+        self.site_code = apk_parser.site_code
+        self.client_id = apk_parser.client_id
+        self.client_secret = apk_parser.client_secret
+        self.country_code = country_code
+        self.user_info = None
+        self.customer_id = None
         try:
-            msg += res.text
-        except BaseException:
-            pass
-        logger.error(msg)
-        raise ConnectionError(msg) from ex
-    try:
-        res2 = requests.post(
-            f"https://mw-{BRAND[package_name]['brand_code'].lower()}-m2c.mym.awsmpsa.com/api/v1/user",
-            params={
-                "culture": apk_parser.culture,
-                "width": 1080,
-                "version": APP_VERSION
-            },
-            data=json.dumps({"site_code": apk_parser.site_code, "ticket": token}),
-            headers={
-                "Connection": "Keep-Alive",
-                "Content-Type": "application/json;charset=UTF-8",
-                "Source-Agent": "App-Android",
-                "Token": token,
-                "User-Agent": "okhttp/4.8.0",
-                "Version": APP_VERSION
-            },
-            cert=("certs/public.pem", "certs/private.pem"),
-            timeout=TIMEOUT_IN_S
-        )
+            res = requests.post(apk_parser.host_brandid_prod + "/GetAccessToken",
+                                headers={
+                                    "Connection": "Keep-Alive",
+                                    "Content-Type": "application/json",
+                                    "User-Agent": "okhttp/2.3.0"
+                                },
+                                params={"jsonRequest": json.dumps(
+                                    {"siteCode": apk_parser.site_code, "culture": "fr-FR", "action": "authenticate",
+                                     "fields": {"USR_EMAIL": {"value": client_email},
+                                                "USR_PASSWORD": {"value": client_password}}
+                                     }
+                                )},
+                                timeout=TIMEOUT_IN_S
+                                )
 
-        res_dict = res2.json()["success"]
-        customer_id = BRAND[package_name]["brand_code"] + "-" + res_dict["id"]
-    except Exception as ex:
-        msg = traceback.format_exc()
+            self.token = res.json()["accessToken"]
+        except Exception as ex:
+            msg = traceback.format_exc() + f"\nHOST_BRANDID : {apk_parser.host_brandid_prod} " \
+                                           f"sitecode: {apk_parser.site_code}"
+            try:
+                msg += res.text
+            except BaseException:
+                pass
+            logger.error(msg)
+            raise ConnectionError(msg) from ex
+
+        # Psacc
+        self.user_info = self.__fetch_user_info()
+        self.customer_id = BRAND[self.package_name]["brand_code"] + "-" + self.user_info["id"]
+
+        self.psacc = PSAClient(None, self.client_id, self.client_secret,
+                               None, self.customer_id, BRAND[self.package_name]["realm"],
+                               self.country_code, BRAND[self.package_name]["brand_code"])
+
+    def __fetch_user_info(self):
         try:
-            msg += res2.text
-        except BaseException:
-            pass
-        logger.error(msg)
-        raise ConnectionError(msg) from ex
-    # Psacc
-    psacc = PSAClient(None, apk_parser.client_id, apk_parser.client_secret,
-                      None, customer_id, BRAND[package_name]["realm"],
-                      country_code, BRAND[package_name]["brand_code"])
-    psacc.connect()
-    psacc.save_config(name=config_prefix + "config.json")
-    res = psacc.get_vehicles()
+            res2 = requests.post(
+                f"https://mw-{BRAND[self.package_name]['brand_code'].lower()}-m2c.mym.awsmpsa.com/api/v1/user",
+                params={
+                    "culture": self.culture,
+                    "width": 1080,
+                    "version": APP_VERSION
+                },
+                data=json.dumps({"site_code": self.site_code, "ticket": self.token}),
+                headers={
+                    "Connection": "Keep-Alive",
+                    "Content-Type": "application/json;charset=UTF-8",
+                    "Source-Agent": "App-Android",
+                    "Token": self.token,
+                    "User-Agent": "okhttp/4.8.0",
+                    "Version": APP_VERSION
+                },
+                cert=("certs/public.pem", "certs/private.pem"),
+                timeout=TIMEOUT_IN_S
+            )
 
-    if len(res) == 0:
-        raise ValueError("No vehicle in your account is compatible with this API, you vehicle is probably too old...")
+            res_dict = res2.json()["success"]
+        except Exception as ex:
+            msg = traceback.format_exc()
+            try:
+                msg += res2.text
+            except BaseException:
+                pass
+            logger.error(msg)
+            raise ConnectionError(msg) from ex
+        return res_dict
 
-    for vehicle in res_dict["vehicles"]:
-        car = psacc.vehicles_list.get_car_by_vin(vehicle["vin"])
-        if car is not None and "short_label" in vehicle and car.label == "unknown":
-            car.label = vehicle["short_label"].split(" ")[-1]  # remove new, nouvelle, neu word....
-    psacc.vehicles_list.save_cars()
+    def connect(self, code, config_prefix=""):
+        self.psacc.connect(code)
+        self.psacc.save_config(name=config_prefix + "config.json")
+        res = self.psacc.get_vehicles()
 
-    logger.info("\nYour vehicles: %s", res)
+        if len(res) == 0:
+            raise ValueError(
+                "No vehicle in your account is compatible with this API, you vehicle is probably too old...")
 
-    # Charge control
-    charge_controls = ChargeControls(config_prefix + "charge_config.json")
-    for vehicle in res:
-        chc = ChargeControl(psacc, vehicle.vin, 100, [0, 0])
-        charge_controls[vehicle.vin] = chc
-    charge_controls.save_config()
-    return "Success !!!"
+        for vehicle in self.user_info["vehicles"]:
+            car = self.psacc.vehicles_list.get_car_by_vin(vehicle["vin"])
+            if car is not None and "short_label" in vehicle and car.label == "unknown":
+                car.label = vehicle["short_label"].split(" ")[-1]  # remove new, nouvelle, neu word....
+        self.psacc.vehicles_list.save_cars()
+
+        logger.info("\nYour vehicles: %s", res)
+
+        # Charge control
+        charge_controls = ChargeControls(config_prefix + "charge_config.json")
+        for vehicle in res:
+            chc = ChargeControl(self.psacc, vehicle.vin, 100, [0, 0])
+            charge_controls[vehicle.vin] = chc
+        charge_controls.save_config()
+        app.load_app()
+        app.start_remote_control()

psa_car_controller/psacc/application/car_controller.py

@@ -4,7 +4,7 @@ import logging
 import socket
 import sys
 import threading
-from os import environ, path
+from os import path
 
 import psa_car_controller
 from oauth2_client.credentials_manager import OAuthError
@@ -35,8 +35,6 @@ def parse_args():
     parser.add_argument("-p", "--port", help="change server listen port", default="5000")
     parser.add_argument("-r", "--record", help="save vehicle data to db", action='store_true')
     parser.add_argument("-R", "--refresh", help="refresh vehicles status every x min", type=int)
-    parser.add_argument("-m", "--mail", default=environ.get('USER_EMAIL', None), help="set the email address")
-    parser.add_argument("-P", "--password", default=environ.get('USER_PASSWORD', None), help="set the password")
     parser.add_argument("--remote-disable", help="disable remote control", action='store_true')
     parser.add_argument("--offline", help="offline limited mode", action='store_true')
     parser.add_argument("--web-conf", help="ignore if config files not existing yet", action='store_true')
@@ -76,7 +74,6 @@ class PSACarController(metaclass=Singleton):
                 logger.error("start_remote_control failed redo otp config")
 
     def load_app(self) -> bool:
-        # pylint: disable=too-many-branches
         my_logger(handler_level=int(self.args.debug))
 
         logger.info("App version %s", __version__)
@@ -105,16 +102,11 @@ class PSACarController(metaclass=Singleton):
                 if self.is_good:
                     logger.info(str(self.myp.get_vehicles()))
             except OAuthError:
-                if self.args.mail and self.args.password:
-                    self.myp.connect(self.args.mail, self.args.password)
-                    logger.info(str(self.myp.get_vehicles()))
-                    self.is_good = True
+                self.is_good = False
+                if self.args.web_conf:
+                    logger.error("Please reconnect by going to config web page")
                 else:
-                    self.is_good = False
-                    if self.args.web_conf:
-                        logger.error("Please reconnect by going to config web page")
-                    else:
-                        logger.error("Connection need to be updated, Please redo authentication process.")
+                    logger.error("Connection need to be updated, Please redo authentication process.")
             if self.args.refresh:
                 self.myp.info_refresh_rate = self.args.refresh * 60
                 if self.is_good:

psa_car_controller/psacc/application/psa_client.py

@@ -31,8 +31,8 @@ logger = CustomLogger.getLogger(__name__)
 
 
 class PSAClient:
-    def connect(self):
-        self.manager.init_with_oauth2_redirect(realm_info[self.realm]["scheme"], self.country_code)
+    def connect(self, code: str):
+        self.manager.connect_with_code(code)
 
     # pylint: disable=too-many-arguments
     def __init__(self, refresh_token, client_id, client_secret, remote_refresh_token, customer_id, realm, country_code,
@@ -44,7 +44,9 @@ class PSAClient:
                                                       client_secret,
                                                       SCOPE, True)
         self.client_id = client_id
-        self.manager = OpenIdCredentialManager(self.service_information)
+        self.country_code = country_code
+        self.manager = OpenIdCredentialManager.create(self.service_information,
+                                                      realm_info[self.realm]["scheme"], self.country_code)
         self.api_config = Oauth2PSACCApiConfig()
         self.api_config.set_refresh_callback(self.manager.refresh_token_now)
         self.manager.refresh_token = refresh_token
@@ -59,7 +61,6 @@ class PSAClient:
         self.remote_token_last_update = None
         self._record_enabled = False
         self.weather_api = weather_api
-        self.country_code = country_code
         self.brand = brand
         self.info_callback = []
         self.info_refresh_rate = 120

psa_car_controller/web/view/config_oauth.py

@@ -0,0 +1,63 @@
+import logging
+
+from dash import callback_context, html, dcc
+from dash.exceptions import PreventUpdate
+from flask import request
+
+from psa_car_controller.web.app import dash_app
+import dash_bootstrap_components as dbc
+from dash.dependencies import Output, Input, State
+
+from psa_car_controller.web.view import config_views
+
+logger = logging.getLogger(__name__)
+
+
+def get_oauth_config_layout(redirect_url):
+    return dbc.Row(dbc.Col(md=12, lg=2, className="m-3", children=[
+        dbc.Row(html.H2('Connection to PSA')),
+        dbc.Row(className="ms-2", children=[
+            html.Div(html.P([
+                html.A("1. Click here", href=redirect_url, target="_blank"), html.Br(),
+                "2. Complete the login procedure there too until you see 'LOGIN SUCCESSFUL'", html.Br(),
+                "3. Open your browser's DevTools (F12) and then the click on 'Network' tab", html.Br(),
+                "4. Hit the final 'OK' button, under 'LOGIN SUCCESSFUL'", html.Br(),
+                "5. Find in the network tab: xxxx://oauth2redirect....?code=<copy this part>&scope=openid... ",
+                html.Br(),
+                html.A("You can find more info here",
+                       href="https://github.com/flobz/psa_car_controller/discussions/779"), html.Br()]
+            )),
+            dbc.Form([
+                html.Div([
+                    dbc.Label("Code", html_for="psa-oauth-code"),
+                    dbc.Input(type="text", id="psa-oauth-code", placeholder="Enter login code"),
+                    dbc.FormText(
+                        "PSA code from step above",
+                        color="secondary",
+                    )]),
+                dbc.Row(dbc.Button("Submit", color="primary", id="finish-oauth")),
+                dcc.Loading(
+                    id="loading-2",
+                    children=[html.Div([html.Div(id="oauth-result")])],
+                    type="circle",
+                ),
+            ])
+        ])]))
+
+
+@dash_app.callback(
+    Output("oauth-result", "children"),
+    Input("finish-oauth", "n_clicks"),
+    State("psa-oauth-code", "value"))
+def finish_oauth(n_clicks, code):  # pylint: disable=unused-argument
+    ctx = callback_context
+    if ctx.triggered:
+        try:
+            config_views.INITIAL_SETUP.connect(code)
+            return dbc.Alert(["PSA login finish !",
+                              html.A(" Go to otp config", href=request.url_root + "config_otp")],
+                             color="success")
+        except Exception as e:
+            logger.exception("finish_oauth:")
+            return dbc.Alert(str(e), color="danger")
+    raise PreventUpdate()

psa_car_controller/web/view/config_views.py

@@ -1,4 +1,5 @@
 import logging
+from urllib import parse
 
 from dash import callback_context, html, dcc
 from dash.exceptions import PreventUpdate
@@ -6,7 +7,7 @@ from flask import request
 
 from psa_car_controller.psa.otp.otp import new_otp_session
 from psa_car_controller.psacc.application.car_controller import PSACarController
-from psa_car_controller.psa.setup.app_decoder import firstLaunchConfig
+from psa_car_controller.psa.setup.app_decoder import InitialSetup
 from psa_car_controller.common.mylogger import LOG_FILE
 from psa_car_controller.web.app import dash_app
 import dash_bootstrap_components as dbc
@@ -15,7 +16,9 @@ from dash.dependencies import Output, Input, State
 logger = logging.getLogger(__name__)
 
 app = PSACarController()
-login_config_layout = dbc.Row(dbc.Col(md=12, lg=2, className="m-3", children=[
+INITIAL_SETUP: InitialSetup = None
+
+setup_config_layout = dbc.Row(dbc.Col(md=12, lg=2, className="m-3", children=[
     dbc.Row(html.H2('Config')),
     dbc.Row(className="ms-2", children=[
         dbc.Form([
@@ -61,16 +64,9 @@ login_config_layout = dbc.Row(dbc.Col(md=12, lg=2, className="m-3", children=[
                     color="secondary",
                 )]),
             dbc.Row(dbc.Button("Submit", color="primary", id="submit-form")),
-            html.Div(html.P([
-                "1. After submit switch back to the CLI and wait for the login URL to be shown", html.Br(),
-                "2. Open that URL in a browser", html.Br(),
-                "3. Complete the login procedure there too", html.Br(),
-                "4. Open your browser's DevTools (F12) and then the 'Network' tab", html.Br(),
-                "5. Hit the final 'OK' button, under 'LOGIN SUCCESSFUL'", html.Br(),
-                "6. Find xxxx://oauth2redirect....?code=<copy this part>&scope=openid... "
-                "in the DevTools and paste it into the command line interface", html.Br(),
-                "7. Switch back to the browser and do the OTP config", html.Br()]
-            )),
+            dbc.Row(dbc.FormText(
+                "After submit be patient it can take some time...",
+                color="secondary")),
             dcc.Loading(
                 id="loading-2",
                 children=[html.Div([html.Div(id="form_result")])],
@@ -121,7 +117,7 @@ def log_layout():
 def config_layout(activeTabs="log"):
     return dbc.Tabs(active_tab=activeTabs, children=[
         dbc.Tab([log_layout()], label="Log", tab_id="log"),
-        dbc.Tab([login_config_layout], label="User config", tab_id="login"),
+        dbc.Tab([setup_config_layout], label="User config", tab_id="login"),
         dbc.Tab([config_otp_layout], label="OTP config", tab_id="otp")])
 
 
@@ -135,11 +131,14 @@ def config_layout(activeTabs="log"):
 def connectPSA(n_clicks, app_name, email, password, countrycode):  # pylint: disable=unused-argument
     ctx = callback_context
     if ctx.triggered:
+        logger.info("Initial setup...")
         try:
-            res = firstLaunchConfig(app_name, email, password, countrycode)
-            app.load_app()
-            app.start_remote_control()
-            return dbc.Alert([res, html.A(" Go to otp config", href=request.url_root + "config_otp")], color="success")
+            global INITIAL_SETUP
+            INITIAL_SETUP = InitialSetup(app_name, email, password, countrycode)
+            redirect_uri = parse.quote(INITIAL_SETUP.psacc.manager.generate_redirect_url())
+            return dbc.Alert(["Success !", html.A(" Go to login",
+                                                  href=f"{request.url_root}config_connect?url={redirect_uri}")],
+                             color="success")
         except Exception as e:
             res = str(e)
             logger.exception(e)

psa_car_controller/web/view/views.py

@@ -22,6 +22,7 @@ from psa_car_controller.web import figures
 from psa_car_controller.web.app import dash_app
 from psa_car_controller.psacc.repository.db import Database
 from psa_car_controller.web.tools.utils import diff_dashtable, unix_time_millis, get_marks_from_start_end, create_card
+from psa_car_controller.web.view.config_oauth import get_oauth_config_layout
 from psa_car_controller.web.view.config_views import log_layout, config_layout
 
 # pylint: disable=invalid-name
@@ -75,6 +76,8 @@ def display_page(pathname, search):
         page = config_layout()
     elif pathname == "/config_login":
         page = config_layout("login")
+    elif pathname == "/config_connect":
+        page = get_oauth_config_layout(query_params["url"])
     elif pathname == "/log":
         page = log_layout()
     elif not APP.is_good: