refactor the auth code according to this article: https://zhile.io/20…

…23/05/19/how-to-get-chatgpt-access-token-via-pkce.html

Signed-off-by: pengzhile <[email protected]>

src/pandora/__init__.py

@@ -1,3 +1,3 @@
 # -*- coding: utf-8 -*-
 
-__version__ = '1.1.3'
+__version__ = '1.1.4'

src/pandora/openai/auth.py

@@ -46,40 +46,34 @@ def auth(self, login_local=True) -> str:
         return self.__part_two() if login_local else self.get_access_token_proxy()
 
     def __part_two(self) -> str:
-        url = '{}/auth/endpoint1'.format(self.api_prefix)
-        headers = {
-            'User-Agent': self.user_agent,
-        }
-        data = {
-            'g-recaptcha-response': 'Pandora is so cool!',
-        }
+        code_challenge = 'w6n3Ix420Xhhu-Q5-mOOEyuPZmAsJHUbBpO8Ub7xBCY'
+        code_verifier = 'yGrXROHx_VazA0uovsxKfE263LMFcrSrdm4SlC-rob8'
 
-        resp = self.session.post(url, headers=headers, data=data, allow_redirects=False, **self.req_kwargs)
-
-        if resp.status_code == 200:
-            json = resp.json()
-            return self.__part_three(json['state'], json['url'])
-        else:
-            raise Exception(resp.text)
+        url = 'https://auth0.openai.com/authorize?client_id=pdlLIX2Y72MIl2rhLhTE9VV9bN905kBh&audience=https%3A%2F' \
+              '%2Fapi.openai.com%2Fv1&redirect_uri=com.openai.chat%3A%2F%2Fauth0.openai.com%2Fios%2Fcom.openai.chat' \
+              '%2Fcallback&scope=openid%20email%20profile%20offline_access%20model.request%20model.read' \
+              '%20organization.read%20offline&response_type=code&code_challenge={}' \
+              '&code_challenge_method=S256&prompt=login'.format(code_challenge)
+        return self.__part_three(code_verifier, url)
 
-    def __part_three(self, state1: str, url: str) -> str:
+    def __part_three(self, code_verifier, url: str) -> str:
         headers = {
             'User-Agent': self.user_agent,
-            'Referer': 'https://chat.openai.com/',
+            'Referer': 'https://ios.chat.openai.com/',
         }
         resp = self.session.get(url, headers=headers, allow_redirects=True, **self.req_kwargs)
 
         if resp.status_code == 200:
             try:
                 url_params = parse_qs(urlparse(resp.url).query)
                 state = url_params['state'][0]
-                return self.__part_four(state1, state)
+                return self.__part_four(code_verifier, state)
             except IndexError as exc:
                 raise Exception('Rate limit hit.') from exc
         else:
             raise Exception('Error request login url.')
 
-    def __part_four(self, state1: str, state: str) -> str:
+    def __part_four(self, code_verifier: str, state: str) -> str:
         url = 'https://auth0.openai.com/u/login/identifier?state=' + state
         headers = {
             'User-Agent': self.user_agent,
@@ -98,11 +92,11 @@ def __part_four(self, state1: str, state: str) -> str:
         resp = self.session.post(url, headers=headers, data=data, allow_redirects=False, **self.req_kwargs)
 
         if resp.status_code == 302:
-            return self.__part_five(state1, state)
+            return self.__part_five(code_verifier, state)
         else:
             raise Exception('Error check email.')
 
-    def __part_five(self, state1: str, state: str) -> str:
+    def __part_five(self, code_verifier: str, state: str) -> str:
         url = 'https://auth0.openai.com/u/login/password?state=' + state
         headers = {
             'User-Agent': self.user_agent,
@@ -122,14 +116,14 @@ def __part_five(self, state1: str, state: str) -> str:
             if not location.startswith('/authorize/resume?'):
                 raise Exception('Login failed.')
 
-            return self.__part_six(state1, location, url)
+            return self.__part_six(code_verifier, location, url)
 
         if resp.status_code == 400:
             raise Exception('Wrong email or password.')
         else:
             raise Exception('Error login.')
 
-    def __part_six(self, state1: str, location: str, ref: str) -> str:
+    def __part_six(self, code_verifier: str, location: str, ref: str) -> str:
         url = 'https://auth0.openai.com' + location
         headers = {
             'User-Agent': self.user_agent,
@@ -139,31 +133,45 @@ def __part_six(self, state1: str, location: str, ref: str) -> str:
         resp = self.session.get(url, headers=headers, allow_redirects=False, **self.req_kwargs)
         if resp.status_code == 302:
             location = resp.headers['Location']
-            if not location.startswith('https://chat.openai.com/api/auth/callback/auth0?'):
+            if not location.startswith('com.openai.chat://auth0.openai.com/ios/com.openai.chat/callback?'):
                 raise Exception('Login callback failed.')
 
-            return self.get_access_token(state1, resp.headers['Location'])
+            return self.get_access_token(code_verifier, resp.headers['Location'])
 
         raise Exception('Error login.')
 
-    def get_access_token(self, state1: str, callback_url: str) -> str:
-        url = '{}/auth/token1'.format(self.api_prefix)
+    def get_access_token(self, code_verifier: str, callback_url: str) -> str:
+        url_params = parse_qs(urlparse(callback_url).query)
+
+        if 'error' in url_params:
+            error = url_params['error'][0]
+            error_description = url_params['error_description'][0] if 'error_description' in url_params else ''
+            raise Exception('{}: {}'.format(error, error_description))
+
+        if 'code' not in url_params:
+            raise Exception('Error get code from callback url.')
+
+        url = 'https://auth0.openai.com/oauth/token'
         headers = {
             'User-Agent': self.user_agent,
         }
         data = {
-            'state': state1,
-            'callbackUrl': callback_url,
+            'redirect_uri': 'com.openai.chat://auth0.openai.com/ios/com.openai.chat/callback',
+            'grant_type': 'authorization_code',
+            'client_id': 'pdlLIX2Y72MIl2rhLhTE9VV9bN905kBh',
+            'code': url_params['code'][0],
+            'code_verifier': code_verifier,
         }
-        resp = self.session.post(url, headers=headers, data=data, allow_redirects=False, **self.req_kwargs)
+        resp = self.session.post(url, headers=headers, json=data, allow_redirects=False, **self.req_kwargs)
 
         if resp.status_code == 200:
             json = resp.json()
-            if 'accessToken' not in json:
+            if 'access_token' not in json:
                 raise Exception('Get access token failed, maybe you need a proxy.')
 
-            self.access_token = json['accessToken']
-            self.expires = dt.strptime(json['expires'], '%Y-%m-%dT%H:%M:%S.%fZ') - datetime.timedelta(minutes=5)
+            self.access_token = json['access_token']
+            expires_at = dt.utcnow() + datetime.timedelta(seconds=json['expires_in']) - datetime.timedelta(minutes=5)
+            self.expires = expires_at.strftime('%Y-%m-%dT%H:%M:%S.%fZ')
             return self.access_token
         else:
             raise Exception(resp.text)