From 715953bc5a20a7bac4a2462368d513cc9ef7f5aa Mon Sep 17 00:00:00 2001 From: Chris Marslender Date: Thu, 9 Jul 2026 12:32:45 -0500 Subject: [PATCH] Checkpoint Merge (#21106) Co-authored-by: Almog De Paz Co-authored-by: Amine Khaldi Co-authored-by: Earle Lowe Co-authored-by: Zachary Brown --- .coveragerc | 1 + .../_tests/core/data_layer/test_data_store.py | 4 +- .../full_node/full_sync/test_full_sync.py | 250 +++++++++- .../core/full_node/stores/test_sync_store.py | 73 ++- .../full_node/test_add_prevalidated_blocks.py | 4 +- chia/_tests/core/full_node/test_full_node.py | 126 ++++- .../_tests/core/server/test_node_discovery.py | 2 +- chia/_tests/util/test_limited_semaphore.py | 27 ++ chia/_tests/util/test_priority_mutex.py | 43 ++ .../_tests/wallet/db_wallet/test_dl_offers.py | 68 ++- chia/_tests/wallet/test_wallet_node.py | 429 +++++++++++++++++- .../wallet/test_wallet_state_manager.py | 206 +++++++++ chia/_tests/weight_proof/test_weight_proof.py | 16 +- chia/consensus/multiprocess_validation.py | 27 +- chia/data_layer/data_layer_wallet.py | 13 + chia/full_node/full_node.py | 56 ++- chia/full_node/full_node_api.py | 15 +- chia/full_node/sync_store.py | 15 + chia/util/initial-config.yaml | 6 + chia/util/limited_semaphore.py | 3 + chia/util/priority_mutex.py | 7 +- chia/wallet/util/peer_request_cache.py | 5 +- chia/wallet/wallet_node.py | 65 ++- chia/wallet/wallet_state_manager.py | 16 + 24 files changed, 1423 insertions(+), 54 deletions(-) diff --git a/.coveragerc b/.coveragerc index c20cff58d2..1d672868b3 100644 --- a/.coveragerc +++ b/.coveragerc @@ -18,6 +18,7 @@ source = [report] precision = 1 exclude_also = + pragma: no cover abc\.abstractmethod typing\.overload ^\s*\.\.\.\s*$ diff --git a/chia/_tests/core/data_layer/test_data_store.py b/chia/_tests/core/data_layer/test_data_store.py index fbc5aa261b..40c158c112 100644 --- a/chia/_tests/core/data_layer/test_data_store.py +++ b/chia/_tests/core/data_layer/test_data_store.py @@ -2726,7 +2726,9 @@ async def test_manage_kv_files( @pytest.mark.anyio async def test_merkle_blob_cache_isolation_across_stores(tmp_path: Path) -> None: - """SEC-470: Verify two stores with identical content don't share cached MerkleBlobs.""" + """ + Verify two stores with identical content don't share cached MerkleBlobs + """ merkle_blobs_path = tmp_path / "merkle" merkle_blobs_path.mkdir() kv_blobs_path = tmp_path / "kv" diff --git a/chia/_tests/core/full_node/full_sync/test_full_sync.py b/chia/_tests/core/full_node/full_sync/test_full_sync.py index ed0d2c81c2..91a651c334 100644 --- a/chia/_tests/core/full_node/full_sync/test_full_sync.py +++ b/chia/_tests/core/full_node/full_sync/test_full_sync.py @@ -8,12 +8,13 @@ from typing import Any, cast import pytest from chia_rs import ConsensusConstants, FullBlock, SubEpochSummary from chia_rs.sized_bytes import bytes32 -from chia_rs.sized_ints import uint16, uint32, uint64 +from chia_rs.sized_ints import uint16, uint32, uint64, uint128 from chia._tests.conftest import ConsensusMode from chia._tests.core.node_height import node_height_between, node_height_exactly from chia._tests.util.time_out_assert import time_out_assert from chia.full_node.full_node_api import FullNodeAPI +from chia.full_node.sync_store import Peak from chia.protocols import full_node_protocol from chia.protocols.shared_protocol import Capability from chia.server.server import ChiaServer @@ -515,3 +516,250 @@ async def test_bad_peak_cache_invalidation( block = blocks[-1] full_node_1.full_node.add_to_bad_peak_cache(block.header_hash, block.height) assert len(full_node_1.full_node.bad_peak_cache) == 1 + + +# Tests below cover the inflated-peak-weight regression: detection of peers +# that advertise an inflated peak weight via NewPeak. The defense lives in +# `_sync()`'s peer-confirmation gather loop: if the actual block weight +# returned via RequestBlock differs from the advertised peak weight, every +# peer that advertised that exact peak via NewPeak is banned. Two defining +# properties of the fix: +# +# 1. Banning happens *before* `request_validate_wp()` runs, so the two +# negative tests below monkeypatch `request_validate_wp` to a sentinel and +# assert it was never reached. On unpatched code, control still reaches +# `request_validate_wp` and the eventual ban happens via the weight-proof +# timeout / mismatch path, so this assertion distinguishes patched from +# unpatched behavior. +# +# 2. The set of peers to ban is snapshotted at peak-selection time (via +# `SyncStore.get_advertisers_of_peak`). A peer that overwrites its +# `peer_to_peak` entry with a fresh `NewPeak` during the peer-confirmation +# round-trip cannot escape banning. See +# `test_long_sync_advertiser_overwriting_peer_to_peak_is_still_banned`. + + +async def _connect_with_quiet_node_2( + server_1: ChiaServer, + server_2: ChiaServer, + full_node_2: FullNodeAPI, + self_hostname: str, + monkeypatch: pytest.MonkeyPatch, +) -> None: + """Connect node_2 to node_1 and silence node_2's NewPeak handler so the + auto-broadcast from node_1 cannot overwrite the peak entries the test + injects directly into node_2's sync_store.""" + + async def noop_new_peak(*_args: object, **_kwargs: object) -> None: + return None + + monkeypatch.setattr(full_node_2.full_node, "new_peak", noop_new_peak) + await server_2.start_client(PeerInfo(self_hostname, server_1.get_port()), full_node_2.full_node.on_connect) + + async def connected() -> bool: + return server_1.node_id in full_node_2.full_node.server.all_connections + + await time_out_assert(10, connected) + + +@pytest.mark.anyio +async def test_long_sync_advertised_weight_lie_bans_advertiser( + two_nodes: tuple[FullNodeAPI, FullNodeAPI, ChiaServer, ChiaServer, BlockTools], + default_1000_blocks: list[FullBlock], + self_hostname: str, + monkeypatch: pytest.MonkeyPatch, +) -> None: + full_node_1, full_node_2, server_1, server_2, _bt = two_nodes + + for block in default_1000_blocks[:200]: + await full_node_1.full_node.add_block(block) + real_peak = full_node_1.full_node.blockchain.get_peak() + assert real_peak is not None + + await _connect_with_quiet_node_2(server_1, server_2, full_node_2, self_hostname, monkeypatch) + full_node_2.full_node.config["max_sync_wait"] = 0 + + # Frame node_1: claim it advertised the real header_hash with a wildly + # inflated weight via NewPeak. Also stage two stale entries so we can verify + # the helper bans only peers whose (header_hash, weight) actually match. + inflated_weight = uint128(real_peak.weight + 10**12) + full_node_2.full_node.sync_store.peer_has_block( + real_peak.header_hash, server_1.node_id, inflated_weight, real_peak.height, True + ) + disconnected_liar = bytes32(b"\x11" * 32) + full_node_2.full_node.sync_store.peer_has_block( + real_peak.header_hash, disconnected_liar, inflated_weight, real_peak.height, True + ) + bystander = bytes32(b"\x22" * 32) + full_node_2.full_node.sync_store.peer_has_block( + real_peak.header_hash, bystander, real_peak.weight, real_peak.height, True + ) + + # Fail loudly if control ever reaches the weight-proof request. On unpatched + # code, the ban happens via that path (after the WP response/timeout); the + # fix's defining property is that we bail at peer-confirmation instead. + reached_wp_validation = False + + async def fail_if_wp_called(*_args: object, **_kwargs: object) -> None: + nonlocal reached_wp_validation # pragma: no cover (sentinel; only fires on unpatched code) + reached_wp_validation = True # pragma: no cover + raise AssertionError( # pragma: no cover + "request_validate_wp must not be reached when a weight lie is detected" + ) + + monkeypatch.setattr(full_node_2.full_node, "request_validate_wp", fail_if_wp_called) + + await full_node_2.full_node._sync() + + # node_1 confirmed the block with its real (smaller) weight, exposing the + # injected lie. node_1 must be banned; the bystander entry advertising the + # honest weight must remain untouched. + async def liar_disconnected() -> bool: + return server_1.node_id not in full_node_2.full_node.server.all_connections + + await time_out_assert(10, liar_disconnected) + assert not reached_wp_validation + assert full_node_2.full_node.blockchain.get_peak() is None + assert bystander in full_node_2.full_node.sync_store.peer_to_peak + assert full_node_2.full_node.sync_store.peer_to_peak[bystander].weight == real_peak.weight + + +@pytest.mark.anyio +async def test_long_sync_no_peer_confirms_peak_bans_advertiser( + two_nodes: tuple[FullNodeAPI, FullNodeAPI, ChiaServer, ChiaServer, BlockTools], + default_1000_blocks: list[FullBlock], + self_hostname: str, + monkeypatch: pytest.MonkeyPatch, +) -> None: + full_node_1, full_node_2, server_1, server_2, _bt = two_nodes + + for block in default_1000_blocks[:200]: + await full_node_1.full_node.add_block(block) + + await _connect_with_quiet_node_2(server_1, server_2, full_node_2, self_hostname, monkeypatch) + full_node_2.full_node.config["max_sync_wait"] = 0 + + # node_1 claims a peak that does not exist on the network, so no peer can + # return a matching block via RequestBlock. + fabricated_hash = std_hash(b"inflated-peak-no-confirm") + fabricated_height = uint32(50_000) + fabricated_weight = uint128(10**15) + full_node_2.full_node.sync_store.peer_has_block( + fabricated_hash, server_1.node_id, fabricated_weight, fabricated_height, True + ) + + # Fail loudly if control ever reaches the weight-proof request. On + # unpatched code, the ban happens via that path (after the WP timeout); + # the fix bails at peer-confirmation when no peer can serve the block. + reached_wp_validation = False + + async def fail_if_wp_called(*_args: object, **_kwargs: object) -> None: + nonlocal reached_wp_validation # pragma: no cover (sentinel; only fires on unpatched code) + reached_wp_validation = True # pragma: no cover + raise AssertionError( # pragma: no cover + "request_validate_wp must not be reached when no peer confirms the peak" + ) + + monkeypatch.setattr(full_node_2.full_node, "request_validate_wp", fail_if_wp_called) + + await full_node_2.full_node._sync() + + async def liar_disconnected() -> bool: + return server_1.node_id not in full_node_2.full_node.server.all_connections + + await time_out_assert(10, liar_disconnected) + assert not reached_wp_validation + assert full_node_2.full_node.blockchain.get_peak() is None + + +@pytest.mark.anyio +async def test_long_sync_honest_advertiser_not_banned( + two_nodes: tuple[FullNodeAPI, FullNodeAPI, ChiaServer, ChiaServer, BlockTools], + default_1000_blocks: list[FullBlock], + self_hostname: str, + monkeypatch: pytest.MonkeyPatch, +) -> None: + full_node_1, full_node_2, server_1, server_2, _bt = two_nodes + + for block in default_1000_blocks[:200]: + await full_node_1.full_node.add_block(block) + real_peak = full_node_1.full_node.blockchain.get_peak() + assert real_peak is not None + + await _connect_with_quiet_node_2(server_1, server_2, full_node_2, self_hostname, monkeypatch) + full_node_2.full_node.config["max_sync_wait"] = 0 + + # Stop _sync once it reaches weight-proof validation; everything beyond + # that point (WP fetch + chain download) is out of scope for this test. + # Reaching it proves peer-confirmation passed without raising. + reached_wp_validation = False + + async def stop_after_peer_confirmation(*_args: object, **_kwargs: object) -> None: + nonlocal reached_wp_validation + reached_wp_validation = True + raise RuntimeError("test: stop after peer confirmation") + + monkeypatch.setattr(full_node_2.full_node, "request_validate_wp", stop_after_peer_confirmation) + + full_node_2.full_node.sync_store.peer_has_block( + real_peak.header_hash, server_1.node_id, real_peak.weight, real_peak.height, True + ) + + await full_node_2.full_node._sync() + + assert reached_wp_validation + assert server_1.node_id in full_node_2.full_node.server.all_connections + + +@pytest.mark.anyio +async def test_ban_uses_snapshot_after_peer_to_peak_mutation( + two_nodes: tuple[FullNodeAPI, FullNodeAPI, ChiaServer, ChiaServer, BlockTools], + self_hostname: str, + monkeypatch: pytest.MonkeyPatch, +) -> None: + """`_ban_peak_weight_liars` uses the snapshot it was given at + peak-selection time, not the current `peer_to_peak` state. A peer that + overwrites its `peer_to_peak` entry with a fresh `NewPeak` between + `_sync()` picking the target peak and the ban firing is still banned. + + On the previous `(header_hash, weight)` lookup the mutated entry would no + longer match and the attacker would escape the ban; on the snapshot fix + the attacker is locked in at peak-selection time. + """ + _full_node_1, full_node_2, server_1, server_2, _bt = two_nodes + + await _connect_with_quiet_node_2(server_1, server_2, full_node_2, self_hostname, monkeypatch) + sync_store = full_node_2.full_node.sync_store + + real_header_hash = std_hash(b"real-peak") + inflated_weight = uint128(10**18) + height = uint32(200) + sync_store.peer_has_block(real_header_hash, server_1.node_id, inflated_weight, height, True) + + # Production code path: snapshot the advertisers exactly as `_sync()` does, + # synchronously, right after picking the target peak. + target_peak = Peak(real_header_hash, height, inflated_weight) + weight_liars = sync_store.get_advertisers_of_peak(target_peak) + assert weight_liars == {server_1.node_id} + + # Simulate the attacker sending a fresh `NewPeak` after the snapshot is + # captured (e.g. during the `asyncio.gather` await in `_sync()`). + sync_store.peer_has_block( + std_hash(b"attacker-fresh-peak"), + server_1.node_id, + uint128(1), + uint32(0), + True, + ) + # The lookup that `_sync()` used to perform at ban time would no longer + # find the attacker; the snapshot still does. This is the property the fix + # relies on. + assert sync_store.get_advertisers_of_peak(target_peak) == set() + assert server_1.node_id in weight_liars + + await full_node_2.full_node._ban_peak_weight_liars(weight_liars) + + async def liar_disconnected() -> bool: + return server_1.node_id not in full_node_2.full_node.server.all_connections + + await time_out_assert(10, liar_disconnected) diff --git a/chia/_tests/core/full_node/stores/test_sync_store.py b/chia/_tests/core/full_node/stores/test_sync_store.py index 7592bbf828..be72c89490 100644 --- a/chia/_tests/core/full_node/stores/test_sync_store.py +++ b/chia/_tests/core/full_node/stores/test_sync_store.py @@ -6,7 +6,7 @@ import pytest from chia_rs.sized_bytes import bytes32 from chia_rs.sized_ints import uint32, uint128 -from chia.full_node.sync_store import SyncStore +from chia.full_node.sync_store import Peak, SyncStore from chia.util.hash import std_hash @@ -169,6 +169,42 @@ async def test_get_heaviest_peak_returns_none_when_peaks_evicted() -> None: assert store.get_heaviest_peak() is None +@pytest.mark.anyio +async def test_peer_peak_change_removes_old_peak_membership() -> None: + """A peer that moves from one peak to another must be removed from the old peak's membership set. + + Before the fix, peak_to_peer[old_hash] retained the peer indefinitely, letting sync + code treat the peer as a valid holder of the old target peak even after it had moved away. + """ + store = SyncStore() + peer_id = std_hash(b"peer") + hash_a = std_hash(b"block_a") + hash_b = std_hash(b"block_b") + + store.peer_has_block(hash_a, peer_id, uint128(100), uint32(10), True) + assert peer_id in store.get_peers_that_have_peak([hash_a]) + + # Peer moves to a different peak. + store.peer_has_block(hash_b, peer_id, uint128(200), uint32(20), True) + + assert peer_id in store.get_peers_that_have_peak([hash_b]) + assert peer_id not in store.get_peers_that_have_peak([hash_a]) + + +@pytest.mark.anyio +async def test_peer_peak_change_same_hash_preserves_membership() -> None: + """Re-advertising the same peak hash does not remove the peer from its membership set.""" + store = SyncStore() + peer_id = std_hash(b"peer") + hash_a = std_hash(b"block_a") + + store.peer_has_block(hash_a, peer_id, uint128(100), uint32(10), True) + store.peer_has_block(hash_a, peer_id, uint128(100), uint32(10), True) + + assert peer_id in store.get_peers_that_have_peak([hash_a]) + assert store.peer_to_peak[peer_id].header_hash == hash_a + + @pytest.mark.anyio async def test_peer_disconnected_cleans_empty_peak_to_peer_entries() -> None: """peer_disconnected() removes peak_to_peer entries that have no remaining peers.""" @@ -189,3 +225,38 @@ async def test_peer_disconnected_cleans_empty_peak_to_peer_entries() -> None: store.peer_disconnected(peer_b) # After both peers disconnect, the empty entry should be cleaned up assert block_hash not in store.peak_to_peer + + +@pytest.mark.anyio +async def test_get_advertisers_of_peak_exact_match() -> None: + """get_advertisers_of_peak() returns exactly the peers whose advertised peak matches.""" + store = SyncStore() + peer_a = std_hash(b"peer_a") + peer_b = std_hash(b"peer_b") + peer_c = std_hash(b"peer_c") + peer_d = std_hash(b"peer_d") + block_hash = std_hash(b"block") + other_hash = std_hash(b"other_block") + + target = Peak(block_hash, uint32(10), uint128(500)) + + # Empty store: no advertisers. + assert store.get_advertisers_of_peak(target) == set() + + # Two peers advertise the exact target peak. + store.peer_has_block(block_hash, peer_a, uint128(500), uint32(10), True) + store.peer_has_block(block_hash, peer_b, uint128(500), uint32(10), True) + # Same header_hash, different weight — not a match. + store.peer_has_block(block_hash, peer_c, uint128(400), uint32(10), True) + # Same weight, different header_hash — not a match. + store.peer_has_block(other_hash, peer_d, uint128(500), uint32(10), True) + + assert store.get_advertisers_of_peak(target) == {peer_a, peer_b} + + # A peer that subsequently overwrites its entry with a different peak is + # no longer reported — but a snapshot taken earlier is unaffected. This + # is the property the fix relies on at peak-selection time. + snapshot = store.get_advertisers_of_peak(target) + store.peer_has_block(other_hash, peer_a, uint128(999), uint32(11), True) + assert store.get_advertisers_of_peak(target) == {peer_b} + assert snapshot == {peer_a, peer_b} diff --git a/chia/_tests/core/full_node/test_add_prevalidated_blocks.py b/chia/_tests/core/full_node/test_add_prevalidated_blocks.py index 1341d355ea..57e89dbfe2 100644 --- a/chia/_tests/core/full_node/test_add_prevalidated_blocks.py +++ b/chia/_tests/core/full_node/test_add_prevalidated_blocks.py @@ -1,8 +1,8 @@ """ Regression tests for FullNode.add_prevalidated_blocks(). -SEC-349: Prevalidation failures must return typed errors instead of raising -AssertionError, so the caller can ban the offending peer. +Prevalidation failures should return typed errors instead of raising +AssertionError, so the caller can handle the peer consistently. """ from __future__ import annotations diff --git a/chia/_tests/core/full_node/test_full_node.py b/chia/_tests/core/full_node/test_full_node.py index faabf03325..e970ba1ba6 100644 --- a/chia/_tests/core/full_node/test_full_node.py +++ b/chia/_tests/core/full_node/test_full_node.py @@ -9,7 +9,7 @@ import platform import random import sqlite3 import time -from collections.abc import Awaitable, Callable, Coroutine +from collections.abc import AsyncIterator, Awaitable, Callable, Coroutine from typing import Any import pytest @@ -996,6 +996,124 @@ async def test_new_peak( await time_out_assert(10, time_out_messages(incoming_queue, "request_block", 1)) +@pytest.mark.anyio +@pytest.mark.limit_consensus_modes( + allowed=[ConsensusMode.HARD_FORK_2_0], + reason="admission control is consensus-mode independent", +) +async def test_new_peak_admission_gate( + one_node_one_block: tuple[FullNodeAPI | FullNodeSimulator, ChiaServer, BlockTools], + self_hostname: str, + monkeypatch: pytest.MonkeyPatch, + caplog: pytest.LogCaptureFixture, +) -> None: + """Exercise every branch of the new_peak admission gate in FullNodeAPI: + + Phase 1 - outbound peer, sem exhausted: outbound skips the gate, hits + acquire() which raises LimitedSemaphoreFullError, API catches it. + Phase 2 - inbound + outbound present, sem locked: inbound is dropped at + the gate; outbound bypasses the gate and queues on the sem. + Phase 3 - inbound only (no outbound), sem locked: gate condition fails + (no outbound peers), so inbound queues on the sem instead of + being dropped. + """ + full_node_api, server, _bt = one_node_one_block + + _, inbound_id = await add_dummy_connection(server, self_hostname, 12316) + inbound_peer = server.all_connections[inbound_id] + assert not inbound_peer.is_outbound + + _, outbound_id = await add_dummy_connection(server, self_hostname, 12317) + outbound_peer = server.all_connections[outbound_id] + + fake_peak = fnp.NewPeak( + bytes32(b"\x00" * 32), + uint32(0), + uint128(0), + uint32(0), + bytes32(b"\x00" * 32), + ) + + async def noop(*_args: object) -> None: + pass + + @contextlib.asynccontextmanager + async def hold_sem(sem: LimitedSemaphore, n: int = 1) -> AsyncIterator[None]: + """Acquire *n* slots on *sem*, wait for them to be held, then yield.""" + event = asyncio.Event() + + async def _hold() -> None: + async with sem.acquire(): + await event.wait() + + tasks = [create_referenced_task(_hold()) for _ in range(n)] + await asyncio.sleep(0) + try: + yield + finally: + event.set() + for t in tasks: + t.cancel() + with contextlib.suppress(asyncio.CancelledError): + await t + + queued_tasks: list[asyncio.Task[None]] = [] + with monkeypatch.context() as m: + m.setattr(full_node_api.full_node, "new_peak", noop) + m.setattr(outbound_peer, "is_outbound", True) + + try: + # -- Phase 1: outbound, sem exhausted -- + # Outbound peers skip the admission gate entirely and go straight to + # acquire(). With zero waiting slots the sem raises + # LimitedSemaphoreFullError; the API catches it and logs. + sem = LimitedSemaphore.create(active_limit=1, waiting_limit=0) + m.setattr(full_node_api.full_node, "_new_peak_sem", sem) + async with hold_sem(sem): + with caplog.at_level(logging.DEBUG, logger="chia.full_node.full_node_api"): + caplog.clear() + await full_node_api.new_peak(fake_peak, outbound_peer) + assert "limited semaphore full" in caplog.text + + # -- Phase 2: inbound + outbound present, sem locked -- + # With an outbound peer present and both active slots taken, the gate + # drops the inbound message. The outbound peer bypasses the gate and + # queues on the sem. + sem = LimitedSemaphore.create(active_limit=2, waiting_limit=4) + m.setattr(full_node_api.full_node, "_new_peak_sem", sem) + initial = sem._available_count + async with hold_sem(sem, n=2): + # Inbound: gate drops it; sem count unchanged. + await full_node_api.new_peak(fake_peak, inbound_peer) + assert sem._available_count == initial - 2 + + # Outbound: bypasses gate, queues on sem. + queued_tasks.append(create_referenced_task(full_node_api.new_peak(fake_peak, outbound_peer))) + await time_out_assert(2, lambda: sem._available_count == initial - 3) + + # -- Phase 3: inbound only (no outbound), sem locked -- + # Flip the outbound peer back to inbound so get_connections reports + # zero outbound peers. The gate condition (outbound count > 0) is + # now false, so inbound messages queue on the sem instead of being + # dropped. + m.setattr(outbound_peer, "is_outbound", False) + + sem = LimitedSemaphore.create(active_limit=2, waiting_limit=4) + m.setattr(full_node_api.full_node, "_new_peak_sem", sem) + initial = sem._available_count + async with hold_sem(sem, n=2): + # Inbound queues instead of being dropped. + inbound_task = create_referenced_task(full_node_api.new_peak(fake_peak, inbound_peer)) + queued_tasks.append(inbound_task) + await time_out_assert(2, lambda: sem._available_count == initial - 3) + assert not inbound_task.done() + finally: + for t in queued_tasks: + t.cancel() + with contextlib.suppress(asyncio.CancelledError): + await t + + @pytest.mark.anyio @pytest.mark.limit_consensus_modes( allowed=[ConsensusMode.HARD_FORK_2_0, ConsensusMode.HARD_FORK_3_0], @@ -1296,9 +1414,9 @@ async def test_respond_transaction_fail( async def test_add_transaction_seen_before_validation( one_node_one_block: tuple[FullNodeSimulator, ChiaServer, BlockTools], ) -> None: - """Regression test for SEC-111: tx must be marked in-flight before the - pre_validate_spendbundle call so concurrent workers don't redundantly - validate the same transaction. + """ + Transaction must be marked in-flight before the pre_validate_spendbundle + call so concurrent workers don't redundantly validate the same transaction. """ full_node_1, _server_1, _bt = one_node_one_block fn = full_node_1.full_node diff --git a/chia/_tests/core/server/test_node_discovery.py b/chia/_tests/core/server/test_node_discovery.py index b70676b19b..2b9f858edc 100644 --- a/chia/_tests/core/server/test_node_discovery.py +++ b/chia/_tests/core/server/test_node_discovery.py @@ -90,7 +90,7 @@ async def test_enable_private_networks( class TestPeerHostValidation: - """Regression tests for SEC-145: unbounded peer list host strings.""" + """Regression tests for oversized peer list host strings.""" @pytest.mark.anyio async def test_add_peers_common_rejects_oversized_host( diff --git a/chia/_tests/util/test_limited_semaphore.py b/chia/_tests/util/test_limited_semaphore.py index 2f9b8c525d..8b7fc32213 100644 --- a/chia/_tests/util/test_limited_semaphore.py +++ b/chia/_tests/util/test_limited_semaphore.py @@ -111,3 +111,30 @@ async def test_stuff() -> None: assert success_results == [None] * total_limit assert semaphore._available_count == total_limit + + +@pytest.mark.anyio +async def test_locked_reflects_active_slot_usage() -> None: + semaphore = LimitedSemaphore.create(active_limit=2, waiting_limit=4) + finish_event = asyncio.Event() + + async def hold(entered: asyncio.Event) -> None: + async with semaphore.acquire(): + entered.set() + await finish_event.wait() + + assert not semaphore.locked() + + entered_1 = asyncio.Event() + holder_1 = create_referenced_task(hold(entered_1)) + await entered_1.wait() + assert not semaphore.locked() + + entered_2 = asyncio.Event() + holder_2 = create_referenced_task(hold(entered_2)) + await entered_2.wait() + assert semaphore.locked() + + finish_event.set() + await asyncio.gather(holder_1, holder_2) + assert not semaphore.locked() diff --git a/chia/_tests/util/test_priority_mutex.py b/chia/_tests/util/test_priority_mutex.py index 6a88d133da..d66d6542c6 100644 --- a/chia/_tests/util/test_priority_mutex.py +++ b/chia/_tests/util/test_priority_mutex.py @@ -356,6 +356,49 @@ async def test_cancellation_while_waiting() -> None: # TODO: do something other than hanging for ever on a, well, a hang +@pytest.mark.anyio +async def test_cancellation_of_non_first_waiter_releases_to_live_waiter() -> None: + # A waiter cancelled while it is not first in line must not strand the live + # waiter queued ahead of it, and the mutex must release to that live waiter. + mutex = PriorityMutex.create(priority_type=MutexPriority) + + blocker_continue_event = asyncio.Event() + blocker_acquired_event = asyncio.Event() + live_waiter_acquired_event = asyncio.Event() + + async def block() -> None: + async with mutex.acquire(priority=MutexPriority.high): + blocker_acquired_event.set() + await blocker_continue_event.wait() + + async def live_waiter() -> None: + async with mutex.acquire(priority=MutexPriority.high): + live_waiter_acquired_event.set() + + block_task = create_referenced_task(block()) + await blocker_acquired_event.wait() + + # Queue the live waiter ahead of the one that will be cancelled. + live_waiter_task = create_referenced_task(live_waiter()) + await wait_queued(mutex=mutex, task=live_waiter_task) + + # Queue and cancel a waiter that is not first in line. + cancel_task = create_referenced_task(to_be_cancelled(mutex=mutex)) + await wait_queued(mutex=mutex, task=cancel_task) + + cancel_task.cancel() + with pytest.raises(asyncio.CancelledError): + await cancel_task + + blocker_continue_event.set() + await block_task + + with anyio.fail_after(delay=adjusted_timeout(timeout=10)): + await live_waiter_task + + assert live_waiter_acquired_event.is_set() + + # testing many repeatable randomization cases @pytest.mark.parametrize(argnames="seed", argvalues=range(100), ids=lambda seed: f"random seed {seed}") @pytest.mark.anyio diff --git a/chia/_tests/wallet/db_wallet/test_dl_offers.py b/chia/_tests/wallet/db_wallet/test_dl_offers.py index d4703a147d..5f1435b56c 100644 --- a/chia/_tests/wallet/db_wallet/test_dl_offers.py +++ b/chia/_tests/wallet/db_wallet/test_dl_offers.py @@ -9,7 +9,8 @@ from chia_rs.sized_ints import uint64 from chia._tests.environments.wallet import WalletStateTransition, WalletTestFramework from chia._tests.util.time_out_assert import time_out_assert from chia.data_layer.data_layer_wallet import DataLayerSummary, DataLayerWallet, SingletonDependencies, SingletonSummary -from chia.wallet.puzzle_drivers import Solver +from chia.wallet.outer_puzzles import AssetType +from chia.wallet.puzzle_drivers import PuzzleInfo, Solver from chia.wallet.trade_record import TradeRecord from chia.wallet.trading.offer import Offer from chia.wallet.trading.trade_status import TradeStatus @@ -511,6 +512,71 @@ async def test_dl_offer_cancellation(wallet_environments: WalletTestFramework) - await time_out_assert(15, get_trade_and_status, TradeStatus.CANCELLED, trade_manager, offer) + # make_update_offer rejects non-DL legs. + dl_driver = await dl_wallet.get_puzzle_info(launcher_id) + base_solver = Solver({launcher_id.hex(): {"new_root": "0x" + root.hex(), "dependencies": []}}) + bogus_asset = bytes32([0xFF] * 32) + cat_asset = bytes32([0xAA] * 32) + cat_driver = PuzzleInfo({"type": AssetType.CAT.value, "tail": "0x" + cat_asset.hex()}) + # Singleton without a METADATA layer — passes check_type([SINGLETON]) but fails the + # full [SINGLETON, METADATA] predicate. + non_dl_singleton_asset = bytes32([0xBB] * 32) + non_dl_singleton_driver = PuzzleInfo( + { + "type": AssetType.SINGLETON.value, + "launcher_id": "0x" + non_dl_singleton_asset.hex(), + "launcher_ph": "0x" + bytes32.zeros.hex(), + "also": { + "type": AssetType.OWNERSHIP.value, + "owner": "()", + "transfer_program": "()", + }, + } + ) + + async with env_maker.wallet_state_manager.new_action_scope( + wallet_environments.tx_config, push=False + ) as action_scope: + with pytest.raises(ValueError, match=r"cannot include an XCH leg"): + await DataLayerWallet.make_update_offer( + env_maker.wallet_state_manager, + {launcher_id: -1, None: 1000}, + {launcher_id: dl_driver}, + base_solver, + action_scope, + fee=uint64(0), + ) + + with pytest.raises(ValueError, match=rf"{cat_asset.hex()} is not a DataLayer singleton"): + await DataLayerWallet.make_update_offer( + env_maker.wallet_state_manager, + {launcher_id: -1, cat_asset: 1}, + {launcher_id: dl_driver, cat_asset: cat_driver}, + base_solver, + action_scope, + fee=uint64(0), + ) + + with pytest.raises(ValueError, match=rf"{non_dl_singleton_asset.hex()} is not a DataLayer singleton"): + await DataLayerWallet.make_update_offer( + env_maker.wallet_state_manager, + {launcher_id: -1, non_dl_singleton_asset: 1}, + {launcher_id: dl_driver, non_dl_singleton_asset: non_dl_singleton_driver}, + base_solver, + action_scope, + fee=uint64(0), + ) + + with pytest.raises(ValueError, match=rf"{bogus_asset.hex()} is not a DataLayer singleton"): + await DataLayerWallet.make_update_offer( + env_maker.wallet_state_manager, + {launcher_id: -1, bogus_asset: 1}, + {launcher_id: dl_driver}, + base_solver, + action_scope, + fee=uint64(0), + ) + @pytest.mark.limit_consensus_modes @pytest.mark.parametrize("wallet_environments", [{"num_environments": 2, "blocks_needed": [3, 3]}], indirect=True) diff --git a/chia/_tests/wallet/test_wallet_node.py b/chia/_tests/wallet/test_wallet_node.py index b1982e996f..51169f2e1b 100644 --- a/chia/_tests/wallet/test_wallet_node.py +++ b/chia/_tests/wallet/test_wallet_node.py @@ -8,10 +8,10 @@ import time import types from pathlib import Path from typing import Any, cast -from unittest.mock import AsyncMock, MagicMock +from unittest.mock import AsyncMock, MagicMock, Mock import pytest -from chia_rs import CoinState, FullBlock, G1Element, PrivateKey +from chia_rs import CoinState, FullBlock, G1Element, HeaderBlock, PrivateKey from chia_rs.sized_bytes import bytes32 from chia_rs.sized_ints import uint8, uint32, uint64, uint128 @@ -541,6 +541,299 @@ async def test_get_timestamp_for_height_from_peer_backtracks_to_tx_block_determi assert cache.get_height_timestamp(uint32(0)) == uint64(123456789) +@pytest.mark.anyio +async def test_get_timestamp_for_height_from_peer_limits_backtrack( + root_path_populated_with_config: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + config = load_config(root_path_populated_with_config, "config.yaml", "wallet") + wallet_node = WalletNode(config, root_path_populated_with_config, test_constants) + start_height = 10_000 + max_backtrack = int(wallet_node.constants.MAX_SUB_SLOT_BLOCKS) * wallet_node.TIMESTAMP_BACKTRACK_SUB_SLOT_MULTIPLIER + last_allowed_height = start_height - max_backtrack + 1 + requested_heights: list[int] = [] + + class CacheWithNonTransactionBlocks: + def get_height_timestamp(self, height: uint32) -> None: + return None + + def get_block(self, height: uint32) -> object: + requested_heights.append(int(height)) + return types.SimpleNamespace(foliage_transaction_block=None) + + monkeypatch.setattr(wallet_node, "get_cache_for_peer", lambda peer: CacheWithNonTransactionBlocks()) + + timestamp = await wallet_node.get_timestamp_for_height_from_peer( + uint32(start_height), cast(WSChiaConnection, object()) + ) + + assert timestamp is None + assert requested_heights == list(range(start_height, start_height - max_backtrack, -1)) + + requested_heights.clear() + timestamp_at_limit = uint64(12_345) + + class CacheWithTransactionBlockAtLimit: + def get_height_timestamp(self, height: uint32) -> None: + return None + + def get_block(self, height: uint32) -> object: + requested_heights.append(int(height)) + foliage_transaction_block = ( + types.SimpleNamespace(timestamp=timestamp_at_limit) if height == last_allowed_height else None + ) + return types.SimpleNamespace(foliage_transaction_block=foliage_transaction_block) + + monkeypatch.setattr(wallet_node, "get_cache_for_peer", lambda peer: CacheWithTransactionBlockAtLimit()) + + assert ( + await wallet_node.get_timestamp_for_height_from_peer(uint32(start_height), cast(WSChiaConnection, object())) + == timestamp_at_limit + ) + assert requested_heights == list(range(start_height, start_height - max_backtrack, -1)) + + +@pytest.mark.anyio +async def test_get_timestamp_for_height_from_peer_rejects_wrong_response_height( + root_path_populated_with_config: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + config = load_config(root_path_populated_with_config, "config.yaml", "wallet") + wallet_node = WalletNode(config, root_path_populated_with_config, test_constants) + expected_hash = bytes32(b"\x01" * 32) + peer = cast(WSChiaConnection, types.SimpleNamespace(get_peer_info=lambda: "peer")) + add_to_blocks = Mock() + cache = types.SimpleNamespace( + get_height_timestamp=Mock(return_value=None), + get_block=Mock(return_value=None), + add_to_blocks=add_to_blocks, + ) + + async def wrong_height_response(peer: WSChiaConnection, start_height: uint32, end_height: uint32) -> list[object]: + return [ + types.SimpleNamespace( + height=uint32(int(start_height) - 1), + header_hash=expected_hash, + prev_header_hash=bytes32(b"\x02" * 32), + foliage_transaction_block=types.SimpleNamespace(timestamp=uint64(12_345)), + ) + ] + + monkeypatch.setattr(wallet_node, "get_cache_for_peer", lambda peer: cache) + monkeypatch.setattr("chia.wallet.wallet_node.request_header_blocks", wrong_height_response) + + assert await wallet_node.get_timestamp_for_height_from_peer(uint32(100), peer, expected_hash) is None + add_to_blocks.assert_not_called() + + +@pytest.mark.anyio +async def test_get_timestamp_for_height_from_peer_rejects_multiple_response_blocks( + root_path_populated_with_config: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + config = load_config(root_path_populated_with_config, "config.yaml", "wallet") + wallet_node = WalletNode(config, root_path_populated_with_config, test_constants) + expected_hash = bytes32(b"\x01" * 32) + peer = cast(WSChiaConnection, types.SimpleNamespace(get_peer_info=lambda: "peer")) + add_to_blocks = Mock() + cache = types.SimpleNamespace( + get_height_timestamp=Mock(return_value=None), + get_block=Mock(return_value=None), + add_to_blocks=add_to_blocks, + ) + + async def multiple_block_response(peer: WSChiaConnection, start_height: uint32, end_height: uint32) -> list[object]: + return [ + types.SimpleNamespace( + height=uint32(start_height), + header_hash=expected_hash, + prev_header_hash=bytes32(b"\x02" * 32), + foliage_transaction_block=None, + ), + types.SimpleNamespace( + height=uint32(start_height), + header_hash=expected_hash, + prev_header_hash=bytes32(b"\x03" * 32), + foliage_transaction_block=types.SimpleNamespace(timestamp=uint64(12_345)), + ), + ] + + monkeypatch.setattr(wallet_node, "get_cache_for_peer", lambda peer: cache) + monkeypatch.setattr("chia.wallet.wallet_node.request_header_blocks", multiple_block_response) + + assert await wallet_node.get_timestamp_for_height_from_peer(uint32(100), peer, expected_hash) is None + add_to_blocks.assert_not_called() + + +@pytest.mark.anyio +async def test_get_timestamp_for_height_from_peer_validates_backtrack_chain( + root_path_populated_with_config: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + config = load_config(root_path_populated_with_config, "config.yaml", "wallet") + wallet_node = WalletNode(config, root_path_populated_with_config, test_constants) + top_hash = bytes32(b"\x01" * 32) + parent_hash = bytes32(b"\x02" * 32) + timestamp = uint64(12_345) + peer = cast(WSChiaConnection, types.SimpleNamespace(get_peer_info=lambda: "peer")) + ns = types.SimpleNamespace + add_to_blocks = Mock() + cache = types.SimpleNamespace( + get_height_timestamp=Mock(return_value=uint64(99_999)), + get_block=Mock(return_value=None), + add_to_blocks=add_to_blocks, + ) + top_block = ns( + height=uint32(100), header_hash=top_hash, prev_header_hash=parent_hash, foliage_transaction_block=None + ) + tx_foliage = types.SimpleNamespace(timestamp=timestamp) + parent_block = ns( + height=uint32(99), header_hash=parent_hash, foliage_transaction_block=tx_foliage + ) # pragma: no cover + + monkeypatch.setattr(wallet_node, "get_cache_for_peer", lambda peer: cache) + request_header_blocks_mock = AsyncMock(side_effect=[[top_block], [parent_block]]) + monkeypatch.setattr("chia.wallet.wallet_node.request_header_blocks", request_header_blocks_mock) + + assert await wallet_node.get_timestamp_for_height_from_peer(uint32(100), peer, top_hash) == timestamp + assert add_to_blocks.call_count == 2 + add_to_blocks.assert_called() + + add_to_blocks.reset_mock() + wrong_parent_hash = bytes32(b"\x04" * 32) + wrong_parent_block = ns( # pragma: no cover + height=uint32(99), header_hash=wrong_parent_hash, foliage_transaction_block=tx_foliage + ) + request_header_blocks_mock.reset_mock(side_effect=True) + request_header_blocks_mock.side_effect = [[top_block], [wrong_parent_block]] + + assert await wallet_node.get_timestamp_for_height_from_peer(uint32(100), peer, top_hash) is None + assert add_to_blocks.call_count == 1 + + +@pytest.mark.anyio +async def test_get_timestamp_for_height_from_peer_refetches_stale_cached_block( + root_path_populated_with_config: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + config = load_config(root_path_populated_with_config, "config.yaml", "wallet") + wallet_node = WalletNode(config, root_path_populated_with_config, test_constants) + top_hash = bytes32(b"\x01" * 32) + parent_hash = bytes32(b"\x02" * 32) + stale_hash = bytes32(b"\x03" * 32) + timestamp = uint64(12_345) + peer = cast(WSChiaConnection, types.SimpleNamespace(get_peer_info=lambda: "peer")) + ns = types.SimpleNamespace + add_to_blocks = Mock() + stale_block = ns( + height=uint32(100), + header_hash=stale_hash, + prev_header_hash=bytes32(b"\x04" * 32), + foliage_transaction_block=ns(timestamp=uint64(99_999)), + ) + cache = types.SimpleNamespace( + get_height_timestamp=Mock(return_value=None), + get_block=Mock(side_effect=[stale_block, None]), + add_to_blocks=add_to_blocks, + ) + top_block = ns( # pragma: no cover + height=uint32(100), header_hash=top_hash, prev_header_hash=parent_hash, foliage_transaction_block=None + ) + parent_block = ns(height=uint32(99), header_hash=parent_hash, foliage_transaction_block=ns(timestamp=timestamp)) + + monkeypatch.setattr(wallet_node, "get_cache_for_peer", lambda peer: cache) + monkeypatch.setattr( + "chia.wallet.wallet_node.request_header_blocks", AsyncMock(side_effect=[[top_block], [parent_block]]) + ) + + assert await wallet_node.get_timestamp_for_height_from_peer(uint32(100), peer, top_hash) == timestamp + assert add_to_blocks.call_count == 2 + + +def test_peer_request_cache_replaces_height_timestamp() -> None: + cache = PeerRequestCache() + stale_timestamp = uint64(99_999) + replacement_timestamp = uint64(12_345) + + stale_transaction_block = cast( + HeaderBlock, + types.SimpleNamespace( + height=uint32(100), + is_transaction_block=True, + foliage_transaction_block=types.SimpleNamespace(timestamp=stale_timestamp), + ), + ) + replacement_non_transaction_block = cast( + HeaderBlock, + types.SimpleNamespace(height=uint32(100), is_transaction_block=False, foliage_transaction_block=None), + ) + replacement_transaction_block = cast( + HeaderBlock, + types.SimpleNamespace( + height=uint32(100), + is_transaction_block=True, + foliage_transaction_block=types.SimpleNamespace(timestamp=replacement_timestamp), + ), + ) + + cache.add_to_blocks(stale_transaction_block) + assert cache.get_height_timestamp(uint32(100)) == stale_timestamp + + cache.add_to_blocks(replacement_non_transaction_block) + assert cache.get_height_timestamp(uint32(100)) is None + + cache.add_to_blocks(replacement_transaction_block) + assert cache.get_height_timestamp(uint32(100)) == replacement_timestamp + + +@pytest.mark.anyio +async def test_new_peak_wallet_anchors_timestamp_lookup_to_peak_hash( + root_path_populated_with_config: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + config = load_config(root_path_populated_with_config, "config.yaml", "wallet") + wallet_node = WalletNode(config, root_path_populated_with_config, test_constants) + peak_hash = bytes32(b"\x01" * 32) + + class Blockchain: + async def get_peak_block(self) -> None: + return None + + class Peer: + closed = False + peer_node_id = bytes32(b"\x02" * 32) + close_time: int | None = None + + async def call_api(self, api: object, request: object) -> wallet_protocol.RespondBlockHeader: + return cast( + wallet_protocol.RespondBlockHeader, + types.SimpleNamespace( + header_block=types.SimpleNamespace( + header_hash=peak_hash, + weight=uint128(100), + height=uint32(100), + ) + ), + ) + + async def close(self, ban_time: int) -> None: + self.closed = True + self.close_time = ban_time + + def get_peer_info(self) -> str: + return "peer" + + wallet_node._wallet_state_manager = cast(Any, types.SimpleNamespace(blockchain=Blockchain())) + timestamp_mock = AsyncMock(return_value=None) + monkeypatch.setattr(wallet_node, "get_timestamp_for_height_from_peer", timestamp_mock) + monkeypatch.setattr(wallet_node, "is_trusted", lambda peer: False) + + peer = Peer() + await wallet_node.new_peak_wallet( + wallet_protocol.NewPeakWallet(peak_hash, uint32(100), uint128(100), uint32(99)), + cast(WSChiaConnection, peer), + ) + + assert timestamp_mock.await_args is not None + assert timestamp_mock.await_args.args[2] == peak_hash # pragma: no cover + assert peer.closed # pragma: no cover + assert peer.close_time == 120 # pragma: no cover + + @pytest.mark.limit_consensus_modes(allowed=[ConsensusMode.HARD_FORK_2_0]) @pytest.mark.standard_block_tools @pytest.mark.anyio @@ -1163,6 +1456,138 @@ async def test_add_states_from_peer_untrusted_shutdown( assert "Terminating receipt and validation due to shut down request" in caplog.text +async def _setup_untrusted_validate_and_add( + simulator_and_wallet: OldSimulatorsAndWallets, + self_hostname: str, + monkeypatch: pytest.MonkeyPatch, + add_coin_states_side_effect: list[bool], +) -> tuple[WalletNode, WSChiaConnection, AsyncMock]: + """Connect an untrusted wallet to a simulator full node and stub + `validate_received_state_from_peer` (always True) and `add_coin_states` + (returns successive booleans from the supplied list).""" + [full_node_api], [(wallet_node, wallet_server)], _ = simulator_and_wallet + await wallet_server.start_client(PeerInfo(self_hostname, full_node_api.server.get_port()), None) + wallet = wallet_node.wallet_state_manager.main_wallet + await full_node_api.farm_rewards_to_wallet(1, wallet) + full_node_peer = next(iter(wallet_server.all_connections.values())) + assert not wallet_node.is_trusted(full_node_peer) + + async def always_valid(*_: Any) -> bool: + return True + + monkeypatch.setattr( + wallet_node, + "validate_received_state_from_peer", + types.MethodType(always_valid, wallet_node), + ) + + failing_add_coin_states = AsyncMock(side_effect=add_coin_states_side_effect) + monkeypatch.setattr( + wallet_node.wallet_state_manager, + "add_coin_states", + failing_add_coin_states, + ) + return wallet_node, full_node_peer, failing_add_coin_states + + +@pytest.mark.limit_consensus_modes(allowed=[ConsensusMode.HARD_FORK_2_0]) +@pytest.mark.anyio +async def test_add_states_from_peer_untrusted_returns_false_when_add_coin_states_fails( + simulator_and_wallet: OldSimulatorsAndWallets, + self_hostname: str, + caplog: pytest.LogCaptureFixture, + monkeypatch: pytest.MonkeyPatch, +) -> None: + # Pins the untrusted-path contract: a False return from `add_coin_states` + # must propagate to `add_states_from_peer`'s return value. + wallet_node, full_node_peer, failing_add_coin_states = await _setup_untrusted_validate_and_add( + simulator_and_wallet, self_hostname, monkeypatch, add_coin_states_side_effect=[False] + ) + + coin_generator = CoinGenerator() + # One batch at the untrusted chunk size (10); `created_height` set so we hit + # the parallel `validate_and_add` path, not the reorged-states fast path. + coin_states = [CoinState(coin_generator.get().coin, uint32(i + 1), uint32(i + 1)) for i in range(10)] + + with caplog.at_level(logging.ERROR): + assert not await wallet_node.add_states_from_peer(coin_states, full_node_peer) + + assert failing_add_coin_states.await_count == 1 + assert "add_coin_states returned False for chunk" in caplog.text + assert "1 chunk(s) failed to apply" in caplog.text + + +@pytest.mark.limit_consensus_modes(allowed=[ConsensusMode.HARD_FORK_2_0]) +@pytest.mark.anyio +async def test_add_states_from_peer_untrusted_returns_false_when_one_of_many_chunks_fails( + simulator_and_wallet: OldSimulatorsAndWallets, + self_hostname: str, + caplog: pytest.LogCaptureFixture, + monkeypatch: pytest.MonkeyPatch, +) -> None: + # Three parallel chunks (30 states / chunk size 10) with only the middle one + # failing: pins that one failed chunk among successes still returns False. + wallet_node, full_node_peer, failing_add_coin_states = await _setup_untrusted_validate_and_add( + simulator_and_wallet, + self_hostname, + monkeypatch, + add_coin_states_side_effect=[True, False, True], + ) + + coin_generator = CoinGenerator() + coin_states = [CoinState(coin_generator.get().coin, uint32(i + 1), uint32(i + 1)) for i in range(30)] + + with caplog.at_level(logging.ERROR): + assert not await wallet_node.add_states_from_peer(coin_states, full_node_peer) + + assert failing_add_coin_states.await_count == 3 + assert "1 chunk(s) failed to apply" in caplog.text + + +@pytest.mark.limit_consensus_modes(allowed=[ConsensusMode.HARD_FORK_2_0]) +@pytest.mark.anyio +async def test_add_states_from_peer_untrusted_returns_false_when_validate_and_add_raises( + simulator_and_wallet: OldSimulatorsAndWallets, + self_hostname: str, + caplog: pytest.LogCaptureFixture, + monkeypatch: pytest.MonkeyPatch, +) -> None: + # Pins the symmetry of the `except Exception` arm: a raised exception in a + # parallel `validate_and_add` task must propagate to the False return. + [full_node_api], [(wallet_node, wallet_server)], _ = simulator_and_wallet + await wallet_server.start_client(PeerInfo(self_hostname, full_node_api.server.get_port()), None) + wallet = wallet_node.wallet_state_manager.main_wallet + await full_node_api.farm_rewards_to_wallet(1, wallet) + full_node_peer = next(iter(wallet_server.all_connections.values())) + assert not wallet_node.is_trusted(full_node_peer) + + async def always_valid(*_: Any) -> bool: + return True + + monkeypatch.setattr( + wallet_node, + "validate_received_state_from_peer", + types.MethodType(always_valid, wallet_node), + ) + + raising_add_coin_states = AsyncMock(side_effect=RuntimeError("simulated writer commit failure")) + monkeypatch.setattr( + wallet_node.wallet_state_manager, + "add_coin_states", + raising_add_coin_states, + ) + + coin_generator = CoinGenerator() + coin_states = [CoinState(coin_generator.get().coin, uint32(i + 1), uint32(i + 1)) for i in range(10)] + + with caplog.at_level(logging.ERROR): + assert not await wallet_node.add_states_from_peer(coin_states, full_node_peer) + + assert raising_add_coin_states.await_count == 1 + assert "validate_and_add failed - exception" in caplog.text + assert "1 chunk(s) failed to apply" in caplog.text + + @pytest.mark.limit_consensus_modes(reason="consensus rules irrelevant") @pytest.mark.anyio async def test_transaction_send_cache(self_hostname: str, simulator_and_wallet: OldSimulatorsAndWallets) -> None: diff --git a/chia/_tests/wallet/test_wallet_state_manager.py b/chia/_tests/wallet/test_wallet_state_manager.py index aa1b5bb816..1699f27057 100644 --- a/chia/_tests/wallet/test_wallet_state_manager.py +++ b/chia/_tests/wallet/test_wallet_state_manager.py @@ -1,8 +1,11 @@ from __future__ import annotations +import logging from collections.abc import AsyncIterator from contextlib import asynccontextmanager from dataclasses import dataclass +from types import SimpleNamespace +from typing import cast import pytest from chia_rs import CoinState, G2Element @@ -18,8 +21,11 @@ from chia.types.blockchain_format.coin import Coin from chia.types.blockchain_format.program import Program from chia.types.coin_spend import make_spend from chia.types.peer_info import PeerInfo +from chia.wallet import wallet_state_manager as wsm_mod from chia.wallet.derivation_record import DerivationRecord from chia.wallet.derive_keys import master_sk_to_wallet_sk, master_sk_to_wallet_sk_unhardened +from chia.wallet.nft_wallet.nft_wallet import NFTWallet +from chia.wallet.nft_wallet.uncurry_nft import NFTCoinData, UncurriedNFT from chia.wallet.remote_wallet.remote_wallet import RemoteWallet from chia.wallet.transaction_record import TransactionRecord from chia.wallet.util.transaction_type import TransactionType @@ -737,3 +743,203 @@ async def test_get_height_info_with_block_record(wallet_environments: WalletTest ) assert response.height == synced_height assert response.is_transaction_block is not None + + +async def _seed_did_scoped_nft_wallets(wsm: WalletStateManager, did_ids: list[bytes32]) -> list[NFTWallet]: + """Create one DID-scoped NFT wallet per ``did_ids`` via the real auto-create path.""" + created: list[NFTWallet] = [] + for index, did_id in enumerate(did_ids): + wallet = await NFTWallet.create_new_nft_wallet(wsm, wsm.main_wallet, did_id=did_id, name=f"NFT {index}") + created.append(wallet) + return created + + +def _build_fake_nft_data( + *, + old_p2_puzhash: bytes32, + singleton_launcher_id: bytes32, +) -> NFTCoinData: + """Build a duck-typed NFTCoinData for ``handle_nft``. + + ``handle_nft`` only accesses a small subset of fields and the helpers it + invokes (``get_metadata_and_phs`` and ``get_new_owner_did``) are patched in + the tests. Constructing real on-chain CoinSpend/UncurriedNFT objects would + require a full NFT mint, which is orthogonal to the cap behavior under test. + """ + uncurried_nft = SimpleNamespace( + supports_did=True, + owner_did=None, + p2_puzzle=SimpleNamespace(get_tree_hash=lambda: old_p2_puzhash), + singleton_launcher_id=singleton_launcher_id, + ) + parent_coin_spend = SimpleNamespace( + solution=bytes(Program.to([])), + coin=SimpleNamespace(), + ) + parent_coin_state = SimpleNamespace(spent_height=None) + return cast( + NFTCoinData, + SimpleNamespace( + uncurried_nft=uncurried_nft, + parent_coin_spend=parent_coin_spend, + parent_coin_state=parent_coin_state, + ), + ) + + +@pytest.mark.limit_consensus_modes(reason="cap logic is consensus-independent") +@pytest.mark.parametrize( + "case, configured_limit, preexisting_did_ids, seed_matching_wallet, expect_new_wallet, expect_warning", + [ + pytest.param( + "at_limit_blocks_creation", + 2, + [bytes32(b"\x01" * 32), bytes32(b"\x02" * 32)], + False, + False, + True, + id="at_limit_blocks_creation", + ), + pytest.param( + "below_limit_creates_wallet", + 2, + [bytes32(b"\x01" * 32)], + False, + True, + False, + id="below_limit_creates_wallet", + ), + pytest.param( + "matching_wallet_skips_cap", + 1, + [bytes32(b"\x01" * 32)], + True, + False, + False, + id="matching_wallet_skips_cap", + ), + pytest.param( + "yaml_default_governs", + None, + [], + False, + True, + False, + id="yaml_default_governs", + ), + ], +) +@pytest.mark.anyio +async def test_handle_nft_auto_add_limit( + simulator_and_wallet: OldSimulatorsAndWallets, + monkeypatch: pytest.MonkeyPatch, + caplog: pytest.LogCaptureFixture, + case: str, + configured_limit: int | None, + preexisting_did_ids: list[bytes32], + seed_matching_wallet: bool, + expect_new_wallet: bool, + expect_warning: bool, +) -> None: + """Regression test for the NFT auto-add cap. + + The inner-puzzle-parsed ``new_did_id`` in NFT transfer data is + attacker-controllable. Without a cap, ``handle_nft`` would create one + NFT wallet per unique foreign DID with no upper bound (in contrast to + the existing ``did_auto_add_limit`` for DID ingestion). This test + pins down four behaviors: + + - at the configured limit, ``handle_nft`` returns ``None`` and emits a + warning (no new wallet is created); + - below the limit, ``handle_nft`` creates a new wallet; + - when an existing NFT wallet already matches ``new_did_id``, the cap + is not consulted at all (a regression that hoisted the cap above + the matching loop would break inbound NFT routing for users at the + cap); + - when ``nft_auto_add_limit`` is not present in the config, the + ``initial-config.yaml`` default of 100 governs — this catches both + a wrong default literal in the code and any drift between the YAML + key name and the code's ``config.get`` key. + """ + _, [(wallet_node, _)], _ = simulator_and_wallet + wsm = wallet_node.wallet_state_manager + + if configured_limit is not None: + wsm.config["nft_auto_add_limit"] = configured_limit + else: + # The YAML-loaded default must reach the wsm.config dict under the exact + # key the handle_nft cap reads. Catches both a wrong default literal in + # initial-config.yaml and any drift between the YAML key and the + # `config.get("nft_auto_add_limit", ...)` call site in handle_nft. + assert wsm.config.get("nft_auto_add_limit") == 100 + + new_p2_puzhash = bytes32(b"\xaa" * 32) + old_p2_puzhash = bytes32(b"\xbb" * 32) + singleton_launcher_id = bytes32(b"\xcc" * 32) + foreign_did_id = bytes32(b"\xff" * 32) + + sk = master_sk_to_wallet_sk_unhardened(wsm.get_master_private_key(), uint32(99999)) + await wsm.puzzle_store.add_derivation_paths( + [ + DerivationRecord( + uint32(99999), + new_p2_puzhash, + sk.get_g1(), + WalletType.STANDARD_WALLET, + uint32(1), + False, + ) + ] + ) + + await _seed_did_scoped_nft_wallets(wsm, preexisting_did_ids) + if seed_matching_wallet: + await _seed_did_scoped_nft_wallets(wsm, [foreign_did_id]) + + def fake_get_metadata_and_phs(_unft: UncurriedNFT, _solution: bytes) -> tuple[Program, bytes32]: + return Program.to(0), new_p2_puzhash + + def fake_get_new_owner_did(_unft: UncurriedNFT, _solution: Program) -> bytes32: + return foreign_did_id + + monkeypatch.setattr(wsm_mod, "get_metadata_and_phs", fake_get_metadata_and_phs) + monkeypatch.setattr(wsm_mod, "get_new_owner_did", fake_get_new_owner_did) + + nft_data = _build_fake_nft_data( + old_p2_puzhash=old_p2_puzhash, + singleton_launcher_id=singleton_launcher_id, + ) + + def nft_wallet_count() -> int: + return sum(1 for w in wsm.wallets.values() if isinstance(w, NFTWallet)) + + before = nft_wallet_count() + with caplog.at_level(logging.WARNING, logger=wsm.log.name): + result = await wsm.handle_nft(nft_data) + after = nft_wallet_count() + + if seed_matching_wallet: + assert result is not None + existing_wallet = wsm.wallets[result.id] + assert isinstance(existing_wallet, NFTWallet) + assert existing_wallet.nft_wallet_info.did_id == foreign_did_id + assert after == before + elif expect_new_wallet: + assert result is not None + assert after == before + 1 + new_wallet = wsm.wallets[result.id] + assert isinstance(new_wallet, NFTWallet) + assert new_wallet.nft_wallet_info.did_id == foreign_did_id + else: + assert result is None + assert after == before + assert not any( + isinstance(w, NFTWallet) and w.nft_wallet_info.did_id == foreign_did_id for w in wsm.wallets.values() + ) + + if expect_warning: + assert any("nft" in rec.message.lower() and "limit" in rec.message.lower() for rec in caplog.records), ( + f"Expected a cap warning to be emitted; got: {[rec.message for rec in caplog.records]}" + ) + else: + assert not any("limit" in rec.message.lower() for rec in caplog.records) diff --git a/chia/_tests/weight_proof/test_weight_proof.py b/chia/_tests/weight_proof/test_weight_proof.py index a4a1b2e667..96ba610d75 100644 --- a/chia/_tests/weight_proof/test_weight_proof.py +++ b/chia/_tests/weight_proof/test_weight_proof.py @@ -665,8 +665,9 @@ class TestWeightProof: async def test_weight_proof_validation_challenge_at_segment_start( self, default_1000_blocks: list[FullBlock], blockchain_constants: ConsensusConstants ) -> None: - """SEC-614: validation must not crash when a segment's challenge block - is the first sub-slot entry (first_idx == 0). + """ + Validation must not crash when a segment's challenge block is the first + sub-slot entry (first_idx == 0). In legitimately-constructed proofs, segment creation always places at least one slot-end entry before the challenge block (first_idx >= 1). @@ -718,8 +719,10 @@ class TestWeightProof: async def test_weight_proof_validation_no_challenge_block_in_segment( self, default_1000_blocks: list[FullBlock], blockchain_constants: ConsensusConstants ) -> None: - """SEC-614: validation returns False when a segment has no challenge - block (every sub-slot has cc_slot_end set).""" + """ + Validation returns False when a segment has no challenge block (every + sub-slot has cc_slot_end set). + """ blocks = default_1000_blocks header_cache, height_to_hash, sub_blocks, summaries = await load_blocks_dont_validate( blocks, blockchain_constants @@ -758,8 +761,9 @@ class TestWeightProof: async def test_weight_proof_validation_missing_rc_slot_end_info( self, default_1000_blocks: list[FullBlock], blockchain_constants: ConsensusConstants ) -> None: - """SEC-614: validation returns False when a segment is missing - rc_slot_end_info.""" + """ + Validation returns False when a segment is missing rc_slot_end_info + """ blocks = default_1000_blocks header_cache, height_to_hash, sub_blocks, summaries = await load_blocks_dont_validate( blocks, blockchain_constants diff --git a/chia/consensus/multiprocess_validation.py b/chia/consensus/multiprocess_validation.py index 9d8b161272..ef222edf40 100644 --- a/chia/consensus/multiprocess_validation.py +++ b/chia/consensus/multiprocess_validation.py @@ -1,6 +1,5 @@ from __future__ import annotations -import copy import logging import time import traceback @@ -23,6 +22,7 @@ from chia_rs.sized_ints import uint16, uint32, uint64 from chia.consensus.augmented_chain import AugmentedBlockchain from chia.consensus.block_header_validation import validate_finished_header_block from chia.consensus.blockchain_interface import BlockRecordsProtocol +from chia.consensus.difficulty_adjustment import get_next_sub_slot_iters_and_difficulty from chia.consensus.full_block_to_block_record import block_to_block_record from chia.consensus.generator_tools import get_block_header, tx_removals_and_additions from chia.consensus.get_block_challenge import get_block_challenge, pre_sp_tx_block_height @@ -216,11 +216,14 @@ async def pre_validate_block( prev_b = curr assert isinstance(block, FullBlock) - if len(block.finished_sub_slots) > 0: - if block.finished_sub_slots[0].challenge_chain.new_difficulty is not None: - vs.difficulty = block.finished_sub_slots[0].challenge_chain.new_difficulty - if block.finished_sub_slots[0].challenge_chain.new_sub_slot_iters is not None: - vs.ssi = block.finished_sub_slots[0].challenge_chain.new_sub_slot_iters + if len(block.finished_sub_slots) > 0 and ( + block.finished_sub_slots[0].challenge_chain.new_sub_slot_iters is not None + or block.finished_sub_slots[0].challenge_chain.new_difficulty is not None + ): + expected_ssi, expected_difficulty = get_next_sub_slot_iters_and_difficulty(constants, True, prev_b, blockchain) + expected_vs = ValidationState(expected_ssi, expected_difficulty, vs.prev_ses_block) + else: + expected_vs = ValidationState(vs.ssi, vs.difficulty, vs.prev_ses_block) overflow = is_overflow_block(constants, block.reward_chain_block.signage_point_index) challenge = get_block_challenge(constants, block, blockchain, prev_b is None, overflow, False) if block.reward_chain_block.challenge_chain_sp_vdf is None: @@ -234,7 +237,7 @@ async def pre_validate_block( challenge, cc_sp_hash, block.height, - vs.difficulty, + expected_vs.difficulty, pre_sp_tx_block_height( constants=constants, blocks=blockchain, @@ -252,8 +255,8 @@ async def pre_validate_block( blockchain, required_iters, block, - sub_slot_iters=vs.ssi, - prev_ses_block=vs.prev_ses_block, + sub_slot_iters=expected_vs.ssi, + prev_ses_block=expected_vs.prev_ses_block, ) except ValueError: log.exception("block_to_block_record()") @@ -286,7 +289,7 @@ async def pre_validate_block( block, previous_generators, conds, - copy.copy(vs), + expected_vs, skip_commitment_validation=skip_commitment_validation, nice=nice, dedicated=dedicated, @@ -294,5 +297,9 @@ async def pre_validate_block( if block_rec.sub_epoch_summary_included is not None: vs.prev_ses_block = block_rec + if block_rec.sub_epoch_summary_included.new_difficulty is not None: + vs.difficulty = block_rec.sub_epoch_summary_included.new_difficulty + if block_rec.sub_epoch_summary_included.new_sub_slot_iters is not None: + vs.ssi = block_rec.sub_epoch_summary_included.new_sub_slot_iters return future diff --git a/chia/data_layer/data_layer_wallet.py b/chia/data_layer/data_layer_wallet.py index 335a066cab..e5b1a39201 100644 --- a/chia/data_layer/data_layer_wallet.py +++ b/chia/data_layer/data_layer_wallet.py @@ -1012,6 +1012,19 @@ class DataLayerWallet: fee: uint64 = uint64(0), extra_conditions: tuple[Condition, ...] = tuple(), ) -> Offer: + for asset_id in offer_dict: + if asset_id is None: + raise ValueError("DataLayer update offers cannot include an XCH leg") + driver = driver_dict.get(asset_id) + if driver is None or not ( + driver.check_type([AssetType.SINGLETON.value, AssetType.METADATA.value]) + and driver.also()["updater_hash"] == ACS_MU_PH # type: ignore + ): + raise ValueError( + f"DataLayer update offers only support DL singleton legs; " + f"asset {asset_id.hex()} is not a DataLayer singleton" + ) + dl_wallet = None for wallet in wallet_state_manager.wallets.values(): if wallet.type() == WalletType.DATA_LAYER.value: diff --git a/chia/full_node/full_node.py b/chia/full_node/full_node.py index 6b7cc7f8c4..3ca5f82783 100644 --- a/chia/full_node/full_node.py +++ b/chia/full_node/full_node.py @@ -101,6 +101,11 @@ from chia.util.profiler import enable_profiler, mem_profile_task, profile_task from chia.util.safe_cancel_task import cancel_task_safe from chia.util.task_referencer import create_referenced_task +# Per-request timeout for block fetches inside short_sync_backtrack. +# Use 15s here instead of the default 60s call_api timeout to bound how +# long short_sync_backtrack can occupy a new_peak slot. +SHORT_SYNC_BACKTRACK_BLOCK_REQUEST_TIMEOUT_SEC: int = 15 + # This is the result of calling peak_post_processing, which is then fed into peak_post_processing_2 @dataclasses.dataclass @@ -744,7 +749,9 @@ class FullNode: # but not the transactions fetch_tx: bool = unfinished_block is None or curr_height != target_height curr = await peer.call_api( - FullNodeAPI.request_block, full_node_protocol.RequestBlock(uint32(curr_height), fetch_tx) + FullNodeAPI.request_block, + full_node_protocol.RequestBlock(uint32(curr_height), fetch_tx), + timeout=SHORT_SYNC_BACKTRACK_BLOCK_REQUEST_TIMEOUT_SEC, ) if curr is None: raise ValueError(f"Failed to fetch block {curr_height} from {peer.get_peer_logging()}, timed out") @@ -1078,6 +1085,12 @@ class FullNode: if target_peak is None: raise RuntimeError("Not performing sync, no peaks collected") + # Snapshot advertisers of the target peak before the first await. A + # peer that overwrites its peer_to_peak entry with a subsequent + # NewPeak during the peer-confirmation round-trip below would + # otherwise slip past a (header_hash, weight) lookup at ban time. + weight_liars = self.sync_store.get_advertisers_of_peak(target_peak) + self.sync_store.target_peak = target_peak self.log.info(f"Selected peak {target_peak}") @@ -1093,6 +1106,7 @@ class FullNode: timeout=10, ) ) + any_peer_confirmed = False for i, target_peak_response in enumerate(await asyncio.gather(*coroutines, return_exceptions=True)): if isinstance(target_peak_response, BaseException): self.log.warning( @@ -1108,10 +1122,25 @@ class FullNode: and isinstance(target_peak_response, RespondBlock) and target_peak_response.block.header_hash == target_peak.header_hash ): + any_peer_confirmed = True + actual_weight = target_peak_response.block.reward_chain_block.weight + if actual_weight != target_peak.weight: + self.log.warning( + f"Peer-confirmed block weight {actual_weight} differs from " + f"advertised peak weight {target_peak.weight}, banning weight liars" + ) + await self._ban_peak_weight_liars(weight_liars) + raise RuntimeError("Advertised peak weight was a lie, banned offending peers") self.sync_store.peer_has_block( - target_peak.header_hash, peers[i].peer_node_id, target_peak.weight, target_peak.height, False + target_peak.header_hash, peers[i].peer_node_id, actual_weight, target_peak.height, False ) - # TODO: disconnect from peer which gave us the heaviest_peak, if nobody has the peak + if not any_peer_confirmed: + self.log.warning( + f"No peer confirmed the advertised peak {target_peak.header_hash} " + f"via RequestBlock, banning the peak advertiser(s)" + ) + await self._ban_peak_weight_liars(weight_liars) + raise RuntimeError("No peer confirmed the advertised peak") fork_point, summaries = await self.request_validate_wp( target_peak.header_hash, target_peak.height, target_peak.weight ) @@ -1192,6 +1221,19 @@ class FullNode: self._state_changed("sync_mode") return fork_point, summaries + async def _ban_peak_weight_liars(self, weight_liars: set[bytes32]) -> None: + """Ban and disconnect the snapshot of peers that advertised the suspect peak. + + The snapshot is captured synchronously at peak-selection time in `_sync()` + so a peer cannot escape banning by overwriting its `peer_to_peak` entry + with a subsequent `NewPeak` during the intervening async work. + """ + for peer_id in weight_liars: + conn = self.server.all_connections.get(peer_id) + if conn is not None: + self.log.warning(f"Banning peer {conn.peer_info.host} for advertising inflated weight") + await conn.close(CONSENSUS_ERROR_BAN_SECONDS) + async def sync_from_fork_point( self, fork_point_height: uint32, @@ -1730,7 +1772,6 @@ class FullNode: vs: ValidationState, # in-out parameter ) -> tuple[StateChangeSummary | None, Err | None]: agg_state_change_summary: StateChangeSummary | None = None - block_record = await self.blockchain.get_block_record_from_db(blocks_to_validate[0].prev_header_hash) for i, block in enumerate(blocks_to_validate): header_hash = block.header_hash assert vs.prev_ses_block is None or vs.prev_ses_block.height < block.height @@ -1754,15 +1795,10 @@ class FullNode: if len(block.finished_sub_slots) > 0: cc_sub_slot = block.finished_sub_slots[0].challenge_chain if cc_sub_slot.new_sub_slot_iters is not None or cc_sub_slot.new_difficulty is not None: - expected_sub_slot_iters, expected_difficulty = get_next_sub_slot_iters_and_difficulty( - self.constants, True, block_record, blockchain - ) assert cc_sub_slot.new_sub_slot_iters is not None vs.ssi = cc_sub_slot.new_sub_slot_iters assert cc_sub_slot.new_difficulty is not None vs.difficulty = cc_sub_slot.new_difficulty - assert expected_sub_slot_iters == vs.ssi - assert expected_difficulty == vs.difficulty block_rec = blockchain.block_record(block.header_hash) result, error, state_change_summary = await self.blockchain.add_block( block, @@ -1798,7 +1834,7 @@ class FullNode: if error is not None: self.log.error(f"Error: {error}, Invalid block from peer: {peer_info} ") return agg_state_change_summary, error - block_record = blockchain.block_record(header_hash) + block_record = block_rec assert block_record is not None if block_record.sub_epoch_summary_included is not None: vs.prev_ses_block = block_record diff --git a/chia/full_node/full_node_api.py b/chia/full_node/full_node_api.py index 7b19921ce6..b9d460f763 100644 --- a/chia/full_node/full_node_api.py +++ b/chia/full_node/full_node_api.py @@ -49,7 +49,7 @@ from chia.full_node.tx_processing_queue import PeerWithTx, TransactionQueueEntry from chia.protocols import farmer_protocol, full_node_protocol, introducer_protocol, timelord_protocol, wallet_protocol from chia.protocols.fee_estimate import FeeEstimate, FeeEstimateGroup, fee_rate_v2_to_v1 from chia.protocols.full_node_protocol import RejectBlock, RejectBlocks -from chia.protocols.outbound_message import Message, make_msg +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, RATE_LIMITER_BAN_SECONDS from chia.protocols.shared_protocol import Capability @@ -209,7 +209,18 @@ class FullNodeAPI: we can ask for it. """ # this semaphore limits the number of tasks that can call new_peak() at - # the same time, since it can be expensive + # the same time, since it can be expensive. + # When all active slots are busy and we have at least one outbound + # full-node peer, drop additional inbound NewPeak requests. + # Without an outbound peer, let inbound peers use the bounded queue. + if ( + not peer.is_outbound + and self.full_node.new_peak_sem.locked() + and len(self.full_node.server.get_connections(NodeType.FULL_NODE, outbound=True)) > 0 + ): + self.log.debug("Dropping inbound NewPeak, active slots busy: %s %s", peer.get_peer_logging(), request) + return None + try: async with self.full_node.new_peak_sem.acquire(): await self.full_node.new_peak(request, peer) diff --git a/chia/full_node/sync_store.py b/chia/full_node/sync_store.py index 8f389e79af..5eb506705a 100644 --- a/chia/full_node/sync_store.py +++ b/chia/full_node/sync_store.py @@ -76,6 +76,10 @@ class SyncStore: self.peak_to_peer[item[0]] = item[1] # Put it back in if it was the sync target self.peak_to_peer.popitem(last=False) # Remove the oldest entry again if new_peak: + old_peak = self.peer_to_peak.get(peer_id) + if old_peak is not None and old_peak.header_hash != header_hash: + if old_peak.header_hash in self.peak_to_peer: + self.peak_to_peer[old_peak.header_hash].discard(peer_id) self.peer_to_peak[peer_id] = Peak(header_hash, height, weight) def get_peers_that_have_peak(self, header_hashes: list[bytes32]) -> set[bytes32]: @@ -102,6 +106,17 @@ class SyncStore: ret[peer_id] = peak return ret + def get_advertisers_of_peak(self, peak: Peak) -> set[bytes32]: + """ + Returns: peer IDs whose currently advertised peak exactly matches `peak`. + + Intended to snapshot advertisers at peak-selection time so a peer that + overwrites its `peer_to_peak` entry with a fresh `NewPeak` afterward + cannot escape downstream banning for the earlier advertisement. + """ + + return {peer_id for peer_id, p in self.peer_to_peak.items() if p == peak} + def get_heaviest_peak(self) -> Peak | None: """ Returns: the header_hash, height, and weight of the heaviest block that one of our peers has notified diff --git a/chia/util/initial-config.yaml b/chia/util/initial-config.yaml index ff1a037773..8f5d502377 100644 --- a/chia/util/initial-config.yaml +++ b/chia/util/initial-config.yaml @@ -600,6 +600,12 @@ wallet: # if an unknown DID is sent to us, a wallet will be automatically created did_auto_add_limit: 10 + # if an NFT is received whose ownership transfer encodes a DID we do not yet + # have a per-DID NFT wallet for, a wallet will be automatically created. + # This cap bounds the number of DID-scoped NFT wallets to prevent an inbound + # NFT spammer from forcing unbounded wallet creation. + nft_auto_add_limit: 100 + # Interval to resend unconfirmed transactions, even if previously accepted into Mempool tx_resend_timeout_secs: 1800 diff --git a/chia/util/limited_semaphore.py b/chia/util/limited_semaphore.py index 5a3c1fb547..fc90c47055 100644 --- a/chia/util/limited_semaphore.py +++ b/chia/util/limited_semaphore.py @@ -28,6 +28,9 @@ class LimitedSemaphore: _available_count=active_limit + waiting_limit, ) + def locked(self) -> bool: + return self._semaphore.locked() + @contextlib.asynccontextmanager async def acquire(self) -> AsyncIterator[int]: if self._available_count < 1: diff --git a/chia/util/priority_mutex.py b/chia/util/priority_mutex.py index 90972516e5..0ed09b1f55 100644 --- a/chia/util/priority_mutex.py +++ b/chia/util/priority_mutex.py @@ -23,10 +23,11 @@ class NestedLockUnsupportedError(Exception): _T_Priority = TypeVar("_T_Priority", bound=IntEnum) -@dataclasses.dataclass(frozen=True) +# eq=False: queued waiters are removed by identity (deque.remove), so distinct waiters must never compare equal. +@dataclasses.dataclass(frozen=True, eq=False) class _Element: - task: asyncio.Task[object] = dataclasses.field(compare=False) - ready_event: asyncio.Event = dataclasses.field(default_factory=asyncio.Event, compare=False) + task: asyncio.Task[object] + ready_event: asyncio.Event = dataclasses.field(default_factory=asyncio.Event) @final diff --git a/chia/wallet/util/peer_request_cache.py b/chia/wallet/util/peer_request_cache.py index 4bd09bbc2c..b9723a07c7 100644 --- a/chia/wallet/util/peer_request_cache.py +++ b/chia/wallet/util/peer_request_cache.py @@ -40,8 +40,9 @@ class PeerRequestCache: self._blocks.put(header_block.height, header_block) if header_block.is_transaction_block: assert header_block.foliage_transaction_block is not None - if self._timestamps.get(header_block.height) is None: - self._timestamps.put(header_block.height, header_block.foliage_transaction_block.timestamp) + self._timestamps.put(header_block.height, header_block.foliage_transaction_block.timestamp) + else: + self._timestamps.cache.pop(header_block.height, None) def get_block_request(self, start: uint32, end: uint32) -> asyncio.Task[Any] | None: return self._block_requests.get((start, end)) diff --git a/chia/wallet/wallet_node.py b/chia/wallet/wallet_node.py index ac6c697d77..18aaca2874 100644 --- a/chia/wallet/wallet_node.py +++ b/chia/wallet/wallet_node.py @@ -161,6 +161,7 @@ class WalletNode: validation_semaphore: asyncio.Semaphore | None = None local_node_synced: bool = False LONG_SYNC_THRESHOLD: int = 300 + TIMESTAMP_BACKTRACK_SUB_SLOT_MULTIPLIER: ClassVar[int] = 4 last_wallet_tx_resend_time: int = 0 # Duration in seconds coin_state_retry_seconds: int = 10 @@ -1009,6 +1010,11 @@ class WalletNode: if num_filtered > 0: self.log.info(f"Filtered {num_filtered} spam transactions") + # Per-task write failures (False return or raised exception) recorded here so + # `add_states_from_peer` can return False after `asyncio.gather`, matching the + # trusted path's `return False` contract on `add_coin_states` failure. + failed_chunks: list[int] = [] + async def validate_and_add(inner_states: list[CoinState], inner_idx_start: int) -> None: try: assert self.validation_semaphore is not None @@ -1020,11 +1026,19 @@ class WalletNode: f"new coin state received ({inner_idx_start}-" f"{inner_idx_start + len(inner_states) - 1}/ {len(updated_coin_states)})" ) - await self.wallet_state_manager.add_coin_states(valid_states, peer, fork_height) + if not await self.wallet_state_manager.add_coin_states(valid_states, peer, fork_height): + log_level = logging.DEBUG if peer.closed or self._shut_down else logging.ERROR + self.log.log( + log_level, + f"add_coin_states returned False for chunk " + f"{inner_idx_start}-{inner_idx_start + len(inner_states) - 1}", + ) + failed_chunks.append(inner_idx_start) except Exception as e: tb = traceback.format_exc() log_level = logging.DEBUG if peer.closed or self._shut_down else logging.ERROR self.log.log(log_level, f"validate_and_add failed - exception: {e}, traceback: {tb}") + failed_chunks.append(inner_idx_start) # Keep chunk size below 1000 just in case, windows has sqlite limits of 999 per query # Untrusted has a smaller batch size since validation has to happen which takes a while @@ -1079,6 +1093,13 @@ class WalletNode: still_connected = self._server is not None and peer.peer_node_id in self.server.all_connections await asyncio.gather(*all_tasks) await self.update_ui() + if failed_chunks: + log_level = logging.DEBUG if peer.closed or self._shut_down else logging.ERROR + self.log.log( + log_level, + f"add_states_from_peer: {len(failed_chunks)} chunk(s) failed to apply for peer {peer.peer_node_id}", + ) + return False return still_connected and self._server is not None and peer.peer_node_id in self.server.all_connections def is_timestamp_in_sync(self, timestamp: uint64) -> bool: @@ -1139,36 +1160,62 @@ class WalletNode: neither.append(node) return synced_and_trusted + synced + trusted + neither - async def get_timestamp_for_height_from_peer(self, height: uint32, peer: WSChiaConnection) -> uint64 | None: + async def get_timestamp_for_height_from_peer( + self, height: uint32, peer: WSChiaConnection, expected_header_hash: bytes32 | None = None + ) -> uint64 | None: """ Returns the timestamp for transaction block at h=height, if not transaction block, backtracks until it finds - a transaction block + a recent transaction block. """ cache = self.get_cache_for_peer(peer) request_height: int = height - while request_height >= 0: - cached_timestamp = cache.get_height_timestamp(uint32(request_height)) + expected_hash: bytes32 | None = expected_header_hash + max_backtrack = int(self.constants.MAX_SUB_SLOT_BLOCKS) * self.TIMESTAMP_BACKTRACK_SUB_SLOT_MULTIPLIER + min_request_height = max(0, request_height - max_backtrack + 1) + while request_height >= min_request_height: + cached_timestamp = cache.get_height_timestamp(uint32(request_height)) if expected_hash is None else None if cached_timestamp is not None: return cached_timestamp block = cache.get_block(uint32(request_height)) + if block is not None and expected_hash is not None and block.header_hash != expected_hash: + self.log.debug( + f"get_timestamp_for_height_from_peer ignore stale cached block for height {request_height}" + ) + block = None + fetched_block = False if block is None: self.log.debug(f"get_timestamp_for_height_from_peer cache miss for height {request_height}") response: list[HeaderBlock] | None = await request_header_blocks( peer, uint32(request_height), uint32(request_height) ) if response is not None and len(response) > 0: - self.log.debug(f"get_timestamp_for_height_from_peer add to cache for height {request_height}") - cache.add_to_blocks(response[0]) + if len(response) != 1: + self.log.warning(f"bad header blocks response from Peer {peer.get_peer_info()}.") + return None block = response[0] + if block.height != request_height: + self.log.warning(f"bad header block height response from Peer {peer.get_peer_info()}.") + return None + fetched_block = True elif request_height < height: # The peer might be slightly behind but still synced, so we should allow fetching one more block break else: self.log.debug(f"get_timestamp_for_height_from_peer use cached block for height {request_height}") + if block is not None and expected_hash is not None and block.header_hash != expected_hash: + self.log.warning(f"bad header block hash response from Peer {peer.get_peer_info()}.") + return None + + if block is not None and fetched_block: + self.log.debug(f"get_timestamp_for_height_from_peer add to cache for height {request_height}") + cache.add_to_blocks(block) + if block is not None and block.foliage_transaction_block is not None: return block.foliage_transaction_block.timestamp + if block is not None and expected_hash is not None: + expected_hash = block.prev_header_hash request_height -= 1 return None @@ -1228,7 +1275,9 @@ class WalletNode: return trusted: bool = self.is_trusted(peer) - latest_timestamp = await self.get_timestamp_for_height_from_peer(new_peak_hb.height, peer) + latest_timestamp = await self.get_timestamp_for_height_from_peer( + new_peak_hb.height, peer, new_peak_hb.header_hash + ) if latest_timestamp is None or not self.is_timestamp_in_sync(latest_timestamp): if trusted: self.log.debug(f"Trusted peer {peer.get_peer_info()} is not synced.") diff --git a/chia/wallet/wallet_state_manager.py b/chia/wallet/wallet_state_manager.py index 91e9e522ed..a79c9aab2c 100644 --- a/chia/wallet/wallet_state_manager.py +++ b/chia/wallet/wallet_state_manager.py @@ -1558,6 +1558,22 @@ class WalletStateManager: if wallet_identifier is None and new_derivation_record is not None: # Cannot find an existed NFT wallet for the new NFT + # Bound the number of auto-created DID-scoped NFT wallets. `new_did_id` + # is parsed from attacker-controllable NFT transfer data; without a cap a + # peer that spams inbound NFTs with unique foreign DIDs can force + # unbounded wallet creation. Mirrors `did_auto_add_limit`. + # Counter excludes the canonical did_id=None wallet so attacker-driven + # fanout cannot displace a user's legitimate no-DID NFT receives. + nft_wallet_count = sum( + 1 for w in self.wallets.values() if isinstance(w, NFTWallet) and w.nft_wallet_info.did_id is not None + ) + nft_limit = self.config.get("nft_auto_add_limit", 100) + if new_did_id is not None and nft_wallet_count >= nft_limit: + self.log.warning( + f"You are at the max configured limit of {nft_limit} NFT wallets. " + f"Ignoring received NFT {uncurried_nft.singleton_launcher_id.hex()} with DID {new_did_id.hex()}" + ) + return None self.log.info( "Cannot find a NFT wallet for NFT_ID: %s DID_ID: %s, creating a new one.", uncurried_nft.singleton_launcher_id,