From e8ca5d9c697399e295c4c26105e931a469b55028 Mon Sep 17 00:00:00 2001 From: siddhant Date: Sun, 30 Aug 2026 23:58:08 +0530 Subject: [PATCH 1/2] fix: make P2P networking actually work, and let a node mine on its own MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Swarm.listen() waits on a nursery that only exists while the swarm runs as a background service, so a node calling host.get_network().listen() directly never finished starting up: no multiaddr printed, no inbound connections possible. Listening now goes through host.run(). The dialing side had the same problem in reverse — it borrowed a nursery attribute the swarm doesn't expose, so a peer that dialed out never registered the connection or read from it. Both now use the nursery this module already opens for itself. Two mining bugs made a fresh chain effectively unusable: mining refused to produce a block when the mempool was empty, but the mining reward is the only source of new coins, so a chain that had never received one could never mine its first coin either. And two blocks minted within the same millisecond were rejected, since consensus requires strictly increasing timestamps. Both are fixed at the source: an empty block still carries the reward and proof-of-work, and the timestamp is clamped to be newer than the parent's instead of just read off the clock. A node that accepted a transaction or block only kept it — it never told any other peer, so a network more than one hop wide couldn't stay in sync. Accepted content is now relayed onward, excluding whoever sent it, with the existing dedup set stopping the echo from circulating. That set was unbounded, and now that it's load-bearing for stopping gossip loops rather than just avoiding duplicate work, its size is capped with an LRU eviction. Smaller fixes bundled in because they're cheap and adjacent: `peers` now lists the peer IDs it's connected to, not just a count; the JSON-RPC server's bind address is a flag instead of hardcoded to loopack; two unused constants (TRUSTED_PEERS, LOCALHOST_PEERS) are removed; and the README's --connect example, which predates the multiaddr-based CLI, is corrected. --- README.md | 4 +- main.py | 37 ++++++++----- minichain/node_config.py | 3 ++ minichain/p2p.py | 70 ++++++++++++++++++++----- tests/test_protocol_hardening.py | 90 ++++++++++++++++++++++++++++++++ 5 files changed, 176 insertions(+), 28 deletions(-) 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}"})) From 14956ebda0d9df0e4de53d01e57dd8f691604e21 Mon Sep 17 00:00:00 2001 From: siddhant Date: Mon, 31 Aug 2026 00:10:50 +0530 Subject: [PATCH 2/2] feat: reach and rejoin peers on other networks without manual reconnects MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Stacked on fix/p2p-core. That branch makes P2P work at all; this makes it work across networks and survive a restart. A node was only dialable if someone had arranged for its address to be reachable, which in practice meant running bore or ngrok alongside it. Three mechanisms replace that, in increasing order of last resort: mDNS finds peers on the same LAN with nothing typed at all; --upnp asks the router for a port mapping; --relay/--relay-addr route through another MiniChain node when neither works, via libp2p's circuit relay v2. --bootstrap dials a list of peers on startup — discovery rather than reachability, but the same path. That solved reaching a peer once. It did not solve staying reachable: the node's own identity was regenerated every restart, which invalidated any address a peer had saved for it, and it forgot every peer it had ever connected to the moment it stopped. Three more pieces close that: identity.py persists the libp2p keypair to /nodekey, so the peer ID a node presents is the same across restarts. known_peers a table beside the existing banned_peers one (persistence.py) (remember_peer / get_known_peers). A node dials it on startup alongside --bootstrap, and redials with backoff when a stream it dialed itself drops. getaddr/addr after a handshake, a node asks its peer for more peers and (main.py) remembers what it's told, so the network stays reachable after whoever it was originally introduced to goes away. Addresses are stored, never relayed onward, so this can't become an amplification vector. Only the dialing side of a connection ever has a real address for the peer to share — an inbound connection reveals nothing but an ephemeral remote port — which is the same asymmetry Bitcoin has via service flags. --announce lets a node advertise a different address than the one it binds to, for when the bind address (0.0.0.0 under --upnp, a NAT'd LAN IP) isn't itself dialable. --fund now defaults to 0: it credits a wallet outside of block validation, so a nonzero value diverges a node's state from every peer's the moment it mines, and the previous default of 100 broke multi-node operation out of the box. Verified live: two nodes reach each other only through a relay, with no direct route ever used; a node's peer ID and its ability to redial its last peer both survive a restart, even for a node that was only ever reachable through that relay; a third node dialing only one existing peer discovers and connects to the other via addr gossip alone. --- main.py | 54 +++++++- minichain/identity.py | 44 +++++++ minichain/network_config.py | 2 +- minichain/p2p.py | 187 ++++++++++++++++++++++++++-- minichain/persistence.py | 53 ++++++++ tests/test_identity_and_peerbook.py | 89 +++++++++++++ tests/test_persistence_runtime.py | 2 +- 7 files changed, 416 insertions(+), 15 deletions(-) create mode 100644 minichain/identity.py create mode 100644 tests/test_identity_and_peerbook.py diff --git a/main.py b/main.py index 2ead628..6b5bab8 100644 --- a/main.py +++ b/main.py @@ -35,6 +35,7 @@ from minichain.rpc import JSONRPCServer from minichain.validators import is_valid_receiver, ValidationStatus from minichain.block import calculate_receipt_root +from minichain.persistence import remember_peer, get_known_peers, is_peer_banned logger = logging.getLogger(__name__) @@ -197,7 +198,7 @@ async def handler(data): payload = data.get("data") peer_addr = data.get("_peer_addr", "unknown") - if payload is None and msg_type in ("hello", "chain_request", "chain_response"): + if payload is None and msg_type in ("hello", "chain_request", "chain_response", "getaddr", "addr"): return if msg_type == "hello": @@ -218,6 +219,10 @@ async def handler(data): logger.info("📡 Peer %s is ahead (%d > %d). Initiating chunked sync...", peer_addr, peer_tip, chain.last_block.index) request_chain(network, chain.last_block.index + 1, 500) + # Ask for more peers, so the network stays reachable after whoever we + # were told about goes away, instead of only ever knowing our seed. + asyncio.create_task(network._unicast_raw(peer_addr, {"type": "getaddr", "data": {}})) + elif msg_type == "tx": try: tx = Transaction.from_dict(payload) @@ -324,6 +329,23 @@ async def handler(data): if new_chain[0].index > 0: request_chain(network, earlier_start, 50) + elif msg_type == "getaddr": + # Share only peers we have ourselves successfully dialed: an inbound + # connection never reveals the far side's own listen address, so that is + # all we have to offer (the same asymmetry Bitcoin has via service flags). + addrs = [p["multiaddr"] for p in get_known_peers(network.data_path)] + asyncio.create_task( + network._unicast_raw(peer_addr, {"type": "addr", "data": {"addrs": addrs}}) + ) + + elif msg_type == "addr": + for addr in payload.get("addrs", []): + candidate_id = addr.rsplit("/p2p/", 1)[-1] + if candidate_id == network._peer_id or is_peer_banned(candidate_id, path=network.data_path): + continue + remember_peer(candidate_id, addr, network.data_path) + asyncio.create_task(network.connect_to_peer(addr)) + return handler @@ -612,7 +634,10 @@ def print_prompt_info(current_pk): # ────────────────────────────────────────────── async def run_node(port: int, host: str, connect_to: str | None, fund: int, datadir: str | None, - rpc_host: str = "127.0.0.1"): + rpc_host: str = "127.0.0.1", + upnp: bool = False, bootstrap: list | None = None, + relay: bool = False, relay_addr: str | None = None, + announce: str | None = None): """Boot the node, optionally connect to a peer, then enter the CLI.""" sk, pk = load_or_create_wallet(datadir) @@ -659,16 +684,25 @@ async def on_peer_connected(writer): network.register_on_peer_connected(on_peer_connected) - await network.start(port=port, host=host) + await network.start(port=port, host=host, upnp=upnp, bootstrap=bootstrap, + relay=relay, relay_addr=relay_addr, announce=announce, datadir=datadir) # 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=rpc_host, port=rpc_port) - # Fund this node's wallet so it can transact in the demo + # Fund this node's wallet so it can transact in the demo. This is applied + # outside of block validation, so it diverges this node's state from every + # peer's — any block built on top of it will be rejected by them on the + # state-root check. Only safe on a single isolated node. if fund > 0: chain.state.credit_mining_reward(pk, reward=fund) logger.info("💰 Funded %s... with %d coins", pk[:12], fund) + if connect_to or bootstrap: + logger.warning( + "⚠️ --fund is nonzero while connecting to peers: this node's state will " + "diverge and its blocks will be rejected. Use --fund 0 and mine instead." + ) # Connect to a seed peer if requested if connect_to: @@ -695,9 +729,14 @@ def main(): parser.add_argument("--host", type=str, default="127.0.0.1", help="Host/IP to bind the P2P server (default: 127.0.0.1)") parser.add_argument("--port", type=int, default=9000, help="TCP port to listen on (default: 9000)") 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("--fund", type=int, default=0, help="Initial coins to fund this wallet outside consensus (default: 0). Nonzero diverges state from every peer's -- see run_node.") 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)") + parser.add_argument("--upnp", action="store_true", help="Ask the router to forward our port, so peers on other networks can dial us") + parser.add_argument("--bootstrap", type=str, nargs="*", default=None, help="Peer multiaddrs to dial on startup") + parser.add_argument("--relay", action="store_true", help="Act as a relay, letting unreachable peers be dialed through this node") + parser.add_argument("--relay-addr", type=str, default=None, help="Multiaddr of a relay to become reachable through") + parser.add_argument("--announce", type=str, default=None, help="Public host:port to advertise instead of the bind address (e.g. behind --upnp or a port-forward)") args = parser.parse_args() logging.basicConfig( @@ -708,7 +747,10 @@ def main(): try: asyncio.run(run_node(args.port, args.host, args.connect, args.fund, args.datadir, - rpc_host=args.rpc_host)) + rpc_host=args.rpc_host, + upnp=args.upnp, bootstrap=args.bootstrap, + relay=args.relay, relay_addr=args.relay_addr, + announce=args.announce)) except KeyboardInterrupt: print("\nNode shut down.") diff --git a/minichain/identity.py b/minichain/identity.py new file mode 100644 index 0000000..5121f4c --- /dev/null +++ b/minichain/identity.py @@ -0,0 +1,44 @@ +""" +Persistent node identity. + +Without this, `new_host()` mints a fresh libp2p keypair every time a node starts, +so its peer ID — the `/p2p/` half of every multiaddr — changes on every restart. +That invalidates any address a peer saved, any bootstrap entry pointing at this node, +and any relay reservation, making the peer book and reconnection logic pointless. +Persisting the key is what lets an address stay valid across restarts, the same role +geth's `nodekey` file plays. +""" + +import logging +import os + +from libp2p.crypto.ed25519 import Ed25519PrivateKey, create_new_key_pair +from libp2p.crypto.keys import KeyPair + +logger = logging.getLogger(__name__) + +_NODEKEY_FILE = "nodekey" + + +def load_or_create_keypair(datadir: str) -> KeyPair: + """Load the node's persistent Ed25519 keypair from *datadir*, creating one + on first run. The same keypair yields the same peer ID every time.""" + path = os.path.join(datadir, _NODEKEY_FILE) + + if os.path.exists(path): + try: + with open(path, "rb") as f: + seed = f.read() + sk = Ed25519PrivateKey.from_bytes(seed) + return KeyPair(sk, sk.get_public_key()) + except Exception as e: + logger.warning("Failed to load node key from %s: %s — generating a new one", path, e) + + os.makedirs(datadir, exist_ok=True) + keypair = create_new_key_pair() + seed = keypair.private_key.to_bytes() + fd = os.open(path, os.O_WRONLY | os.O_CREAT | os.O_TRUNC, 0o600) + with os.fdopen(fd, "wb") as f: + f.write(seed) + logger.info("Created new node identity at %s", path) + return keypair diff --git a/minichain/network_config.py b/minichain/network_config.py index 5fc5371..3ffbf7a 100644 --- a/minichain/network_config.py +++ b/minichain/network_config.py @@ -5,7 +5,7 @@ # to maintain consensus. # P2P Network Rules -SUPPORTED_MESSAGE_TYPES = {"hello", "tx", "block", "chain_request", "chain_response"} +SUPPORTED_MESSAGE_TYPES = {"hello", "tx", "block", "chain_request", "chain_response", "getaddr", "addr"} PROTOCOL_ID = TProtocol("/minichain/1.0.0") MAX_FRAME_BYTES = 1 * 1024 * 1024 # 1 MB diff --git a/minichain/p2p.py b/minichain/p2p.py index f44e16d..1909030 100644 --- a/minichain/p2p.py +++ b/minichain/p2p.py @@ -21,11 +21,19 @@ from libp2p import new_host TProtocol = str +from libp2p.discovery.events.peerDiscovery import peerDiscovery +from libp2p.discovery.mdns.mdns import MDNSDiscovery +from libp2p.peer.id import ID from libp2p.peer.peerinfo import info_from_p2p_addr +from libp2p.relay.circuit_v2 import CircuitV2Protocol, CircuitV2Transport +from libp2p.relay.circuit_v2.config import RelayConfig, RelayRole +from libp2p.relay.circuit_v2.protocol import PROTOCOL_ID as CIRCUIT_PROTOCOL_ID +from libp2p.tools.async_service import background_trio_service from multiaddr import Multiaddr from .serialization import canonical_json_hash, canonical_json_dumps from .validators import ValidationStatus -from .persistence import ban_peer, is_peer_banned +from .persistence import ban_peer, is_peer_banned, remember_peer, get_known_peers +from .identity import load_or_create_keypair logger = logging.getLogger(__name__) @@ -63,6 +71,7 @@ def __init__( self._peer_count = 0 self._peer_count_lock = threading.Lock() self._peer_ids: set = set() + self._peer_id = None # our own peer id, set once _trio_main starts the host # Misbehavior tracking, keyed directly by ValidationStatus so there is a # single vocabulary for statuses (no parallel string keys to convert). @@ -88,11 +97,46 @@ def register_handler(self, handler_callback): def register_on_peer_connected(self, handler_callback): self._on_peer_connected = handler_callback - async def start(self, port: int = 9000, host: str = "127.0.0.1"): + async def start( + self, + port: int = 9000, + host: str = "127.0.0.1", + *, + upnp: bool = False, + bootstrap: list | None = None, + relay: bool = False, + relay_addr: str | None = None, + announce: str | None = None, + datadir: str | None = None, + ): + """ + Bring the node online. + + Reaching a node on another network needs its address to be dialable, which + this handles in three escalating ways: mDNS finds peers on the same LAN with + no addresses typed at all, `upnp` asks the router for a port mapping, and + `relay`/`relay_addr` route through another MiniChain node when neither works. + `bootstrap` is discovery rather than reachability — a list of peers to dial + on startup instead of typing `connect`. `announce` overrides the address + printed for others to dial, for when the bind address (e.g. 0.0.0.0 under + --upnp) is not itself dialable. `datadir` persists this node's identity and + the peers it has connected to, so both survive a restart. + """ self.port = port self.host_addr = host + self.upnp = upnp + self.bootstrap = bootstrap or [] + self.relay = relay + self.relay_addr = relay_addr + self.announce = announce + self.data_path = datadir or self.data_path self.loop = asyncio.get_running_loop() + # mDNS and bootstrap only put peers in the peerstore and open a transport + # connection; opening the MiniChain stream on top is our job, so every + # discovery is routed through the same CONNECT path as a manual dial. + peerDiscovery.register_peer_discovered_handler(self._on_peer_discovered) + threading.Thread(target=trio.run, args=(self._trio_main,), daemon=True).start() asyncio.create_task(self._asyncio_reader()) asyncio.create_task(self._decay_counters()) @@ -106,6 +150,25 @@ async def connect_to_peer(self, maddr_str: str) -> bool: self._to_trio.put(("CONNECT", maddr_str)) return True + def _on_peer_discovered(self, peer_info): + """ + Queue dials for a peer announced by mDNS or bootstrap. Runs on the discovery + thread, so it only touches the thread-safe command queue. Every address is + queued because any one of them may be unreachable; CONNECT skips a peer we + already hold a stream to, so the extras cost nothing once one succeeds. + + Both ends see the same discovery event, so only the lower peer ID dials. Left + to themselves both would dial and the pair would end up with two streams. The + equal case is our own announcement coming back to us. + """ + if self._peer_id is None or self._peer_id >= peer_info.peer_id.to_string(): + return + for addr in peer_info.addrs: + addr_str = str(addr) + if "/p2p/" not in addr_str: + addr_str = f"{addr_str}/p2p/{peer_info.peer_id.to_string()}" + self._to_trio.put(("CONNECT", addr_str)) + def _message_id(self, msg_type, payload): if msg_type == "tx": return canonical_json_hash(payload) if msg_type == "block": return payload["hash"] @@ -290,7 +353,21 @@ async def drain(self): pass # ── trio main ──────────────────────────────────────────────────────────── async def _trio_main(self): - host = new_host() + # A persisted keypair keeps our peer ID stable across restarts. Without it + # every address a peer saved for us, or that we advertise, goes stale the + # moment we restart. + key_pair = load_or_create_keypair(self.data_path) + host = new_host( + key_pair=key_pair, + enable_upnp=self.upnp, + bootstrap=self.bootstrap, + ) + # new_host(enable_mDNS=True) builds the discovery with libp2p's default port + # of 8000 rather than the port we listen on, so peers would find us and then + # dial nothing. Attaching it ourselves is the only way to advertise the real + # port; host.run() starts whatever is set here. + host.mDNS = MDNSDiscovery(host.get_network(), port=self.port) + self._peer_id = host.get_id().to_string() listen_addr = Multiaddr(f"/ip4/{self.host_addr}/tcp/{self.port}") # 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 @@ -299,10 +376,55 @@ async def _trio_main(self): async with host.run(listen_addrs=[listen_addr]): await self._serve(host, listen_addr) + async def _run_circuit(self, protocol, nursery, host, circuit): + """ + Run the circuit relay service, and reserve a slot on a relay if one was given. + + A reservation is what makes an unreachable node dialable: the relay agrees to + accept connections addressed to us and forward them, so peers dial our circuit + address instead of an address our NAT would drop. + """ + async with background_trio_service(protocol): + await protocol.event_started.wait() + if self.relay_addr: + try: + info = info_from_p2p_addr(Multiaddr(self.relay_addr)) + await host.connect(info) + stream = await host.new_stream(info.peer_id, [CIRCUIT_PROTOCOL_ID]) + if await circuit.reserve(stream, info.peer_id, nursery): + print( + f" Reachable via relay: {self.relay_addr}" + f"/p2p-circuit/p2p/{host.get_id().to_string()}" + ) + else: + logger.error("Relay %s refused our reservation", info.peer_id) + except Exception as e: + logger.error(f"Relay reservation failed: {e}") + await trio.sleep_forever() + async def _serve(self, host, listen_addr): - print(f" Network Multiaddr: {listen_addr}/p2p/{host.get_id().to_string()}") + # The bind address is not necessarily dialable (0.0.0.0, a NAT'd LAN IP), so + # --announce lets the operator substitute the address peers should actually use. + advertise_addr = listen_addr + if self.announce: + try: + announce_host, announce_port = self.announce.rsplit(":", 1) + advertise_addr = Multiaddr(f"/ip4/{announce_host}/tcp/{announce_port}") + except Exception: + logger.warning("Ignoring malformed --announce %r, expected host:port", self.announce) + print(f" Network Multiaddr: {advertise_addr}/p2p/{host.get_id().to_string()}") streams = [] + circuit = None + # Peers we have dialed but whose stream is not in `streams` yet: the reader + # task that registers it only starts on the next scheduler pass, so without + # this a peer announcing several addresses gets dialed once per address. + dialing = set() + # Redial peers we dialed ourselves when their stream drops. Only populated for + # outbound connections: an inbound one never reveals an address to redial with. + dialed_addrs: dict = {} + redial_backoff: dict = {} + REDIAL_BACKOFF_MAX = 300 # seconds async def stream_handler(stream): peer_id = str(stream.muxed_conn.peer_id) @@ -347,14 +469,31 @@ async def stream_handler(stream): except Exception: pass + dialing.discard(stream.muxed_conn.peer_id) if stream in streams: streams.remove(stream) with self._peer_count_lock: self._peer_count -= 1 self._peer_ids.discard(peer_id) + # Redial a peer we dialed ourselves once its stream drops, with backoff so a + # peer that is genuinely gone does not get hammered. The task ends right + # after, so sleeping here does not block anything else. + redial_addr = dialed_addrs.pop(stream.muxed_conn.peer_id, None) + if redial_addr is not None and not is_peer_banned(peer_id, path=self.data_path): + delay = redial_backoff.get(stream.muxed_conn.peer_id, 1.0) + redial_backoff[stream.muxed_conn.peer_id] = min(delay * 2, REDIAL_BACKOFF_MAX) + await trio.sleep(delay) + self._to_trio.put(("CONNECT", redial_addr)) + host.set_stream_handler(PROTOCOL_ID, stream_handler) + # Redial peers we have connected to before, so a restart with no --connect + # or --bootstrap still rejoins the network it was already part of. Same + # CONNECT path as a manual dial, so the in-flight dedup above applies. + for known in get_known_peers(self.data_path): + self._to_trio.put(("CONNECT", known["multiaddr"])) + async def check_queue(nursery): while True: try: @@ -363,16 +502,38 @@ async def check_queue(nursery): if cmd == "STOP": return True elif cmd == "CONNECT": + peer_id = None try: + # The peer we end up talking to is always the last + # /p2p/ hop, whether the address is direct or routed + # through a relay. + peer_id = ID.from_base58(arg.split("/p2p/")[-1]) + if peer_id in dialing or any( + s.muxed_conn.peer_id == peer_id for s in streams + ): + continue + dialing.add(peer_id) maddr = Multiaddr(arg) - info = info_from_p2p_addr(maddr) - await host.connect(info) - stream = await host.new_stream(info.peer_id, [PROTOCOL_ID]) + if "/p2p-circuit" in arg: + if circuit is None: + raise ValueError("node was not started with relay support") + await circuit.dial(maddr) + else: + await host.connect(info_from_p2p_addr(maddr)) + stream = await host.new_stream(peer_id, [PROTOCOL_ID]) # 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) + # Only the dialer ends up with a dialable address for the + # peer — an inbound connection reveals nothing but an + # ephemeral remote port — so this is the only place a + # peer can be remembered for a future redial. + remember_peer(peer_id.to_string(), arg, self.data_path) + dialed_addrs[peer_id] = arg + redial_backoff.pop(peer_id, None) except Exception as e: + dialing.discard(peer_id) logger.error(f"Dial error: {e}") elif cmd == "BROADCAST": payload, exclude = arg @@ -412,6 +573,18 @@ async def check_queue(nursery): await trio.sleep(0.1) async with trio.open_nursery() as nursery: + # STOP and CLIENT are always on so that any node can accept a relayed + # connection and dial a /p2p-circuit address without extra flags. HOP is + # the one that costs something — it lets others push traffic through us — + # so it stays opt-in. + roles = RelayRole.STOP | RelayRole.CLIENT + if self.relay: + roles |= RelayRole.HOP + config = RelayConfig(roles=roles) + protocol = CircuitV2Protocol(host, limits=config.limits, allow_hop=self.relay) + circuit = CircuitV2Transport(host, protocol, config) + nursery.start_soon(self._run_circuit, protocol, nursery, host, circuit) + async def run_monitor(): if await check_queue(nursery): await host.close() diff --git a/minichain/persistence.py b/minichain/persistence.py index dc989ba..4ae108f 100644 --- a/minichain/persistence.py +++ b/minichain/persistence.py @@ -320,6 +320,59 @@ def get_banned_peers(path: str = ".") -> list[dict[str, Any]]: return [{"peer_id": r["peer_id"], "reason": r["reason"], "timestamp": r["timestamp"]} for r in rows] +# --------------------------------------------------------------------------- +# Known Peers (peer book) +# +# Lets a node redial peers it has previously connected to without needing +# --connect or --bootstrap again. Mirrors the banned_peers table above. +# --------------------------------------------------------------------------- + + +def _ensure_known_peers_table(conn: sqlite3.Connection) -> None: + conn.execute( + "CREATE TABLE IF NOT EXISTS known_peers (peer_id TEXT PRIMARY KEY, multiaddr TEXT, last_seen REAL)" + ) + + +@contextmanager +def _known_peers_conn(path: str, create: bool): + """Same shape as _banned_peers_conn: yields None for read-only callers when + no DB exists yet, so they can short-circuit without touching the filesystem.""" + db_path = os.path.join(path, _DB_FILE) + if not create and not os.path.exists(db_path): + yield None + return + if create: + os.makedirs(path, exist_ok=True) + conn = _connect(db_path) + try: + _ensure_known_peers_table(conn) + yield conn + finally: + conn.close() + + +def remember_peer(peer_id: str, multiaddr: str, path: str = ".") -> None: + """Record that we successfully connected to *peer_id* at *multiaddr*, so a + future startup (or the reconnect loop) can redial it without being told.""" + with _known_peers_conn(path, create=True) as conn, conn: + conn.execute( + "INSERT OR REPLACE INTO known_peers (peer_id, multiaddr, last_seen) VALUES (?, ?, ?)", + (peer_id, multiaddr, time.time()) + ) + + +def get_known_peers(path: str = ".", limit: int = 50) -> list[dict[str, Any]]: + with _known_peers_conn(path, create=False) as conn: + if conn is None: + return [] + rows = conn.execute( + "SELECT peer_id, multiaddr, last_seen FROM known_peers ORDER BY last_seen DESC LIMIT ?", + (limit,) + ).fetchall() + return [{"peer_id": r["peer_id"], "multiaddr": r["multiaddr"], "last_seen": r["last_seen"]} for r in rows] + + # --------------------------------------------------------------------------- # Legacy JSON helpers # --------------------------------------------------------------------------- diff --git a/tests/test_identity_and_peerbook.py b/tests/test_identity_and_peerbook.py new file mode 100644 index 0000000..ebb303b --- /dev/null +++ b/tests/test_identity_and_peerbook.py @@ -0,0 +1,89 @@ +import os +import shutil +import tempfile +import unittest + +from libp2p.peer.id import ID + +from minichain.identity import load_or_create_keypair +from minichain.persistence import get_known_peers, remember_peer + + +class TestPersistentIdentity(unittest.TestCase): + def setUp(self): + self.datadir = tempfile.mkdtemp() + self.addCleanup(shutil.rmtree, self.datadir, ignore_errors=True) + + def test_keypair_survives_reload(self): + """A node must present the same peer ID across restarts, or every address + a peer saved for it (and every address it saved for a peer) goes stale.""" + first = load_or_create_keypair(self.datadir) + second = load_or_create_keypair(self.datadir) + + self.assertEqual( + ID.from_pubkey(first.public_key), + ID.from_pubkey(second.public_key), + ) + + def test_nodekey_file_is_created(self): + load_or_create_keypair(self.datadir) + + self.assertTrue(os.path.exists(os.path.join(self.datadir, "nodekey"))) + + def test_corrupt_nodekey_falls_back_to_a_new_identity(self): + """A damaged key file must not crash the node on startup.""" + nodekey_path = os.path.join(self.datadir, "nodekey") + os.makedirs(self.datadir, exist_ok=True) + with open(nodekey_path, "wb") as f: + f.write(b"not a valid seed") + + keypair = load_or_create_keypair(self.datadir) + + self.assertIsNotNone(keypair) + + +class TestPeerBook(unittest.TestCase): + def setUp(self): + self.datadir = tempfile.mkdtemp() + self.addCleanup(shutil.rmtree, self.datadir, ignore_errors=True) + + def test_empty_peer_book_before_any_peer_is_remembered(self): + self.assertEqual(get_known_peers(self.datadir), []) + + def test_remember_and_retrieve_peer(self): + remember_peer("peerA", "/ip4/1.2.3.4/tcp/9000/p2p/peerA", self.datadir) + + known = get_known_peers(self.datadir) + + self.assertEqual(len(known), 1) + self.assertEqual(known[0]["peer_id"], "peerA") + self.assertEqual(known[0]["multiaddr"], "/ip4/1.2.3.4/tcp/9000/p2p/peerA") + + def test_remembering_a_peer_again_updates_its_address(self): + """A peer's multiaddr can change (different port, moved networks); the + book must track the latest one rather than accumulate stale entries.""" + remember_peer("peerA", "/ip4/1.2.3.4/tcp/9000/p2p/peerA", self.datadir) + remember_peer("peerA", "/ip4/1.2.3.4/tcp/9001/p2p/peerA", self.datadir) + + known = get_known_peers(self.datadir) + + self.assertEqual(len(known), 1) + self.assertEqual(known[0]["multiaddr"], "/ip4/1.2.3.4/tcp/9001/p2p/peerA") + + def test_multiple_peers_are_all_retrievable(self): + remember_peer("peerA", "/ip4/1.2.3.4/tcp/9000/p2p/peerA", self.datadir) + remember_peer("peerB", "/ip4/5.6.7.8/tcp/9000/p2p/peerB", self.datadir) + + known = get_known_peers(self.datadir) + + self.assertEqual({p["peer_id"] for p in known}, {"peerA", "peerB"}) + + def test_limit_caps_the_number_returned(self): + for i in range(5): + remember_peer(f"peer{i}", f"/ip4/1.2.3.4/tcp/900{i}/p2p/peer{i}", self.datadir) + + self.assertEqual(len(get_known_peers(self.datadir, limit=2)), 2) + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/test_persistence_runtime.py b/tests/test_persistence_runtime.py index fa1dcca..53b918b 100644 --- a/tests/test_persistence_runtime.py +++ b/tests/test_persistence_runtime.py @@ -23,7 +23,7 @@ def register_handler(self, handler): def register_on_peer_connected(self, callback): self._on_peer_connected = callback - async def start(self, port=9000, host="127.0.0.1"): + async def start(self, port=9000, host="127.0.0.1", **kwargs): self.port = port self.host = host