Check Sighash Flag

counterparty-core/counterpartycore/lib/api/api_v1.py

@@ -968,6 +968,7 @@ def get_tx_info(tx_hex, block_index=None):
                         db,
                         deserialize.deserialize_tx(tx_hex, use_txid=use_txid),
                         block_index=block_index,
+                        composing=True,
                     )
                 )
             return (

counterparty-core/counterpartycore/lib/deserialize.py

@@ -58,15 +58,21 @@ def read_transaction(vds, use_txid=True):
         offset_before_tx_witnesses = vds.read_cursor - start_pos
         for vin in transaction["vin"]:  # noqa: B007
             witnesses_count = vds.read_compact_size()
-            for i in range(witnesses_count):  # noqa: B007
-                witness_length = vds.read_compact_size()
-                witness = vds.read_bytes(witness_length)
-                transaction["vtxinwit"].append(witness)
+            if witnesses_count == 0:
+                transaction["vtxinwit"].append([])
+            else:
+                vin_witnesses = []
+                for i in range(witnesses_count):  # noqa: B007
+                    witness_length = vds.read_compact_size()
+                    witness = vds.read_bytes(witness_length)
+                    vin_witnesses.append(witness)
+                transaction["vtxinwit"].append(vin_witnesses)
 
     transaction["lock_time"] = vds.read_uint32()
     data = vds.input[start_pos : vds.read_cursor]
 
     transaction["tx_hash"] = ib2h(double_hash(data))
+    transaction["tx_id"] = transaction["tx_hash"]
     if transaction["segwit"]:
         hash_data = data[:4] + data[6:offset_before_tx_witnesses] + data[-4:]
         transaction["tx_id"] = ib2h(double_hash(hash_data))

counterparty-core/counterpartycore/lib/gettxinfo.py

@@ -146,7 +146,8 @@ def get_vin_info(vin):
     if "value" in vin:
         return vin["value"], vin["script_pub_key"], vin["is_segwit"]
 
-    # Note: We don't know what block the `vin` is in, and the block might have been from a while ago, so this call may not hit the cache.
+    # Note: We don't know what block the `vin` is in, and the block might
+    # have been from a while ago, so this call may not hit the cache.
     vin_ctx = backend.bitcoind.get_decoded_transaction(vin["hash"])
 
     is_segwit = len(vin_ctx["vtxinwit"]) > 0
@@ -155,11 +156,118 @@ def get_vin_info(vin):
     return vout["value"], vout["script_pub_key"], is_segwit
 
 
+def get_der_signature_sighash_flag(value):
+    if not isinstance(value, bytes):
+        return None
+    lenght_by_prefix = {
+        "3044": 71,
+        "3045": 72,
+        "3046": 73,
+        "3041": 68,
+        "3042": 69,
+        "3043": 70,
+    }
+    for prefix, length in lenght_by_prefix.items():
+        if value.startswith(binascii.unhexlify(prefix)) and len(value) == length:
+            return value[-1:]
+    return None
+
+
+def get_schnorr_signature_sighash_flag(value):
+    if not isinstance(value, bytes):
+        return None
+    if len(value) not in [64, 65]:
+        return None
+    if len(value) == 65:
+        return value[-1:]
+    return b"\x01"  # SIGHASH_ALL by default
+
+
+def collect_sighash_flags(script_sig, witnesses):
+    flags = []
+
+    # P2PK, P2PKH, P2MS
+    if script_sig != b"":
+        asm = script.script_to_asm(script_sig)
+        for item in asm:
+            flag = get_der_signature_sighash_flag(item)
+            if flag is not None:
+                flags.append(flag)
+
+    if len(witnesses) == 0:
+        return flags
+
+    witnesses = [
+        binascii.unhexlify(witness) if isinstance(witness, str) else witness
+        for witness in witnesses
+    ]
+
+    # P2WPKH
+    if len(witnesses) == 2:
+        flag = get_der_signature_sighash_flag(witnesses[0])
+        if flag is not None:
+            flags.append(flag)
+        return flags
+
+    # P2TR key path spend
+    if len(witnesses) == 1:
+        flag = get_schnorr_signature_sighash_flag(witnesses[0])
+        if flag is not None:
+            flags.append(flag)
+        return flags
+
+    # Other cases
+    if len(witnesses) >= 3:
+        for item in witnesses:
+            flag = get_schnorr_signature_sighash_flag(item) or get_der_signature_sighash_flag(item)
+            if flag is not None:
+                flags.append(flag)
+        return flags
+
+    return flags
+
+
+# class SighashFlagError(DecodeError):
+class SighashFlagError(Exception):
+    pass
+
+
+# known transactions with invalid SIGHASH flag
+SIGHASH_FLAG_TRANSACTION_WHITELIST = [
+    "c8091f1ef768a2f00d48e6d0f7a2c2d272a5d5c8063db78bf39977adcb12e103"
+]
+
+
+def check_signatures_sighash_flag(decoded_tx):
+    if decoded_tx["tx_id"] in SIGHASH_FLAG_TRANSACTION_WHITELIST:
+        return
+
+    script_sig = decoded_tx["vin"][0]["script_sig"]
+    witnesses = []
+    if decoded_tx["segwit"]:
+        witnesses = decoded_tx["vtxinwit"][0]
+
+    flags = collect_sighash_flags(script_sig, witnesses)
+
+    if len(flags) == 0:
+        error = f"impossible to determine SIGHASH flag for transaction {decoded_tx['tx_id']}"
+        logger.debug(error)
+        raise SighashFlagError(error)
+
+    # first input must be signed with SIGHASH_ALL or SIGHASH_ALL|SIGHASH_ANYONECANPAY
+    authorized_flags = [b"\x01", b"\x81"]
+    for flag in flags:
+        if flag not in authorized_flags:
+            error = f"invalid SIGHASH flag for transaction {decoded_tx['tx_id']}"
+            logger.debug(error)
+            raise SighashFlagError(error)
+
+
 def get_transaction_sources(decoded_tx):
     sources = []
     outputs_value = 0
 
-    for vin in decoded_tx["vin"][:]:  # Loop through inputs.
+    for vin in decoded_tx["vin"]:  # Loop through inputs.
         vout_value, script_pubkey, _is_segwit = get_vin_info(vin)
 
         outputs_value += vout_value
@@ -394,6 +502,8 @@ def get_tx_info_new(db, decoded_tx, block_index, p2sh_is_segwit=False, composing
     # Collect all (unique) source addresses.
     #   if we haven't found them yet
     if p2sh_encoding_source is None:
+        if not composing:
+            check_signatures_sighash_flag(decoded_tx)
         sources, outputs_value = get_transaction_sources(decoded_tx)
         if not fee_added:
             fee += outputs_value
@@ -524,7 +634,7 @@ def get_tx_info_legacy(decoded_tx, block_index):
     return source, destination, btc_amount, fee, data, []
 
 
-def _get_tx_info(db, decoded_tx, block_index, p2sh_is_segwit=False):
+def _get_tx_info(db, decoded_tx, block_index, p2sh_is_segwit=False, composing=False):
     """Get the transaction info. Calls one of two subfunctions depending on signature type."""
     if not block_index:
         block_index = util.CURRENT_BLOCK_INDEX
@@ -535,12 +645,14 @@ def _get_tx_info(db, decoded_tx, block_index, p2sh_is_segwit=False):
             decoded_tx,
             block_index,
             p2sh_is_segwit=p2sh_is_segwit,
+            composing=composing,
         )
     elif util.enabled("multisig_addresses", block_index=block_index):  # Protocol change.
         return get_tx_info_new(
             db,
             decoded_tx,
             block_index,
+            composing=composing,
         )
     else:
         return get_tx_info_legacy(decoded_tx, block_index)
@@ -604,7 +716,7 @@ def get_utxos_info(db, decoded_tx):
     ]
 
 
-def get_tx_info(db, decoded_tx, block_index):
+def get_tx_info(db, decoded_tx, block_index, composing=False):
     """Get the transaction info. Returns normalized None data for DecodeError and BTCOnlyError."""
     if util.enabled("utxo_support", block_index=block_index):
         # utxos_info is a space-separated list of UTXOs, last element is the destination,
@@ -618,7 +730,7 @@ def get_tx_info(db, decoded_tx, block_index):
         utxos_info = []
     try:
         source, destination, btc_amount, fee, data, dispensers_outs = _get_tx_info(
-            db, decoded_tx, block_index
+            db, decoded_tx, block_index, composing=composing
         )
         return source, destination, btc_amount, fee, data, dispensers_outs, utxos_info
     except DecodeError as e:  # noqa: F841

counterparty-core/counterpartycore/lib/message_type.py

@@ -96,8 +96,12 @@
         return "unknown"
 
     if message_type_id == messages.utxo.ID:
-        message_data = messages.utxo.unpack(message, return_dict=True)
-        if util.is_utxo_format(message_data["source"]):
-            return "detach"
-        return "attach"
+        try:
+            message_data = messages.utxo.unpack(message, return_dict=True)
+            if util.is_utxo_format(message_data["source"]):
+                return "detach"
+            return "attach"
+        except Exception:
+            return "unknown"
+
     return TRANSACTION_TYPE_BY_ID.get(message_type_id, "unknown")

counterparty-core/counterpartycore/test/deserialize_test.py

@@ -48,6 +48,7 @@ def test_deserialize():
         "vtxinwit": [],
         "lock_time": 0,
         "tx_hash": "54cc399879446c4eaa7774bb764b319a2680709f99704ce60344587f49ff97e8",
+        "tx_id": "54cc399879446c4eaa7774bb764b319a2680709f99704ce60344587f49ff97e8",
         "__data__": "0100000001db3acf37743ac015808f7911a88761530c801819b3b907340aa65dfb6d98ce24030000006a473044022002961f4800cb157f8c0913084db0ee148fa3e1130e0b5e40c3a46a6d4f83ceaf02202c3dd8e631bf24f4c0c5341b3e1382a27f8436d75f3e0a095915995b0bf7dc8e01210395c223fbf96e49e5b9e06a236ca7ef95b10bf18c074bd91a5942fc40360d0b68fdffffff040000000000000000536a4c5058325bd61325dc633fadf05bec9157c23106759cee40954d39d9dbffc17ec5851a2d1feb5d271da422e0e24c7ae8ad29d2eeabf7f9ca3de306bd2bc98e2a39e47731aa000caf400053000c1283000149c8000000000000001976a91462bef4110f98fdcb4aac3c1869dbed9bce8702ed88acc80000000000000017a9144317f779c0a2ccf8f6bc3d440bd9e536a5bff75287fa3e5100000000001976a914bf2646b8ba8b4a143220528bde9c306dac44a01c88ac00000000",
     }