TallyBox Wallet Transaction - Python Edition

Code Description

This Python script manages transactions for a TallyBox wallet. It loads an existing wallet from an XML file, decrypts the private key using a user-provided password, and allows users to either view token balances (2PN, 2ZR, TLH) or send transactions. The script interacts with the TallyBox network via HTTP POST requests to fetch balances or broadcast transactions, ensuring secure signing with ECDSA (secp256r1).

tallybox_wallet_transaction.py

# @title
!pip install ecdsa

"""
Tallybox Wallet Transaction Script
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.

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

# 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')  # Match Java's password.toCharArray() (UTF-8 equivalent)
    aes_iv = secret[32:48].encode('ascii')     # Match Java's aes_iv.getBytes("ASCII")
    aes_salt = secret[48:64].encode('ascii')   # Match Java's salt.getBytes() (UTF-8 equivalent)

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

    # Match Java's Base64.DEFAULT (may include newlines for long strings)
    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:
        # Handle Java's Base64.DEFAULT (strip newlines if present)
        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')  # Match Java's password.toCharArray() (UTF-8 equivalent)
    aes_iv = secret[32:48].encode('ascii')     # Match Java's aes_iv.getBytes("ASCII")
    aes_salt = secret[48:64].encode('ascii')   # Match Java's salt.getBytes() (UTF-8 equivalent)

    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)}")

def prompt_user_action(wallet_state):
    """Step 2: Prompt user to view history or sign transaction."""
    if not wallet_state.get("wallet_address"):
        raise ValueError("No wallet loaded. Load a wallet first.")

    print("Select action:")
    print("[1] View History")
    print("[2] Sign Transaction")
    action = input("Enter choice (1 or 2): ")

    if action == "1":
        return fetch_balances(wallet_state)
    elif action == "2":
        return prepare_transaction(wallet_state)
    else:
        raise ValueError("Invalid action selected")

def fetch_balances(wallet_state):
    """Step 3: Fetch token balances via POST request."""
    url = f"{wallet_state['protocol']}://{wallet_state['graph']}/archive.asmx/order_history"
    post_data = (
        f"app_name=tallybox&app_version=2.0&in_graph={quote(wallet_state['graph'])}"
        f"&wallet_address={quote(wallet_state['wallet_address'])}"
    )

    try:
        response = requests.post(url, data=post_data, headers={"Content-Type": "application/x-www-form-urlencoded"})
        response.raise_for_status()
        parts = response.text.split("~")
        balances = {
            "2PN": "0.00000000",
            "2ZR": "0.00000000",
            "TLH": "0.00000000"
        }

        for i in range(6, len(parts) - 1, 2):
            token = parts[i]
            amount = parts[i + 1]
            if token in balances and amount and re.match(r"^-?\d*\.\d+$", amount):
                balances[token] = f"{float(amount):.8f}"

        return balances
    except Exception as e:
        raise ValueError(f"Failed to fetch balances: {str(e)}")

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 prepare_transaction(wallet_state):
    """Step 4: Prepare and sign transaction."""
    target_address = input("Enter target wallet address: ")
    token = input("Select token (2PN, 2ZR, TLH): ")
    amount = input("Enter amount (positive number): ")
    order_id = input("Enter optional order ID (press Enter to skip): ")
    graph = wallet_state["graph"]
    utc_unix = int(time.time())

    if not validate_wallet_address(target_address):
        raise ValueError("Invalid target wallet address")
    if token not in ["2PN", "2ZR", "TLH"]:
        raise ValueError("Invalid token")
    try:
        amount_float = float(amount)
        if amount_float <= 0:
            raise ValueError
    except ValueError:
        raise ValueError("Invalid amount. Please enter a positive number.")

    transaction_data = (
        f"{graph}~{graph}~"
        f"{wallet_state['wallet_address']}~{target_address}~"
        f"{token}~{amount_float:.8f}~{order_id}~{utc_unix}"
    )
    msg_hash = hashlib.sha256(transaction_data.encode()).digest()

    try:
        # Extract private key from cryptography key_pair
        private_key = wallet_state["key_pair"].private_numbers().private_value
        private_key_bytes = private_key.to_bytes(32, byteorder='big')

        # Create ecdsa SigningKey for secp256r1 (NIST256p)
        sk = SigningKey.from_string(private_key_bytes, curve=NIST256p)

        # Sign the message hash with RFC 6979 deterministic signature (canonical DER encoding)
        # RFC 6979 ensures deterministic signatures for the same private key and message, enabling cross-platform compatibility
        signature_der = sk.sign_digest(msg_hash, sigencode=sigencode_der_canonize)

        # Encode signature in Base64 to match original output format
        sig_base64 = base64.b64encode(signature_der).decode()
    except Exception as e:
        raise ValueError(f"Failed to sign transaction: {str(e)}")

    broadcast_data = "~".join([
        "tallybox", "parcel_of_transaction",
        "graph_from", graph,
        "graph_to", graph,
        "wallet_from", wallet_state['wallet_address'],
        "wallet_to", target_address,
        "order_currency", token,
        "order_amount", f"{amount_float:.8f}",
        "order_id", order_id,
        "order_utc_unix", str(utc_unix),
        "the_sign", sig_base64,  # No quote() for offline file
        "publicKey_xy_compressed", wallet_state['public_key_b58']
    ])

    return broadcast_data

def broadcast_transaction(wallet_state, broadcast_data):
    """Step 5: Broadcast transaction via POST request."""
    url = f"{wallet_state['protocol']}://{wallet_state['graph']}/broadcast.asmx/order_accept"
    cleaned_broadcast_data = broadcast_data.replace('\n', '').replace('\r', '')
    post_data = f"app_name=tallybox&app_version=2.0&order_csv={cleaned_broadcast_data}"

    try:
        response = requests.post(url, data=post_data, headers={"Content-Type": "application/x-www-form-urlencoded"})
        response.raise_for_status()
        if response.text.startswith("submitted~200~"):
            return f"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:
        raise ValueError(f"Failed to broadcast transaction: {str(e)}")

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

        # Prompt user for log visibility
        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!")

        result = prompt_user_action(wallet_state)

        if isinstance(result, dict):
            print("Token Balances:")
            for token, amount in result.items():
                print(f"{token}: {amount}")
        else:
            # Transaction signed, save to file
            broadcast_data = result
            print("Transaction signed:", broadcast_data)

            # Extract token from broadcast_data (field 11, after 'order_currency')
            parts = broadcast_data.split("~")
            token = parts[11] if len(parts) > 11 else "UNKNOWN"

            # Generate timestamp for filename
            timestamp = datetime.now().strftime("%Y_%m_%d_%H_%M_%S")
            filename = f"offline_{token}_{timestamp}.txt"

            # Write broadcast_data to file (sig_base64 is unquoted)
            with open(filename, "w") as f:
                f.write(broadcast_data)
            print(f"Transaction saved to {filename}")

            # Prompt for broadcast or quit
            print("\nSelect action:")
            print("[1] Broadcast")
            print("[2] Quit")
            action = input("Enter choice (1 or 2): ")

            if action == "1":
                # Reconstruct broadcast_data with quoted signature for broadcast
                broadcast_parts = broadcast_data.split("~")
                if len(broadcast_parts) != 22:
                    raise ValueError(f"Invalid broadcast data format for broadcasting: expected 22 fields, got {len(broadcast_parts)}")

                # Extract fields from broadcast_data
                # Expected format: tallybox~parcel_of_transaction~graph_from~graph~graph_to~graph~wallet_from~wallet_address~wallet_to~target_address~order_currency~token~order_amount~amount~order_id~order_id~order_utc_unix~utc_unix~the_sign~sig_base64~publicKey_xy_compressed~public_key_b58
                tallybox, parcel_type, 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

                # Validate extracted fields
                if not all([
                    tallybox == "tallybox",
                    parcel_type == "parcel_of_transaction",
                    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")

                # Reconstruct broadcast_data with quoted signature for the_sign field
                broadcast_data_quoted = "~".join([
                    tallybox, parcel_type,
                    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, quote(sig_base64),  # Apply quote() to signature for the_sign field
                    public_key_label, public_key_b58
                ])

                print("broadcast_data_quoted:", broadcast_data_quoted)

                broadcast_result = broadcast_transaction(wallet_state, broadcast_data_quoted)
                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()