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..6b5bab8 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 @@ -34,13 +35,11 @@ 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__) -TRUSTED_PEERS = set() -LOCALHOST_PEERS = {"127.0.0.1", "::1", "localhost", "0:0:0:0:0:0:0:1"} - # ────────────────────────────────────────────── # Wallet helpers @@ -120,11 +119,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 +150,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, @@ -193,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": @@ -214,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) @@ -320,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 @@ -505,6 +531,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 +633,11 @@ 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", + 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) @@ -652,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="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 + # 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: @@ -688,8 +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( @@ -699,7 +746,11 @@ 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, + 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/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..1909030 100644 --- a/minichain/p2p.py +++ b/minichain/p2p.py @@ -11,16 +11,29 @@ 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 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__) @@ -48,12 +61,17 @@ 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() + 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). @@ -79,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()) @@ -97,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"] @@ -111,11 +183,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 +221,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 +319,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] @@ -260,12 +353,78 @@ 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}") - await host.get_network().listen(listen_addr) - print(f" Network Multiaddr: {listen_addr}/p2p/{host.get_id().to_string()}") + # 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 _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): + # 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) @@ -283,6 +442,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: @@ -309,14 +469,32 @@ 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) - async def check_queue(): + # 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: while not self._to_trio.empty(): @@ -324,17 +502,45 @@ async def check_queue(): 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]) - host.get_network().nursery.start_soon(stream_handler, stream) + 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": - 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 +567,26 @@ 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: + # 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(): + if await check_queue(nursery): await host.close() nursery.cancel_scope.cancel() nursery.start_soon(run_monitor) 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 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}"}))