Make sure the fee and cost specified in a NewTransaction match the ones from validating its spend bundle.

Author: AmineKhaldi

chia/_tests/core/full_node/test_full_node.py

@@ -3362,18 +3362,21 @@ async def test_pending_tx_cache_retry_on_new_peak(
 @pytest.mark.anyio
 @pytest.mark.parametrize("mismatch_cost", [True, False])
 @pytest.mark.parametrize("mismatch_fee", [True, False])
+@pytest.mark.parametrize("tx_already_seen", [True, False])
 async def test_ban_for_mismatched_tx_cost_fee(
     setup_two_nodes_fixture: tuple[list[FullNodeSimulator], list[tuple[WalletNode, ChiaServer]], BlockTools],
     self_hostname: str,
     mismatch_cost: bool,
     mismatch_fee: bool,
+    tx_already_seen: bool,
 ) -> None:
     """
     Tests that a peer gets banned if it sends a `NewTransaction` message with a
     cost and/or fee that doesn't match the transaction's validation cost/fee.
-    We setup two full nodes with the test transaction as already seen, and we
-    check its validation cost and fee against the ones specified in the
-    `NewTransaction` message.
+    We setup two full nodes, and with `tx_already_seen` we control whether the
+    first full node has this transaction already or it needs to request it.
+    In both cases we check the transaction's validation cost and fee against
+    the ones specified in the `NewTransaction` message.
     """
     nodes, _, bt = setup_two_nodes_fixture
     full_node_1, full_node_2 = nodes
@@ -3384,17 +3387,26 @@ async def test_ban_for_mismatched_tx_cost_fee(
     ws_con_2 = next(iter(server_2.all_connections.values()))
     wallet = WalletTool(test_constants)
     wallet_ph = wallet.get_new_puzzlehash()
+    # If we're covering that the first full node has this transaction already
+    # we must add it accordingly, otherwise we'll add it to the second node so
+    # that the first node requests it, reacting to the NewTransaction message.
+    if tx_already_seen:
+        node = full_node_1.full_node
+        ws_con = ws_con_1
+    else:
+        node = full_node_2.full_node
+        ws_con = ws_con_2
     blocks = bt.get_consecutive_blocks(
         3, guarantee_transaction_block=True, farmer_reward_puzzle_hash=wallet_ph, pool_reward_puzzle_hash=wallet_ph
     )
     for block in blocks:
-        await full_node_1.full_node.add_block(block)
+        await node.add_block(block)
     # Create a transaction and add it to the relevant full node's mempool
     coin = blocks[-1].get_included_reward_coins()[0]
     sb = wallet.generate_signed_transaction(uint64(42), wallet_ph, coin)
     sb_name = sb.name()
-    await full_node_1.full_node.add_transaction(sb, sb_name, ws_con_1)
-    mempool_item = full_node_1.full_node.mempool_manager.get_mempool_item(sb_name)
+    await node.add_transaction(sb, sb_name, ws_con)
+    mempool_item = node.mempool_manager.get_mempool_item(sb_name)
     assert mempool_item is not None
     # Now send a NewTransaction with a cost and/or fee mismatch from the second
     # full node.
@@ -3407,6 +3419,14 @@ async def test_ban_for_mismatched_tx_cost_fee(
     ws_con_1.peer_info = PeerInfo(full_node_2_ip, ws_con_1.peer_info.port)
     # Send the NewTransaction message from the second node to the first
     await ws_con_2.send_message(msg)
+    if not tx_already_seen:
+        # When the first full node receives the NewTransaction message and it
+        # hasen't seen the transaction before, it will issue a transaction
+        # request. We need to wait until it receives the transaction and add it
+        # to its mempool.
+        await time_out_assert(
+            30, lambda: full_node_1.full_node.mempool_manager.get_mempool_item(mempool_item.name) is not None
+        )
     # Make sure the first full node has banned the second as the item it has
     # already seen has a different validation cost and/or fee than the one from
     # the NewTransaction message.

chia/full_node/full_node.py

@@ -68,7 +68,7 @@
 from chia.protocols.full_node_protocol import RequestBlocks, RespondBlock, RespondBlocks, RespondSignagePoint
 from chia.protocols.outbound_message import Message, NodeType, make_msg
 from chia.protocols.protocol_message_types import ProtocolMessageTypes
-from chia.protocols.protocol_timing import CONSENSUS_ERROR_BAN_SECONDS
+from chia.protocols.protocol_timing import CONSENSUS_ERROR_BAN_SECONDS, RATE_LIMITER_BAN_SECONDS
 from chia.protocols.shared_protocol import Capability
 from chia.protocols.wallet_protocol import CoinStateUpdate, RemovedMempoolItem
 from chia.rpc.rpc_server import StateChangedProtocol
@@ -498,7 +498,9 @@ def _set_state_changed_callback(self, callback: StateChangedProtocol) -> None:
     async def _handle_one_transaction(self, entry: TransactionQueueEntry) -> None:
         peer = entry.peer
         try:
-            inc_status, err = await self.add_transaction(entry.transaction, entry.spend_name, peer, entry.test)
+            inc_status, err = await self.add_transaction(
+                entry.transaction, entry.spend_name, peer, entry.test, entry.peers_with_tx
+            )
             entry.done.set((inc_status, err))
         except asyncio.CancelledError:
             error_stack = traceback.format_exc()
@@ -2761,7 +2763,12 @@ async def add_end_of_sub_slot(
         return None, False
 
     async def add_transaction(
-        self, transaction: SpendBundle, spend_name: bytes32, peer: Optional[WSChiaConnection] = None, test: bool = False
+        self,
+        transaction: SpendBundle,
+        spend_name: bytes32,
+        peer: Optional[WSChiaConnection] = None,
+        test: bool = False,
+        peers_with_tx: dict[bytes32, tuple[uint64, uint64]] = {},
     ) -> tuple[MempoolInclusionStatus, Optional[Err]]:
         if self.sync_store.get_sync_mode():
             return MempoolInclusionStatus.FAILED, Err.NO_TRANSACTIONS_WHILE_SYNCING
@@ -2810,10 +2817,23 @@ async def add_transaction(
                 f"{self.mempool_manager.mempool.total_mempool_cost() / 5000000}"
             )
 
-            # Only broadcast successful transactions, not pending ones. Otherwise it's a DOS
-            # vector.
             mempool_item = self.mempool_manager.get_mempool_item(spend_name)
             assert mempool_item is not None
+            # Now that we validated this transaction, check what fees and
+            # costs the peers have advertised for it.
+            for peer_id, (advertised_fee, advertised_cost) in peers_with_tx.items():
+                if advertised_fee == mempool_item.fee and advertised_cost == mempool_item.cost:
+                    continue
+                if peer_id not in self.server.all_connections:
+                    continue
+                self.log.warning(
+                    f"Banning peer {peer_id}. Sent us a new tx {spend_name} with mismatch "
+                    f"on cost {advertised_cost} vs validation cost {mempool_item.cost} and/or "
+                    f"fee {advertised_fee} vs {mempool_item.fee}."
+                )
+                await self.server.all_connections[peer_id].close(RATE_LIMITER_BAN_SECONDS)
+            # Only broadcast successful transactions, not pending ones. Otherwise it's a DOS
+            # vector.
             await self.broadcast_removed_tx(info.removals)
             await self.broadcast_added_tx(mempool_item, current_peer=peer)
 

chia/full_node/full_node_api.py

@@ -1,6 +1,7 @@
 from __future__ import annotations
 
 import asyncio
+import copy
 import logging
 import time
 import traceback
@@ -89,6 +90,7 @@ async def tx_request_and_timeout(full_node: FullNode, transaction_id: bytes32, t
     receive it or timeout.
     """
     counter = 0
+    peers_with_tx = copy.copy(full_node.full_node_store.peers_with_tx.get(transaction_id, {}))
     try:
         while True:
             # Limit to asking a few peers, it's possible that this tx got included on chain already
@@ -98,10 +100,9 @@ async def tx_request_and_timeout(full_node: FullNode, transaction_id: bytes32, t
                 break
             if transaction_id not in full_node.full_node_store.peers_with_tx:
                 break
-            peers_with_tx: set[bytes32] = full_node.full_node_store.peers_with_tx[transaction_id]
             if len(peers_with_tx) == 0:
                 break
-            peer_id = peers_with_tx.pop()
+            peer_id, _ = peers_with_tx.popitem()
             assert full_node.server is not None
             if peer_id not in full_node.server.all_connections:
                 continue
@@ -111,7 +112,7 @@ async def tx_request_and_timeout(full_node: FullNode, transaction_id: bytes32, t
             await random_peer.send_message(msg)
             await asyncio.sleep(5)
             counter += 1
-            if full_node.mempool_manager.seen(transaction_id):
+            if full_node.mempool_manager.get_mempool_item(transaction_id) is not None:
                 break
     except asyncio.CancelledError:
         pass
@@ -228,21 +229,22 @@ async def new_transaction(
             # If there's current pending request just add this peer to the set of peers that have this tx
             if transaction.transaction_id in self.full_node.full_node_store.pending_tx_request:
                 if transaction.transaction_id in self.full_node.full_node_store.peers_with_tx:
-                    current_set = self.full_node.full_node_store.peers_with_tx[transaction.transaction_id]
-                    if peer.peer_node_id in current_set:
+                    current_map = self.full_node.full_node_store.peers_with_tx[transaction.transaction_id]
+                    if peer.peer_node_id in current_map:
                         return None
-                    current_set.add(peer.peer_node_id)
+                    current_map[peer.peer_node_id] = (transaction.fees, transaction.cost)
                     return None
                 else:
-                    new_set = set()
-                    new_set.add(peer.peer_node_id)
-                    self.full_node.full_node_store.peers_with_tx[transaction.transaction_id] = new_set
+                    self.full_node.full_node_store.peers_with_tx[transaction.transaction_id] = {
+                        peer.peer_node_id: (transaction.fees, transaction.cost)
+                    }
                     return None
 
             self.full_node.full_node_store.pending_tx_request[transaction.transaction_id] = peer.peer_node_id
-            new_set = set()
-            new_set.add(peer.peer_node_id)
-            self.full_node.full_node_store.peers_with_tx[transaction.transaction_id] = new_set
+            self.full_node.full_node_store.peers_with_tx[transaction.transaction_id] = {
+                peer.peer_node_id: (transaction.fees, transaction.cost)
+            }
+
             task_id: bytes32 = bytes32.secret()
             fetch_task = create_referenced_task(
                 tx_request_and_timeout(self.full_node, transaction.transaction_id, task_id)
@@ -282,13 +284,15 @@ async def respond_transaction(
         spend_name = std_hash(tx_bytes)
         if spend_name in self.full_node.full_node_store.pending_tx_request:
             self.full_node.full_node_store.pending_tx_request.pop(spend_name)
+        peers_with_tx = {}
         if spend_name in self.full_node.full_node_store.peers_with_tx:
-            self.full_node.full_node_store.peers_with_tx.pop(spend_name)
+            peers_with_tx = self.full_node.full_node_store.peers_with_tx.pop(spend_name)
 
         # TODO: Use fee in priority calculation, to prioritize high fee TXs
         try:
             await self.full_node.transaction_queue.put(
-                TransactionQueueEntry(tx.transaction, tx_bytes, spend_name, peer, test), peer.peer_node_id
+                TransactionQueueEntry(tx.transaction, tx_bytes, spend_name, peer, test, peers_with_tx),
+                peer.peer_node_id,
             )
         except TransactionQueueFull:
             pass  # we can't do anything here, the tx will be dropped. We might do something in the future.

chia/full_node/full_node_store.py

@@ -132,7 +132,7 @@ class FullNodeStore:
     recent_eos: LRUCache[bytes32, tuple[EndOfSubSlotBundle, float]]
 
     pending_tx_request: dict[bytes32, bytes32]  # tx_id: peer_id
-    peers_with_tx: dict[bytes32, set[bytes32]]  # tx_id: set[peer_ids}
+    peers_with_tx: dict[bytes32, dict[bytes32, tuple[uint64, uint64]]]  # tx_id: dict[peer_id, (fee, cost)]
     tx_fetch_tasks: dict[bytes32, asyncio.Task[None]]  # Task id: task
     serialized_wp_message: Optional[Message]
     serialized_wp_message_tip: Optional[bytes32]

chia/full_node/tx_processing_queue.py

@@ -9,6 +9,7 @@
 
 from chia_rs import SpendBundle
 from chia_rs.sized_bytes import bytes32
+from chia_rs.sized_ints import uint64
 
 from chia.server.ws_connection import WSChiaConnection
 from chia.types.mempool_inclusion_status import MempoolInclusionStatus
@@ -56,6 +57,9 @@ class TransactionQueueEntry:
     spend_name: bytes32
     peer: Optional[WSChiaConnection] = field(compare=False)
     test: bool = field(compare=False)
+    # IDs of peers that advertised this transaction via new_transaction, along
+    # with their fee and cost.
+    peers_with_tx: dict[bytes32, tuple[uint64, uint64]] = field(default_factory=dict, compare=False)
     done: ValuedEvent[tuple[MempoolInclusionStatus, Optional[Err]]] = field(
         default_factory=ValuedEvent,
         compare=False,