feat: retrieve redirect scheme from constant and allow retry auth

Signed-off-by: Florian Bezannier <florian.bezannier@hotmail.fr>

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",

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:

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,