253 lines
10 KiB
Python
Executable File
253 lines
10 KiB
Python
Executable File
#!/usr/bin/env python3
|
|
|
|
import argparse
|
|
import base64
|
|
import json
|
|
import os
|
|
import re
|
|
import subprocess
|
|
import sys
|
|
import time
|
|
import requests
|
|
from hashlib import sha256
|
|
from Crypto.Hash import RIPEMD
|
|
|
|
BASE58_ALPHABET = "123456789ABCDEFGHJKLMNPQRSTUVWXYZabcdefghijkmnopqrstuvwxyz"
|
|
BASE58_MAP = {c: i for i, c in enumerate(BASE58_ALPHABET)}
|
|
DEFAULT_FEE = 0.01
|
|
BASE_URL = os.getenv("BASE_URL", "http://localhost:12391")
|
|
|
|
TRANSACTION_TYPES = {
|
|
# Payments
|
|
"payment": {"url": "payments/pay", "required": ["recipient", "amount"], "key_name": "senderPublicKey"},
|
|
|
|
# Groups
|
|
"set_group": {"url": "groups/setdefault", "required": ["defaultGroupId"], "key_name": "creatorPublicKey"},
|
|
"create_group": {"url": "groups/create", "required": ["groupName", "description", "isOpen", "approvalThreshold"], "defaults": {"minimumBlockDelay": 10, "maximumBlockDelay": 30}, "key_name": "creatorPublicKey"},
|
|
"update_group": {"url": "groups/update", "required": ["groupId", "newOwner", "newDescription", "newIsOpen", "newApprovalThreshold"], "key_name": "ownerPublicKey"},
|
|
"join_group": {"url": "groups/join", "required": ["groupId"], "key_name": "joinerPublicKey"},
|
|
"leave_group": {"url": "groups/leave", "required": ["groupId"], "key_name": "leaverPublicKey"},
|
|
"group_invite": {"url": "groups/invite", "required": ["groupId", "txGroupId", "invitee"], "key_name": "adminPublicKey"},
|
|
"group_kick": {"url": "groups/kick", "required": ["groupId", "member", "reason"], "key_name": "adminPublicKey"},
|
|
"add_group_admin": {"url": "groups/addadmin", "required": ["groupId", "txGroupId", "member"], "key_name": "ownerPublicKey"},
|
|
"remove_group_admin": {"url": "groups/removeadmin", "required": ["groupId", "txGroupId", "admin"], "key_name": "ownerPublicKey"},
|
|
"group_approval": {"url": "groups/approval", "required": ["pendingSignature", "approval"], "key_name": "adminPublicKey"},
|
|
|
|
# Assets
|
|
"issue_asset": {"url": "assets/issue", "required": ["assetName", "description", "quantity", "isDivisible"], "key_name": "issuerPublicKey"},
|
|
"update_asset": {"url": "assets/update", "required": ["assetId", "newOwner"], "key_name": "ownerPublicKey"},
|
|
"transfer_asset": {"url": "assets/transfer", "required": ["recipient", "amount", "assetId"], "key_name": "senderPublicKey"},
|
|
"create_order": {"url": "assets/order", "required": ["haveAssetId", "wantAssetId", "amount", "price"], "key_name": "creatorPublicKey"},
|
|
|
|
# Names
|
|
"register_name": {"url": "names/register", "required": ["name", "data"], "key_name": "registrantPublicKey"},
|
|
"update_name": {"url": "names/update", "required": ["name", "newName", "newData"], "key_name": "ownerPublicKey"},
|
|
|
|
# Reward Shares
|
|
"reward_share": {"url": "addresses/rewardshare", "required": ["recipient", "rewardSharePublicKey", "sharePercent"], "key_name": "minterPublicKey"},
|
|
|
|
# Arbitrary
|
|
"arbitrary": {"url": "arbitrary", "required": ["service", "dataType", "data"], "key_name": "senderPublicKey"},
|
|
|
|
# Chat
|
|
"chat": {"url": "chat", "required": ["data"], "optional": ["recipient", "isText", "isEncrypted"], "defaults": {"isText": "true"}, "key_name": "senderPublicKey", "pow_url": "chat/compute"},
|
|
|
|
# Misc
|
|
"publicize": {"url": "addresses/publicize", "required": [], "key_name": "senderPublicKey", "pow_url": "addresses/publicize/compute"},
|
|
|
|
# Automated Transactions (AT)
|
|
"deploy_at": {"url": "at", "required": ["name", "description", "aTType", "tags", "creationBytes", "amount"], "optional": ["assetId"], "defaults": {"assetId": 0}, "key_name": "creatorPublicKey"},
|
|
|
|
# Cross-chain trading
|
|
"create_trade": {"url": "crosschain/tradebot/create", "required": ["qortAmount", "fundingQortAmount", "foreignAmount", "receivingAddress"], "optional": ["tradeTimeout", "foreignBlockchain"], "defaults": {"tradeTimeout": 1440, "foreignBlockchain": "LITECOIN"}, "key_name": "creatorPublicKey"},
|
|
"trade_recipient": {"url": "crosschain/tradeoffer/recipient", "required": ["atAddress", "recipient"], "key_name": "creatorPublicKey", "remove": ["timestamp", "reference", "fee"]},
|
|
"trade_secret": {"url": "crosschain/tradeoffer/secret", "required": ["atAddress", "secret"], "key_name": "recipientPublicKey", "remove": ["timestamp", "reference", "fee"]},
|
|
|
|
# Signing only
|
|
"sign": {"url": "transactions/sign", "required": ["transactionBytes"]}
|
|
}
|
|
|
|
def get_last_reference(address):
|
|
try:
|
|
resp = json.loads(api_call(f"addresses/{address}", method='GET'))
|
|
return resp.get("lastReference") or api_call("utils/random?length=64", method='GET')
|
|
except requests.HTTPError as e:
|
|
if e.response.status_code == 404:
|
|
return api_call("utils/random?length=64", method='GET')
|
|
raise
|
|
|
|
|
|
def get_pub_key(privkey):
|
|
pubkey = api_call("utils/publickey", privkey)
|
|
if not pubkey or pubkey.strip().lower() == "none":
|
|
raise ValueError("❌ Invalid public key returned. Is the private key correct?")
|
|
return pubkey.strip()
|
|
|
|
|
|
def encode_base58(hex_str):
|
|
return api_call("utils/toBase58", hex_str)
|
|
|
|
def decode_base58(b58_str):
|
|
return api_call("utils/fromBase58", b58_str)
|
|
|
|
def get_group_owner_pubkey(group_id):
|
|
try:
|
|
group_info = json.loads(api_call(f"groups/{group_id}", method='GET'))
|
|
owner_address = group_info.get("owner")
|
|
if not owner_address:
|
|
raise Exception(f"Group {group_id} has no owner address.")
|
|
address_info = json.loads(api_call(f"addresses/{owner_address}", method='GET'))
|
|
return address_info.get("publicKey")
|
|
except requests.HTTPError as e:
|
|
print(f"Error fetching group owner publicKey for group ID {group_id}: {e}")
|
|
raise
|
|
|
|
def pubkey_to_address(pubkey):
|
|
return api_call(f"addresses/convert/{pubkey}", method='GET')
|
|
|
|
|
|
def api_call(endpoint, data=None, method='POST'):
|
|
url = f"{BASE_URL}/{endpoint}"
|
|
print("🌐 API CALL:", method, url)
|
|
if data:
|
|
print("📨 Payload:", json.dumps(data, indent=2) if isinstance(data, dict) else data)
|
|
|
|
headers = {}
|
|
resp = None
|
|
|
|
try:
|
|
if method == 'GET':
|
|
resp = requests.get(url)
|
|
else:
|
|
if isinstance(data, str):
|
|
headers['Content-Type'] = 'text/plain'
|
|
resp = requests.post(url, headers=headers, data=data)
|
|
else:
|
|
headers['Content-Type'] = 'application/json'
|
|
resp = requests.post(url, headers=headers, json=data)
|
|
|
|
resp.raise_for_status()
|
|
content_type = resp.headers.get('Content-Type', '')
|
|
if 'application/json' in content_type:
|
|
return json.dumps(resp.json())
|
|
return resp.text.strip('"')
|
|
|
|
except requests.HTTPError as e:
|
|
print(f"❌ API call failed: {e}")
|
|
try:
|
|
print("🧾 Response:", resp.json())
|
|
except:
|
|
print("🧾 Raw response:", resp.text)
|
|
raise
|
|
|
|
|
|
|
|
def build_raw(tx_type, priv_key, values):
|
|
tx_info = TRANSACTION_TYPES[tx_type]
|
|
pub_key = get_pub_key(priv_key)
|
|
address = pubkey_to_address(pub_key)
|
|
|
|
reference = get_last_reference(address) if tx_type != 'sign' else None
|
|
data = {
|
|
"timestamp": int(time.time() * 1000),
|
|
"fee": DEFAULT_FEE,
|
|
}
|
|
if reference:
|
|
data["reference"] = reference
|
|
|
|
data.update(values)
|
|
data.update(tx_info.get("defaults", {}))
|
|
|
|
# Always use the signer's pubkey for txGroupId-based approval-required transactions
|
|
if "txGroupId" in data and str(data["txGroupId"]) != "0" and "key_name" in tx_info:
|
|
data[tx_info["key_name"]] = pub_key
|
|
elif "key_name" in tx_info:
|
|
data[tx_info["key_name"]] = pub_key
|
|
|
|
if "remove" in tx_info:
|
|
for key in tx_info["remove"]:
|
|
data.pop(key, None)
|
|
|
|
if "optional" in tx_info:
|
|
for opt_key in tx_info["optional"]:
|
|
if opt_key not in data:
|
|
data[opt_key] = tx_info.get("defaults", {}).get(opt_key, None)
|
|
|
|
# Auto-convert numeric-looking strings to int
|
|
for key in list(data.keys()):
|
|
if isinstance(data[key], str) and data[key].isdigit():
|
|
data[key] = int(data[key])
|
|
|
|
print("📦 Final JSON Payload:")
|
|
print(json.dumps(data, indent=2))
|
|
|
|
raw_tx = api_call(tx_info["url"], data)
|
|
|
|
if "pow_url" in tx_info:
|
|
raw_tx = api_call(tx_info["pow_url"], raw_tx)
|
|
|
|
print("🧪 Parsed input values:")
|
|
print(json.dumps(values, indent=2))
|
|
|
|
return raw_tx
|
|
|
|
def sign_tx(privkey, raw_tx):
|
|
data = {
|
|
"privateKey": privkey,
|
|
"transactionBytes": raw_tx
|
|
}
|
|
return api_call("transactions/sign", data)
|
|
|
|
def process_tx(signed_tx):
|
|
return api_call("transactions/process", signed_tx)
|
|
|
|
def main():
|
|
parser = argparse.ArgumentParser(description="Qortal Transaction Tool")
|
|
parser.add_argument("tx_type", help="Transaction type")
|
|
parser.add_argument("privkey", help="Private key")
|
|
parser.add_argument("params", nargs="*", help="Transaction parameters and optional key=value pairs")
|
|
parser.add_argument("-s", "--sign", action="store_true")
|
|
parser.add_argument("-p", "--process", action="store_true")
|
|
parser.add_argument("-d", "--debug", action="store_true")
|
|
parser.add_argument("--dry-run", action="store_true", help="Only print the raw transaction JSON without signing or broadcasting")
|
|
args = parser.parse_args()
|
|
|
|
tx_type = args.tx_type.lower()
|
|
if tx_type not in TRANSACTION_TYPES:
|
|
print(f"Unknown transaction type: {tx_type}")
|
|
sys.exit(1)
|
|
|
|
required_keys = TRANSACTION_TYPES[tx_type].get("required", [])
|
|
values = {}
|
|
|
|
if len(args.params) < len(required_keys):
|
|
print(f"Missing required parameters for {tx_type}: {required_keys}")
|
|
sys.exit(1)
|
|
|
|
for i, key in enumerate(required_keys):
|
|
values[key] = args.params[i]
|
|
|
|
for param in args.params[len(required_keys):]:
|
|
if '=' in param:
|
|
k, v = param.split('=', 1)
|
|
values[k] = v
|
|
|
|
raw = build_raw(tx_type, args.privkey, values)
|
|
|
|
if args.dry_run:
|
|
print("[Dry Run] Raw transaction built successfully. Not signing or broadcasting.")
|
|
return
|
|
|
|
if args.sign:
|
|
signed = sign_tx(args.privkey, raw)
|
|
print("Signed:", signed)
|
|
if args.process:
|
|
result = process_tx(signed)
|
|
print("Processed:", result)
|
|
elif args.debug:
|
|
print("Raw:", raw)
|
|
|
|
if __name__ == "__main__":
|
|
main()
|
|
|