new pack and unpack for sweep

counterparty-core/counterpartycore/lib/messages/sweep.py

@@ -8,7 +8,7 @@
 )
 from counterpartycore.lib.ledger.currentstate import CurrentState
 from counterpartycore.lib.parser import messagetype, protocol
-from counterpartycore.lib.utils import address
+from counterpartycore.lib.utils import address, helpers
 
 logger = logging.getLogger(config.LOGGER_NAME)
 
@@ -70,28 +70,69 @@ def compose(
     db, source: str, destination: str, flags: int, memo: str, skip_validation: bool = False
 ):
     if memo is None:
-        memo = b""
+        memo_bytes = b""
     elif flags & FLAG_BINARY_MEMO:
-        memo = bytes.fromhex(memo)
+        memo_bytes = bytes.fromhex(memo)
     else:
-        memo = memo.encode("utf-8")
-        memo = struct.pack(f">{len(memo)}s", memo)
+        memo_bytes = memo.encode("utf-8")
+        memo_bytes = struct.pack(f">{len(memo)}s", memo)
 
     block_index = CurrentState().current_block_index()
-    problems, _total_fee = validate(db, source, destination, flags, memo, block_index)
+    problems, _total_fee = validate(db, source, destination, flags, memo_bytes, block_index)
     if problems and not skip_validation:
         raise exceptions.ComposeError(problems)
 
     short_address_bytes = address.pack(destination)
 
-    data = messagetype.pack(ID)
-    data += struct.pack(FORMAT, short_address_bytes, flags)
-    data += memo
+    if protocol.enabled("taproot_support"):
+        data = struct.pack(config.SHORT_TXTYPE_FORMAT, ID)
+        data_content = b"|".join(
+            [
+                short_address_bytes,
+                helpers.int_to_bytes(flags),
+                memo_bytes,
+            ]
+        )
+        logger.warning(f"data_content: {data_content}")
+        data += struct.pack(f">{len(data_content)}s", data_content)
+    else:
+        data = messagetype.pack(ID)
+        data += struct.pack(FORMAT, short_address_bytes, flags)
+        data += memo
 
     return (source, [], data)
 
 
+def new_unpack(message):
+    try:
+        data_content = struct.unpack(f">{len(message)}s", message)[0].split(b"|")
+        logger.warning(f"data_content unpack: {data_content}")
+        arg_count = len(data_content)
+        (
+            short_address_bytes,
+            flags_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 :])
+
+        flags = helpers.bytes_to_int(flags_bytes)
+
+        full_address = address.unpack(short_address_bytes)
+
+        return {
+            "destination": full_address,
+            "flags": flags,
+            "memo": memo_bytes,
+        }
+    except Exception as e:  # pylint: disable=broad-exception-caught
+        logger.error(f"sweep unpack error: {e}")
+        raise exceptions.UnpackError("could not unpack") from e
+
+
 def unpack(message):
+    if protocol.enabled("taproot_support"):
+        return new_unpack(message)
+
     try:
         memo_bytes_length = len(message) - LENGTH
         if memo_bytes_length < 0:
@@ -105,7 +146,6 @@ def unpack(message):
             memo_bytes = None
         elif not flags & FLAG_BINARY_MEMO:
             memo_bytes = memo_bytes.decode("utf-8")
-
         # unpack address
         full_address = address.unpack(short_address_bytes)
     except struct.error as e:
@@ -128,6 +168,7 @@ def parse(db, tx, message):
     # Unpack message.
     try:
         unpacked = unpack(message)
+        logger.warning("unpacked: %s", unpacked)
         destination, flags, memo_bytes = (
             unpacked["destination"],
             unpacked["flags"],

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

@@ -3,8 +3,6 @@
 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
@@ -21,6 +19,7 @@
 def new_unpack(message):
     try:
         data_content = struct.unpack(f">{len(message)}s", message)[0].split(b"|")
+        logger.warning(f"data_content unpack: {data_content}")
         arg_count = len(data_content)
         (
             asset_id_bytes,
@@ -37,7 +36,7 @@ def new_unpack(message):
 
         quantity = helpers.bytes_to_int(quantity_bytes)
 
-        full_address = utils.unpack_address(short_address_bytes)
+        full_address = address.unpack(short_address_bytes)
 
         return {
             "asset": asset,
@@ -46,6 +45,7 @@ def new_unpack(message):
             "memo": memo_bytes,
         }
     except Exception as e:  # pylint: disable=broad-exception-caught
+        logger.error(f"enhanced send unpack error: {e}")
         raise exceptions.UnpackError("could not unpack") from e
 
 
@@ -195,6 +195,7 @@ def compose(
                 memo_bytes,
             ]
         )
+        logger.warning(f"data_content: {data_content}")
         data += struct.pack(f">{len(data_content)}s", data_content)
     else:
         memo_bytes = struct.pack(f">{len(memo)}s", memo)

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

@@ -80,11 +80,8 @@ def pack(address):
     """
     if enabled("taproot_support"):
         try:
-            packed = bytes(utils.pack_address(address, config.NETWORK_NAME))
-            logger.warning(f"{address} packed: {packed}")
-            return packed
+            return bytes(utils.pack_address(address, config.NETWORK_NAME))
         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
@@ -123,12 +120,8 @@ def unpack(short_address_bytes):
     """
     if enabled("taproot_support"):
         try:
-            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
+            return utils.unpack_address(short_address_bytes, config.NETWORK_NAME)
         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/test/functionals/taproot_scenarios_test.py

@@ -19,5 +19,5 @@ def test_taproot_scenario_2(empty_ledger_db, bitcoind_mock, defaults):
         defaults,
         defaults["addresses"][0],
         defaults["p2tr_addresses"][0],
-        "",
+        "4330fdfa070eb72f65742a23fe9e68bbf2e67c55d28ad4b85ad11288f2eeb7ab",
     )