diff --git a/README.md b/README.md index 4c41e20..e1f4433 100644 --- a/README.md +++ b/README.md @@ -89,9 +89,9 @@ python main.py --port 9000 --datadir ./node1_data *Note: Keep this terminal open to interact with the node via the CLI.* ### 2. Connecting to an Existing Chain -To connect a secondary node to the network, start a new instance on a different port and point it to the seed node using the `--connect` flag. +To connect a secondary node to the network, start a new instance on a different port and point it to the seed node using the `--connect` flag with the full multiaddress the seed node printed on startup (it includes the seed's peer ID, not just its host and port). ```bash -python main.py --port 9001 --connect 127.0.0.1:9000 --datadir ./node2_data +python main.py --port 9001 --connect /ip4/127.0.0.1/tcp/9000/p2p/ --datadir ./node2_data ``` The node will automatically sync the blockchain state via the P2P network using the Fork-Choice rule. diff --git a/main.py b/main.py index b37fce4..2ead628 100644 --- a/main.py +++ b/main.py @@ -26,6 +26,7 @@ import sys import os import json +import time from nacl.signing import SigningKey from nacl.encoding import HexEncoder @@ -38,9 +39,6 @@ logger = logging.getLogger(__name__) -TRUSTED_PEERS = set() -LOCALHOST_PEERS = {"127.0.0.1", "::1", "localhost", "0:0:0:0:0:0:0:1"} - # ────────────────────────────────────────────── # Wallet helpers @@ -120,11 +118,15 @@ async def submit_and_broadcast(chain, mempool, network, tx, ok_msg, reject_msg): # ────────────────────────────────────────────── def mine_and_process_block(chain, mempool, miner_pk): - """Mine pending transactions into a new block.""" + """ + Mine pending transactions into a new block. + + A block with no transactions is still mined: it carries the mining reward and + adds proof-of-work, so the chain keeps advancing while the network is idle. + Requiring a transaction would also make a fresh chain unusable, since coins + only come into existence through the mining reward. + """ pending_txs = mempool.get_transactions_for_block() - if not pending_txs: - logger.info("Mempool is empty — nothing to mine.") - return None # Filter queue candidates against a temporary state snapshot. temp_state = chain.state.copy() @@ -147,17 +149,19 @@ def mine_and_process_block(chain, mempool, miner_pk): if stale_txs: mempool.remove_transactions(stale_txs) - if not mineable_txs: - logger.info("No mineable transactions in current queue window.") - return None - total_fees = sum(getattr(r, 'gas_used', 0) * getattr(tx, 'fee_per_gas', 0) for r, tx in zip(receipts, mineable_txs)) temp_state.credit_mining_reward(miner_pk, reward=temp_state.DEFAULT_MINING_REWARD + total_fees) + # Timestamps are milliseconds and a block must be strictly newer than its parent. + # Blocks can be mined inside the same millisecond, so clamp forward rather than + # letting the miner build a block the chain will reject. + timestamp = max(round(time.time() * 1000), chain.last_block.timestamp + 1) + block = Block( index=chain.last_block.index + 1, previous_hash=chain.last_block.hash, transactions=mineable_txs, + timestamp=timestamp, state_root=temp_state.state_root(), receipt_root=calculate_receipt_root(receipts), receipts=receipts, @@ -505,6 +509,8 @@ def print_prompt_info(current_pk): # ── peers ── elif cmd == "peers": print(f" Connected peers: {network.peer_count}") + for peer_id in network.peer_ids: + print(f" {peer_id}") # ── connect ── elif cmd == "connect": @@ -605,7 +611,8 @@ def print_prompt_info(current_pk): # Main entry point # ────────────────────────────────────────────── -async def run_node(port: int, host: str, connect_to: str | None, fund: int, datadir: str | None): +async def run_node(port: int, host: str, connect_to: str | None, fund: int, datadir: str | None, + rpc_host: str = "127.0.0.1"): """Boot the node, optionally connect to a peer, then enter the CLI.""" sk, pk = load_or_create_wallet(datadir) @@ -656,7 +663,7 @@ async def on_peer_connected(writer): # Start RPC server on a port correlated to the node port (e.g. 8545 if P2P is 9000) rpc_port = 8545 + (port - 9000) - await rpc_server.start(host="127.0.0.1", port=rpc_port) + await rpc_server.start(host=rpc_host, port=rpc_port) # Fund this node's wallet so it can transact in the demo if fund > 0: @@ -690,6 +697,7 @@ def main(): parser.add_argument("--connect", type=str, default=None, help="Peer address to connect to (multiaddr)") parser.add_argument("--fund", type=int, default=100, help="Initial coins to fund this wallet (default: 100)") parser.add_argument("--datadir", type=str, default=".minichain", help="Directory to save/load blockchain state (enables persistence)") + parser.add_argument("--rpc-host", type=str, default="127.0.0.1", help="Host/IP to bind the JSON-RPC server (default: 127.0.0.1, loopback-only)") args = parser.parse_args() logging.basicConfig( @@ -699,7 +707,8 @@ def main(): ) try: - asyncio.run(run_node(args.port, args.host, args.connect, args.fund, args.datadir)) + asyncio.run(run_node(args.port, args.host, args.connect, args.fund, args.datadir, + rpc_host=args.rpc_host)) except KeyboardInterrupt: print("\nNode shut down.") diff --git a/minichain/node_config.py b/minichain/node_config.py index 784b54e..0128139 100644 --- a/minichain/node_config.py +++ b/minichain/node_config.py @@ -8,6 +8,9 @@ INVALID_THRESHOLD = 1 # L: accumulated invalid messages before ban (1 = immediate) DECAY_INTERVAL_MINUTES = 10 # T: counter half-life period in minutes +# Gossip dedup: max tx/block ids remembered per set before the oldest is evicted. +SEEN_CACHE_MAX = 10_000 + # Mempool Config MEMPOOL_MAX_SIZE = 1000 MEMPOOL_TX_PER_BLOCK = 100 diff --git a/minichain/p2p.py b/minichain/p2p.py index 1171eb1..f44e16d 100644 --- a/minichain/p2p.py +++ b/minichain/p2p.py @@ -11,7 +11,12 @@ import struct import trio import queue -from .node_config import MALFORMED_THRESHOLD, FAILED_THRESHOLD, INVALID_THRESHOLD, DECAY_INTERVAL_MINUTES +from collections import OrderedDict + +from .node_config import ( + MALFORMED_THRESHOLD, FAILED_THRESHOLD, INVALID_THRESHOLD, DECAY_INTERVAL_MINUTES, + SEEN_CACHE_MAX, +) from .network_config import SUPPORTED_MESSAGE_TYPES, PROTOCOL_ID, MAX_FRAME_BYTES from libp2p import new_host @@ -48,12 +53,16 @@ def __init__( ): self._handler_callback = handler_callback self._on_peer_connected = None - self._seen_tx_ids = set() - self._seen_block_hashes = set() + # OrderedDict as an LRU: unbounded sets here would let a live node accumulate + # tx/block ids forever, and since relaying now depends on this set to stop + # echoes circulating, its size is no longer just a memory concern. + self._seen_tx_ids = OrderedDict() + self._seen_block_hashes = OrderedDict() self._to_trio = queue.Queue() self._to_asyncio = queue.Queue() self._peer_count = 0 self._peer_count_lock = threading.Lock() + self._peer_ids: set = set() # Misbehavior tracking, keyed directly by ValidationStatus so there is a # single vocabulary for statuses (no parallel string keys to convert). @@ -111,11 +120,16 @@ def _is_duplicate(self, msg_type, payload): def _mark_seen(self, msg_type, payload): mid = self._message_id(msg_type, payload) - if mid: - self._seen_set(msg_type).add(mid) + if not mid: + return + seen = self._seen_set(msg_type) + seen[mid] = True + seen.move_to_end(mid) + if len(seen) > SEEN_CACHE_MAX: + seen.popitem(last=False) - async def _broadcast_raw(self, payload: dict): - self._to_trio.put(("BROADCAST", payload)) + async def _broadcast_raw(self, payload: dict, exclude: str | None = None): + self._to_trio.put(("BROADCAST", (payload, exclude))) async def _unicast_raw(self, target_addr: str, payload: dict): self._to_trio.put(("UNICAST", (target_addr, payload))) @@ -144,6 +158,11 @@ def peer_count(self) -> int: with self._peer_count_lock: return self._peer_count + @property + def peer_ids(self) -> list: + with self._peer_count_lock: + return list(self._peer_ids) + # ── misbehavior helpers ────────────────────────────────────────────────── def _increment_counter(self, peer_id: str, status: ValidationStatus) -> bool: @@ -237,6 +256,17 @@ async def _asyncio_reader(self): except Exception: pass + # Relay accepted content onward, so gossip travels further than one + # hop. Only VALID is forwarded: rejected content is never amplified, + # and a None status (hello/chain_request/chain_response) is not + # relayable at all. Excluding the sender avoids a pointless echo, and + # _mark_seen above means an echo arriving from another peer is dropped + # as a duplicate, so cycles in the peer graph terminate. + if msg_type in ("tx", "block") and status == ValidationStatus.VALID: + await self._broadcast_raw( + {"type": msg_type, "data": payload}, exclude=peer_addr + ) + elif msg[0] == "MALFORMED": # JSON parse failure signalled from the Trio thread. peer_addr = msg[1] @@ -262,7 +292,14 @@ async def drain(self): pass async def _trio_main(self): host = new_host() listen_addr = Multiaddr(f"/ip4/{self.host_addr}/tcp/{self.port}") - await host.get_network().listen(listen_addr) + # Swarm.listen() waits on a nursery that only exists while the swarm runs as + # a background service, so listening must go through host.run(). Calling + # host.get_network().listen() directly blocks forever and the node comes up + # with no networking at all. + async with host.run(listen_addrs=[listen_addr]): + await self._serve(host, listen_addr) + + async def _serve(self, host, listen_addr): print(f" Network Multiaddr: {listen_addr}/p2p/{host.get_id().to_string()}") streams = [] @@ -283,6 +320,7 @@ async def stream_handler(stream): streams.append(stream) with self._peer_count_lock: self._peer_count += 1 + self._peer_ids.add(peer_id) self._to_asyncio.put(("PEER_CONNECTED", None)) try: @@ -313,10 +351,11 @@ async def stream_handler(stream): streams.remove(stream) with self._peer_count_lock: self._peer_count -= 1 + self._peer_ids.discard(peer_id) host.set_stream_handler(PROTOCOL_ID, stream_handler) - async def check_queue(): + async def check_queue(nursery): while True: try: while not self._to_trio.empty(): @@ -329,12 +368,18 @@ async def check_queue(): info = info_from_p2p_addr(maddr) await host.connect(info) stream = await host.new_stream(info.peer_id, [PROTOCOL_ID]) - host.get_network().nursery.start_soon(stream_handler, stream) + # Read the outbound stream in our own nursery: the swarm + # exposes no nursery to borrow, and without this the + # dialing side never registers the peer or reads from it. + nursery.start_soon(stream_handler, stream) except Exception as e: logger.error(f"Dial error: {e}") elif cmd == "BROADCAST": - msg = (canonical_json_dumps(arg) + "\n").encode() + payload, exclude = arg + msg = (canonical_json_dumps(payload) + "\n").encode() for s in list(streams): + if exclude and f"peer:{s.muxed_conn.peer_id}" == exclude: + continue try: await s.write(msg) except Exception: @@ -361,13 +406,14 @@ async def check_queue(): streams.remove(s) with self._peer_count_lock: self._peer_count -= 1 + self._peer_ids.discard(str(s.muxed_conn.peer_id)) except Exception: pass await trio.sleep(0.1) async with trio.open_nursery() as nursery: async def run_monitor(): - if await check_queue(): + if await check_queue(nursery): await host.close() nursery.cancel_scope.cancel() nursery.start_soon(run_monitor) diff --git a/tests/test_protocol_hardening.py b/tests/test_protocol_hardening.py index d34de14..a7786a2 100644 --- a/tests/test_protocol_hardening.py +++ b/tests/test_protocol_hardening.py @@ -1,10 +1,13 @@ +import asyncio import unittest from nacl.encoding import HexEncoder from nacl.signing import SigningKey from minichain import Block, Mempool, P2PNetwork, State, Transaction, calculate_hash +from minichain.node_config import SEEN_CACHE_MAX from minichain.serialization import canonical_json_dumps +from minichain.validators import ValidationStatus class TestDeterministicConsensus(unittest.TestCase): @@ -162,3 +165,90 @@ async def test_duplicate_tx_and_block_detection(self): self.assertFalse(network._is_duplicate("block", block_message["data"])) network._mark_seen("block", block_message["data"]) self.assertTrue(network._is_duplicate("block", block_message["data"])) + + + +class TestP2PRelayAndPeers(unittest.IsolatedAsyncioTestCase): + """ + A node relays content it accepts, so gossip travels past one hop, but never + back to the peer it came from and never anything it rejected. + """ + + SOURCE = "peer:source" + TX_DATA = { + "sender": "a" * 64, "receiver": "b" * 64, "amount": 1, "nonce": 0, + "data": None, "timestamp": 123, "signature": "c" * 128, + } + + def _network(self, status): + network = P2PNetwork(malformed_threshold=1000, failed_threshold=1000, invalid_threshold=1000) + network.loop = asyncio.get_running_loop() + + async def handler(_data): + return status + + network.register_handler(handler) + return network + + async def _deliver(self, network, msg_type, data): + network._to_asyncio.put(("MSG", {"type": msg_type, "data": data, "_peer_addr": self.SOURCE})) + task = asyncio.create_task(network._asyncio_reader()) + for _ in range(30): + await asyncio.sleep(0.01) + if not network._to_trio.empty(): + break + task.cancel() + network._to_asyncio.put(("MALFORMED", self.SOURCE)) + try: + await task + except asyncio.CancelledError: + pass + commands = [] + while not network._to_trio.empty(): + commands.append(network._to_trio.get_nowait()) + return [arg for cmd, arg in commands if cmd == "BROADCAST"] + + async def test_accepted_tx_is_relayed_excluding_sender(self): + network = self._network(ValidationStatus.VALID) + + broadcasts = await self._deliver(network, "tx", self.TX_DATA) + + self.assertEqual(len(broadcasts), 1) + payload, exclude = broadcasts[0] + self.assertEqual(payload, {"type": "tx", "data": self.TX_DATA}) + self.assertEqual(exclude, self.SOURCE) + + async def test_rejected_content_is_never_relayed(self): + network = self._network(ValidationStatus.INVALID) + + broadcasts = await self._deliver(network, "tx", self.TX_DATA) + + self.assertEqual(broadcasts, []) + + async def test_control_messages_are_never_relayed(self): + network = self._network(None) + + broadcasts = await self._deliver(network, "hello", {"some": "payload"}) + + self.assertEqual(broadcasts, []) + + async def test_duplicate_is_relayed_only_once(self): + network = self._network(ValidationStatus.VALID) + + first = await self._deliver(network, "tx", self.TX_DATA) + second = await self._deliver(network, "tx", self.TX_DATA) + + self.assertEqual(len(first), 1) + self.assertEqual(second, []) + + +class TestSeenCacheIsBounded(unittest.TestCase): + def test_oldest_entry_is_evicted_past_the_cap(self): + network = P2PNetwork() + + for i in range(SEEN_CACHE_MAX + 10): + network._mark_seen("block", {"hash": f"h{i}"}) + + self.assertEqual(len(network._seen_block_hashes), SEEN_CACHE_MAX) + self.assertFalse(network._is_duplicate("block", {"hash": "h0"})) + self.assertTrue(network._is_duplicate("block", {"hash": f"h{SEEN_CACHE_MAX + 9}"}))