qortal-scripts/tx.py

#!/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", "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):
    return api_call("utils/publickey", privkey)

def encode_base58(hex_str):
    num = int(hex_str, 16)
    encode = ""
    while num > 0:
        num, rem = divmod(num, 58)
        encode = BASE58_ALPHABET[rem] + encode
    return encode

def decode_base58(b58_str):
    num = 0
    for char in b58_str:
        num *= 58
        num += BASE58_MAP[char]
    hex_str = hex(num)[2:]
    return hex_str.zfill(66)
    
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):
    pubkey_bytes = bytes.fromhex(decode_base58(pubkey))
    pkh = RIPEMD.new(sha256(pubkey_bytes).digest()).hexdigest()
    raw = bytes.fromhex("3a" + pkh)
    checksum = sha256(sha256(raw).digest()).hexdigest()[:8]
    address_hex = "3a" + pkh + checksum
    return encode_base58(address_hex)

def api_call(endpoint, data=None, method='POST'):
    url = f"{BASE_URL}/{endpoint}"
    if method == 'GET':
        resp = requests.get(url)
    else:
        if isinstance(data, str) and endpoint.startswith("utils/"):
            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()
    return resp.text.strip('"')



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,
        tx_info.get("key_name", "senderPublicKey"): pub_key,
    }
    if reference:
        data["reference"] = reference

    data.update(values)
    data.update(tx_info.get("defaults", {}))

    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)

    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)

    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()