new pack and unpack for enhancedsend

counterparty-core/counterpartycore/lib/messages/versions/enhancedsend.py

@@ -3,6 +3,8 @@
 import logging
 import struct
 
+from counterparty_rs import utils  # pylint: disable=no-name-in-module
+
 from counterpartycore.lib import config, exceptions, ledger
 from counterpartycore.lib.messages.versions import send1
 from counterpartycore.lib.parser import messagetype, protocol
@@ -16,7 +18,41 @@
 ID = 2  # 0x02
 
 
+def new_unpack(message):
+    try:
+        data_content = struct.unpack(f">{len(message)}s", message)[0].split(b"|")
+        arg_count = len(data_content)
+        (
+            asset_id_bytes,
+            quantity_bytes,
+            short_address_bytes,
+        ) = data_content[0 : arg_count - 1]
+        # The memo is placed last to be able to contain `|`.
+        memo_bytes = b"|".join(data_content[arg_count - 1 :])
+
+        asset_id = helpers.bytes_to_int(asset_id_bytes)
+        asset = ledger.issuances.generate_asset_name(asset_id)
+        if asset == config.BTC:
+            raise exceptions.AssetNameError(f"{config.BTC} not allowed")
+
+        quantity = helpers.bytes_to_int(quantity_bytes)
+
+        full_address = utils.unpack_address(short_address_bytes)
+
+        return {
+            "asset": asset,
+            "quantity": quantity,
+            "address": full_address,
+            "memo": memo_bytes,
+        }
+    except Exception as e:  # pylint: disable=broad-exception-caught
+        raise exceptions.UnpackError("could not unpack") from e
+
+
 def unpack(message):
+    if protocol.enabled("taproot_support"):
+        return new_unpack(message)
+
     try:
         # account for memo bytes
         memo_bytes_length = len(message) - LENGTH
@@ -136,8 +172,7 @@ def compose(
     elif memo_is_hex:
         memo_bytes = bytes.fromhex(memo)
     else:
-        memo = memo.encode("utf-8")
-        memo_bytes = struct.pack(f">{len(memo)}s", memo)
+        memo_bytes = memo.encode("utf-8")
 
     problems = validate(db, destination, asset, quantity, memo_bytes)
     if problems and not skip_validation:
@@ -150,9 +185,22 @@ def compose(
 
     short_address_bytes = address.pack(destination)
 
-    data = messagetype.pack(ID)
-    data += struct.pack(FORMAT, asset_id, quantity, short_address_bytes)
-    data += memo_bytes
+    if protocol.enabled("taproot_support"):
+        data = struct.pack(config.SHORT_TXTYPE_FORMAT, ID)
+        data_content = b"|".join(
+            [
+                helpers.int_to_bytes(asset_id),
+                helpers.int_to_bytes(quantity),
+                short_address_bytes,
+                memo_bytes,
+            ]
+        )
+        data += struct.pack(f">{len(data_content)}s", data_content)
+    else:
+        memo_bytes = struct.pack(f">{len(memo)}s", memo)
+        data = messagetype.pack(ID)
+        data += struct.pack(FORMAT, asset_id, quantity, short_address_bytes)
+        data += memo_bytes
 
     cursor.close()
     # return an empty array as the second argument because we don't need to send BTC dust to the recipient

counterparty-core/counterpartycore/lib/utils/address.py

@@ -1,3 +1,5 @@
+import logging
+
 import bitcoin
 from bitcoin.bech32 import CBech32Data
 from bitcoinutils.keys import P2pkhAddress, P2shAddress, P2trAddress, P2wpkhAddress
@@ -7,6 +9,8 @@
 from counterpartycore.lib.parser.protocol import enabled
 from counterpartycore.lib.utils import base58, helpers, multisig
 
+logger = logging.getLogger(config.LOGGER_NAME)
+
 
 def is_pubkeyhash(monosig_address):
     """Check if PubKeyHash is valid P2PKH address."""
@@ -77,9 +81,10 @@ def pack(address):
     if enabled("taproot_support"):
         try:
             packed = bytes(utils.pack_address(address, config.NETWORK_NAME))
-            print("packed", packed)
+            logger.warning(f"{address} packed: {packed}")
             return packed
         except Exception as e:  # pylint: disable=broad-except  # noqa: F841
+            logger.error(f"Error packing address: {e}")
             raise exceptions.AddressError(  # noqa: B904
                 f"The address {address} is not a valid bitcoin address ({config.NETWORK_NAME})"
             ) from e
@@ -118,8 +123,12 @@ def unpack(short_address_bytes):
     """
     if enabled("taproot_support"):
         try:
-            return utils.unpack_address(short_address_bytes, config.NETWORK_NAME)
+            logger.warning(f"short_address_bytes: {short_address_bytes}")
+            unpacked = utils.unpack_address(short_address_bytes, config.NETWORK_NAME)
+            logger.warning(f"unpacked: {unpacked}")
+            return unpacked
         except Exception as e:  # pylint: disable=broad-except  # noqa: F841
+            logger.error(f"Error unpacking address: {e}")
             raise exceptions.DecodeError(  # noqa: B904
                 f"T{short_address_bytes} is not a valid packed bitcoin address ({config.NETWORK_NAME})"
             ) from e

counterparty-core/counterpartycore/lib/utils/helpers.py

@@ -139,3 +139,15 @@ def dhash(text):
 
 def dhash_string(text):
     return binascii.hexlify(dhash(text)).decode()
+
+
+def int_to_bytes(integer_in: int) -> bytes:
+    if integer_in == 0:
+        return b"\x00"
+    binary = bin(integer_in)[2:]
+    byte_length = (len(binary) + 7) // 8
+    return integer_in.to_bytes(byte_length, "little")
+
+
+def bytes_to_int(bytes_in: bytes) -> int:
+    return int.from_bytes(bytes_in, "little")

counterparty-core/counterpartycore/test/functionals/taproot_scenarios_test.py

@@ -10,3 +10,14 @@ def test_taproot_scenario_1(empty_ledger_db, bitcoind_mock, defaults):
         defaults["addresses"][0],
         "6f488077d790fa369c53bccd1d09a548201fad6c56f48cf150f6c2f0e7062a01",
     )
+
+
+def test_taproot_scenario_2(empty_ledger_db, bitcoind_mock, defaults):
+    check_standard_scenario(
+        empty_ledger_db,
+        bitcoind_mock,
+        defaults,
+        defaults["addresses"][0],
+        defaults["p2tr_addresses"][0],
+        "",
+    )