Insert dummy signatures before needed fee calculation

counterparty-core/counterpartycore/lib/backend/bitcoind.py

@@ -118,7 +118,6 @@ def rpc_call(payload, retry=0):
 
 
 def is_api_request():
-    print("is_api_request", current_process().name)
     if current_process().name != "API":
         return False
     thread_name = current_thread().name

counterparty-core/counterpartycore/lib/composer.py

@@ -10,7 +10,7 @@
 from bitcoinutils.keys import P2pkhAddress, P2shAddress, P2wpkhAddress, PublicKey
 from bitcoinutils.script import Script, b_to_h
 from bitcoinutils.setup import setup
-from bitcoinutils.transactions import Transaction, TxInput, TxOutput
+from bitcoinutils.transactions import Transaction, TxInput, TxOutput, TxWitnessInput
 
 from counterpartycore.lib import (
     arc4,
@@ -546,9 +546,66 @@ def utxos_to_txins(utxos: list):
 #   Composition   #
 ##################
 
-
-def get_needed_fee(tx, satoshis_per_vbyte):
-    virtual_size = tx.get_vsize()
+DUMMY_DER_SIG = "3045" + "00" * 70
+DUMMY_REEDEM_SCRIPT = DUMMY_DER_SIG
+DUMMY_PUBKEY = "03" + 32 * "00"
+DUMMY_SCHNORR_SIG = "00" * 65
+OP_0 = "00"
+OP_PUSHBYTES_72 = "48"
+OP_PUSHBYTES_33 = "21"
+
+
+# dummies script_sig from https://learnmeabitcoin.com/technical/script/
+def get_dummy_script_sig(script_pub_key):
+    output_type = script.get_output_type(script_pub_key)
+    script_sig = None
+    if output_type == "P2PK":
+        script_sig = OP_PUSHBYTES_72 + DUMMY_DER_SIG
+    elif output_type == "P2PKH":
+        script_sig = OP_PUSHBYTES_72 + DUMMY_DER_SIG + OP_PUSHBYTES_33 + DUMMY_PUBKEY
+    elif output_type == "P2MS":
+        asm = script.script_to_asm(script_pub_key)
+        required_signatures = asm[0]
+        script_sig = OP_0 + (required_signatures * DUMMY_DER_SIG)
+    elif output_type == "P2SH":
+        script_sig = OP_0 + OP_PUSHBYTES_72 + DUMMY_DER_SIG + OP_PUSHBYTES_72 + DUMMY_REEDEM_SCRIPT
+    if script_sig is not None:
+        return Script.from_raw(script_sig)
+    return None
+
+
+def get_dummy_witness(script_pub_key):
+    output_type = script.get_output_type(script_pub_key)
+    witness = None
+    if output_type == "P2WPKH":
+        witness = [DUMMY_DER_SIG, DUMMY_PUBKEY]
+    elif output_type == "P2WSH":
+        witness = [
+            OP_PUSHBYTES_72 + DUMMY_DER_SIG + OP_PUSHBYTES_33 + DUMMY_PUBKEY,  # P2PKH unlock script
+            script_pub_key,
+        ]
+    elif output_type == "P2TR":
+        witness = [DUMMY_SCHNORR_SIG]
+    if witness is not None:
+        return TxWitnessInput(witness)
+    return None
+
+
+def generate_dummy_signed_tx(tx, selected_utxos):
+    dummy_signed_tx = Transaction.copy(tx)
+    for i, utxo in enumerate(selected_utxos):
+        dummy_script_sig = get_dummy_script_sig(utxo["script_pub_key"])
+        if dummy_script_sig is not None:
+            dummy_signed_tx.inputs[i].script_sig = dummy_script_sig
+        dummy_witness = get_dummy_witness(utxo["script_pub_key"])
+        if dummy_witness is not None:
+            dummy_signed_tx.witnesses.append(dummy_witness)
+    return dummy_signed_tx
+
+
+def get_needed_fee(tx, satoshis_per_vbyte, selected_utxos):
+    dummy_signed_tx = generate_dummy_signed_tx(tx, selected_utxos)
+    virtual_size = dummy_signed_tx.get_vsize()
     return satoshis_per_vbyte * virtual_size
 
 
@@ -625,7 +682,7 @@ def prepare_inputs_and_change(db, source, outputs, unspent_list, construct_param
             + [create_tx_output(change_amount, change_address, unspent_list, construct_params)],
             has_segwit=has_segwit,
         )
-        needed_fee = get_needed_fee(tx, sat_per_vbyte)
+        needed_fee = get_needed_fee(tx, sat_per_vbyte, selected_utxos)
         if max_fee is not None:
             needed_fee = min(needed_fee, max_fee)
         # if change is enough for needed fee, add change output and break

counterparty-core/counterpartycore/lib/script.py

@@ -584,6 +584,7 @@ def get_output_type(script_pub_key):
         return "P2WPKH"
     if len(asm) == 2 and asm[0] == b"\x01":
         return "P2TR"
+    return "UNKNOWN"
 
 
 def is_segwit_output(script_pub_key):

counterparty-core/counterpartycore/test/regtest/regtestnode.py

@@ -877,7 +877,96 @@ def test_asset_conservation(self):
         self.test_command("check-db")
 
     def test_fee_calculation(self):
-        pass
+        unsigned_tx = self.compose(
+            self.addresses[0],
+            "send",
+            {
+                "destination": self.addresses[1],
+                "quantity": 10,
+                "asset": "XCP",
+                "sat_per_vbyte": 1,
+                "verbose": True,
+            },
+        )["result"]
+        print("Unsigned transaction 1: ", unsigned_tx)
+
+        signed_tx = json.loads(
+            self.bitcoin_wallet(
+                "signrawtransactionwithwallet", unsigned_tx["rawtransaction"]
+            ).strip()
+        )["hex"]
+        transaction2 = Transaction.from_raw(signed_tx)
+
+        print("Fees: ", unsigned_tx["btc_fee"])
+        print("VSize After signing: ", transaction2.get_vsize())
+        assert unsigned_tx["btc_fee"] == transaction2.get_vsize()
+
+        unsigned_tx = self.compose(
+            self.addresses[0],
+            "send",
+            {
+                "destination": self.addresses[1],
+                "quantity": 10,
+                "asset": "XCP",
+                "sat_per_vbyte": 2,
+                "verbose": True,
+                "encoding": "multisig",
+            },
+        )["result"]
+        print("Unsigned transaction 2: ", unsigned_tx)
+
+        signed_tx = json.loads(
+            self.bitcoin_wallet(
+                "signrawtransactionwithwallet", unsigned_tx["rawtransaction"]
+            ).strip()
+        )["hex"]
+        transaction3 = Transaction.from_raw(signed_tx)
+
+        print("Fees: ", unsigned_tx["btc_fee"])
+        print("VSize After signing: ", transaction3.get_vsize())
+        assert unsigned_tx["btc_fee"] == transaction3.get_vsize() * 2
+
+        legacy_address = self.bitcoin_wallet("getnewaddress", WALLET_NAME, "legacy").strip()
+        self.send_transaction(
+            self.addresses[0],
+            "send",
+            {
+                "destination": legacy_address,
+                "quantity": 10,
+                "asset": "XCP",
+            },
+        )
+        self.bitcoin_wallet("sendtoaddress", legacy_address, 0.1)
+        self.mine_blocks(1)
+
+        unsigned_tx = self.compose(
+            legacy_address,
+            "send",
+            {
+                "destination": self.addresses[1],
+                "quantity": 5,
+                "asset": "XCP",
+                "sat_per_vbyte": 3,
+                "verbose": True,
+            },
+        )["result"]
+        print("Unsigned transaction 3: ", unsigned_tx)
+
+        signed_tx = json.loads(
+            self.bitcoin_wallet(
+                "signrawtransactionwithwallet", unsigned_tx["rawtransaction"]
+            ).strip()
+        )["hex"]
+        print("signed_tx: ", signed_tx)
+        transaction4 = Transaction.from_raw(signed_tx)
+
+        print("Fees: ", unsigned_tx["btc_fee"])
+        print("VSize After signing: ", transaction4.get_vsize())
+        assert (
+            transaction4.get_vsize() * 3 - 3
+            <= unsigned_tx["btc_fee"]
+            <= transaction4.get_vsize() * 3 + 3
+        )
 
 
 class RegtestNodeThread(threading.Thread):

counterparty-core/counterpartycore/test/regtest/testscenarios.py

@@ -394,9 +394,6 @@ def run_scenarios(serve=False, wsgi_server="gunicorn"):
         while not regtest_node_thread.ready():
             time.sleep(1)
 
-        # regtest_node_thread.node.test_transaction_chaining()
-        # raise KeyboardInterrupt
-
         context = {}
 
         check_api_v1(regtest_node_thread.node)
@@ -434,6 +431,8 @@ def run_scenarios(serve=False, wsgi_server="gunicorn"):
             regtest_node_thread.node.test_invalid_detach()
             print("Testing transaction chaining...")
             regtest_node_thread.node.test_transaction_chaining()
+            print("Testing fee calculation...")
+            regtest_node_thread.node.test_fee_calculation()
             print("Tesing asset conservation checking...")
             regtest_node_thread.node.test_asset_conservation()
             print("Tesing reparse...")