[electrum] implement TxInput.to_coin_dict and TxInput.is_complete

Summary:
Move the serialization to the legacy dict format into TxInput

This opens the way to using TxInput internally in the Transaction class while still being able to convert to the legacy dict where it is really needed (saving coins to JSON)

Add tests related to input completeness (incomplete inputs have a value and a xpub)
Depends on D14459

Test Plan: `python test_runner.py`

Reviewers: #bitcoin_abc, Fabien

Reviewed By: #bitcoin_abc, Fabien

Differential Revision: https://reviews.bitcoinabc.org/D14461

electrumabc/tests/test_transaction.py

@@ -808,6 +808,7 @@ def _deser_test(
         expected_sigs: List[str],
         expected_pubkeys: Optional[List[str]] = None,
         expected_address: Optional[Address] = None,
+        expected_value: Optional[int] = None,
     ):
         tx = transaction.Transaction(tx_hex)
         input_dict = tx.inputs()[0]
@@ -851,6 +852,18 @@ def _deser_test(
         # None (p2pk and coinbase) or an address instance
         self.assertEqual(txinput.address, expected_address)
 
+        self.assertEqual(txinput.is_complete(), tx.is_txin_complete(input_dict))
+        if not txinput.is_complete():
+            self.assertIsNotNone(txinput.get_value())
+            self.assertEqual(txinput.get_value(), input_dict["value"])
+            self.assertEqual(txinput.get_value(), expected_value)
+
+            self.assertTrue(any(xpub[0] == 0xFF for xpub in txinput.x_pubkeys))
+            self.assertNotEqual(txinput.pubkeys, txinput.x_pubkeys)
+        else:
+            # Signed transactions don't bother storing the xpub and derivation path
+            self.assertEqual(txinput.pubkeys, txinput.x_pubkeys)
+
     def test_multisig_p2sh_deserialization(self):
         self._deser_test(
             tx_hex="0100000001b98d550fa331da21038952d6931ffd3607c440ab2985b75477181b577de118b10b000000fdfd0000483045022100a26ea637a6d39aa27ea7a0065e9691d477e23ad5970b5937a9b06754140cf27102201b00ed050b5c468ee66f9ef1ff41dfb3bd64451469efaab1d4b56fbf92f9df48014730440220080421482a37cc9a98a8dc3bf9d6b828092ad1a1357e3be34d9c5bbdca59bb5f02206fa88a389c4bf31fa062977606801f3ea87e86636da2625776c8c228bcd59f8a014c69522102420e820f71d17989ed73c0ff2ec1c1926cf989ad6909610614ee90cf7db3ef8721036eae8acbae031fdcaf74a824f3894bf54881b42911bd3ad056ea59a33ffb3d312103752669b75eb4dc0cca209af77a59d2c761cbb47acc4cf4b316ded35080d92e8253aeffffffff0101ac3a00000000001976a914a6b6bcc85975bf6a01a0eabb2ac97d5a418223ad88ac00000000",
@@ -915,6 +928,7 @@ def test_multisig_incomplete(self):
             expected_address=Address.from_string(
                 "ecash:ppfhzqryfq5u9y3ccqw3j9qaa9rsyz746sar20zk99"
             ),
+            expected_value=100_900,
         )
 
 

electrumabc/transaction.py

@@ -30,7 +30,7 @@
 import struct
 import warnings
 from io import BytesIO
-from typing import List, NamedTuple, Optional, Tuple, Union
+from typing import Any, Dict, List, NamedTuple, Optional, Tuple, Union
 
 import ecdsa
 
@@ -159,11 +159,20 @@ def __str__(self):
 
 
 class TxInput:
-    def __init__(self, outpoint: OutPoint, scriptsig: bytes, sequence: int):
+    def __init__(
+        self,
+        outpoint: OutPoint,
+        scriptsig: bytes,
+        sequence: int,
+        value: Optional[int] = None,
+    ):
         self.outpoint = outpoint
         self.scriptsig = scriptsig
         self.sequence = sequence
 
+        # Must be defined for partially signed inputs (needed for signing)
+        self._value: Optional[int] = value
+
         # Properties that are lazily computed on demand
         self._x_pubkeys: Optional[List[bytes]] = None
         self._pubkeys: Optional[List[bytes]] = None
@@ -172,6 +181,12 @@ def __init__(self, outpoint: OutPoint, scriptsig: bytes, sequence: int):
         self._signatures: List[Optional[bytes]] = []
         self._address: Optional[Address] = None
 
+    def set_value(self, value: int):
+        self._value = value
+
+    def get_value(self) -> Optional[int]:
+        return self._value
+
     def size(self):
         scriptsig_nbytes = len(self.scriptsig)
         assert scriptsig_nbytes <= MAX_SCRIPT_SIZE
@@ -304,6 +319,69 @@ def address(self) -> Optional[Address]:
             self.parse_scriptsig()
         return self._address
 
+    def is_complete(self) -> bool:
+        if self.type == ScriptType.coinbase or self.num_sig == 0:
+            return True
+        non_null_signatures = list(filter(None, self.signatures))
+        return len(non_null_signatures) == self.num_sig
+
+    def to_coin_dict(self) -> Dict[str, Any]:
+        """Return a legacy coin dict for this TxInput"""
+        d = {
+            "prevout_hash": self.outpoint.txid.get_hex(),
+            "prevout_n": self.outpoint.n,
+            "sequence": self.sequence,
+            # default address, updated later for p2pkh and p2sh
+            "address": UnknownAddress(),
+            "scriptSig": bh2u(self.scriptsig),
+        }
+        if self.is_coinbase():
+            d["type"] = "coinbase"
+            return d
+
+        d["x_pubkeys"] = []
+        d["pubkeys"] = []
+        d["signatures"] = {}
+        d["address"] = None
+        d["num_sig"] = 0
+        try:
+            self.parse_scriptsig()
+        except Exception as e:
+            print_error(
+                f"{__name__}: Failed to parse tx input {self.outpoint}, probably a "
+                f"p2sh (non multisig?). Exception was: {repr(e)}"
+            )
+            # that whole heuristic codepath is fragile; just ignore it when it dies.
+            # failing tx examples:
+            # 1c671eb25a20aaff28b2fa4254003c201155b54c73ac7cf9c309d835deed85ee
+            # 08e1026eaf044127d7103415570afd564dfac3131d7a5e4b645f591cd349bb2c
+            # override these once more just to make sure
+            d["address"] = UnknownAddress()
+            d["type"] = "unknown"
+            return d
+
+        d["type"] = self.type.name
+        d["signatures"] = [
+            (sig.hex() if sig is not None else None) for sig in self.signatures
+        ]
+        d["num_sig"] = self.num_sig
+        if self.type == ScriptType.p2pk:
+            # TODO: remove this entirely if unused
+            d["x_pubkeys"] = ["(pubkey)"]
+            d["pubkeys"] = ["(pubkey)"]
+        else:
+            d["x_pubkeys"] = [xpub.hex() for xpub in self.x_pubkeys]
+            d["pubkeys"] = [pub.hex() for pub in self.pubkeys]
+            d["address"] = self.address
+        if self.type == ScriptType.p2sh:
+            d["redeemScript"] = self.scriptsig
+        if not self.is_complete():
+            del d["scriptSig"]
+            # The amount is needed for signing, in case of partially signed inputs.
+            d["value"] = self.get_value()
+            assert d["value"] is not None
+        return d
+
 
 class BCDataStream(object):
     def __init__(self):
@@ -635,61 +713,20 @@ def get_address_from_output_script(
 
 
 def parse_input(vds):
-    d = {}
     prevout_hash = bitcoin.hash_encode(vds.read_bytes(32))
     prevout_n = vds.read_uint32()
     scriptSig = vds.read_bytes(vds.read_compact_size())
     sequence = vds.read_uint32()
-    d["prevout_hash"] = prevout_hash
-    d["prevout_n"] = prevout_n
-    d["sequence"] = sequence
-    d["address"] = UnknownAddress()
+
     txin = TxInput(
         OutPoint(UInt256.from_hex(prevout_hash), prevout_n), scriptSig, sequence
     )
-    d["type"] = txin.type.name
-    d["scriptSig"] = bh2u(scriptSig)
-    if txin.is_coinbase():
-        return d
-    d["x_pubkeys"] = []
-    d["pubkeys"] = []
-    d["signatures"] = {}
-    d["address"] = None
-    d["num_sig"] = 0
-    try:
-        txin.parse_scriptsig()
-    except Exception as e:
-        print_error(
-            "{}: Failed to parse tx input {}:{}, probably a p2sh (non multisig?)."
-            " Exception was: {}".format(__name__, prevout_hash, prevout_n, repr(e))
-        )
-        # that whole heuristic codepath is fragile; just ignore it when it dies.
-        # failing tx examples:
-        # 1c671eb25a20aaff28b2fa4254003c201155b54c73ac7cf9c309d835deed85ee
-        # 08e1026eaf044127d7103415570afd564dfac3131d7a5e4b645f591cd349bb2c
-        # override these once more just to make sure
-        d["address"] = UnknownAddress()
-        d["type"] = "unknown"
-    else:
-        d["signatures"] = [
-            (sig.hex() if sig is not None else None) for sig in txin.signatures
-        ]
-        d["num_sig"] = txin.num_sig
-        if txin.type == bitcoin.ScriptType.p2pk:
-            # TODO: remove this entirely if unused
-            d["x_pubkeys"] = ["(pubkey)"]
-            d["pubkeys"] = ["(pubkey)"]
-        else:
-            d["x_pubkeys"] = [xpub.hex() for xpub in txin.x_pubkeys]
-            d["pubkeys"] = [pub.hex() for pub in txin.pubkeys]
-            d["address"] = txin.address
-        if txin.type == bitcoin.ScriptType.p2sh:
-            d["redeemScript"] = txin.scriptsig
-    if not Transaction.is_txin_complete(d):
-        del d["scriptSig"]
-        # The amount is needed for signing, in case of partially signed inputs.
-        d["value"] = vds.read_uint64()
-    return d
+    if not txin.is_complete():
+        # This is a partially signed transaction in the custom Electrum format, the
+        # amount is appended to the serialized input.
+        txin.set_value(vds.read_uint64())
+
+    return txin.to_coin_dict()
 
 
 def parse_output(vds: BCDataStream, i: int):
@@ -788,6 +825,7 @@ def txinputs(self, estimate_size: bool = False) -> List[TxInput]:
                     self.input_script(inp, estimate_size, self._sign_schnorr)
                 ),
                 inp.get("sequence", DEFAULT_TXIN_SEQUENCE),
+                inp.get("value", None),
             )
             for inp in self.inputs()
         ]