diff --git a/requirements.txt b/requirements.txt new file mode 100644 index 0000000..5d84d4a --- /dev/null +++ b/requirements.txt @@ -0,0 +1,3 @@ +requests +pycryptodome + diff --git a/tx.py b/tx.py new file mode 100755 index 0000000..8e82f8b --- /dev/null +++ b/tx.py @@ -0,0 +1,229 @@ +#!/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() +