firewalletbrowser/account.py

770 lines
19 KiB
Python
Raw Normal View History

2024-01-26 03:51:52 +11:00
from datetime import datetime, timedelta
2023-12-28 13:34:48 +11:00
from handywrapper import api
import os
import dotenv
import requests
2023-12-28 16:04:45 +11:00
import re
import domainLookup
import json
2023-12-28 13:34:48 +11:00
dotenv.load_dotenv()
APIKEY = os.getenv("hsd_api")
2024-01-29 15:48:36 +11:00
ip = os.getenv("hsd_ip")
if ip is None:
ip = "localhost"
hsd = api.hsd(APIKEY,ip)
hsw = api.hsw(APIKEY,ip)
2024-01-29 15:42:54 +11:00
2023-12-28 13:34:48 +11:00
# Verify the connection
response = hsd.getInfo()
def check_account(cookie: str):
2023-12-28 16:04:45 +11:00
if cookie is None:
return False
2023-12-28 13:34:48 +11:00
# Check the account
if cookie.count(":") < 1:
return False
account = cookie.split(":")[0]
# Check if the account is valid
info = hsw.getAccountInfo(account, 'default')
if 'error' in info:
return False
return account
2024-01-26 03:51:52 +11:00
def check_password(cookie: str, password: str):
account = check_account(cookie)
if account == False:
return False
# Check if the password is valid
info = hsw.rpc_selectWallet(account)
if info['error'] is not None:
return False
info = hsw.rpc_walletPassphrase(password,10)
if info['error'] is not None:
return False
return True
def createWallet(account: str, password: str):
# Create the account
# Python wrapper doesn't support this yet
2024-01-29 15:48:36 +11:00
response = requests.put(f"http://x:{APIKEY}@{ip}:12039/wallet/{account}")
print(response)
print(response.json())
if response.status_code != 200:
return {
"error": {
"message": "Error creating account"
}
}
# Get seed
seed = hsw.getMasterHDKey(account)
seed = seed['mnemonic']['phrase']
# Encrypt the wallet (python wrapper doesn't support this yet)
2024-01-29 15:48:36 +11:00
response = requests.post(f"http://x:{APIKEY}@{ip}:12039/wallet/{account}/passphrase",
json={"passphrase": password})
print(response)
return {
"seed": seed,
"account": account,
"password": password
}
2024-02-04 16:55:28 +11:00
def importWallet(account: str, password: str,seed: str):
# Import the wallet
data = {
"passphrase": password,
"mnemonic": seed,
}
response = requests.put(f"http://x:{APIKEY}@{ip}:12039/wallet/{account}",json=data)
print(response)
print(response.json())
if response.status_code != 200:
return {
"error": {
"message": "Error creating account"
}
}
return {
"seed": seed,
"account": account,
"password": password
}
def listWallets():
# List the wallets
response = hsw.listWallets()
# Check if response is json or an array
if isinstance(response, list):
return response
return ['Wallet not connected']
2023-12-28 13:34:48 +11:00
def getBalance(account: str):
# Get the total balance
info = hsw.getBalance('default',account)
if 'error' in info:
return {'available': 0, 'total': 0}
total = info['confirmed']
available = total - info['lockedConfirmed']
2024-02-13 11:18:16 +11:00
locked = info['lockedConfirmed'] / 1000000
2023-12-28 13:34:48 +11:00
# Convert to HNS
total = total / 1000000
available = available / 1000000
domains = getDomains(account)
domainValue = 0
for domain in domains:
if domain['state'] == "CLOSED":
domainValue += domain['value']
total = total - (domainValue/1000000)
2023-12-28 13:34:48 +11:00
# Only keep 2 decimal places
total = round(total, 2)
available = round(available, 2)
2024-02-13 11:18:16 +11:00
return {'available': available, 'total': total, 'locked': locked}
2023-12-28 13:34:48 +11:00
2024-01-26 03:51:52 +11:00
def getBlockHeight():
# Get the block height
info = hsd.getInfo()
if 'error' in info:
return 0
return info['chain']['height']
2023-12-28 16:58:11 +11:00
def getAddress(account: str):
# Get the address
info = hsw.getAccountInfo(account, 'default')
if 'error' in info:
return ''
return info['receiveAddress']
2023-12-28 13:34:48 +11:00
def getPendingTX(account: str):
# Get the pending transactions
2023-12-28 18:31:44 +11:00
info = hsw.getWalletTxHistory(account)
2023-12-28 13:34:48 +11:00
if 'error' in info:
return 0
pending = 0
for tx in info:
if tx['confirmations'] < 1:
pending += 1
return pending
2024-02-13 11:18:16 +11:00
def getDomains(account,own=True):
if own:
response = requests.get(f"http://x:{APIKEY}@{ip}:12039/wallet/{account}/name?own=true")
else:
response = requests.get(f"http://x:{APIKEY}@{ip}:12039/wallet/{account}/name")
2023-12-28 13:34:48 +11:00
info = response.json()
return info
def getTransactions(account):
# Get the transactions
info = hsw.getWalletTxHistory(account)
if 'error' in info:
return []
2023-12-28 16:04:45 +11:00
return info
def check_address(address: str, allow_name: bool = True, return_address: bool = False):
# Check if the address is valid
if address.startswith('@'):
# Check if the address is a name
if not allow_name and not return_address:
return 'Invalid address'
elif not allow_name and return_address:
return False
return check_hip2(address[1:])
# Check if the address is a valid HNS address
2024-01-29 15:48:36 +11:00
response = requests.post(f"http://x:{APIKEY}@{ip}:12037",json={
2023-12-28 16:04:45 +11:00
"method": "validateaddress",
"params": [address]
}).json()
if response['error'] is not None:
if return_address:
return False
return 'Invalid address'
if response['result']['isvalid'] == True:
if return_address:
return address
return 'Valid address'
if return_address:
return False
return 'Invalid address'
def check_hip2(domain: str):
# Check if the domain is valid
domain = domain.lower()
if re.match(r'^[a-zA-Z0-9\-\.]{1,63}$', domain) is None:
return 'Invalid address'
address = domainLookup.hip2(domain)
2023-12-28 16:32:20 +11:00
if address.startswith("Hip2: "):
return address
2023-12-28 16:04:45 +11:00
if not check_address(address, False,True):
return 'Hip2: Lookup succeeded but address is invalid'
return address
def send(account,address,amount):
account_name = check_account(account)
password = ":".join(account.split(":")[1:])
2023-12-28 16:32:20 +11:00
response = hsw.rpc_selectWallet(account_name)
if response['error'] is not None:
2023-12-28 16:04:45 +11:00
return {
2024-01-26 03:51:52 +11:00
"error": {
"message": response['error']['message']
}
2023-12-28 16:04:45 +11:00
}
2023-12-28 16:32:20 +11:00
response = hsw.rpc_walletPassphrase(password,10)
# Unlock the account
2024-01-29 15:48:36 +11:00
# response = requests.post(f"http://x:{APIKEY}@{ip}:12039/wallet/{account_name}/unlock",
2023-12-28 16:32:20 +11:00
# json={"passphrase": password,"timeout": 10})
if response['error'] is not None:
2023-12-28 16:04:45 +11:00
return {
2024-01-26 03:51:52 +11:00
"error": {
"message": response['error']['message']
}
2023-12-28 16:04:45 +11:00
}
2023-12-28 16:32:20 +11:00
response = hsw.rpc_sendToAddress(address,amount)
if response['error'] is not None:
2023-12-28 16:04:45 +11:00
return {
2024-01-26 03:51:52 +11:00
"error": {
"message": response['error']['message']
}
2023-12-28 16:04:45 +11:00
}
return {
"tx": response['result']
2023-12-28 18:04:38 +11:00
}
def getDomain(domain: str):
# Get the domain
response = hsd.rpc_getNameInfo(domain)
if response['error'] is not None:
return {
2024-01-26 03:51:52 +11:00
"error": {
"message": response['error']['message']
}
2023-12-28 18:04:38 +11:00
}
2023-12-28 18:31:44 +11:00
return response['result']
def renewDomain(account,domain):
account_name = check_account(account)
password = ":".join(account.split(":")[1:])
if account_name == False:
return {
2024-01-26 03:51:52 +11:00
"error": {
"message": "Invalid account"
}
2023-12-28 18:31:44 +11:00
}
response = hsw.sendRENEW(account_name,password,domain)
return response
def getDNS(domain: str):
# Get the DNS
response = hsd.rpc_getNameResource(domain)
if response['error'] is not None:
return {
"error": response['error']['message']
}
return response['result']['records']
2023-12-29 11:42:20 +11:00
2024-01-25 23:15:59 +11:00
def setDNS(account,domain,records):
account_name = check_account(account)
password = ":".join(account.split(":")[1:])
if account_name == False:
return {
2024-01-26 03:51:52 +11:00
"error": {
"message": "Invalid account"
}
2024-01-25 23:15:59 +11:00
}
records = json.loads(records)
newRecords = []
TXTRecords = []
for record in records:
if record['type'] == 'TXT':
if 'txt' not in record:
TXTRecords.append(record['value'])
else:
for txt in record['txt']:
TXTRecords.append(txt)
2024-01-25 23:15:59 +11:00
elif record['type'] == 'NS':
newRecords.append({
'type': 'NS',
'ns': record['value']
})
elif record['type'] in ['GLUE4','GLUE6',"SYNTH4","SYNTH6"]:
newRecords.append({
'type': record['type'],
'ns': str(record['value']).split(' ')[0],
'address': str(record['value']).split(' ')[1]
})
else:
newRecords.append(record)
if len(TXTRecords) > 0:
newRecords.append({
'type': 'TXT',
'txt': TXTRecords
2024-02-13 13:41:56 +11:00
})
2024-01-25 23:15:59 +11:00
data = '{"records":'+str(newRecords).replace("'","\"")+'}'
response = hsw.sendUPDATE(account_name,password,domain,data)
return response
2023-12-29 11:42:20 +11:00
def getNodeSync():
response = hsd.getInfo()
if 'error' in response:
return 0
2023-12-29 13:57:20 +11:00
sync = response['chain']['progress']*100
sync = round(sync, 2)
return sync
2023-12-29 12:42:07 +11:00
2024-02-13 11:18:16 +11:00
def getBids(account, domain="NONE"):
if domain == "NONE":
return hsw.getWalletBids(account)
2023-12-29 12:42:07 +11:00
response = hsw.getWalletBidsByName(domain,account)
return response
2023-12-29 13:57:20 +11:00
def getReveals(account,domain):
return hsw.getWalletRevealsByName(domain,account)
def getRevealTX(reveal):
prevout = reveal['prevout']
hash = prevout['hash']
index = prevout['index']
tx = hsd.getTxByHash(hash)
2023-12-29 14:24:43 +11:00
return tx['inputs'][index]['prevout']['hash']
2023-12-29 13:57:20 +11:00
2023-12-29 14:24:43 +11:00
def revealAuction(account,domain):
account_name = check_account(account)
password = ":".join(account.split(":")[1:])
if account_name == False:
return {
2024-01-26 03:51:52 +11:00
"error": {
"message": "Invalid account"
}
2023-12-29 14:24:43 +11:00
}
try:
response = hsw.sendREVEAL(account_name,password,domain)
return response
except Exception as e:
return {
"error": str(e)
}
2023-12-29 13:57:20 +11:00
2024-02-13 11:18:16 +11:00
def revealAll(account):
account_name = check_account(account)
password = ":".join(account.split(":")[1:])
if account_name == False:
return {
"error": {
"message": "Invalid account"
}
}
try:
# Try to select and login to the wallet
response = hsw.rpc_selectWallet(account_name)
if response['error'] is not None:
return
response = hsw.rpc_walletPassphrase(password,10)
if response['error'] is not None:
return
# Try to send the batch of all renew, reveal and redeem actions
return requests.post(f"http://x:{APIKEY}@{ip}:12039",json={"method": "sendbatch","params": [[["REVEAL"]]]}).json()
except Exception as e:
return {
"error": {
"message": str(e)
}
}
2023-12-29 12:42:07 +11:00
def rescan_auction(account,domain):
# Get height of the start of the auction
response = hsw.rpc_selectWallet(account)
response = hsd.rpc_getNameInfo(domain)
if 'result' not in response:
return {
"error": "Invalid domain"
}
if 'bidPeriodStart' not in response['result']['info']['stats']:
return {
"error": "Not in auction"
}
2023-12-29 13:57:20 +11:00
height = response['result']['info']['height']-1
2023-12-29 12:42:07 +11:00
response = hsw.rpc_importName(domain,height)
return response
def bid(account,domain,bid,blind):
account_name = check_account(account)
password = ":".join(account.split(":")[1:])
if account_name == False:
return {
2024-01-26 03:51:52 +11:00
"error": {
"message": "Invalid account"
}
2023-12-29 12:42:07 +11:00
}
bid = int(bid)*1000000
lockup = int(blind)*1000000 + bid
try:
response = hsw.sendBID(account_name,password,domain,bid,lockup)
return response
except Exception as e:
return {
2024-01-26 03:51:52 +11:00
"error": {
"message": str(e)
}
2023-12-29 12:42:07 +11:00
}
2023-12-29 12:50:15 +11:00
def openAuction(account,domain):
account_name = check_account(account)
password = ":".join(account.split(":")[1:])
if account_name == False:
return {
2024-01-26 03:51:52 +11:00
"error": {
"message": "Invalid account"
}
2023-12-29 12:50:15 +11:00
}
try:
response = hsw.sendOPEN(account_name,password,domain)
return response
except Exception as e:
return {
2024-01-26 03:51:52 +11:00
"error": {
"message": str(e)
}
}
def transfer(account,domain,address):
account_name = check_account(account)
password = ":".join(account.split(":")[1:])
if account_name == False:
return {
"error": {
"message": "Invalid account"
}
}
try:
response = hsw.sendTRANSFER(account_name,password,domain,address)
return response
except Exception as e:
return {
"error": {
"message": str(e)
}
}
def finalize(account,domain):
account_name = check_account(account)
password = ":".join(account.split(":")[1:])
if account_name == False:
return {
"error": {
"message": "Invalid account"
}
}
try:
2024-02-04 16:55:28 +11:00
response = hsw.rpc_selectWallet(account_name)
if response['error'] is not None:
return {
"error": {
"message": response['error']['message']
}
}
response = hsw.rpc_walletPassphrase(password,10)
if response['error'] is not None:
return {
"error": {
"message": response['error']['message']
}
}
response = hsw.rpc_sendFINALIZE(domain)
2024-01-26 03:51:52 +11:00
return response
except Exception as e:
return {
"error": {
"message": str(e)
}
}
def cancelTransfer(account,domain):
account_name = check_account(account)
password = ":".join(account.split(":")[1:])
if account_name == False:
return {
"error": {
"message": "Invalid account"
}
}
try:
response = hsw.rpc_selectWallet(account_name)
if response['error'] is not None:
return {
"error": {
"message": response['error']['message']
}
}
response = hsw.rpc_walletPassphrase(password,10)
if response['error'] is not None:
return {
"error": {
"message": response['error']['message']
}
}
response = hsw.rpc_sendCANCEL(domain)
return response
except Exception as e:
return {
"error": {
"message": str(e)
}
}
def revoke(account,domain):
account_name = check_account(account)
password = ":".join(account.split(":")[1:])
if account_name == False:
return {
"error": {
"message": "Invalid account"
}
}
try:
response = hsw.rpc_selectWallet(account_name)
if response['error'] is not None:
return {
"error": {
"message": response['error']['message']
}
}
response = hsw.rpc_walletPassphrase(password,10)
if response['error'] is not None:
return {
"error": {
"message": response['error']['message']
}
}
response = hsw.rpc_sendREVOKE(domain)
return response
2024-01-26 03:51:52 +11:00
except Exception as e:
return {
"error": {
"message": str(e)
}
}
#region settingsAPIs
def rescan():
try:
response = hsw.walletRescan(0)
return response
except Exception as e:
return {
"error": {
"message": str(e)
}
}
def resendTXs():
try:
response = hsw.walletResend()
return response
except Exception as e:
return {
"error": {
"message": str(e)
}
}
def zapTXs(account):
age = 60 * 20 # 20 minutes
account_name = check_account(account)
if account_name == False:
return {
"error": {
"message": "Invalid account"
}
}
try:
2024-01-29 15:48:36 +11:00
response = requests.post(f"http://x:{APIKEY}@{ip}:12039/wallet/{account_name}/zap",
json={"age": age,
"account": "default"
})
return response
except Exception as e:
return {
"error": {
"message": str(e)
}
}
def getxPub(account):
account_name = check_account(account)
if account_name == False:
return {
"error": {
"message": "Invalid account"
}
}
try:
print(account_name)
response = hsw.getAccountInfo(account_name,"default")
if 'error' in response:
return {
"error": {
"message": response['error']['message']
}
}
return response['accountKey']
return response
except Exception as e:
return {
"error": {
"message": str(e)
}
}
def signMessage(account,domain,message):
account_name = check_account(account)
password = ":".join(account.split(":")[1:])
if account_name == False:
return {
"error": {
"message": "Invalid account"
}
}
try:
response = hsw.rpc_selectWallet(account_name)
if response['error'] is not None:
return {
"error": {
"message": response['error']['message']
}
}
response = hsw.rpc_walletPassphrase(password,10)
if response['error'] is not None:
return {
"error": {
"message": response['error']['message']
}
}
response = hsw.rpc_signMessageWithName(domain,message)
return response
except Exception as e:
return {
"error": {
"message": str(e)
}
}
#endregion
2024-01-26 03:51:52 +11:00
2024-02-05 22:40:13 +11:00
def generateReport(account,format="{name},{expiry},{value},{maxBid}"):
2024-01-26 03:51:52 +11:00
domains = getDomains(account)
lines = [format.replace("{","").replace("}","")]
for domain in domains:
line = format.replace("{name}",domain['name'])
expiry = "N/A"
expiryBlock = "N/A"
if 'daysUntilExpire' in domain['stats']:
days = domain['stats']['daysUntilExpire']
# Convert to dateTime
expiry = datetime.now() + timedelta(days=days)
expiry = expiry.strftime("%d/%m/%Y %H:%M:%S")
expiryBlock = str(domain['stats']['renewalPeriodEnd'])
line = line.replace("{expiry}",expiry)
line = line.replace("{state}",domain['state'])
line = line.replace("{expiryBlock}",expiryBlock)
line = line.replace("{value}",str(domain['value']/1000000))
line = line.replace("{maxBid}",str(domain['highest']/1000000))
line = line.replace("{openHeight}",str(domain['height']))
lines.append(line)
2024-02-05 22:40:13 +11:00
return lines
def convertHNS(value: int):
return value/1000000