107 lines
3.4 KiB
Python
107 lines
3.4 KiB
Python
import requests
|
|
from dotenv import load_dotenv
|
|
import os
|
|
import time
|
|
import logging
|
|
import threading
|
|
from urllib import parse
|
|
|
|
from eth_account.messages import encode_typed_data
|
|
from eth_account import Account
|
|
from eth_account.datastructures import SignedMessage
|
|
|
|
load_dotenv()
|
|
|
|
USER: str = os.getenv(key="RABBY_WALLET") # ty:ignore[invalid-assignment]
|
|
SIGNER: str = os.getenv(key="ASTER_API_WALLET_ADDRESS") # ty:ignore[invalid-assignment]
|
|
PRIVATE_KEY: str = os.getenv(key="ASTER_API_PRIVATE_KEY") # ty:ignore[invalid-assignment]
|
|
|
|
_last_ms = 0
|
|
_i = 0
|
|
|
|
async def post_authenticated_url(req: dict) -> list | dict:
|
|
typed_data: dict = {
|
|
"types": {
|
|
"EIP712Domain": [
|
|
{"name": "name", "type": "string"},
|
|
{"name": "version", "type": "string"},
|
|
{"name": "chainId", "type": "uint256"},
|
|
{"name": "verifyingContract", "type": "address"}
|
|
],
|
|
"Message": [
|
|
{ "name": "msg", "type": "string" },
|
|
]
|
|
},
|
|
"primaryType": "Message",
|
|
"domain": {
|
|
"name": "AsterSignTransaction",
|
|
"version": "1",
|
|
"chainId": 1666,
|
|
"verifyingContract": "0x0000000000000000000000000000000000000000"
|
|
},
|
|
"message": {
|
|
"msg": "$msg"
|
|
}
|
|
}
|
|
|
|
headers: dict[str, str] = {
|
|
'Content-Type': 'application/x-www-form-urlencoded',
|
|
'User-Agent': 'PythonApp/1.0'
|
|
}
|
|
host: str = 'https://fapi.asterdex.com'
|
|
|
|
def get_nonce():
|
|
_nonce_lock = threading.Lock()
|
|
global _last_ms, _i
|
|
with _nonce_lock:
|
|
now_ms = int(time.time())
|
|
|
|
if now_ms == _last_ms:
|
|
_i += 1
|
|
else:
|
|
_last_ms = now_ms
|
|
_i = 0
|
|
|
|
return now_ms * 1_000_000 + _i
|
|
|
|
def sign_typed_data(data: dict, private_key: str):
|
|
"""Sign EIP-712 typed data using encode_typed_data."""
|
|
message = encode_typed_data(
|
|
domain_data=data["domain"],
|
|
message_types={"Message": data["types"]["Message"]},
|
|
message_data=data["message"],
|
|
)
|
|
return Account.sign_message(message, private_key=private_key)
|
|
|
|
async def send_by_url(req) -> list | dict: # ty:ignore[invalid-return-type]
|
|
my_dict = req['params'].copy()
|
|
url = host + req['url']
|
|
method = req['method']
|
|
|
|
my_dict['nonce'] = str(object=get_nonce())
|
|
my_dict['user'] = USER
|
|
my_dict['signer'] = SIGNER
|
|
|
|
param: str = parse.urlencode(query=my_dict)
|
|
|
|
typed_data['message']['msg'] = param
|
|
signed: SignedMessage = sign_typed_data(data=typed_data, private_key=PRIVATE_KEY)
|
|
|
|
full_url: str = url + '?' + param + '&signature=' + signed.signature.hex()
|
|
|
|
if method == 'GET':
|
|
res: requests.Response = requests.get(url=full_url, headers=headers)
|
|
# logging.warning(res.status_code, res.text)
|
|
return res.json()
|
|
elif method == 'POST':
|
|
res: requests.Response = requests.post(url=full_url, headers=headers)
|
|
return res.json()
|
|
elif method == 'PUT':
|
|
res: requests.Response = requests.put(url=full_url, headers=headers)
|
|
return res.json()
|
|
elif method == 'DELETE':
|
|
res: requests.Response = requests.delete(url=full_url, headers=headers)
|
|
return res.json()
|
|
|
|
return await send_by_url(req=req)
|