Accomodate for Stellantis' Jan 23 OAuth changes

psa_car_controller/psa/constants.py

@@ -28,7 +28,7 @@
     "program3": {"day": [0, 0, 0, 0, 0, 0, 0], "hour": 34, "minute": 7, "on": 0},
     "program4": {"day": [0, 0, 0, 0, 0, 0, 0], "hour": 34, "minute": 7, "on": 0}
 }
-AUTHORIZE_SERVICE = "https://api.mpsa.com/api/connectedcar/v2/oauth/authorize"
+AUTHORIZE_SERVICE = "https://idpcvs.opel.com/am/oauth2/authorize"
 REMOTE_URL = "https://api.groupe-psa.com/connectedcar/v4/virtualkey/remoteaccess/token?client_id="
 BRAND = {"com.psa.mym.myopel": {"realm": "clientsB2COpel", "brand_code": "OP", "app_name": "MyOpel"},
          "com.psa.mym.mypeugeot": {"realm": "clientsB2CPeugeot", "brand_code": "AP", "app_name": "MyPeugeot"},

psa_car_controller/psa/oauth.py

@@ -1,4 +1,9 @@
 import logging
+import hashlib
+import secrets
+import base64
+from typing import Tuple
+
 from http import HTTPStatus
 from typing import Optional
 
@@ -22,8 +27,25 @@
         return {"grant_type": 'password', "username": login, "scope": ' '.join(self.service_information.scopes),
                 "password": password, "realm": realm}
 
-    def init_with_user_credentials_realm(self, login: str, password: str, realm: str):
-        self._token_request(self._grant_password_request_realm(login, password, realm), True)
+    def generate_sha256_pkce(self, length: int) -> Tuple[str, str]:
+        if not (43 <= length <= 128):
+            raise Exception("Invalid length: " % str(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 init_with_user_credentials_realm(self, country_code: str, realm: str):
+        redir_uri = "mymop://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")
+
+        ret = ""
+        while len(ret) != 36:
+            logger.info("Now login to this URL in a browser: " + url)
+            ret = input("\nCopy+paste the resulting mymop-code (in F12 > Network when you hit the final OK button, 36 chars, UUID format): ")
+
+        self._token_request(dict(grant_type='authorization_code', code=ret, redirect_uri=redir_uri, code_verifier=code_verifier), False)
 
     @staticmethod
     def _is_token_expired(response: Response) -> bool:

psa_car_controller/psa/setup/app_decoder.py

@@ -92,7 +92,7 @@ def firstLaunchConfig(package_name, client_email, client_password, country_code,
     psacc = PSAClient(None, apk_parser.client_id, apk_parser.client_secret,
                       None, customer_id, BRAND[package_name]["realm"],
                       country_code)
-    psacc.connect(client_email, client_password)
+    psacc.connect(country_code)
     psacc.save_config(name=config_prefix + "config.json")
     res = psacc.get_vehicles()
 

psa_car_controller/psacc/application/psa_client.py

@@ -31,8 +31,8 @@
 
 
 class PSAClient:
-    def connect(self, user, password):
-        self.manager.init_with_user_credentials_realm(user, password, self.realm)
+    def connect(self, country_code):
+        self.manager.init_with_user_credentials_realm(country_code, self.realm)
 
     # pylint: disable=too-many-arguments
     def __init__(self, refresh_token, client_id, client_secret, remote_refresh_token, customer_id, realm, country_code,