Files
kisstron/kisstron.py
T
2024-02-15 17:46:13 +02:00

529 lines
21 KiB
Python
Executable File

#!/usr/bin/env python3
# kisstron: a simple to use TRON CLI wallet based on tronpy
#
# Supports:
# * BIP39 mnemonic wallet generation/restoration
# * converting private keys to BIP39 mnemonics and vice versa
# * basic TRX parameters display (balance, energy, bandwidth)
# * basic TRX operations (transfer, freeze for energy, unfreeze)
# * basic TRC20 token operations (balance display, transfer) for select tokens
# (contract addresses are hardcoded, see KT_TRC20_TOKENS dictionary)
# * ability to select testnets (nile, shasta) if necessary
# * ability to connect to custom mainnet nodes if necessary
# * switching to TronGrid API by providing your own API key
# * offline mode with -o switch and bro[adcast] operation
#
# As of now, the main goal is to be able to operate on TRX and stablecoins.
# For mnemonic phrase generation, only English is currently supported.
#
# See README for more information, run with -h flag for usage instructions
#
# Created by Luxferre in 2024, released into public domain
import sys, json
from tronpy import Tron
from tronpy.tron import Transaction
from tronpy.keys import PrivateKey
from tronpy.providers import HTTPProvider
from tronpy.exceptions import (BadAddress, AddressNotFound,
BadKey, ValidationError, TransactionError)
from json.decoder import JSONDecodeError
import mnemonic, eth_utils
# some global parameters (hardcoded)
KT_TIMEOUT = 20.0 # network operation timeout
KT_FEE_LIMIT = 10000000 # transaction fee limit for mainnet (in Sun)
KT_TEST_FEE_LIMIT = 1000000000 # transaction fee limit for testnets (in Sun)
TRX_SCALE = 1000000.0 # TRX scale (million Suns in 1 TRX)
TG_API_KEY = None # TronGrid API Key (unused by default)
NODE_ADDR = 'http://18.196.99.16:8090' # public Tron node (Germany)
# TRC20 token definitions (also hardcoded)
# Make sure the mainnet tokens match the contract addresses
# specified on these pages:
# USDT: https://tron.network/usdt
# USDC: https://tron.network/usdc
KT_TRC20_TOKENS = {
'mainnet': {
'usdt': {
'contract': 'TR7NHqjeKQxGTCi8q8ZY4pL8otSzgjLj6t',
'ticker': 'USDT',
'show_balance': True
},
'usdc': {
'contract': 'TEkxiTehnzSmSe2XqrBj4w32RUN966rdz8',
'ticker': 'USDC',
'show_balance': True
}
},
'nile': {
'usdt': {
'contract': 'TXLAQ63Xg1NAzckPwKHvzw7CSEmLMEqcdj',
'ticker': 'USDT',
'show_balance': True
},
'usdc': {
'contract': 'TEMVynQpntMqkPxP6wXTW2K7e4sM3cRmWz',
'ticker': 'USDC',
'show_balance': True
}
},
'shasta': {
'usdt': {
'contract': 'TG3XXyExBkPp9nzdajDZsozEu4BkaSJozs',
'ticker': 'USDT',
'show_balance': True
},
'usdc': {
'contract': 'TSdZwNqpHofzP6BsBKGQUWdBeJphLmF6id',
'ticker': 'USDC',
'show_balance': True
}
}
}
# helper functions
# print operation result according to the below convention
def presult(resdict: dict):
if 'success' not in resdict:
resdict['success'] = False
if 'message' not in resdict:
resdict['message'] = ''
if resdict['success'] == True:
print ('Success:\n%s\n' % resdict['message'])
else:
print ('Error:\n%s\n' % resdict['message'])
# confirm transactions
def trconfirm(promptstr: str):
choice = input(promptstr).lower()
if choice != 'yes':
print('Transaction not confirmed, aborting!')
sys.exit(0)
# instantiate tronpy client
def tronclient(network='mainnet'):
network = network.lower() # lowercase network name
if network == 'mainnet': # use a public node or TronGrid API for mainnet
if TG_API_KEY is not None: # TG_API_KEY must be set
client = Tron(network=network, conf={'fee_limit': KT_FEE_LIMIT},
provider=HTTPProvider(api_key=TG_API_KEY, timeout=KT_TIMEOUT))
else: # NODE_ADDR must be set
client = Tron(network=network, conf={'fee_limit': KT_FEE_LIMIT},
provider=HTTPProvider(NODE_ADDR, timeout=KT_TIMEOUT))
else: # any default API is fine for a testnet
client = Tron(network=network,
conf={'fee_limit': KT_TEST_FEE_LIMIT, 'timeout': KT_TIMEOUT})
return client
# get TRC20 token balance of an address (in float)
def trc20balance(client: Tron, contractaddr: str, targetaddr: str):
cntr = client.get_contract(contractaddr)
precision = cntr.functions.decimals()
return cntr.functions['balanceOf'](targetaddr) / (10 ** precision)
# operation definitions
# all operations return a dictionary with the following fields:
# success: True/False
# message: string
# (optional) result: dictionary
# create a new wallet
# input: network, passphrase, word count (12, 15, 18, 21, 24), derivation path
# output: private key, address, mnemonic
def newwallet(network: str, passphrase: str, wordcount: int, dpath: str):
res = {'success': False, 'message': '', 'result': {}}
if wordcount not in [12, 15, 18, 21, 24]:
res['message'] = 'Invalid word count for mnemonic generation'
return res
client = tronclient(network)
wallet, mnemo = client.generate_address_with_mnemonic(passphrase = passphrase,
num_words = wordcount, language = 'english', account_path = dpath)
wallet['mnemonic'] = mnemo
res['success'] = True
res['result'] = wallet
res['message'] = 'Wallet created, save this info in a secure place!\n\n'
res['message'] += 'Address: %s\n' % wallet['base58check_address']
res['message'] += 'Address (hex): %s\n' % wallet['hex_address']
res['message'] += 'Private key (hex): %s\n' % wallet['private_key']
res['message'] += 'Public key (hex): %s\n' % wallet['public_key']
res['message'] += 'Mnemonic: %s' % wallet['mnemonic']
return res
# import a wallet
# input: network, mnemonic, passphrase, derivation path
# output: private key, address
def importwallet(network: str, mnemo: str, passphrase: str, dpath: str):
res = {'success': False, 'message': '', 'result': {}}
client = tronclient(network)
try:
wallet = client.generate_address_from_mnemonic(mnemonic = mnemo,
passphrase = passphrase, account_path = dpath)
except (mnemonic.mnemonic.ConfigurationError,
eth_utils.exceptions.ValidationError):
res['message'] = 'Invalid mnemonic phrase!'
return res
wallet['mnemonic'] = mnemo
res['success'] = True
res['result'] = wallet
res['message'] = 'Wallet imported, save this info in a secure place!\n\n'
res['message'] += 'Address: %s\n' % wallet['base58check_address']
res['message'] += 'Address (hex): %s\n' % wallet['hex_address']
res['message'] += 'Private key (hex): %s\n' % wallet['private_key']
res['message'] += 'Public key (hex): %s\n' % wallet['public_key']
res['message'] += 'Mnemonic: %s' % wallet['mnemonic']
return res
# get account information by Tron address
# input: network, address
# output: info(dictionary)
def infobyaddr(network: str, address: str):
res = {'success': False, 'message': '', 'result': {}}
client = tronclient(network)
try:
rawinfo = client.get_account(address)
except BadAddress:
res['message'] = 'Invalid address! Set the correct one with -addr flag'
return res
except AddressNotFound:
res['message'] = 'Address not found on the blockchain! Set the correct one with -addr flag'
return res
# get bandwidth and energy information
rawresinfo = client.get_account_resource(address)
# some values used in bandwidth calculation
freenetlimit = 0
freenetused = 0
netlimit = 0
netused = 0
if 'freeNetLimit' in rawresinfo:
freenetlimit = rawresinfo['freeNetLimit']
if 'freeNetUsed' in rawresinfo:
freenetlimit = rawresinfo['freeNetUsed']
if 'NetLimit' in rawresinfo:
netlimit = rawresinfo['NetLimit']
if 'NetUsed' in rawresinfo:
netused = rawresinfo['NetUsed']
bandwidth = freenetlimit - freenetused + netlimit - netused
# some values used in energy calculation
energylimit = 0
energyused = 0
if 'EnergyLimit' in rawresinfo:
energylimit = rawresinfo['EnergyLimit']
if 'EnergyUsed' in rawresinfo:
energyused = rawresinfo['EnergyUsed']
# info dictionary received, parse it
info = {
'address': rawinfo['address'], # wallet address
'trxbalance': rawinfo['balance'], # TRX wallet balance (in Sun)
'frozen': [], # hold frozen TRX info here
'trc20balance': {}, # hold TRC20 token balances here
'bandwidth': bandwidth, # bandwidth
'energy': energylimit - energyused # energy
}
# iterate over TRX freeze records
for el in rawinfo['frozenV2']:
if 'type' in el and 'amount' in el and el['amount'] > 0:
info['frozen'].append({'type': el['type'], 'amount': el['amount']})
# iterate over hardcoded TRC20 tokens
for tkn in KT_TRC20_TOKENS[network]:
tknobj = KT_TRC20_TOKENS[network][tkn]
if tknobj['show_balance'] == True:
info['trc20balance'][tknobj['ticker']] = trc20balance(client,
tknobj['contract'], address)
res['success'] = True
res['result'] = info
res['message'] = 'Address: %s\nBalance:\n' % info['address']
res['message'] += '%f TRX\n' % (info['trxbalance'] / TRX_SCALE)
for ticker in info['trc20balance']:
res['message'] += '%f %s\n' % (info['trc20balance'][ticker], ticker)
res['message'] += 'Frozen assets:\n'
for el in info['frozen']:
res['message'] += '%s\t%f TRX\n' % (el['type'], el['amount'] / TRX_SCALE)
res['message'] += 'Remaining bandwidth: %s\n' % str(info['bandwidth'])
res['message'] += 'Remaining energy: %s' % str(info['energy'])
return res
# get own account information by private key
# input: network, private key
# output: info(dictionary)
def infobypk(network: str, pk: str):
res = {'success': False, 'message': '', 'result': {}}
try:
rpk = PrivateKey(bytes.fromhex(pk))
except (BadKey, ValueError):
res['message'] = 'Invalid private key!'
return res
client = tronclient(network)
addr = client.generate_address(rpk)['base58check_address']
return infobyaddr(network, addr)
# send TRX transaction (and wait for its completion)
# input: network, private key, target address, amount (in TRX)
# output: transaction receipt object
def sendtrx(network: str, pk: str, toaddr: str, amt: float,
offline: bool = False):
res = {'success': False, 'message': '', 'result': {}}
if amt < 1:
res['message'] = 'Nothing to transfer!'
return res
try:
rpk = PrivateKey(bytes.fromhex(pk))
except (BadKey, ValueError):
res['message'] = 'Invalid private key!'
return res
client = tronclient(network)
fromaddr = client.generate_address(rpk)['base58check_address']
# build and sign the transaction
txn = client.trx.transfer(fromaddr, toaddr,
int(amt * TRX_SCALE)).build().sign(rpk)
if offline == True: # generate transaction JSON
res['success'] = True
res['result'] = txn
res['message'] = json.dumps(txn.to_json(), separators=(',', ':'))
else: # send the transaction
try:
receipt = txn.broadcast().wait()
except ValidationError as e:
res['message'] = str(e)
return res
res['success'] = True
res['message'] = 'Sent %f TRX to %s\nTransaction ID %s' % (amt,
toaddr, receipt['id'])
res['result'] = receipt
return res
# send TRC20 token transaction (and wait for its completion)
# input: network, private key, token ticker, target address, amount (in tokens)
# output: transaction receipt object
def sendtrc20(network: str, pk: str, ticker: str, toaddr: str,
amt: float, offline: bool = False):
res = {'success': False, 'message': '', 'result': {}}
ticker = ticker.lower()
if ticker not in KT_TRC20_TOKENS[network]:
res['message'] = 'Token not supported by kisstron for %s!' % network
return res
if amt < 1:
res['message'] = 'Nothing to transfer!'
return res
try:
rpk = PrivateKey(bytes.fromhex(pk))
except (BadKey, ValueError):
res['message'] = 'Invalid private key!'
return res
client = tronclient(network)
fromaddr = client.generate_address(rpk)['base58check_address']
# instantiate the contract
contractaddr = KT_TRC20_TOKENS[network][ticker]['contract']
cntr = client.get_contract(contractaddr)
precision = cntr.functions.decimals()
scaler = 10 ** precision
# build and sign the transaction
txn = cntr.functions.transfer(toaddr,
int(amt * scaler)).with_owner(fromaddr).build().sign(rpk)
if offline == True: # generate transaction JSON
res['success'] = True
res['result'] = txn
res['message'] = json.dumps(txn.to_json(), separators=(',', ':'))
else: # send the transaction
try:
receipt = txn.broadcast().wait()
except ValidationError as e:
res['message'] = str(e)
return res
res['result'] = receipt
if receipt['receipt']['result'] == 'OUT_OF_ENERGY':
res['message'] = 'Not enough energy for transaction!'
return res
res['success'] = True
res['message'] = 'Sent %f %s to %s\nTransaction ID %s' % (amt,
KT_TRC20_TOKENS[network][ticker]['ticker'],
toaddr, receipt['id'])
return res
# freeze TRX for energy (and wait for its completion)
# input: network, private key, amount
# output: transaction receipt object
def freezetrx(network: str, pk: str, amt: float, offline: bool = False):
res = {'success': False, 'message': '', 'result': {}}
if amt < 1:
res['message'] = 'Nothing to freeze!'
return res
try:
rpk = PrivateKey(bytes.fromhex(pk))
except (BadKey, ValueError):
res['message'] = 'Invalid private key!'
return res
client = tronclient(network)
myaddr = client.generate_address(rpk)['base58check_address']
# build and sign the transaction
txn = client.trx.freeze_balance(myaddr, int(amt * TRX_SCALE),
"ENERGY").build().sign(rpk)
if offline == True: # generate transaction JSON
res['success'] = True
res['result'] = txn
res['message'] = json.dumps(txn.to_json(), separators=(',', ':'))
else: # send the transaction
try:
receipt = txn.broadcast().wait()
except ValidationError as e:
res['message'] = str(e)
return res
res['success'] = True
res['message'] = 'Frozen %f TRX at %s\nTransaction ID %s' % (amt,
myaddr, receipt['id'])
res['result'] = receipt
return res
# unfreeze TRX for energy (and wait for its completion)
# input: network, private key, amount
# output: transaction receipt object
def unfreezetrx(network: str, pk: str, amt: float, offline: bool = False):
res = {'success': False, 'message': '', 'result': {}}
if amt < 1:
res['message'] = 'Nothing to unfreeze!'
return res
try:
rpk = PrivateKey(bytes.fromhex(pk))
except (BadKey, ValueError):
res['message'] = 'Invalid private key!'
return res
client = tronclient(network)
myaddr = client.generate_address(rpk)['base58check_address']
# build and sign the transaction
txn = client.trx.unfreeze_balance(owner = myaddr,
unfreeze_balance = int(amt * TRX_SCALE),
resource = "ENERGY").build().sign(rpk)
if offline == True: # generate transaction JSON
res['success'] = True
res['result'] = txn
res['message'] = json.dumps(txn.to_json(), separators=(',', ':'))
else: # send the transaction
try:
receipt = txn.broadcast().wait()
except ValidationError as e:
res['message'] = str(e)
return res
res['success'] = True
res['message'] = 'Unfrozen %f TRX at %s\nTransaction ID %s' % (amt,
myaddr, receipt['id'])
res['result'] = receipt
return res
# Broadcast transaction JSON saved in kisstron offline mode
# input: network, transaction JSON string
# output: transaction receipt object
def broadcastjson(network, tjson):
res = {'success': False, 'message': '', 'result': {}}
# import the transaction from JSON
try:
txn = Transaction.from_json(json.loads(tjson))
except (TypeError, ValidationError, TransactionError,
JSONDecodeError, KeyError):
res['message'] = 'Malformed transaction JSON!'
return res
# instantiate a client for the transaction
txn._client = tronclient(network)
# send the transaction
try:
receipt = txn.broadcast().wait()
except ValidationError as e:
res['message'] = str(e)
return res
except TransactionError as e:
res['message'] = str(e)
return res
res['result'] = receipt
if ('receipt' in receipt and 'result' in receipt['receipt']
and receipt['receipt']['result'] == 'OUT_OF_ENERGY'):
res['message'] = 'Not enough energy for transaction!'
return res
res['success'] = True
res['message'] = 'Transaction sent\nTransaction ID %s' % receipt['id']
return res
if __name__ == '__main__': # main app start
from argparse import ArgumentParser
parser = ArgumentParser(description='kisstron: a simple CLI wallet for Tron blockchain', epilog='(c) Luxferre 2024 --- No rights reserved <https://unlicense.org>')
parser.add_argument('op', help='Operation: new, import, info, send, fre[eze], unf[reeze], bro[adcast]')
parser.add_argument('coin', help='Coin to operate on: trx (default), usdt, usdc', default='trx')
parser.add_argument('-net', '--network', default='mainnet', help='(op: all) Network: mainnet (default), nile, shasta')
parser.add_argument('-node', '--tron-node', default=NODE_ADDR, help='(op: all) Custom Tron node HTTP address (default %s)' % NODE_ADDR)
parser.add_argument('-tgkey', '--trongrid-api-key', default=None, help='(op: all) TronGrid API key (default None)')
parser.add_argument('-addr', '--address', default=None, help='(op: info, send) Tron address to send to or display information about (-pk overrides this to display own info)')
parser.add_argument('-amt', '--amount', type=float, default=0, help='(op: send, fre, unf) Amount to send/freeze/unfreeze')
parser.add_argument('-pk', '--private-key', default=None, help='(op: info, send, fre, unf) Private key for transactions or own information, in hex')
parser.add_argument('-o', '--offline', action='store_true', help='(op: send, fre, unf) Build and sign the transaction in offline mode and output its JSON')
parser.add_argument('-pp', '--passphrase', default='', help='(op: new, import) Mnemonic encryption passphrase (default empty)')
parser.add_argument('-wc', '--words-count', type=int, default=12, help='(op: new) Word count in the mnemonic: 12 (default), 15, 18, 21, 24')
parser.add_argument('-dp', '--derivation-path', default="m/44'/195'/0'/0/0", help="(op: new, import) BIP-39 derivation path for mnemonic (default is m/44'/195'/0'/0/0)")
parser.add_argument('-nc', '--no-confirm', action='store_true', help='(op: send, fre, unf) Do not ask to type "yes" for transaction confirmation (dangerous)')
args = parser.parse_args()
# fix all relevant parameters (and lowercase where applicable)
op = args.op.lower() # operation
coin = args.coin.lower() # coin/token
tronnet = args.network.lower() # Tron network (mainnet, nile, shasta)
targetaddr = args.address # address parameter
amount = float(args.amount) # amount to send/freeze/unfreeze
pk = args.private_key # private key (in hex)
passphrase = args.passphrase # mnemonic encryption passphrase
wcount = int(args.words_count) # mnemonic word count (for new operation)
dpath = args.derivation_path # mnemonic derivation path (for new/import)
trans_confirm = True # ask for transaction confirmation ("yes")
NODE_ADDR = args.tron_node # custom Tron node HTTP address
TG_API_KEY = args.trongrid_api_key # optional TronGrid API key
offline_mode = False # offline mode
if args.offline == True:
offline_mode = True
if args.no_confirm == True:
trans_confirm = False
# check the constraints for every action and call appropriate functions
opres = {} # operation result placeholder
if op == 'new': # new wallet
opres = newwallet(tronnet, passphrase, wcount, dpath)
elif op == 'import': # import wallet
mnemo = input('Your mnemonic phrase: ')
opres = importwallet(tronnet, mnemo, passphrase, dpath)
elif op == 'info': # info on own or foreign address
if offline_mode == True:
print('Info operation is not available in offline mode!')
sys.exit(1)
if pk == None: # public info, address must be set
opres = infobyaddr(tronnet, targetaddr)
else: # own info, private key must be set
opres = infobypk(tronnet, pk)
elif op == 'send': # send transaction
if trans_confirm == True:
trconfirm('Enter "yes" to confirm sending funds: ')
if coin == 'trx': # send TRX
opres = sendtrx(tronnet, pk, targetaddr, amount, offline_mode)
else: # send one of the predefined TRC20 tokens
opres = sendtrc20(tronnet, pk, coin, targetaddr, amount, offline_mode)
elif op == 'fre' or op == 'freeze': # freeze TRX for energy
if trans_confirm == True:
trconfirm('Enter "yes" to confirm freezing funds: ')
opres = freezetrx(tronnet, pk, amount, offline_mode)
elif op == 'unf' or op == 'unfreeze': # unfreeze TRX
if trans_confirm == True:
trconfirm('Enter "yes" to confirm unfreezing funds: ')
opres = unfreezetrx(tronnet, pk, amount, offline_mode)
elif op == 'bro' or op == 'broadcast': # broadcast signed JSON transaction
if offline_mode == True:
print('Broadcast operation is not available in offline mode!')
sys.exit(1)
transjson = input('Input transaction JSON: ')
if trans_confirm == True:
trconfirm('Enter "yes" to confirm sending the transaction: ')
opres = broadcastjson(tronnet, transjson)
else: # unknown operation
print('Unknown operation!')
print('Run the script with -h parameter to see available options')
sys.exit(1)
presult(opres) # print out the result