From 6f81149f4539db356ffa368c8542b4008cbb98e5 Mon Sep 17 00:00:00 2001 From: Florian Bezannier Date: Mon, 26 Feb 2024 11:56:59 +0100 Subject: [PATCH] feat: add UI to provide oauth2 code --- .../psa/connected_car_api/configuration.py | 2 +- psa_car_controller/psa/oauth.py | 58 +++--- psa_car_controller/psa/setup/apk_parser.py | 3 + psa_car_controller/psa/setup/app_decoder.py | 181 ++++++++++-------- .../psacc/application/car_controller.py | 18 +- .../psacc/application/psa_client.py | 9 +- psa_car_controller/web/view/config_oauth.py | 63 ++++++ psa_car_controller/web/view/config_views.py | 33 ++-- psa_car_controller/web/view/views.py | 3 + 9 files changed, 224 insertions(+), 146 deletions(-) create mode 100644 psa_car_controller/web/view/config_oauth.py diff --git a/psa_car_controller/psa/connected_car_api/configuration.py b/psa_car_controller/psa/connected_car_api/configuration.py index 4b2fe62..8f1a4a4 100644 --- a/psa_car_controller/psa/connected_car_api/configuration.py +++ b/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 diff --git a/psa_car_controller/psa/oauth.py b/psa_car_controller/psa/oauth.py index 079fea4..c3a959e 100644 --- a/psa_car_controller/psa/oauth.py +++ b/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: diff --git a/psa_car_controller/psa/setup/apk_parser.py b/psa_car_controller/psa/setup/apk_parser.py index f9a8286..6a3e3b8 100644 --- a/psa_car_controller/psa/setup/apk_parser.py +++ b/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): diff --git a/psa_car_controller/psa/setup/app_decoder.py b/psa_car_controller/psa/setup/app_decoder.py index cd475ec..3fcf797 100755 --- a/psa_car_controller/psa/setup/app_decoder.py +++ b/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() diff --git a/psa_car_controller/psacc/application/car_controller.py b/psa_car_controller/psacc/application/car_controller.py index 64e6e7a..cc49fe1 100644 --- a/psa_car_controller/psacc/application/car_controller.py +++ b/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: diff --git a/psa_car_controller/psacc/application/psa_client.py b/psa_car_controller/psacc/application/psa_client.py index 7b08ffa..1dc3458 100644 --- a/psa_car_controller/psacc/application/psa_client.py +++ b/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 diff --git a/psa_car_controller/web/view/config_oauth.py b/psa_car_controller/web/view/config_oauth.py new file mode 100644 index 0000000..0be4487 --- /dev/null +++ b/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=&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() diff --git a/psa_car_controller/web/view/config_views.py b/psa_car_controller/web/view/config_views.py index 5634d58..bc8c065 100644 --- a/psa_car_controller/web/view/config_views.py +++ b/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=&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) diff --git a/psa_car_controller/web/view/views.py b/psa_car_controller/web/view/views.py index 4990873..7ccdad2 100644 --- a/psa_car_controller/web/view/views.py +++ b/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: