diff --git a/Cargo.lock b/Cargo.lock index 0d26ba4c..9f0774db 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -3031,7 +3031,7 @@ dependencies = [ [[package]] name = "dig-node-service" -version = "0.217.0" +version = "0.224.0" dependencies = [ "async-trait", "axum", diff --git a/Cargo.toml b/Cargo.toml index 2a394e81..b99a5b09 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -32,7 +32,7 @@ edition = "2021" # the ROOT manifest (`[workspace.package].version`), so it MUST be set here for a # release to fire (§3.6). The library crates (dig-node-core/dig-runtime/dig-wallet) # keep their own independent versions — only the released binary tracks the workspace version. -version = "0.217.0" +version = "0.224.0" # Release hardening, matching digstore: keep integer-overflow checks ON in release. # The node parses untrusted serialized input and does offset/length arithmetic over diff --git a/crates/dig-node-core/src/peer.rs b/crates/dig-node-core/src/peer.rs index b80d0c88..ca728bc4 100644 --- a/crates/dig-node-core/src/peer.rs +++ b/crates/dig-node-core/src/peer.rs @@ -2928,6 +2928,9 @@ async fn run_peer_network(node: Arc) -> Result<(), String> { stun_server, ); + // #3124: the listener registers every peer it ACCEPTS in this same pool, so clone the handle + // before the responder takes ownership of it. + let handle_for_pool = handle.clone(); let mut node_responder = NodeResponder::with_pool(node, handle); if let Some(dht) = dht { node_responder = node_responder.with_dht(dht); @@ -2947,7 +2950,7 @@ async fn run_peer_network(node: Arc) -> Result<(), String> { ); } - serve_peer_rpc_listener_with(listener, identity, responder, Some(pex)).await + serve_peer_rpc_listener_with(listener, identity, responder, Some(pex), Some(handle_for_pool)).await } /// Bring up the content-location DHT (#163) for a running node: build a [`crate::dht::NatDhtTransport`] @@ -3290,7 +3293,7 @@ pub async fn serve_peer_rpc_listener( node: Arc, responder: Arc, ) -> Result<(), String> { - serve_peer_rpc_listener_with(listener, node, responder, None).await + serve_peer_rpc_listener_with(listener, node, responder, None, None).await } /// Like [`serve_peer_rpc_listener`] but additionally running the node↔node **PEX** peer-sharing layer @@ -3298,11 +3301,17 @@ pub async fn serve_peer_rpc_listener( /// stream (handshake→snapshot→deltas) and serves the peer's incoming PEX stream, feeding discovered /// peers into the pool as dial candidates. `None` disables PEX (the FFI/base path + existing callers), /// leaving the serve path byte-identical to before. +/// +/// `gossip` is the connected pool every accepted peer is registered in for the life of its serve loop +/// (**dig_ecosystem#3124**). `None` leaves the peer unregistered — correct for the test and FFI +/// callers below, which run no pool — and is why the 3-argument +/// [`serve_peer_rpc_listener`] keeps its signature. pub async fn serve_peer_rpc_listener_with( listener: tokio::net::TcpListener, node: Arc, responder: Arc, pex: Option>, + gossip: Option, ) -> Result<(), String> { let server_config = build_server_tls_config(&node)?; let acceptor = tokio_rustls::TlsAcceptor::from(server_config); @@ -3325,6 +3334,7 @@ pub async fn serve_peer_rpc_listener_with( let acceptor = acceptor.clone(); let responder = responder.clone(); let pex = pex.clone(); + let gossip = gossip.clone(); let spawned = spawn_with_permit(&conn_permits, async move { // mTLS handshake (client cert required by build_server_tls_config; a peer with no cert or // a failed handshake is dropped here — no unauthenticated peer traffic reaches the RPC). @@ -3337,8 +3347,39 @@ pub async fn serve_peer_rpc_listener_with( // body. `None` if (defensively) no client cert is present, which the verifier // should already have rejected. let caller = caller_from_tls(&tls, peer_addr); + // #3124: the identity the pool slot is keyed on comes from the CERTIFICATE the + // handshake just verified — the same derivation `caller_from_tls` uses — never + // from the wire body and never from the DHT contact (whose `peer_id` is a + // display-hex String, not this type). Read before the stream is consumed by the + // mux. `None` when no client cert parsed: there is then no authenticated identity + // to key a slot on, and serving uncounted is safer than inventing one. + let authenticated = peer_id_from_tls(&tls); let mut session = dig_nat::mux::PeerSession::server(tls); + + // COUNT the peer for as long as we serve it. `peer_addr` is its EPHEMERAL SOURCE + // port, which is why this goes through the direct-INBOUND entry point — the pool + // records the address for observability and never offers it as a dial target. The + // tier is `Direct`: this connection arrived over TCP with no relay. + let adopted = match authenticated { + Some(peer_id) => { + adopt_inbound_peer_in_pool( + gossip.as_ref(), + &peer_id, + peer_addr, + dig_nat::TraversalKind::Direct, + &session, + ) + .await + } + None => None, + }; + serve_peer_session_from_with(caller, &mut session, responder, pex).await; + + // The serve loop has returned, so this node is no longer serving the peer and must + // stop counting it. `disconnect` only ends the ACCOUNTING — the session is ours and + // is closed by dropping it here. + release_inbound_pool_slot(gossip.as_ref(), adopted).await; } Err(e) => tracing::debug!(error = %e, "peer mTLS handshake failed; dropped"), } @@ -3352,6 +3393,72 @@ pub async fn serve_peer_rpc_listener_with( } } +/// Register an ACCEPTED inbound peer in the dig-gossip connected pool for as long as this node serves +/// it, and stop counting it when the serve loop ends (**dig_ecosystem#3124**). +/// +/// # Why this exists +/// +/// The pool is what every subsystem reads to answer "am I connected", and until this call the node +/// registered NOTHING it accepted — only peers it dialed. A node serving inbound peers perfectly well +/// reported `connected_peers` as if it had none. +/// +/// # Why it registers by HANDLE +/// +/// The serve loop below needs `&mut PeerSession` to answer the peer's L7 RPC, and `PeerSession` is not +/// `Clone`. Handing the session to the pool would buy the count and stop serving the peer — strictly +/// worse than being uncounted. `adopt_direct_inbound_handle` takes a `ClosedHandle` instead, so +/// ownership stays here and the peer is both counted and served. +/// +/// Adoption is best-effort by design: it is ACCOUNTING, and every refusal the pool can return (the +/// accepted-direct cap, a ban, a full pool, a peer already holding a dialable slot) is a decision this +/// node made on purpose. None of them is a reason to refuse SERVICE to a peer whose handshake already +/// succeeded, so a refusal is logged and the connection is served uncounted — the behaviour that +/// shipped before this call existed. +/// +/// Returns the `PeerId` to deregister once serving ends, or `None` when nothing was registered. +async fn adopt_inbound_peer_in_pool( + gossip: Option<&dig_gossip::GossipHandle>, + peer_id: &dig_nat::PeerId, + remote: std::net::SocketAddr, + method: dig_nat::TraversalKind, + session: &dig_nat::mux::PeerSession, +) -> Option { + let gossip = gossip?; + // dig-nat reports the identity as its own `PeerId`; the pool keys on the gossip `PeerId` (chia + // `Bytes32`) over the SAME 32 bytes the mTLS handshake proved. + let pool_id = dig_gossip::PeerId::from(*peer_id.as_bytes()); + let observed = dig_gossip::ObservedSession::new(session.closed_handle(), move || { + // A newer connection for this identity displaced the slot. This session is now obsolete to the + // pool, but it is still OURS to end, and the serve loop below ends it on return — so there is + // nothing to do here but record it. Dropping the observer silently is what #71 forbids. + tracing::debug!(peer_id = %pool_id, "inbound pool slot superseded by a newer connection"); + }); + + match gossip + .adopt_direct_inbound_handle(pool_id, remote, method, observed, None) + .await + { + Ok(id) => Some(id), + Err(e) => { + tracing::debug!(peer_id = %pool_id, error = %e, "inbound peer not adopted into the pool; serving it uncounted"); + None + } + } +} + +/// Stop counting an inbound peer once its serve loop has ended. +/// +/// `disconnect` only stops ACCOUNTING for a slot registered by handle — it does not close the +/// transport, which is this side's to end and is ended by the serve loop returning. +async fn release_inbound_pool_slot( + gossip: Option<&dig_gossip::GossipHandle>, + adopted: Option, +) { + if let (Some(gossip), Some(peer_id)) = (gossip, adopted) { + let _ = gossip.disconnect(&peer_id).await; + } +} + /// Build the authenticated caller [`dig_dht::Contact`] from an accepted mTLS server connection: read /// the client's leaf certificate, derive its `peer_id = SHA-256(SPKI DER)` (the SAME derivation /// dig-nat enforces), and pair it with the remote socket address. Returns `None` if no client cert is @@ -3360,10 +3467,25 @@ fn caller_from_tls( tls: &tokio_rustls::server::TlsStream, remote_addr: std::net::SocketAddr, ) -> Option { + let peer_id = peer_id_from_tls(tls)?; + Some(crate::dht::caller_contact(&peer_id, remote_addr)) +} + +/// The AUTHENTICATED `peer_id = SHA-256(SPKI DER)` of an accepted mTLS peer, read from the leaf +/// certificate rustls verified during the handshake (**dig_ecosystem#3124**). +/// +/// Split out of [`caller_from_tls`] because the pool keys a slot on the identity itself, while the DHT +/// contact carries it as display hex — the two are not interchangeable, and deriving both from this +/// ONE function is what keeps them from drifting onto different sources. +/// +/// `None` when the peer presented no certificate or it does not parse; the client-cert verifier should +/// already have rejected such a peer. +fn peer_id_from_tls( + tls: &tokio_rustls::server::TlsStream, +) -> Option { let (_io, conn) = tls.get_ref(); let leaf = conn.peer_certificates()?.first()?; - let peer_id = dig_nat::peer_id_from_leaf_cert_der(leaf.as_ref())?; - Some(crate::dht::caller_contact(&peer_id, remote_addr)) + dig_nat::peer_id_from_leaf_cert_der(leaf.as_ref()) } /// Build the rustls `ServerConfig` for the mTLS peer-RPC listener from the node's CA-signed diff --git a/crates/dig-node-core/tests/inbound_pool_membership.rs b/crates/dig-node-core/tests/inbound_pool_membership.rs new file mode 100644 index 00000000..ec07e48d --- /dev/null +++ b/crates/dig-node-core/tests/inbound_pool_membership.rs @@ -0,0 +1,218 @@ +//! Integration test for **dig_ecosystem#3124 — an ACCEPTED inbound peer must appear in +//! `connected_peers`.** +//! +//! ## The defect +//! +//! `serve_peer_rpc_listener_with` accepted an inbound mTLS connection, derived the authenticated +//! `peer_id`, wrapped it in a yamux session and served it — and registered it nowhere. The connected +//! pool is what every subsystem reads to answer "am I connected", so a node happily serving inbound +//! peers reported `connected_peers = 0`. +//! +//! ## Why this test drives a REAL connection +//! +//! The wiring under test sits between the mTLS handshake and the serve loop, on a code path reachable +//! only by accepting a socket. A unit test around the pool cannot see it — the pool was always able to +//! hold the slot; nothing was calling it. So this dials the real listener with a real `dig-nat` client +//! over loopback, exactly as a peer would. +//! +//! ## The three things asserted, and why none implies the next +//! +//! * **COUNTED** — the pool sees the peer. This was zero. +//! * **still SERVED** — the peer's L7 RPC still answers AFTER adoption. Registering by value would buy +//! the count and silently stop serving the peer, which is strictly worse than being uncounted; the +//! count alone cannot distinguish the two. +//! * **RELEASED** — the slot goes away when the peer does. A membership that is never released reports +//! a peer count that only ever grows, which is a different lie from the one being fixed. + +use std::sync::Arc; +use std::time::Duration; + +use dig_node_core::peer::{ + load_or_generate_node_cert, serve_peer_rpc_listener_with, write_framed, PeerRpcResponder, +}; +use serde_json::{json, Value}; + +/// A deterministic 32-byte identity seed derived from a label — no hard-coded key material. +fn node_seed(label: &str) -> [u8; 32] { + use sha2::{Digest, Sha256}; + Sha256::digest(label.as_bytes()).into() +} + +fn test_identity(label: &str) -> Arc { + let dir = tempfile::tempdir().expect("cert tempdir"); + load_or_generate_node_cert(dir.path(), &node_seed(label)).expect("node cert") +} + +struct TestResponder; + +#[async_trait::async_trait] +impl PeerRpcResponder for TestResponder { + async fn handle_json_rpc(&self, req: Value, _conn_key: &str) -> Value { + let id = req.get("id").cloned().unwrap_or(json!(1)); + let method = req.get("method").and_then(Value::as_str).unwrap_or(""); + json!({"jsonrpc":"2.0","id":id,"result":{"served_method": method}}) + } + async fn handle_availability(&self, _items: Value, _conn_key: &str) -> Value { + json!({"items": []}) + } + async fn stream_range( + &self, + _req: Value, + _conn_key: &str, + out: &mut (dyn tokio::io::AsyncWrite + Send + Unpin), + ) -> std::io::Result<()> { + write_framed(out, &json!({"complete": true})).await + } +} + +/// Start a real gossip service with an EMPTY pool, on its own ephemeral port. +async fn running_gossip() -> (dig_gossip::GossipService, dig_gossip::GossipHandle, tempfile::TempDir) +{ + let dir = tempfile::tempdir().expect("gossip tempdir"); + let cfg = dig_gossip::GossipConfig { + network_id: chia_protocol::Bytes32::new([1u8; 32]), + cert_path: dir.path().join("node.cert").display().to_string(), + key_path: dir.path().join("node.key").display().to_string(), + peers_file_path: dir.path().join("peers.json"), + peer_pool: Some(dig_gossip::PeerPoolConfig::default()), + listen_addr: "127.0.0.1:0".parse().expect("listen addr"), + ..Default::default() + }; + let service = dig_gossip::GossipService::new(cfg).expect("gossip config"); + let handle = service.start().await.expect("gossip start"); + (service, handle, dir) +} + +/// Poll `peer_count` until it reaches `want`, or fail. Adoption happens on the listener's spawned +/// task, so it is concurrent with the client's `connect` returning — a bare read races it. +async fn await_peer_count(handle: &dig_gossip::GossipHandle, want: usize, what: &str) { + let deadline = std::time::Instant::now() + Duration::from_secs(10); + loop { + let got = handle.peer_count().await; + if got == want { + return; + } + assert!( + std::time::Instant::now() < deadline, + "{what}: peer_count settled at {got}, expected {want}" + ); + tokio::time::sleep(Duration::from_millis(25)).await; + } +} + +/// **#3124 end to end: a peer that DIALS this node becomes a counted pool member, keeps being served, +/// and is released when it leaves.** +#[tokio::test] +async fn an_accepted_inbound_peer_is_counted_served_and_released() { + dig_node_core::peer::install_crypto_provider(); + + let (service, gossip, _gdir) = running_gossip().await; + assert_eq!( + gossip.peer_count().await, + 0, + "the pool starts empty, so any count below is caused by the inbound connection" + ); + + let server_identity = test_identity("3124-inbound-server"); + let server_peer_id = server_identity.peer_id(); + let listener = tokio::net::TcpListener::bind("127.0.0.1:0") + .await + .expect("bind"); + let listen_addr = listener.local_addr().expect("local addr"); + + let responder: Arc = Arc::new(TestResponder); + let server = tokio::spawn(serve_peer_rpc_listener_with( + listener, + server_identity, + responder, + None, + Some(gossip.clone()), + )); + + // A peer DIALS this node — the direction that was never counted. + let client_identity = test_identity("3124-inbound-client"); + let client_peer_id = client_identity.peer_id(); + let target = dig_nat::PeerTarget::with_addr(server_peer_id, listen_addr, "DIG_MAINNET"); + let config = dig_nat::NatConfig::builder() + .enabled_methods(vec![dig_nat::TraversalKind::Direct]) + .per_method_timeout(Duration::from_secs(5)) + .build(); + let mut conn = dig_nat::connect(&target, &client_identity, &config) + .await + .expect("the peer connects over mTLS"); + + // (a) COUNTED — this was zero for every inbound peer, on every node. + await_peer_count(&gossip, 1, "after an inbound peer connects").await; + + let pool_id = dig_gossip::PeerId::from(*client_peer_id.as_bytes()); + assert!( + gossip.is_pool_peer(&pool_id), + "the slot is keyed on the identity the CLIENT's certificate proved, not the server's" + ); + + let detailed = gossip.connected_pool_peers_detailed(); + let peer = detailed + .iter() + .find(|p| p.peer_id == pool_id) + .expect("the inbound peer is in the pool"); + assert!( + !peer.is_outbound, + "this node never dialed the peer, so the slot must not be charged outbound diversity" + ); + assert_eq!( + peer.dial_addr, None, + "the peer's source port is ephemeral and must never be offered as a dial target" + ); + assert_ne!( + peer.session_addr.port(), + listen_addr.port(), + "the recorded address is the CLIENT's source port, which is exactly why it is not dialable" + ); + assert!( + gossip.dialable_pool_peers().is_empty(), + "an accepted peer contributes no dial target to peer selection" + ); + + // (b) STILL SERVED — adopting must not cost the serve loop its session. A count-only assertion + // passes against the shape that buys the count and stops answering the peer. + { + let mut stream = conn.session.open_stream().await.expect("open stream"); + let req = json!({"jsonrpc":"2.0","id":7,"method":"dig.getNetworkInfo"}); + write_framed(&mut stream, &req).await.expect("write"); + let resp = read_one_frame(&mut stream).await; + assert_eq!( + resp["result"]["served_method"], "dig.getNetworkInfo", + "the node must still answer the peer it just registered" + ); + } + assert_eq!( + gossip.peer_count().await, + 1, + "serving the peer neither duplicates nor drops its slot" + ); + + // (c) RELEASED — the peer goes away and the pool stops counting it. A slot that is never released + // makes `connected_peers` a high-water mark rather than a count. + drop(conn); + await_peer_count(&gossip, 0, "after the inbound peer disconnects").await; + + server.abort(); + service.stop().await.expect("stop"); +} + +/// Read one length-framed JSON value. +async fn read_one_frame(stream: &mut S) -> Value { + use tokio::io::AsyncReadExt; + let mut len = [0u8; 4]; + tokio::time::timeout(Duration::from_secs(5), stream.read_exact(&mut len)) + .await + .expect("frame header arrives") + .expect("frame header reads"); + let n = u32::from_be_bytes(len) as usize; + let mut body = vec![0u8; n]; + tokio::time::timeout(Duration::from_secs(5), stream.read_exact(&mut body)) + .await + .expect("frame body arrives") + .expect("frame body reads"); + serde_json::from_slice(&body).expect("frame is JSON") +}