Files
to_session/text/sign绑推.py
Administrator a0720d80dc fefdwef
2025-11-12 12:54:37 +08:00

209 lines
7.2 KiB
Python

import threading
import time
from urllib.parse import urlparse, parse_qs
from curl_cffi import requests
from models.hub_web import HubWeb
from models.x_token import XToken
from 推特.接口.main import AuthXNew
import datetime
import random
import string
from eth_account.messages import encode_defunct
from web3 import AsyncWeb3
def get_evm_wallet_singed_message(original_msg, private_key):
"""
:param original_msg: 签名时生成的message参数
:param private_key: 私钥
:return:
"""
w3 = AsyncWeb3(AsyncWeb3.AsyncHTTPProvider('https://cloudflare-eth.com'))
account = w3.eth.account.from_key(private_key)
signature = account.sign_message(encode_defunct(text=original_msg))
return signature.signature.hex()
def generate_random_string(length=10):
# 定义可以使用的字符集,包含大写字母、小写字母和数字
characters = string.ascii_letters + string.digits
# 随机选择字符并组合成字符串
random_string = ''.join(random.choice(characters) for _ in range(length))
return random_string
def get_format_time(date):
now_str = date.strftime('%Y-%m-%dT%H:%M:%S.%f')[:-3] + 'Z'
return now_str
def login_sign(private_key, address):
from curl_cffi import requests
cookies = {
'_ga': 'GA1.1.762544975.1745577118',
'connect.sid': 's%3A4dpBOk9k_gJ9LTYz43EDuMJtapBRmcdf.uSmX6Oe9%2BeiOXSTRki%2FJlikI6JmSovEUOmQx4ebVaC8',
'_ga_JZT6JMXTV3': 'GS1.1.1745577118.1.1.1745577724.0.0.0',
'_ga_7L1GCL5RHD': 'GS1.1.1745577138.1.1.1745578489.0.0.0',
}
original_str = f"app.sign.global wants you to sign in with your Ethereum account:\n{address}\n\nSign Protocol\n\nURI: https://app.sign.global\nVersion: 1\nChain ID: 11155111\nNonce: {generate_random_string()}\nIssued At: {get_format_time(datetime.datetime.utcnow())}"
singed_message = get_evm_wallet_singed_message(original_str,
private_key)
headers = {
'accept': '*/*',
'accept-language': 'zh-CN,zh;q=0.9',
'content-type': 'application/json',
'origin': 'https://app.sign.global',
'priority': 'u=1, i',
'referer': 'https://app.sign.global/create-schema',
'sec-ch-ua': '"Not(A:Brand";v="99", "Google Chrome";v="133", "Chromium";v="133"',
'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/133.0.0.0 Safari/537.36',
# 'cookie': '_ga=GA1.1.762544975.1745577118; connect.sid=s%3A4dpBOk9k_gJ9LTYz43EDuMJtapBRmcdf.uSmX6Oe9%2BeiOXSTRki%2FJlikI6JmSovEUOmQx4ebVaC8; _ga_JZT6JMXTV3=GS1.1.1745577118.1.1.1745577724.0.0.0; _ga_7L1GCL5RHD=GS1.1.1745577138.1.1.1745578489.0.0.0',
}
json_data = {
'message': original_str,
'signature': '0x' + singed_message,
'chainType': 'evm',
'client': 'MetaMask',
'key': address,
}
response = requests.post('https://app.sign.global/api/signin', cookies=cookies, headers=headers, json=json_data)
# print(response.headers["Set-Cookie"].split(";")[0].split("=")[-1])
return response.headers["Set-Cookie"].split(";")[0].split("=")[-1]
def get_x_url(cookie):
from curl_cffi import requests
cookies = {
# '_ga': 'GA1.1.1132140841.1745478459',
# '_ga_JZT6JMXTV3': 'GS1.1.1745548073.2.1.1745548104.0.0.0',
'connect.sid': cookie,
# '_ga_7L1GCL5RHD': 'GS1.1.1745569854.6.1.1745570795.0.0.0',
}
headers = {
'accept': '*/*',
'accept-language': 'zh,zh-CN;q=0.9,zh-HK;q=0.8,en;q=0.7',
'cache-control': 'no-cache',
'content-type': 'application/json',
'pragma': 'no-cache',
'priority': 'u=1, i',
'referer': 'https://app.sign.global/profile',
'sec-ch-ua': '"Google Chrome";v="135", "Not-A.Brand";v="8", "Chromium";v="135"',
'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/135.0.0.0 Safari/537.36',
# 'cookie': '_ga=GA1.1.1132140841.1745478459; _ga_JZT6JMXTV3=GS1.1.1745548073.2.1.1745548104.0.0.0; connect.sid=s%3Ao1jl01ZYX_UfSMh958QL0w2GUwhHOVgU.aDCxm8VOxNnuXHydlPwBkkmxIgt27%2FHm6eWe0REYy90; _ga_7L1GCL5RHD=GS1.1.1745569854.6.1.1745570795.0.0.0',
}
response = requests.get('https://app.sign.global/api/my/third-party-accounts', cookies=cookies, headers=headers)
for i in response.json()["data"]["accounts"]:
if i["accountType"] == "twitter":
return i["authorizeUrl"], i["accountInfo"]["id"]
def get111(url, cookie):
cookies = {
# '_ga': 'GA1.1.1924581846.1745478256',
# '_ga_JZT6JMXTV3': 'GS1.1.1745572095.2.0.1745572095.0.0.0',
'connect.sid': cookie,
# '_ga_7L1GCL5RHD': 'GS1.1.1745579764.5.1.1745579996.0.0.0',
}
headers = {
'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/135.0.0.0 Safari/537.36 Edg/135.0.0.0',
# 'Accept-Encoding': 'gzip, deflate, br, zstd',
'Content-Type': 'application/json',
'sec-ch-ua-platform': '"Windows"',
'sec-ch-ua': '"Microsoft Edge";v="135", "Not-A.Brand";v="8", "Chromium";v="135"',
'dnt': '1',
'sec-ch-ua-mobile': '?0',
'origin': 'https://app.sign.global',
'sec-fetch-site': 'same-origin',
'sec-fetch-mode': 'cors',
'sec-fetch-dest': 'empty',
'referer': url,
'accept-language': 'zh-CN,zh;q=0.9,en;q=0.8,en-GB;q=0.7,en-US;q=0.6',
'priority': 'u=1, i',
}
# 解析 URL
parsed_url = urlparse(url)
# 解析查询字符串
query_params = parse_qs(parsed_url.query)
# 获取 code 参数的值
code = query_params.get('code')
json_data = {
'code': code[0],
'accountType': 'twitter',
'redirectUri': 'https://app.sign.global/bind-callback?accountType=twitter',
}
print(
json_data
)
response = requests.post('https://app.sign.global/api/my/bind-oauth', cookies=cookies, headers=headers,
json=json_data)
print(response.text)
def main(sql_pro):
try:
cookie = login_sign(sql_pro.private_key, sql_pro.address)
url1, type1 = get_x_url(cookie=cookie)
if not type1:
x_info = XToken.get_or_none(
XToken.id == sql_pro.x_id,
)
x_new = AuthXNew(auth_token=x_info.token)
x_new.login_x_token()
url = x_new.auth(
url=f"{url1}&redirect_uri=https%3A%2F%2Fapp.sign.global%2Fbind-callback%3FaccountType%3Dtwitter")
get111(url, cookie)
else:
print(sql_pro.id, type1)
except Exception as e:
print(e)
if __name__ == '__main__':
hub_web_infos = HubWeb.select()
for hub_web_info in hub_web_infos:
threading.Thread(target=main, args=(hub_web_info,)).start()
time.sleep(0.5)