329 lines
11 KiB
Python
329 lines
11 KiB
Python
import time
|
|
import json
|
|
import base64
|
|
import urllib
|
|
import hashlib
|
|
import requests
|
|
|
|
from urllib.parse import unquote
|
|
|
|
from typing import Optional, List
|
|
from nacl.signing import SigningKey
|
|
from tonsdk.utils import bytes_to_b64str
|
|
from tonsdk.contract.wallet import Wallets, WalletVersionEnum
|
|
|
|
|
|
def get_user_data1(
|
|
phone,
|
|
bot_name,
|
|
url,
|
|
invite_code="",
|
|
platform="",
|
|
start_params="",
|
|
channel_name='',
|
|
device_model=""
|
|
):
|
|
# data = {
|
|
# "phone": phone, # 区号加号码
|
|
# "bot_name": bot_name, # 机器人名称
|
|
# "url": url,
|
|
# "platform": platform, # 平台 web_3 tdesktop
|
|
# "start_params": start_params,
|
|
# 'channel_name': channel_name, # 频道名称
|
|
# }
|
|
|
|
data = {
|
|
"phone": phone, # 区号加号码
|
|
"bot_name": bot_name, # 机器人名称
|
|
"url": url,
|
|
"invite_code": invite_code, # 邀请码
|
|
"platform": platform, # 平台 web_3 tdesktop
|
|
"start_params": start_params,
|
|
"channel_name": channel_name, # 频道名称
|
|
# 'device_model': device_model,
|
|
}
|
|
|
|
resp = requests.post("http://192.168.11.30:9000/api/get_token", json=data)
|
|
# print(resp.json())
|
|
if resp.status_code == 200:
|
|
return resp.json().get("data")
|
|
|
|
return False
|
|
|
|
|
|
def send_click1(phone, bot_name, datas):
|
|
json_data = {
|
|
"phone": phone,
|
|
"bot_name": bot_name,
|
|
"datas": datas
|
|
}
|
|
|
|
try:
|
|
res = requests.post("http://192.168.11.30:9000/api/send_click", json=json_data)
|
|
print(res.json())
|
|
|
|
return True
|
|
except:
|
|
pass
|
|
return False
|
|
|
|
|
|
def join(phone, bot_name):
|
|
json_data = {
|
|
"phone": phone,
|
|
"bot_name": bot_name,
|
|
}
|
|
|
|
try:
|
|
res = requests.post("http://192.168.11.30:9000/api/join_channe", json=json_data)
|
|
print(res.json())
|
|
|
|
if res.json()["code"] == 200:
|
|
return True
|
|
except:
|
|
pass
|
|
return False
|
|
|
|
|
|
def get_bot_message(phone, bot_name):
|
|
json_data = {
|
|
"phone": phone,
|
|
"bot_name": bot_name,
|
|
}
|
|
|
|
try:
|
|
res = requests.post("http://192.168.11.30:9000/api/get_bot_message", json=json_data)
|
|
print(res.json())
|
|
|
|
if res.json()["code"] == 200:
|
|
return True
|
|
except:
|
|
pass
|
|
return False
|
|
|
|
|
|
# 解析url
|
|
def url_to_json(data):
|
|
# 解码URL编码的数据为字典
|
|
decoded_data = urllib.parse.parse_qs(data)
|
|
|
|
# 提取单一值并解析嵌套的JSON数据
|
|
parsed_data = {k: json.loads(v[0]) if v[0].startswith('{') else v[0] for k, v in decoded_data.items()}
|
|
|
|
return parsed_data
|
|
|
|
|
|
async def wallet(mnemonic: Optional[List[str]] = None):
|
|
"""
|
|
This function creates or loads a wallet depending on whether a mnemonic is provided.
|
|
|
|
- If a mnemonic is provided, it loads the wallet using the mnemonic phrase.
|
|
- If no mnemonic is provided, it creates a new wallet.
|
|
|
|
The function returns the following details:
|
|
- Wallet address
|
|
- Secret key (first 32 bytes of private key)
|
|
- Mnemonic phrase as a string
|
|
- Ton address in user-friendly format
|
|
- Public key in hexadecimal format
|
|
- Private key in hexadecimal format
|
|
- State initialization data in base64 format
|
|
|
|
Args:
|
|
- mnemonic (Optional[List[str]]): A list of words representing the mnemonic phrase to load an existing wallet. Defaults to None.
|
|
|
|
Returns:
|
|
- Tuple: Contains wallet address, secret key, mnemonic, user-friendly Ton address, public key, private key, and state_init.
|
|
"""
|
|
if mnemonic is None:
|
|
mnemonic, pub_k, priv_k, wallet = Wallets.create(WalletVersionEnum.v4r2, workchain=0)
|
|
else:
|
|
mnemonic, pub_k, priv_k, wallet = Wallets.from_mnemonics(mnemonic, WalletVersionEnum.v4r2, workchain=0)
|
|
|
|
state = wallet.create_state_init()
|
|
init = state["state_init"]
|
|
|
|
address = wallet.address.to_string(is_user_friendly=False)
|
|
secret_key = priv_k[:32]
|
|
mnemonics = " ".join(mnemonic)
|
|
address_ton = wallet.address.to_string(True, True, False)
|
|
public_key = pub_k.hex()
|
|
private_key = priv_k.hex()
|
|
state_init = bytes_to_b64str(init.to_boc(has_idx=False))
|
|
|
|
return address, secret_key, mnemonics, address_ton, public_key, private_key, state_init
|
|
|
|
|
|
async def proof(manifest_url: str, payload: str, mnemonic: Optional[str] = None):
|
|
"""
|
|
This function creates a proof by signing a message and returning the proof details in a structured format.
|
|
|
|
- The proof is generated by hashing the address, domain (manifest URL), timestamp, and payload.
|
|
- It then signs the resulting hash using the wallet's private key.
|
|
- The final proof contains the signature and other details such as the wallet information, timestamp, and state initialization.
|
|
|
|
Args:
|
|
- manifest_url (str): The URL from the manifest (domain) to include in the proof.
|
|
- payload (str): The payload (data) to be included in the proof.
|
|
- mnemonic (Optional[str]): A string representing the mnemonic phrase for the wallet. Defaults to None.
|
|
|
|
Returns:
|
|
- str: A JSON string containing the wallet information (mnemonics, Ton address, public/private keys)
|
|
and the proof details (signature, timestamp, domain, state init, etc.).
|
|
"""
|
|
if mnemonic is not None:
|
|
phrase = mnemonic.split()
|
|
else:
|
|
phrase = None
|
|
|
|
address, secret_key, mnemonics, address_ton, public_key, private_key, state_init = await wallet(phrase)
|
|
try:
|
|
timestamp = int(time.time())
|
|
timestamp_bytes = timestamp.to_bytes(8, 'little')
|
|
|
|
domain = manifest_url
|
|
domain_bytes = domain.encode('utf-8')
|
|
domain_len_bytes = len(domain_bytes).to_bytes(4, 'little')
|
|
|
|
workchain, addr_hash = address.split(':')
|
|
workchain_bytes = int(workchain).to_bytes(4, 'big')
|
|
address_bytes = workchain_bytes + bytes.fromhex(addr_hash)
|
|
|
|
msg_bytes = b''.join([
|
|
b'ton-proof-item-v2/',
|
|
address_bytes,
|
|
domain_len_bytes,
|
|
domain_bytes,
|
|
timestamp_bytes,
|
|
payload.encode('utf-8') if payload else b'',
|
|
])
|
|
|
|
msg_hash = hashlib.sha256(msg_bytes).digest()
|
|
|
|
buffer_bytes = b''.join([
|
|
bytes.fromhex('ffff'),
|
|
b'ton-connect',
|
|
msg_hash
|
|
])
|
|
|
|
key = SigningKey(secret_key)
|
|
data = key.sign(hashlib.sha256(buffer_bytes).digest())
|
|
signature = base64.b64encode(data.signature).decode('utf-8')
|
|
|
|
proof = {
|
|
"wallet": {
|
|
"mnemonics": mnemonics,
|
|
"ton_address": address_ton
|
|
},
|
|
"address": address,
|
|
"network": "-239",
|
|
"public_key": public_key,
|
|
"private_key": private_key,
|
|
"proof": {
|
|
"name": "ton_proof",
|
|
"timestamp": timestamp,
|
|
"domain": {
|
|
"lengthBytes": len(domain_bytes),
|
|
"value": domain
|
|
},
|
|
"signature": signature,
|
|
"payload": payload,
|
|
"state_init": state_init
|
|
}
|
|
}
|
|
|
|
return proof
|
|
|
|
except Exception as e:
|
|
print(f"CreateTonProof: {e} | {str(e)}")
|
|
return None
|
|
|
|
# def url_decode_in_query_string(query_string):
|
|
# """
|
|
# 将包含 URL 编码数据的 URL 查询字符串解码
|
|
#
|
|
# :param query_string: 包含 URL 编码数据的 URL 查询字符串
|
|
# :return: 解码后的字典
|
|
# """
|
|
#
|
|
# def decode_url_in_value(value):
|
|
# if isinstance(value, str):
|
|
# try:
|
|
# # 尝试将字符串解析为 JSON 对象
|
|
# json_value = json.loads(value)
|
|
# # 递归处理 JSON 对象中的每个值
|
|
# return decode_url_in_value(json_value)
|
|
# except json.JSONDecodeError:
|
|
# # 如果不是 JSON 格式，则直接解码 URL 编码字符串
|
|
# return urllib.parse.unquote(value)
|
|
# elif isinstance(value, dict):
|
|
# return {k: decode_url_in_value(v) for k, v in value.items()}
|
|
# elif isinstance(value, list):
|
|
# return [decode_url_in_value(v) for v in value]
|
|
# else:
|
|
# return value
|
|
#
|
|
# try:
|
|
# # 将 URL 查询字符串解析为字典
|
|
# query_params = urllib.parse.parse_qs(query_string)
|
|
# # 将字典的值转换为单个元素
|
|
# query_params = {k: v[0] for k, v in query_params.items()}
|
|
# # 对字典中的每个 URL 编码字符串进行解码
|
|
# decoded_params = decode_url_in_value(query_params)
|
|
# return decoded_params
|
|
# except Exception as e:
|
|
# return f"解码失败: {e}"
|
|
|
|
|
|
# 解析url
|
|
|
|
# if __name__ == '__main__':
|
|
# print(get_user_data1(
|
|
# phone="15798297810",
|
|
# bot_name="collection_hunt_bot",
|
|
# url="https://clicker.liveart.io"
|
|
# ))
|
|
|
|
|
|
# a = 'query_id=AAGCcmcdAwAAAIJyZx0_LSS7&user={"id":6935769730,"first_name":"🐦 SUI Alexander","last_name":"Pratt","username":"dJQQPM996","language_code":"zh-hans","allows_write_to_pm":true,"photo_url":"https:\/\/t.me\/i\/userpic\/320\/SFt9SCZ5oRgBLi1Wg8yMBGegf6qQK7_7N8_dcT3j1BH1fryjg0sFfEw0NzKeXBX8.svg"}&auth_date=1751355393&signature=Bcdr-HzJZEwpY_Tq0nRIoffh4ZiK0YeZBTi2SB1IrzUb3mZKP0KjlbM8r6kPP2xMfitXUgK0HS6XXCSZPS8oAw&hash=94f06b976fbf295a29f999a490a7c7974f6a809143c3527f5e7866d8a29b6d68'
|
|
# print(url_to_json(data=a))
|
|
# join_the_group(phone="15798298193", channel_name="timefarmcommunitychat",device_model="")
|
|
#
|
|
# # url = "query_id=AAG0N3o0AwAAALQ3ejRk9rhk&user=%7B%22id%22%3A7322875828%2C%22first_name%22%3A%22%F0%9F%90%A6%20SUI%20Rodney%22%2C%22last_name%22%3A%22%22%2C%22username%22%3A%22rcXbpC4996%22%2C%22language_code%22%3A%22en%22%2C%22allows_write_to_pm%22%3Atrue%2C%22photo_url%22%3A%22https%3A%5C%2F%5C%2Ft.me%5C%2Fi%5C%2Fuserpic%5C%2F320%5C%2FZz_aXp8Ip-umZPJhOWom2KTGBs3Xe8dwwgEqc-A5KG7J3Aq3CRmiv097xryPafqp.svg%22%7D&auth_date=1735561646&signature=cL2Y2F58ZhwSf47V4OhUAnaKQJlan_IFcYAec8EOUjK-f-jNtRVlgB6cIHu3IV62xCXsW6w3zoemmLSpRao3DA&hash=765a84035331eb5b2d379bb23e62ee72f8a12813e4156e2d3831a0a6996656da"
|
|
# # print(url_to_json(data=url)["user"]["username"])
|
|
#
|
|
# # tg_interactive_bot(
|
|
# # phone="972525788299",
|
|
# # device_model="P8Z68-V LX",
|
|
# # bot_name="BlumCryptoTradingBot",
|
|
# # message="/start",
|
|
# # btn=""
|
|
# # )
|
|
#
|
|
# json_data = {
|
|
# "phone": "8619260226414",
|
|
# "device_model": "USOPPBH",
|
|
# "bot_name": "BlumCryptoTradingBot",
|
|
# "data": [
|
|
# # {"message": "/start", "btn_lst": ["Settings", "Language", "English"]},
|
|
# # {"message": "/start", "btn_lst": ["BUY", "Token", "Blum hot", "FPIBANK", "slippage"]},
|
|
# # {"message": "3", "btn_lst": ["amount", ]},
|
|
# # {"message": "4", "btn_lst": ["BUY", ]},
|
|
# {"message": "/start", "btn_lst": ["Positions", "FPIBANK", "SELL"]},
|
|
# ],
|
|
# }
|
|
# url = "http://192.168.50.180:9696/bot/click"
|
|
# resp = requests.post(url, json=json_data)
|
|
# print(resp.text)
|
|
|
|
# json_data = {
|
|
# "phone": "8613641930695",
|
|
# "bot_name": "pengu_clash_bot",
|
|
# "datas": [
|
|
# {"send_message": ["/start"], "click_button": ["now"], },
|
|
# ]
|
|
# }
|
|
#
|
|
# res = requests.post("http://192.168.50.175:9000/api/send_click", json=json_data)
|
|
# print(res.json())
|