From ce3c7a4ae4a02d2b7d8c3e9b5eb8195c5f48fff3 Mon Sep 17 00:00:00 2001 From: Florian Bezannier Date: Sat, 24 Feb 2024 11:36:32 +0100 Subject: [PATCH] feat: retrieve redirect scheme from constant and allow retry auth Signed-off-by: Florian Bezannier --- psa_car_controller/psa/constants.py | 18 +++++++--- psa_car_controller/psa/oauth.py | 35 +++++++++++-------- .../psacc/application/psa_client.py | 2 +- 3 files changed, 35 insertions(+), 20 deletions(-) diff --git a/psa_car_controller/psa/constants.py b/psa_car_controller/psa/constants.py index a41a858..9e86ba7 100644 --- a/psa_car_controller/psa/constants.py +++ b/psa_car_controller/psa/constants.py @@ -3,12 +3,20 @@ IMMEDIATE_CHARGE = "immediate" PSA_CORRELATION_DATE_FORMAT = "%Y%m%d%H%M%S%f" PSA_DATE_FORMAT = "%Y-%m-%dT%H:%M:%SZ" realm_info = { - "clientsB2CPeugeot": {"oauth_url": "https://idpcvs.peugeot.com/am/oauth2/access_token", "app_name": "MyPeugeot"}, - "clientsB2CCitroen": {"oauth_url": "https://idpcvs.citroen.com/am/oauth2/access_token", "app_name": "MyCitroen"}, - "clientsB2CDS": {"oauth_url": "https://idpcvs.driveds.com/am/oauth2/access_token", "app_name": "MyDS"}, - "clientsB2COpel": {"oauth_url": "https://idpcvs.opel.com/am/oauth2/access_token", "app_name": "MyOpel"}, + "clientsB2CPeugeot": {"oauth_url": "https://idpcvs.peugeot.com/am/oauth2/access_token", "app_name": "MyPeugeot", + "scheme": "mymap"}, + "clientsB2CCitroen": {"oauth_url": "https://idpcvs.citroen.com/am/oauth2/access_token", "app_name": "MyCitroen", + "scheme": "mymacsdk"}, + "clientsB2CDS": {"oauth_url": "https://idpcvs.driveds.com/am/oauth2/access_token", "app_name": "MyDS", + "scheme": "mymdssdk"}, + + "clientsB2COpel": {"oauth_url": "https://idpcvs.opel.com/am/oauth2/access_token", "app_name": "MyOpel", + "scheme": "mymopsdk"}, + "clientsB2CVauxhall": {"oauth_url": "https://idpcvs.vauxhall.co.uk/am/oauth2/access_token", - "app_name": "MyVauxhall"} + "app_name": "MyVauxhall", + "scheme": "mymvxsdk", + } } MQTT_BRANDCODE = {"AP": "AP", "AC": "AC", diff --git a/psa_car_controller/psa/oauth.py b/psa_car_controller/psa/oauth.py index 607c0d2..079fea4 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 +from oauth2_client.credentials_manager import CredentialManager, ServiceInformation, OAuthError from requests import Response, RequestException from psa_car_controller.common.utils import rate_limit @@ -35,21 +35,28 @@ class OpenIdCredentialManager(CredentialManager): challenge = encoded.decode('ascii')[:-1] return verifier, challenge - def init_with_brand_country_code(self, brand: str, country_code: str): - redir_uri = "mym" + brand.lower() + "://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) - + def init_with_oauth2_redirect(self, scheme: str, country_code: str): ret = "" - while len(ret) != 36: - ret = input("\nCopy+paste the resulting mymXX-code (in F12 > Network, " \ - "when you hit the final OK button, 36 chars, UUID format): ") + 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") - self._token_request({ "grant_type": 'authorization_code', "code": ret, - "redirect_uri": redir_uri, "code_verifier": code_verifier}, False) + 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 @staticmethod def _is_token_expired(response: Response) -> bool: diff --git a/psa_car_controller/psacc/application/psa_client.py b/psa_car_controller/psacc/application/psa_client.py index 67c3c62..7b08ffa 100644 --- a/psa_car_controller/psacc/application/psa_client.py +++ b/psa_car_controller/psacc/application/psa_client.py @@ -32,7 +32,7 @@ logger = CustomLogger.getLogger(__name__) class PSAClient: def connect(self): - self.manager.init_with_brand_country_code(self.brand, self.country_code) + self.manager.init_with_oauth2_redirect(realm_info[self.realm]["scheme"], self.country_code) # pylint: disable=too-many-arguments def __init__(self, refresh_token, client_id, client_secret, remote_refresh_token, customer_id, realm, country_code,