Update x-signature for forefront, python implementation

[ZachKLYeh]

你好，我參考了你的代碼想要用python修復forefront在gpt4free的問題，但是我一直得不到response，想請問您有關於這邊編碼的部分，是不是有誤？

[ZachKLYeh]

from json import loads
from re import findall
from time import time, sleep
from typing import Generator, Optional
from uuid import uuid4

import base64
from Crypto.Cipher import AES
from Crypto.Util.Padding import pad
from Crypto import Random
from Crypto.Hash import SHA256

from mailgw_temporary_email import Email
from fake_useragent import UserAgent
from requests import post
from tls_client import Session

from .typing import ForeFrontResponse


def make_signature(user_id, default_persona, chat_id, session_id):
    data = user_id + default_persona + chat_id
    data = data.encode('ascii')
    base64data = base64.b64encode(data)
    signature = aes_encrypt(base64data, session_id)
    return signature

def aes_encrypt(data, key):
    iv = Random.get_random_bytes(16)
    hash_obj = SHA256.new()
    hash_obj.update(key.encode('utf-8'))
    hash_val = hash_obj.digest()
    cipher = AES.new(hash_val, AES.MODE_CBC, iv)
    res = cipher.encrypt(pad(data, AES.block_size))
    return str(iv.hex()) + str(res) 

class Account:
    @staticmethod
    def create(proxy: Optional[str] = None, logging: bool = False):
        proxies = {'http': 'http:https://' + proxy, 'https': 'http:https://' + proxy} if proxy else False

        start = time()

        mail_client = Email()
        mail_client.register()
        mail_address = mail_client.address

        agent = str(UserAgent().random)
        client = Session(client_identifier='chrome110')
        client.proxies = proxies
        client.headers = {
            'origin': 'https://accounts.forefront.ai',
            'user-agent': agent,
        }

        response = client.post(
            'https://clerk.forefront.ai/v1/client/sign_ups?_clerk_js_version=4.38.4',
            data={'email_address': mail_address},
        )
        print(response.json()['response']['id'])

        try:
            trace_token = response.json()['response']['id']
            if logging:
                print(trace_token)
        except KeyError:
            return 'Failed to create account!'

        response = client.post(
            f'https://clerk.forefront.ai/v1/client/sign_ups/{trace_token}/prepare_verification?_clerk_js_version=4.38.4',
            data={
                'strategy': 'email_link',
                'redirect_url': 'https://accounts.forefront.ai/sign-up/verify'
            },
        )

        if logging:
            print(response.text)

        if 'sign_up_attempt' not in response.text:
            return 'Failed to create account!'

        while True:
            sleep(5)
            message_id = mail_client.message_list()[0]['id']
            message = mail_client.message(message_id)
            
            new_message: Message = message
            verification_url = findall(r'https:\/\/clerk\.forefront\.ai\/v1\/verify\?token=\w.+', new_message["text"])[0]

            if verification_url:
                break

        if logging:
            print(verification_url)

        response = client.get(verification_url)

        response = client.get('https://clerk.forefront.ai/v1/client?_clerk_js_version=4.38.4')

        token = response.json()['response']['sessions'][0]['last_active_token']['jwt']
        session_id = response.json()['response']['sessions'][0]['id']
        user_id = response.json()['response']['sessions'][0]['user']['id']

        with open('accounts.txt', 'a') as f:
            f.write(f'{mail_address}:{token}\n')

        if logging:
            print(time() - start)

        return token, session_id, user_id, agent

class StreamingCompletion:
    @staticmethod
    def create(
        token=None,
        chat_id=None,
        prompt='',
        action_type='new',
        default_persona='607e41fe-95be-497e-8e97-010a59b2e2c0',  # default
        model='gpt-4',
        proxy=None,
        session_id=None,
        user_id=None,
        agent=None
    ) -> Generator[ForeFrontResponse, None, None]:
        if not token:
            raise Exception('Token is required!')
        if not chat_id:
            chat_id = str(uuid4())

        proxies = { 'http': 'http:https://' + proxy, 'https': 'http:https://' + proxy } if proxy else None
        x_signature = make_signature(user_id, default_persona, chat_id, session_id)

        headers = {
            'authority': 'chat-server.tenant-forefront-default.knative.chi.coreweave.com',
            'accept': '*/*',
            'accept-language': 'en,fr-FR;q=0.9,fr;q=0.8,es-ES;q=0.7,es;q=0.6,en-US;q=0.5,am;q=0.4,de;q=0.3',
            'authorization': 'Bearer ' + token,
            'cache-control': 'no-cache',
            'content-type': 'application/json',
            'origin': 'https://chat.forefront.ai',
            'pragma': 'no-cache',
            'referer': 'https://chat.forefront.ai/',
            'sec-ch-ua': '"Chromium";v="112", "Google Chrome";v="112", "Not:A-Brand";v="99"',
            'sec-ch-ua-mobile': '?0',
            'sec-ch-ua-platform': '"macOS"',
            'sec-fetch-dest': 'empty',
            'sec-fetch-mode': 'cors',
            'sec-fetch-site': 'cross-site',
            'user-agent': agent,
            'x-signature': x_signature, 
        }

        json_data = {
            'text': prompt,
            'action': action_type,
            'parentId': chat_id,
            'workspaceId': chat_id,
            'messagePersona': default_persona,
            'model': model,
        }

        for chunk in post(
            'https://streaming.tenant-forefront-default.knative.chi.coreweave.com/chat',    
            headers=headers,
            proxies=proxies,
            json=json_data,
            stream=True,
        ).iter_lines():
            if b'finish_reason":null' in chunk:
                data = loads(chunk.decode('utf-8').split('data: ')[1])
                token = data['choices'][0]['delta'].get('content')

                if token is not None:
                    yield ForeFrontResponse(
                        **{
                            'id': chat_id,
                            'object': 'text_completion',
                            'created': int(time()),
                            'text': token,
                            'model': model,
                            'choices': [{'text': token, 'index': 0, 'logprobs': None, 'finish_reason': 'stop'}],
                            'usage': {
                                'prompt_tokens': len(prompt),
                                'completion_tokens': len(token),
                                'total_tokens': len(prompt) + len(token),
                            },
                        }
                    )


class Completion:
    @staticmethod
    def create(
        token=None,
        chat_id=None,
        prompt='',
        action_type='new',
        default_persona='607e41fe-95be-497e-8e97-010a59b2e2c0',  # default
        model='gpt-4',
        proxy=None,
        session_id=None,
        user_id=None,
        agent=None
    ) -> ForeFrontResponse:
        text = ''
        final_response = None
        for response in StreamingCompletion.create(
            token=token,
            chat_id=chat_id,
            prompt=prompt,
            action_type=action_type,
            default_persona=default_persona,
            model=model,
            proxy=proxy,
            session_id=session_id,
            user_id=user_id,
            agent=agent
        ):
            if response:
                final_response = response
                text += response.text

        if final_response:
            final_response.text = text
        else:
            raise Exception('Unable to get the response, Please try again')

        return final_response

[xiangsx]

有人提过pr修复了