Add bitcoincash support for prepare and run scripts, add bitcoincash to testing suite, groundwork for bch-xmr atomic swap protocol

basicswap/interface/bch.py

@@ -0,0 +1,119 @@
+#!/usr/bin/env python
+# -*- coding: utf-8 -*-
+
+# Copyright (c) 2020-2023 tecnovert
+# Distributed under the MIT software license, see the accompanying
+# file LICENSE or http://www.opensource.org/licenses/mit-license.php.
+
+from typing import Union
+from basicswap.contrib.test_framework.messages import COutPoint, CTransaction, CTxIn, CTxOut
+from basicswap.util import ensure, i2h
+from .btc import BTCInterface, findOutput
+from basicswap.rpc import make_rpc_func
+from basicswap.chainparams import Coins, chainparams
+from basicswap.interface.contrib.bch_test_framework.cashaddress import Address
+from basicswap.util.crypto import hash160, sha256
+from basicswap.interface.contrib.bch_test_framework.script import OP_EQUAL, OP_EQUALVERIFY, OP_HASH256, OP_DUP, OP_HASH160, OP_CHECKSIG
+from basicswap.contrib.test_framework.script import (
+    CScript, CScriptOp,
+)
+
+class BCHInterface(BTCInterface):
+    @staticmethod
+    def coin_type():
+        return Coins.BCH
+
+    def __init__(self, coin_settings, network, swap_client=None):
+        super(BCHInterface, self).__init__(coin_settings, network, swap_client)
+
+
+    def decodeAddress(self, address: str) -> bytes:
+        return bytes(Address.from_string(address).payload)
+
+    def pubkey_to_segwit_address(self, pk: bytes) -> str:
+        raise NotImplementedError()
+
+    def pkh_to_address(self, pkh: bytes) -> str:
+        # pkh is ripemd160(sha256(pk))
+        assert (len(pkh) == 20)
+        prefix = self.chainparams_network()['hrp']
+        address = Address("P2PKH", b'\x76\xa9\x14' + pkh + b'\x88\xac')
+        address.prefix = prefix
+        return address.cash_address()
+
+    def getNewAddress(self, use_segwit: bool = False, label: str = 'swap_receive') -> str:
+        args = [label]
+        return self.rpc_wallet('getnewaddress', args)
+
+    def addressToLockingBytecode(self, address: str) -> bytes:
+        return b'\x76\xa9\x14' + bytes(Address.from_string(address).payload) + b'\x88\xac'
+    
+    def getScriptDest(self, script):
+        return self.scriptToP2SH32LockingBytecode(script)
+
+    def scriptToP2SH32LockingBytecode(self, script: Union[bytes, str]) -> bytes:
+        if isinstance(script, str):
+            script = bytes.fromhex(script)
+
+        return CScript([
+            CScriptOp(OP_HASH256),
+            sha256(sha256(script)),
+            CScriptOp(OP_EQUAL),
+        ])
+    
+    def createSCLockTx(self, value: int, script: bytearray, vkbv: bytes = None) -> bytes:
+        tx = CTransaction()
+        tx.nVersion = self.txVersion()
+        tx.vout.append(self.txoType()(value, self.getScriptDest(script)))
+        return tx.serialize_without_witness()
+
+    def getScriptForPubkeyHash(self, pkh: bytes) -> CScript:
+        return CScript([
+            CScriptOp(OP_DUP),
+            CScriptOp(OP_HASH160),
+            pkh,
+            CScriptOp(OP_EQUALVERIFY),
+            CScriptOp(OP_CHECKSIG),
+        ])
+
+    def getTxSize(self, tx: CTransaction) -> int:
+        return len(tx.serialize_without_witness())
+
+    def getScriptScriptSig(self, script: bytes, ves: bytes) -> bytes:
+        if ves is not None:
+            return CScript([ves, script])
+        else:
+            return CScript([script])
+
+    def createSCLockSpendTx(self, tx_lock_bytes, script_lock, pkh_dest, tx_fee_rate, ves=None, fee_info={}):
+        # tx_fee_rate in this context is equal to `mining_fee` contract param
+        tx_lock = self.loadTx(tx_lock_bytes)
+        output_script = self.getScriptDest(script_lock)
+        locked_n = findOutput(tx_lock, output_script)
+        ensure(locked_n is not None, 'Output not found in tx')
+        locked_coin = tx_lock.vout[locked_n].nValue
+
+        tx_lock.rehash()
+        tx_lock_id_int = tx_lock.sha256
+
+        tx = CTransaction()
+        tx.nVersion = self.txVersion()
+        tx.vin.append(CTxIn(COutPoint(tx_lock_id_int, locked_n),
+                            scriptSig=self.getScriptScriptSig(script_lock, ves),
+                            nSequence=0))
+
+        tx.vout.append(self.txoType()(locked_coin, self.getScriptForPubkeyHash(pkh_dest)))
+        pay_fee = tx_fee_rate
+        tx.vout[0].nValue = locked_coin - pay_fee
+
+        size = self.getTxSize(tx)
+
+        fee_info['fee_paid'] = pay_fee
+        fee_info['rate_used'] = tx_fee_rate
+        fee_info['size'] = size
+
+        tx.rehash()
+        self._log.info('createSCLockSpendTx %s:\n    fee_rate, size, fee: %ld, %ld, %ld.',
+                       i2h(tx.sha256), tx_fee_rate, size, pay_fee)
+
+        return tx.serialize_without_witness()

basicswap/interface/contrib/bch_test_framework/cashaddress.py

@@ -0,0 +1,247 @@
+import unittest
+
+
+CHARSET = "qpzry9x8gf2tvdw0s3jn54khce6mua7l"
+
+def polymod(values):
+    chk = 1
+    generator = [
+        (0x01, 0x98F2BC8E61),
+        (0x02, 0x79B76D99E2),
+        (0x04, 0xF33E5FB3C4),
+        (0x08, 0xAE2EABE2A8),
+        (0x10, 0x1E4F43E470),
+    ]
+    for value in values:
+        top = chk >> 35
+        chk = ((chk & 0x07FFFFFFFF) << 5) ^ value
+        for i in generator:
+            if top & i[0] != 0:
+                chk ^= i[1]
+    return chk ^ 1
+
+
+def calculate_checksum(prefix, payload):
+    poly = polymod(prefix_expand(prefix) + payload + [0, 0, 0, 0, 0, 0, 0, 0])
+    out = list()
+    for i in range(8):
+        out.append((poly >> 5 * (7 - i)) & 0x1F)
+    return out
+
+
+def verify_checksum(prefix, payload):
+    return polymod(prefix_expand(prefix) + payload) == 0
+
+
+def b32decode(inputs):
+    out = list()
+    for letter in inputs:
+        out.append(CHARSET.find(letter))
+    return out
+
+
+def b32encode(inputs):
+    out = ""
+    for char_code in inputs:
+        out += CHARSET[char_code]
+    return out
+
+
+def convertbits(data, frombits, tobits, pad=True):
+    acc = 0
+    bits = 0
+    ret = []
+    maxv = (1 << tobits) - 1
+    max_acc = (1 << (frombits + tobits - 1)) - 1
+    for value in data:
+        if value < 0 or (value >> frombits):
+            return None
+        acc = ((acc << frombits) | value) & max_acc
+        bits += frombits
+        while bits >= tobits:
+            bits -= tobits
+            ret.append((acc >> bits) & maxv)
+    if pad:
+        if bits:
+            ret.append((acc << (tobits - bits)) & maxv)
+    elif bits >= frombits or ((acc << (tobits - bits)) & maxv):
+        return None
+    return ret
+
+
+def prefix_expand(prefix):
+    return [ord(x) & 0x1F for x in prefix] + [0]
+
+
+class Address:
+    """
+    Class to handle CashAddr.
+
+    :param version: Version of CashAddr
+    :type version: ``str``
+    :param payload: Payload of CashAddr as int list of the bytearray
+    :type payload: ``list`` of ``int``
+    """
+
+    VERSIONS = {
+        "P2SH20": {"prefix": "bitcoincash", "version_bit": 8, "network": "mainnet"},
+        "P2SH32": {"prefix": "bitcoincash", "version_bit": 11, "network": "mainnet"},
+        "P2PKH": {"prefix": "bitcoincash", "version_bit": 0, "network": "mainnet"},
+        "P2SH20-TESTNET": {"prefix": "bchtest", "version_bit": 8, "network": "testnet"},
+        "P2SH32-TESTNET": {
+            "prefix": "bchtest",
+            "version_bit": 11,
+            "network": "testnet",
+        },
+        "P2PKH-TESTNET": {"prefix": "bchtest", "version_bit": 0, "network": "testnet"},
+        "P2SH20-REGTEST": {"prefix": "bchreg", "version_bit": 8, "network": "regtest"},
+        "P2SH32-REGTEST": {"prefix": "bchreg", "version_bit": 11, "network": "regtest"},
+        "P2PKH-REGTEST": {"prefix": "bchreg", "version_bit": 0, "network": "regtest"},
+        "P2SH20-CATKN": {
+            "prefix": "bitcoincash",
+            "version_bit": 24,
+            "network": "mainnet",
+        },
+        "P2SH32-CATKN": {
+            "prefix": "bitcoincash",
+            "version_bit": 27,
+            "network": "mainnet",
+        },
+        "P2PKH-CATKN": {
+            "prefix": "bitcoincash",
+            "version_bit": 16,
+            "network": "mainnet",
+        },
+        "P2SH20-CATKN-TESTNET": {
+            "prefix": "bchtest",
+            "version_bit": 24,
+            "network": "testnet",
+        },
+        "P2SH32-CATKN-TESTNET": {
+            "prefix": "bchtest",
+            "version_bit": 27,
+            "network": "testnet",
+        },
+        "P2PKH-CATKN-TESTNET": {
+            "prefix": "bchtest",
+            "version_bit": 16,
+            "network": "testnet",
+        },
+        "P2SH20-CATKN-REGTEST": {
+            "prefix": "bchreg",
+            "version_bit": 24,
+            "network": "regtest",
+        },
+        "P2SH32-CATKN-REGTEST": {
+            "prefix": "bchreg",
+            "version_bit": 27,
+            "network": "regtest",
+        },
+        "P2PKH-CATKN-REGTEST": {
+            "prefix": "bchreg",
+            "version_bit": 16,
+            "network": "regtest",
+        },
+    }
+
+    VERSION_SUFFIXES = {"bitcoincash": "", "bchtest": "-TESTNET", "bchreg": "-REGTEST"}
+
+    ADDRESS_TYPES = {
+        0: "P2PKH",
+        8: "P2SH20",
+        11: "P2SH32",
+        16: "P2PKH-CATKN",
+        24: "P2SH20-CATKN",
+        27: "P2SH32-CATKN",
+    }
+
+    def __init__(self, version, payload):
+        if version not in Address.VERSIONS:
+            raise ValueError("Invalid address version provided")
+
+        self.version = version
+        self.payload = payload
+        self.prefix = Address.VERSIONS[self.version]["prefix"]
+
+    def __str__(self):
+        return (
+            f"version: {self.version}\npayload: {self.payload}\nprefix: {self.prefix}"
+        )
+
+    def __repr__(self):
+        return f"Address('{self.cash_address()}')"
+
+    def __eq__(self, other):
+        if isinstance(other, str):
+            return self.cash_address() == other
+        elif isinstance(other, Address):
+            return self.cash_address() == other.cash_address()
+        else:
+            raise ValueError(
+                "Address can be compared to a string address"
+                " or an instance of Address"
+            )
+
+    def cash_address(self):
+        """
+        Generate CashAddr of the Address
+
+        :rtype: ``str``
+        """
+        version_bit = Address.VERSIONS[self.version]["version_bit"]
+        payload = [version_bit] + self.payload
+        payload = convertbits(payload, 8, 5)
+        checksum = calculate_checksum(self.prefix, payload)
+        return self.prefix + ":" + b32encode(payload + checksum)
+
+    @staticmethod
+    def from_string(address):
+        """
+        Generate Address from a cashadress string
+
+        :param scriptcode: The cashaddress string
+        :type scriptcode: ``str``
+        :returns: Instance of :class:~bitcash.cashaddress.Address
+        """
+        try:
+            address = str(address)
+        except Exception:
+            raise ValueError("Expected string as input")
+
+        if address.upper() != address and address.lower() != address:
+            raise ValueError(
+                "Cash address contains uppercase and lowercase characters"
+            )
+
+        address = address.lower()
+        colon_count = address.count(":")
+        if colon_count == 0:
+            raise ValueError("Cash address is missing prefix")
+        if colon_count > 1:
+            raise ValueError("Cash address contains more than one colon character")
+
+        prefix, base32string = address.split(":")
+        decoded = b32decode(base32string)
+
+        if not verify_checksum(prefix, decoded):
+            raise ValueError(
+                "Bad cash address checksum for address {}".format(address)
+            )
+        converted = convertbits(decoded, 5, 8)
+
+        try:
+            version = Address.ADDRESS_TYPES[converted[0]]
+        except Exception:
+            raise ValueError("Could not determine address version")
+
+        version += Address.VERSION_SUFFIXES[prefix]
+
+        payload = converted[1:-6]
+        return Address(version, payload)
+
+class TestFrameworkScript(unittest.TestCase):
+    def test_base58encodedecode(self):
+        def check_cashaddress(address: str):
+            self.assertEqual(Address.from_string(address).cash_address(), address)
+    
+        check_cashaddress("bitcoincash:qzfyvx77v2pmgc0vulwlfkl3uzjgh5gnmqk5hhyaa6")

basicswap/interface/contrib/bch_test_framework/script.py

@@ -0,0 +1,40 @@
+# -*- coding: utf-8 -*-
+
+# Copyright (c) 2024 tecnovert
+# Distributed under the MIT software license, see the accompanying
+# file LICENSE or http://www.opensource.org/licenses/mit-license.php.
+
+
+OP_TXINPUTCOUNT = 0xc3
+OP_1 = 0x51
+OP_NUMEQUALVERIFY = 0x9d
+OP_TXOUTPUTCOUNT = 0xc4
+OP_0 = 0x00
+OP_UTXOVALUE = 0xc6
+OP_OUTPUTVALUE = 0xcc
+OP_SUB = 0x94
+OP_UTXOTOKENCATEGORY = 0xce
+OP_OUTPUTTOKENCATEGORY = 0xd1
+OP_EQUALVERIFY = 0x88
+OP_UTXOTOKENCOMMITMENT = 0xcf
+OP_OUTPUTTOKENCOMMITMENT = 0xd2
+OP_UTXOTOKENAMOUNT = 0xd0
+OP_OUTPUTTOKENAMOUNT = 0xd3
+OP_INPUTSEQUENCENUMBER = 0xcb
+OP_NOTIF = 0x64
+OP_OUTPUTBYTECODE = 0xcd
+OP_OVER = 0x78
+OP_CHECKDATASIG = 0xba
+OP_CHECKDATASIGVERIFY = 0xbb
+OP_ELSE = 0x67
+OP_CHECKSEQUENCEVERIFY = 0xb2
+OP_DROP = 0x75
+OP_EQUAL = 0x87
+OP_ENDIF = 0x68
+OP_HASH256 = 0xaa
+OP_PUSHBYTES_32 = 0x20
+OP_DUP = 0x76
+OP_HASH160 = 0xa9
+OP_CHECKSIG = 0xac
+OP_SHA256 = 0xa8
+OP_VERIFY = 0x69