TallyBox Wallet Transaction Group - Python Edition

Code Description

This Python script manages Transaction Groups for a TallyBox wallet. It loads an existing wallet from an XML file, decrypts the private key using a user-provided password, and processes multiple transactions specified in a text file. The script validates the transaction file, checks for duplicate target wallet addresses, signs a consolidated Transaction Group using ECDSA (secp256r1) with RFC 6979-compliant signatures, and saves it to an offline file. Users can then broadcast the signed transaction to the TallyBox network via HTTP POST requests. The script supports tokens (2PN, 2ZR, TLH) on the TallyBox DAG network. To ensure consistency and prevent sorting issues, the script hashes the concatenated wallet_to~order_amount~ strings for the transaction group, and uses a unique order ID formatted as # followed by the Unix timestamp. The transaction file format excludes order IDs, aligning with the C# implementation for consistent data flow.

tallybox_wallet_transaction_group.py

!pip install ecdsa

"""
Tallybox Wallet Transaction Script - Group Transaction Edition
Updated: 2025-05-02

This script provisions a Tallybox wallet (https://tallybox.mixoftix.net) for secure management of
tokens (e.g., 2PN, 2ZR, TLH) on a DAG network. This implementation provides AES-256-CBC encryption
for private keys, RFC 6979-compliant ECDSA signatures, and offline transaction signing for groups
of transactions. The transaction file format excludes order IDs, but the signed data includes an
order ID as #. The total_wallet_to is computed as a concatenation of
wallet_to~order_amount~ per transaction, matching the C# edition data flow. It also checks for
duplicate wallet_to addresses and raises an error if found.

Licensed under the GNU General Public License v3 (GPL-3), this software is open-source, ensuring
freedom to use, modify, and distribute. Derivative works must also be open-source under GPL-3,
and source code must be provided with distributions.

MixofTix Was Here!
by shahiN Noursalehi
"""

import xml.etree.ElementTree as ET
import hashlib
import base64
from cryptography.hazmat.primitives.ciphers import Cipher, algorithms, modes
from cryptography.hazmat.primitives.asymmetric import ec
from cryptography.hazmat.primitives import hashes, serialization
from cryptography.hazmat.backends import default_backend
import requests
from urllib.parse import quote
import time
import re
from getpass import getpass
import random
import string
from ecdsa import SigningKey, NIST256p
from ecdsa.util import sigencode_der_canonize
from datetime import datetime
import os.path

# Base58 encoding function
def base58_encode(bytes_data):
    """
    Encode a byte array in Base58 as per the first script's implementation.
    Args:
        bytes_data (bytes): The byte array to encode.
    Returns: Base58-encoded string.
    """
    alphabet = '123456789ABCDEFGHJKLMNPQRSTUVWXYZabcdefghijkmnopqrstuvwxyz'
    value = int.from_bytes(bytes_data, byteorder='big')
    result = ''

    while value > 0:
        remainder = value % 58
        result = alphabet[remainder] + result
        value //= 58

    for byte in bytes_data:
        if byte == 0:
            result = '1' + result
        else:
            break

    return result

def aes256_cbc_encrypt_js_compatible(data: str, secret: str, show_logs: bool = False) -> str:
    """
    Encrypt data using AES-256-CBC to match Java's AES_Encrypt_by_secret_with_custom_padding.
    Args:
        data (str): The data to encrypt (string, UTF-8 encoded).
        secret (str): Secret key (at least 64 characters, hex string).
        show_logs (bool): Whether to print debug logs.
    Returns: Base64-encoded ciphertext (no IV prepended, may include newlines).
    """
    if len(secret) < 64:
        raise ValueError("Secret key must be at least 64 characters")
    if not re.match(r'^[0-9a-fA-F]{64,}$', secret):
        raise ValueError("Secret key must be a hex string")

    aes_password = secret[:32].encode('ascii')
    aes_iv = secret[32:48].encode('ascii')
    aes_salt = secret[48:64].encode('ascii')

    key = hashlib.pbkdf2_hmac('sha256', aes_password, aes_salt, 3, 32)

    if show_logs:
        print(f"Encrypt - Secret: {secret}")
        print(f"Encrypt - Key (hex): {key.hex()}")
        print(f"Encrypt - IV (hex): {aes_iv.hex()}")

    left_padding_size = random.randint(0, 99)
    left_padding = ''.join(random.choice(string.ascii_letters) for _ in range(left_padding_size))
    right_padding_size = random.randint(0, 99)
    right_padding = ''.join(random.choice(string.ascii_letters) for _ in range(right_padding_size))
    padded_data = f"{left_padding}|{data}|{right_padding}".encode('utf-8')

    padding_length = 16 - (len(padded_data) % 16)
    padded_data += bytes([padding_length] * padding_length)

    cipher = Cipher(algorithms.AES(key), modes.CBC(aes_iv), backend=default_backend())
    encryptor = cipher.encryptor()
    ciphertext = encryptor.update(padded_data) + encryptor.finalize()

    ciphertext_b64 = base64.b64encode(ciphertext).decode('utf-8')
    return ciphertext_b64

def aes256_cbc_decrypt_js_compatible(encrypted_base64: str, secret: str, show_logs: bool = False) -> str:
    """
    Decrypt Base64-encoded AES-256-CBC data to match Java's AES_Decrypt_by_secret_with_custom_padding.
    Args:
        encrypted_base64 (str): Base64-encoded ciphertext (no IV prepended, may include newlines).
        secret (str): Secret key (at least 64 characters, hex string).
        show_logs (bool): Whether to print debug logs.
    Returns: Decrypted data as string.
    """
    if len(secret) < 64:
        raise ValueError("Secret key must be at least 64 characters")
    if not re.match(r'^[0-9a-fA-F]{64,}$', secret):
        raise ValueError("Secret key must be a hex string")

    try:
        ciphertext = base64.b64decode(encrypted_base64.replace('\n', ''))
        if show_logs:
            print(f"Decrypt - Ciphertext length: {len(ciphertext)}")
    except Exception as e:
        raise ValueError(f"Invalid Base64 ciphertext: {e}")

    aes_password = secret[:32].encode('ascii')
    aes_iv = secret[32:48].encode('ascii')
    aes_salt = secret[48:64].encode('ascii')

    key = hashlib.pbkdf2_hmac('sha256', aes_password, aes_salt, 3, 32)

    if show_logs:
        print(f"Decrypt - Secret: {secret}")
        print(f"Decrypt - Key (hex): {key.hex()}")
        print(f"Decrypt - IV (hex): {aes_iv.hex()}")

    cipher = Cipher(algorithms.AES(key), modes.CBC(aes_iv), backend=default_backend())
    decryptor = cipher.decryptor()
    try:
        padded_data = decryptor.update(ciphertext) + decryptor.finalize()
    except Exception as e:
        raise ValueError(f"Decryption failed: {e}")

    try:
        padding_length = padded_data[-1]
        if padding_length > 16 or padding_length == 0:
            raise ValueError("Invalid PKCS#7 padding")
        padded_data = padded_data[:-padding_length]
    except IndexError:
        raise ValueError("Invalid padding length")

    if show_logs:
        print(f"Decrypt - Padded data (hex): {padded_data.hex()}")

    try:
        padded_text = padded_data.decode('utf-8')
        if show_logs:
            print(f"Decrypt - Padded text: {padded_text}")
        parts = padded_text.split('|')
        if len(parts) != 3:
            raise ValueError("Invalid padding format in decrypted text")
        plaintext = parts[1]
    except UnicodeDecodeError:
        if show_logs:
            print("Decrypt - Warning: Decrypted data is not UTF-8, attempting byte split")
        parts = padded_data.split(b'|')
        if len(parts) != 3:
            raise ValueError("Invalid padding format in decrypted bytes")
        try:
            plaintext = parts[1].decode('ascii')
            if show_logs:
                print(f"Decrypt - Extracted plaintext: {plaintext}")
        except UnicodeDecodeError:
            raise ValueError("Decrypted plaintext is not a valid ASCII string")

    if not re.match(r'^[0-9a-fA-F]{64}$', plaintext):
        raise ValueError(f"Decrypted private key is not a 64-character hex string: {plaintext}")

    return plaintext

def load_wallet(file_path, password, protocol="https", graph="tallybox.mixoftix.net", show_logs: bool = False):
    """Step 1: Load wallet XML, decrypt private key, and reconstruct key pair."""
    try:
        tree = ET.parse(file_path)
        root = tree.getroot()
        wallet_name = root.find("wallet_name").text
        public_key_b58 = root.find("public_key_b58_compressed").text
        private_key_aes_b64 = root.find("private_key_aes_b64").text
        wallet_address = root.find("wallet_address").text

        if not all([wallet_name, public_key_b58, private_key_aes_b64, wallet_address]):
            raise ValueError("Invalid XML format")

        key_components = f"{wallet_name}~{password}~{wallet_address}"
        secret = hashlib.sha256(key_components.encode()).hexdigest()
        if show_logs:
            print(f"Load Wallet - Key components: {key_components}")
            print(f"Load Wallet - Secret: {secret}")

        private_key_hex = aes256_cbc_decrypt_js_compatible(private_key_aes_b64, secret, show_logs)
        if show_logs:
            print(f"Load Wallet - Decrypted private key (hex): {private_key_hex}")

        try:
            private_key_int = int(private_key_hex, 16)
            SECP256R1_ORDER = 0xFFFFFFFF00000000FFFFFFFFFFFFFFFFBCE6FAADA7179E84F3B9CAC2FC632551
            if not (1 <= private_key_int < SECP256R1_ORDER):
                raise ValueError(f"Private key out of range for secp256r1: {private_key_hex}")
        except ValueError:
            raise ValueError(f"Decrypted private key is not a valid hex string: {private_key_hex}")

        key_pair = ec.derive_private_key(
            private_key_int,
            ec.SECP256R1(),
            default_backend()
        )

        public_key = key_pair.public_key()
        public_key_bytes = public_key.public_bytes(
            encoding=serialization.Encoding.X962,
            format=serialization.PublicFormat.UncompressedPoint
        )
        x_bytes = public_key_bytes[1:33]
        y_bytes = public_key_bytes[33:]
        y_int = int.from_bytes(y_bytes, byteorder='big')
        y_parity = y_int % 2
        suffix = '1' if y_parity == 1 else '2'
        compressed_key = f"{x_bytes.hex()}*{suffix}"

        return {
            "key_pair": key_pair,
            "compressed_key": compressed_key,
            "public_key_b58": public_key_b58,
            "wallet_address": wallet_address,
            "wallet_name": wallet_name,
            "protocol": protocol,
            "graph": graph
        }
    except Exception as e:
        raise ValueError(f"Decryption failed or invalid file: {str(e)}")

import hashlib

def validate_wallet_address(address):
    """
    Validate a Tallybox wallet address, matching C# logic.
    Args:
        address (str): Wallet address in the format box.
    Returns:
        bool: True if valid, False otherwise.
    """
    # Null/empty and length check
    if not address or len(address) < 40:
        return False

    # Prefix check
    if not address.startswith("box"):
        return False

    # Algorithm check
    curve_char = address[3]
    if curve_char not in ["A", "B", "C"]:
        return False

    # Checksum check
    checksum_md5 = address[4:15]  # 11 characters
    base58_part = address[15:]
    computed_md5 = hashlib.md5(base58_part.encode()).hexdigest()[:11]

    return checksum_md5 == computed_md5

def is_numeric(value):
    """
    Check if a string is a valid numeric value (integer or decimal).
    Args:
        value (str): String to check.
    Returns:
        bool: True if numeric, False otherwise.
    """
    try:
        float(value)
        return True
    except ValueError:
        return False

def read_transaction_file(file_path, show_logs=False):
    """
    Read and validate a transaction file containing multiple transactions (no order IDs).
    Checks for duplicate wallet_to addresses and raises an error if found.
    Args:
        file_path (str): Path to the transaction text file.
        show_logs (bool): Whether to print debug logs.
    Returns:
        tuple: (transactions, total_wallet_to, total_order_amount)
               where transactions is a list of transaction dictionaries,
               total_wallet_to is wallet_to~order_amount~ per transaction.
    """
    try:
        with open(file_path, 'r') as f:
            content = f.read().strip()

        if show_logs:
            print(f"Read Transaction File - Raw content: {content}")

        parts = [part.strip() for part in content.split('~')]

        if show_logs:
            print(f"Read Transaction File - Total fields after splitting: {len(parts)}")
            print(f"Read Transaction File - Fields: {parts}")

        if len(parts) < 2 or parts[0] != "tallybox" or parts[1] != "parcel_of_transactions":
            raise ValueError("Invalid file format: must start with tallybox~parcel_of_transactions")

        if (len(parts) - 2) % 4 != 0:
            raise ValueError(f"Invalid file format: incorrect number of transaction fields, expected 2 + 4N, got {len(parts)}")

        transactions = []
        total_wallet_to = ""

        total_order_amount = 0.0
        wallet_to_set = set()  # Track wallet_to addresses for duplicate checking

        for i in range(2, len(parts), 4):
            if i + 3 >= len(parts):
                raise ValueError("Incomplete transaction data")

            wallet_to_label = parts[i]
            wallet_to = parts[i + 1]
            order_amount_label = parts[i + 2]
            order_amount = parts[i + 3]

            if show_logs:
                print(f"Read Transaction File - Transaction {(i-2)//4 + 1}:")
                print(f"  Wallet To Label: '{wallet_to_label}'")
                print(f"  Wallet To: '{wallet_to}'")
                print(f"  Order Amount Label: '{order_amount_label}'")
                print(f"  Order Amount: '{order_amount}'")

            # Check for duplicate wallet_to
            if wallet_to in wallet_to_set:
                if show_logs:
                    print(f"Read Transaction File - Duplicate wallet address: {wallet_to}")
                raise ValueError(f"error~304~duplicate wallet address found~{wallet_to}")
            wallet_to_set.add(wallet_to)

            transaction_num = (i - 2) // 4 + 1
            expected_wallet_to_label = f"wallet_to_{transaction_num}"
            expected_order_amount_label = f"order_amount_{transaction_num}"
            if (wallet_to_label != expected_wallet_to_label or
                order_amount_label != expected_order_amount_label):
                raise ValueError(f"Invalid field labels at transaction {transaction_num}: "
                               f"expected {expected_wallet_to_label}, {expected_order_amount_label}, "
                               f"got '{wallet_to_label}', '{order_amount_label}'")

            if not validate_wallet_address(wallet_to):
                if show_logs:
                    print(f"Read Transaction File - Invalid wallet address: {wallet_to}")
                raise ValueError(f"error~301~invalid wallet format~{wallet_to}")

            if show_logs:
                print(f"Read Transaction File - Wallet address {wallet_to} is valid")

            if not is_numeric(order_amount):
                if show_logs:
                    print(f"Read Transaction File - Invalid order_amount: {order_amount}")
                raise ValueError(f"error~303~invalid numeric data~order_amount_{order_amount}")
            try:
                amount = float(order_amount)
                if amount <= 0:
                    raise ValueError
            except ValueError:
                if show_logs:
                    print(f"Read Transaction File - Invalid order_amount (non-positive): {order_amount}")
                raise ValueError(f"error~303~invalid order_amount, must be positive~{order_amount}")

            if show_logs:
                print(f"Read Transaction File - Order amount {order_amount} is valid")

            transactions.append({
                "wallet_to": wallet_to,
                "order_amount": order_amount
            })

            total_wallet_to += f"{wallet_to}~{order_amount}~"
            total_order_amount += amount

        if not transactions:
            raise ValueError("No valid transactions found in file")

        if show_logs:
            print(f"Read Transaction File - Total transactions: {len(transactions)}")
            print(f"Read Transaction File - Total wallet_to: {total_wallet_to}")
            print(f"Read Transaction File - Total order_amount: {total_order_amount:.8f}")

        return transactions, total_wallet_to, total_order_amount
    except Exception as e:
        if show_logs:
            print(f"Read Transaction File - Error: {str(e)}")
        raise ValueError(f"Failed to read transaction file: {str(e)}")

def prepare_group_transaction(wallet_state, transactions, total_wallet_to, total_order_amount, show_logs=False):
    """
    Prepare and sign a group transaction based on multiple transactions.
    Args:
        wallet_state (dict): Wallet state from load_wallet.
        transactions (list): List of transaction dictionaries (no order_id).
        total_wallet_to (str): Concatenated wallet_to~order_amount~ strings.
        total_order_amount (float): Sum of order_amounts.
        show_logs (bool): Whether to print debug logs.
    Returns:
        tuple: (broadcast_data, token) where broadcast_data is the signed transaction and token is the selected token.
    """
    graph = wallet_state["graph"]
    utc_unix = int(time.time())
    token = input("Select token (2PN, 2ZR, TLH): ")

    if token not in ["2PN", "2ZR", "TLH"]:
        raise ValueError("Invalid token")

    if show_logs:
        print(f"Prepare Group Transaction - Selected token: {token}")
        print(f"Prepare Group Transaction - Number of transactions: {len(transactions)}")
        print(f"Prepare Group Transaction - Total wallet_to: {total_wallet_to}")
        print(f"Prepare Group Transaction - Total order_amount: {total_order_amount:.8f}")

    sha256_total_wallet_to = hashlib.sha256(total_wallet_to.encode()).hexdigest()
    order_id = f"#{utc_unix}"

    if show_logs:
        print(f"Prepare Group Transaction - SHA256 wallet_to: {sha256_total_wallet_to}")
        print(f"Prepare Group Transaction - order_id: {order_id}")

    transaction_data = (
        f"{graph}~{graph}~"
        f"{wallet_state['wallet_address']}~{sha256_total_wallet_to}~"
        f"{token}~{total_order_amount:.8f}~{order_id}~{utc_unix}"
    )

    if show_logs:
        print(f"Prepare Group Transaction - Transaction data: {transaction_data}")

    msg_hash = hashlib.sha256(transaction_data.encode()).digest()

    if show_logs:
        print(f"Prepare Group Transaction - Message hash: {msg_hash.hex()}")

    try:
        private_key = wallet_state["key_pair"].private_numbers().private_value
        private_key_bytes = private_key.to_bytes(32, byteorder='big')
        sk = SigningKey.from_string(private_key_bytes, curve=NIST256p)
        signature_der = sk.sign_digest(msg_hash, sigencode=sigencode_der_canonize)
        sig_base64 = base64.b64encode(signature_der).decode()

        if show_logs:
            print(f"Prepare Group Transaction - Signature (Base64): {sig_base64}")
    except Exception as e:
        if show_logs:
            print(f"Prepare Group Transaction - Signature failed: {str(e)}")
        raise ValueError(f"Failed to sign transaction: {str(e)}")

    broadcast_data = "~".join([
        "tallybox", "parcel_of_transaction",
        "number_of_transactions", str(len(transactions)),
        "graph_from", graph,
        "graph_to", graph,
        "wallet_from", wallet_state['wallet_address'],
        "wallet_to", sha256_total_wallet_to,
        "order_currency", token,
        "order_amount", f"{total_order_amount:.8f}",
        "order_id", order_id,
        "order_utc_unix", str(utc_unix),
        "the_sign", sig_base64,
        "publicKey_xy_compressed", wallet_state['public_key_b58']
    ])

    if show_logs:
        print(f"Prepare Group Transaction - Broadcast data: {broadcast_data}")

    return broadcast_data, token

def broadcast_group_transaction(wallet_state, broadcast_data, transactions, show_logs=False):
    """
    Broadcast a group transaction to the Tallybox web service.
    Args:
        wallet_state (dict): Wallet state from load_wallet.
        broadcast_data (str): Signed group transaction data.
        transactions (list): List of original transactions (no order_id).
        show_logs (bool): Whether to print debug logs.
    Returns:
        str: Broadcast result message.
    """
    url = f"{wallet_state['protocol']}://{wallet_state['graph']}/broadcast.asmx/order_accept_multiple"

    order_csv_multiple_parts = ["tallybox", "parcel_of_transactions"]
    for i, txn in enumerate(transactions, 1):
        order_csv_multiple_parts.extend([
            f"wallet_to_{i}", txn["wallet_to"],
            f"order_amount_{i}", txn["order_amount"]
        ])
    order_csv_multiple = "~".join(order_csv_multiple_parts)

    if show_logs:
        print(f"Broadcast Group Transaction - order_csv_multiple: {order_csv_multiple}")

    broadcast_parts = broadcast_data.split("~")

    tallybox, parcel_type, num_label, num_txs, graph_from_label, graph_from, graph_to_label, graph_to, \
    wallet_from_label, wallet_from, wallet_to_label, wallet_to, order_currency_label, order_currency, \
    order_amount_label, order_amount, order_id_label, order_id, order_utc_unix_label, order_utc_unix, \
    the_sign_label, sig_base64, public_key_label, public_key_b58 = broadcast_parts

    if not all([
        tallybox == "tallybox",
        parcel_type == "parcel_of_transaction",
        num_label == "number_of_transactions",
        graph_from_label == "graph_from",
        graph_to_label == "graph_to",
        wallet_from_label == "wallet_from",
        wallet_to_label == "wallet_to",
        order_currency_label == "order_currency",
        order_amount_label == "order_amount",
        order_id_label == "order_id",
        order_utc_unix_label == "order_utc_unix",
        the_sign_label == "the_sign",
        public_key_label == "publicKey_xy_compressed"
    ]):
        raise ValueError("Invalid broadcast data structure")

    broadcast_data_quoted = "~".join([
        tallybox, parcel_type, num_label, num_txs, graph_from_label, graph_from, graph_to_label, graph_to,
        wallet_from_label, wallet_from, wallet_to_label, wallet_to, order_currency_label, order_currency,
        order_amount_label, order_amount, order_id_label, quote(order_id), order_utc_unix_label, order_utc_unix,
        the_sign_label, quote(sig_base64), public_key_label, public_key_b58
    ])

    if show_logs:
        print(f"Broadcast Group Transaction - broadcast_data_quoted: {broadcast_data_quoted}")

    post_data = (
        f"app_name=tallybox&app_version=2.0&"
        f"order_csv={broadcast_data_quoted}&"
        f"order_csv_multiple={order_csv_multiple}"
    )

    if show_logs:
        print(f"Broadcast Group Transaction - POST data: {post_data}")

    try:
        response = requests.post(url, data=post_data, headers={"Content-Type": "application/x-www-form-urlencoded"})
        response.raise_for_status()
        if show_logs:
            print(f"Broadcast Group Transaction - Response: {response.text}")
        if response.text.startswith("submitted~200~"):
            return f"Group transaction broadcast successfully: {response.text}"
        else:
            parts = response.text.split("~")
            error_code = parts[1] if len(parts) > 1 else "unknown"
            error_message = parts[2] if len(parts) > 2 else "no details provided"
            return f"Broadcast failed with error {error_code}: {error_message}"
    except Exception as e:
        if show_logs:
            print(f"Broadcast Group Transaction - Error: {str(e)}")
        raise ValueError(f"Failed to broadcast group transaction: {str(e)}")

def main():
    """Main function to run the group transaction workflow."""
    try:
        file_path = input("Enter wallet XML file path: ")
        password = getpass("Enter wallet password: ")

        print("Do you want to see the log of key extraction and decryption process?")
        print("[1] Yes")
        print("[2] No")
        log_choice = input("Enter choice (1 or 2): ")

        if log_choice not in ["1", "2"]:
            raise ValueError("Invalid choice for log visibility")

        show_logs = log_choice == "1"
        wallet_state = load_wallet(file_path, password, show_logs=show_logs)

        if not show_logs:
            print("Wallet loaded successfully!")

        txn_file_path = input("Enter path to transaction text file: ")
        transactions, total_wallet_to, total_order_amount = read_transaction_file(txn_file_path, show_logs)
        print(f"Loaded {len(transactions)} transactions from {txn_file_path}")

        broadcast_data, token = prepare_group_transaction(wallet_state, transactions, total_wallet_to, total_order_amount, show_logs)
        print("Group transaction signed:", broadcast_data)

        timestamp = datetime.now().strftime("%Y_%m_%d_%H_%M_%S")
        txn_file_base = os.path.splitext(os.path.basename(txn_file_path))[0]
        filename = f"{txn_file_base}_offline_{token}_{timestamp}.txt"

        with open(filename, "w") as f:
            f.write(broadcast_data)
        print(f"Group transaction saved to {filename}")

        print("\nSelect action:")
        print("[1] Broadcast")
        print("[2] Quit")
        action = input("Enter choice (1 or 2): ")

        if action == "1":
            broadcast_result = broadcast_group_transaction(wallet_state, broadcast_data, transactions, show_logs)
            print(broadcast_result)
        elif action == "2":
            print("bye")
            return
        else:
            raise ValueError("Invalid action selected")

    except Exception as e:
        print(f"Error: {str(e)}")

if __name__ == "__main__":
    main()