Fix decoding op from scripts with TapLeafs pattern but no scripts found

electrumx/lib/util_atomicals.py

@@ -30,7 +30,7 @@
 import math
 import re
 import sys
-from typing import Optional
+from typing import Dict
 
 import krock32
 from cbor2 import CBORTag, dumps, loads
@@ -1255,39 +1255,6 @@ def is_op_return_dmitem_payment_marker_atomical_id(script):
     return script[start_index + 5 + 2 + 1 : start_index + 5 + 2 + 1 + 36]
 
 
-def parse_atomicals_operations_from_tap_leafs(scripts, allow_args_bytes: bool):
-    # All inputs are parsed but further upstream most operations will only function if placed in the 0'th input
-    op_name, payload, index = parse_protocols_operations_from_witness_for_input(scripts)
-    if not op_name:
-        return None
-    decoded_object = {}
-    if payload:
-        # Ensure that the payload is cbor encoded dictionary or empty
-        try:
-            decoded_object = loads(payload)
-            if not isinstance(decoded_object, dict):
-                return None
-        except Exception as e:
-            return None
-        # Also enforce that if there are meta, args, or ctx fields that they must be dicts
-        # This is done to ensure that these fields are always easily parseable and do not contain unexpected data
-        # which could cause parsing problems later.
-        # Ensure that they are not allowed to contain bytes like objects
-        if (
-            not is_sanitized_dict_whitelist_only(decoded_object.get("meta", {}))
-            or not is_sanitized_dict_whitelist_only(decoded_object.get("args", {}), allow_args_bytes)
-            or not is_sanitized_dict_whitelist_only(decoded_object.get("ctx", {}))
-            or not is_sanitized_dict_whitelist_only(decoded_object.get("init", {}), True)
-        ):
-            return None
-        return {
-            "op": op_name,
-            "payload": decoded_object,
-            "input_index": index,
-        }
-    return None
-
-
 # Parses and detects valid Atomicals protocol operations in a witness script
 # Stops when it finds the first operation in the first input
 def parse_protocols_operations_from_witness_for_input(txinwitness):
@@ -1325,7 +1292,7 @@ def parse_protocols_operations_from_witness_for_input(txinwitness):
 
 
 # Parses and detects the witness script array and detects the Atomicals operations
-def parse_protocols_operations_from_witness_array(tx, tx_hash, allow_args_bytes) -> Optional[dict]:
+def parse_protocols_operations_from_witness_array(tx, tx_hash, allow_args_bytes) -> Dict:
     """Detect and parse all operations of atomicals across the witness input arrays (inputs 0 and 1) from a tx"""
     if not hasattr(tx, "witness"):
         return {}
@@ -1335,7 +1302,6 @@ def parse_protocols_operations_from_witness_array(tx, tx_hash, allow_args_bytes)
         op_name, payload, _ = parse_protocols_operations_from_witness_for_input(txinwitness)
         if not op_name:
             continue
-        decoded_object = {}
         if payload:
             # Ensure that the payload is cbor encoded dictionary or empty
             try:
@@ -1387,6 +1353,43 @@ def parse_protocols_operations_from_witness_array(tx, tx_hash, allow_args_bytes)
     return {}
 
 
+def parse_atomicals_operations_from_tap_leafs(scripts, allow_args_bytes: bool) -> Dict:
+    # All inputs are parsed but further upstream most operations will only function if placed in the 0'th input
+    op_name, payload, index = parse_protocols_operations_from_witness_for_input(scripts)
+    if not op_name:
+        return {}
+    if payload:
+        # Ensure that the payload is cbor encoded dictionary or empty
+        try:
+            decoded_object = loads(payload)
+            if not isinstance(decoded_object, dict):
+                return {}
+        except Exception as e:
+            print(
+                f"parse_atomicals_operations_from_tap_leafs found {op_name} "
+                f"but CBOR payload parsing failed for {scripts}. "
+                f"Skipping tx input...{e}"
+            )
+            return {}
+        # Also enforce that if there are meta, args, or ctx fields that they must be dicts
+        # This is done to ensure that these fields are always easily parseable and do not contain unexpected data
+        # which could cause parsing problems later.
+        # Ensure that they are not allowed to contain bytes like objects
+        if (
+            not is_sanitized_dict_whitelist_only(decoded_object.get("meta", {}))
+            or not is_sanitized_dict_whitelist_only(decoded_object.get("args", {}), allow_args_bytes)
+            or not is_sanitized_dict_whitelist_only(decoded_object.get("ctx", {}))
+            or not is_sanitized_dict_whitelist_only(decoded_object.get("init", {}), True)
+        ):
+            return {}
+        return {
+            "op": op_name,
+            "payload": decoded_object,
+            "input_index": index,
+        }
+    return {}
+
+
 def encode_atomical_ids_hex(state):
     if isinstance(state, bytes):
         if is_atomical_id_long_form_bytes(state):

tests/lib/test_atomicals_blueprint_builder.py

@@ -2,11 +2,10 @@
 
 from electrumx.lib.atomicals_blueprint_builder import AtomicalsTransferBlueprintBuilder
 from electrumx.lib.coins import Bitcoin
-from electrumx.lib.hash import HASHX_LEN, hash_to_hex_str, hex_str_to_hash
-from electrumx.lib.tx import Tx, TxInput, TxOutput
+from electrumx.lib.psbt import parse_psbt_hex_and_operations
 from electrumx.lib.util_atomicals import (
-    compact_to_location_id_bytes,
     location_id_bytes_to_compact,
+    parse_atomicals_operations_from_tap_leafs,
     parse_protocols_operations_from_witness_array,
 )
 
@@ -1774,3 +1773,16 @@ def mock_mint_fetcher(self, atomical_id):
     }
     payment_valid = blueprint_builder.are_payments_satisfied(rules)
     assert not payment_valid
+
+
+def test_parse_operations_from_empty_tap_leafs():
+    psbt = (
+        "70736274ff01005e010000000100000000000000000000000000000000000000000000000000000000000000000000000000ffffff"
+        "ff0122020000000000002251202b2e6c7946ede6a9e76ea8dc599b375a1899cf7ba784754fa9ab91486ad56fb3000000000001012b"
+        "62d0000000000000225120ecbc068d696bf671b51d45a892a6777a9e4a624bbb14aa4f3040c1a0d95786b72215c0486ff77b86a935"
+        "ed21a35a48ee5fa00cec653dcfcc6f3f93cd9b9232287870963220486ff77b86a935ed21a35a48ee5fa00cec653dcfcc6f3f93cd9b"
+        "923228787096ac00630477697a7a013604b4f5493a68c00000"
+    )
+    tx, tap_leafs = parse_psbt_hex_and_operations(psbt)
+    op = parse_atomicals_operations_from_tap_leafs(tap_leafs, True)
+    assert isinstance(op, dict)