email/password login process reverts to local

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

src/pandora/__init__.py

@@ -1,3 +1,3 @@
 # -*- coding: utf-8 -*-
 
-__version__ = '0.9.5'
+__version__ = '0.9.6'

src/pandora/launcher.py

@@ -161,7 +161,6 @@ def main():
     access_token, need_save = confirm_access_token(args.token_file, args.server, args.api)
     if not access_token:
         Console.info_b('Please enter your email and password to log in ChatGPT!')
-        Console.warn('We login via https://chat.gateway.do, but it doesn\'t retain your data.')
         email = getenv('OPENAI_EMAIL') or Prompt.ask('  Email')
         password = getenv('OPENAI_PASSWORD') or Prompt.ask('  Password', password=True)
         Console.warn('### Do login, please wait...')

src/pandora/openai/auth.py

@@ -3,9 +3,11 @@
 import datetime
 import re
 from datetime import datetime as dt
+from urllib.parse import urlparse, parse_qs
 
 import requests
 from certifi import where
+from requests.exceptions import SSLError
 
 
 class Auth0:
@@ -40,18 +42,98 @@ def auth(self) -> str:
         if not self.__check_email(self.email) or not self.password:
             raise Exception('invalid email or password.')
 
-        return self.get_access_token()
+        return self.__part_two()
 
-    def get_access_token(self) -> str:
-        url = 'https://chat.gateway.do/api/auth/login'
+    def __part_two(self) -> str:
+        url = 'https://chat.gateway.do/auth/endpoint'
         headers = {
             'User-Agent': self.user_agent,
         }
+        resp = self.session.get(url, headers=headers, 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)
+
+    def __part_three(self, state1: str, url: str) -> str:
+        headers = {
+            'User-Agent': self.user_agent,
+            'Referer': 'https://explorer.api.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)
+            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:
+        url = 'https://auth0.openai.com/u/login/identifier?state=' + state
+        headers = {
+            'User-Agent': self.user_agent,
+            'Referer': url,
+            'Origin': 'https://auth0.openai.com',
+        }
+        data = {
+            'state': state,
+            'username': self.email,
+            'js-available': 'true',
+            'webauthn-available': 'true',
+            'is-brave': 'false',
+            'webauthn-platform-available': 'true',
+            'action': 'default',
+        }
+        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)
+        else:
+            raise Exception('Error check email.')
+
+    def __part_five(self, state1: str, state: str) -> str:
+        url = 'https://auth0.openai.com/u/login/password?state=' + state
+        headers = {
+            'User-Agent': self.user_agent,
+            'Referer': url,
+            'Origin': 'https://auth0.openai.com',
+        }
         data = {
+            'state': state,
             'username': self.email,
             'password': self.password,
+            'action': 'default',
+        }
+
+        try:
+            resp = self.session.post(url, headers=headers, data=data, allow_redirects=True, **self.req_kwargs)
+        except SSLError as e:
+            if not hasattr(e.request, 'url') or not e.request.url:
+                raise Exception('Error login')
+
+            return self.get_access_token(state1, e.request.url)
+
+        if resp.status_code == 400:
+            raise Exception('Wrong email or password.')
+        else:
+            raise Exception('Error login.')
+
+    def get_access_token(self, state1: str, callback_url: str) -> str:
+        url = 'https://chat.gateway.do/auth/token'
+        headers = {
+            'User-Agent': self.user_agent,
+        }
+        data = {
+            'state': state1,
+            'callbackUrl': callback_url,
         }
-        resp = self.session.post(url=url, headers=headers, data=data, allow_redirects=False, **self.req_kwargs)
+        resp = self.session.post(url, headers=headers, data=data, allow_redirects=False, **self.req_kwargs)
 
         if resp.status_code == 200:
             json = resp.json()
@@ -62,4 +144,4 @@ def get_access_token(self) -> str:
             self.expires = dt.strptime(json['expires'], '%Y-%m-%dT%H:%M:%S.%fZ') - datetime.timedelta(minutes=5)
             return self.access_token
         else:
-            raise Exception('Error get access token.')
+            raise Exception(resp.text)