Init commit

Author: akdev-tech

.gitignore

@@ -0,0 +1,2 @@
+env
+__pycache__

auth.py

@@ -0,0 +1,129 @@
+import os
+import base64
+import pickle
+import logging
+from bs4 import BeautifulSoup
+
+
+class Auth:
+    CAPTCHA_FILENAME = "captcha.svg"
+    COOKIES_FILENAME = "cookies.pk"
+
+    def __init__(self, username, password):
+        self.username = username
+        self.password = password
+
+    @classmethod
+    def get_token(cls, session):
+        req = session.get("https://chat.openai.com/api/auth/session")
+        return req.json().get("accessToken")
+
+    @classmethod
+    def save_cookies(cls, session):
+        with open(cls.COOKIES_FILENAME, "wb") as f:
+            pickle.dump(session.cookies.jar._cookies, f)
+
+    @classmethod
+    def load_cookies(cls, cookies):
+        if not os.path.isfile(cls.COOKIES_FILENAME):
+            return None
+        with open(cls.COOKIES_FILENAME, "rb") as f:
+            jar_cookies = pickle.load(f)
+        for domain, pc in jar_cookies.items():
+            for path, c in pc.items():
+                for k, v in c.items():
+                    cookies.set(k, v.value, domain=domain, path=path)
+        return cookies
+
+    @classmethod
+    def check_auth(cls, session):
+        token = cls.get_token(session)
+        return token
+
+    def _get_state(self, resp):
+        return resp.find("input", {"name": "state", "type": "hidden"}).get("value")
+
+    def _captcha_handler(self, resp):
+        captcha = resp.find("img", {"alt": "captcha"})
+        if captcha:
+            src = captcha.get("src").split(",")[1]
+            with open(self.CAPTCHA_FILENAME, "wb") as fh:
+                fh.write(base64.urlsafe_b64decode(src))
+            logging.info("Please enter captcha code: ")
+            code = input()
+            if os.path.isfile(self.CAPTCHA_FILENAME):
+                os.remove(self.CAPTCHA_FILENAME)
+            return code
+
+    def _block_handler(self, resp):
+        s = "We have detected a potential security issue with this account"
+        return s in str(resp)
+
+    def perform_login(self, session):
+        req = session.get("https://chat.openai.com/api/auth/csrf")
+        csrf = req.json().get("csrfToken")
+
+        session.headers.update({"content-type": "application/x-www-form-urlencoded"})
+        params = {
+            "callbackUrl": "/",
+            "csrfToken": csrf,
+            "json": "true"
+        }
+        req = session.post(
+            "https://chat.openai.com/api/auth/signin/auth0?prompt=login", data=params
+        )
+        login_link = req.json().get("url")
+
+        req = session.get(login_link)
+        response = BeautifulSoup(req.text, features="html.parser")
+
+        while True:  # try to auth
+            state = self._get_state(response)
+            code = self._captcha_handler(response)
+
+            session.headers.update({
+                "content-type": "application/x-www-form-urlencoded"
+            })
+            params = {
+                "state": state,
+                "username": self.username,
+                "captcha": code,
+                "js-available": "true",
+                "webauthn-available": "true",
+                "is-brave": "false",
+                "webauthn-platform-available": "false",
+                "action": "default"
+            }
+            url = f"https://auth0.openai.com/u/login/identifier?state={state}"
+            req = session.post(url, data=params)
+            response = BeautifulSoup(req.text, features="html.parser")
+
+            # break if password field is found on page
+            if response.find("input", {"name": "password", "id": "password"}):
+                break
+            logging.info("Incorrect captcha or another error, trying again...")
+
+        state = self._get_state(response)
+
+        is_blocked = self._block_handler(response)
+
+        if is_blocked:
+            logging.warning("We are blocked")
+            return
+
+        req = session.get(f"https://auth0.openai.com/u/login/password?state={state}")
+        response = BeautifulSoup(req.text, features="html.parser")
+        state = self._get_state(response)
+
+        session.headers.update({
+            "content-type": "application/x-www-form-urlencoded"
+        })
+        params = {
+            "state": state,
+            "username": self.username,
+            "password": self.password,
+            "action": "default"
+        }
+        url = f"https://auth0.openai.com/u/login/password?state={state}"
+        req = session.post(url, data=params)
+        return req.url == "https://chat.openai.com/chat"

config.py

@@ -0,0 +1,20 @@
+initial_headers = {
+    "accept": "*/*",
+    "accept-encoding": "gzip, deflate, br",
+    "accept-language": "ru-RU,ru;q=0.9,en-US;q=0.8,en;q=0.7,la;q=0.6",
+    "cache-control": "no-cache",
+    "content-type": "application/x-www-form-urlencoded",
+    "pragma": "no-cache",
+    "referer": "https://chat.openai.com/auth/login",
+    "sec-ch-ua": '"Not?A_Brand";v="8", "Chromium";v="108", "Google Chrome";v="108"',
+    "sec-ch-ua-mobile": "?0",
+    "sec-ch-ua-platform": '"Windows"',
+    "sec-fetch-dest": "empty",
+    "sec-fetch-mode": "cors",
+    "sec-fetch-site": "same-origin",
+    "user-agent": (
+        "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 "
+        "(KHTML, like Gecko) Chrome/108.0.0.0 Safari/537.36"
+    )
+}
+model = "text-davinci-002-render"

errors.py

@@ -0,0 +1,12 @@
+class RequestError(Exception):
+    "Raised when a request wasn't successfull"
+
+    def __init__(self, message=None):
+        super().__init__(message)
+
+
+class ResponseError(Exception):
+    "Raised when the response wasn't successfull"
+
+    def __init__(self, message=None):
+        super().__init__(message)

main.py

@@ -0,0 +1,83 @@
+import httpx
+import logging
+import argparse
+
+import config
+from auth import Auth
+from stream import Stream
+
+parser = argparse.ArgumentParser(description="ChatGPT dialog without browser")
+parser.add_argument(
+    "-u",
+    "--username",
+    dest="username",
+    help="Your username (email) for openai.com",
+    required=True,
+)
+parser.add_argument(
+    "-p",
+    "--password",
+    dest="password",
+    help="Your password for openai.com",
+    required=True,
+)
+parser.add_argument(
+    "-m",
+    "--message",
+    dest="message",
+    help="Message for ChatGPT",
+    required=True,
+)
+parser.add_argument(
+    "-vvv",
+    "--verbose",
+    action="store_true",
+    dest="verbose",
+    help="Enable debug logging level",
+)
+
+args = parser.parse_args()
+
+
+level = (logging.DEBUG if args.verbose else logging.INFO)
+logging.basicConfig(level=level)
+
+
+def log_request(req):
+    logging.debug(f"Request event hook: {req.method} {req.url}")
+
+
+def log_response(resp):
+    req = resp.request
+    logging.debug(f"Response event hook: {req.method} {req.url}")
+    logging.debug(f"Status - {resp.status_code}")
+    logging.debug(resp.read())
+    for cookie in resp.cookies:
+        logging.debug("Cookie", cookie, resp.cookies[cookie])
+    for header in resp.headers:
+        logging.debug("Header", header, resp.headers[header])
+
+
+if args.verbose:
+    event_hooks = {"request": [log_request], "response": [log_response]}
+else:
+    event_hooks = None
+
+session = httpx.Client(
+    http2=True,
+    headers=config.initial_headers,
+    event_hooks=event_hooks,
+    follow_redirects=True,
+    cookies=Auth.load_cookies(httpx.Cookies()),
+)
+session.cookies.jar.clear_expired_cookies()
+
+while not Auth.check_auth(session):
+    logging.info("Start the loging process")
+    auth_instance = Auth(args.username, args.password)
+    auth_instance.perform_login(session)
+
+
+stream = Stream(session)
+response = stream.send_message(args.message)
+print(response)

requirements.txt

@@ -0,0 +1,22 @@
+anyio==3.6.2
+beautifulsoup4==4.11.1
+certifi==2022.12.7
+cffi==1.15.1
+charset-normalizer==2.1.1
+cryptography==38.0.4
+h11==0.14.0
+h2==4.1.0
+hpack==4.0.0
+httpcore==0.16.2
+httpx==0.23.1
+hyperframe==6.0.1
+idna==3.4
+ndg-httpsclient==0.5.1
+pyasn1==0.4.8
+pycparser==2.21
+pyOpenSSL==22.1.0
+requests==2.28.1
+rfc3986==1.5.0
+sniffio==1.3.0
+soupsieve==2.3.2.post1
+urllib3==1.26.13

stream.py

@@ -0,0 +1,64 @@
+import uuid
+import json
+
+import config
+from auth import Auth
+from errors import ResponseError
+
+
+class Stream:
+    def __init__(self, session):
+        self.session = session
+
+    def _process_chunks(self, chunks):
+        if type(chunks) == bytes:
+            chunks = chunks.decode()
+            chunks = chunks.split("\n\n")
+        elif type(chunks) == list:
+            chunks = [c.decode() for c in chunks]
+        else:
+            raise ResponseError("Unknown response")
+        chunks = [c for c in chunks if c and "message" in c]
+        if not chunks:
+            raise ResponseError("No response in stream")
+        chunk = chunks[-1]
+        chunk = chunk.replace("data: ", "").strip()
+        chunk = json.loads(chunk)
+        return chunk.get("message", {}).get("content", {}).get("parts", [None])[0]
+
+    def send_message(self, message):
+        token = Auth.get_token(self.session)
+
+        self.session.headers.update({
+            "accept": "text/event-stream",
+            "X-OpenAI-Assistant-App-Id": "",
+            "Authorization": f"Bearer {token}"
+        })
+
+        params = {
+            "action": "next",
+            "messages": [
+                {
+                    "id": str(uuid.uuid4()),
+                    "role": "user",
+                    "content": {
+                        "content_type": "text",
+                        "parts": [message]
+                    }
+                }
+            ],
+            "parent_message_id": "",
+            "model": config.model
+        }
+
+        with self.session.stream(
+            "POST", "https://chat.openai.com/backend-api/conversation",
+            json=params, headers={"content-type": "application/json"},
+            timeout=None
+        ) as response:
+            chunks = []
+            for chunk in response.iter_bytes():
+                chunks.append(chunk)
+
+            response = self._process_chunks(chunks)
+            return response