diff --git a/crates/dig-node-core/Cargo.toml b/crates/dig-node-core/Cargo.toml index 21a5ecf4..86f87fb2 100644 --- a/crates/dig-node-core/Cargo.toml +++ b/crates/dig-node-core/Cargo.toml @@ -349,7 +349,7 @@ dig-nat = "0.20" # certificate at all — a mutual-TLS bypass that no consumer could see from its own manifest. Its # publish.yml guard is unconditional and deliberate: do not weaken it, do not `--allow-dirty` past it. # dig_ecosystem#900 ("publish dig-gossip") was closed as superseded by #2647. -dig-gossip = { git = "https://github.com/DIG-Network/dig-gossip", rev = "48a14ed2ed395a41c913d5c8f944e6ea0b747075", default-features = false, features = ["rustls", "relay"] } +dig-gossip = { git = "https://github.com/DIG-Network/dig-gossip", rev = "f21f993e7c46d0f506dd200743b11b3b19f4eef6", default-features = false, features = ["rustls", "relay"] } # The canonical DIG network id (genesis challenge) for the gossip handshake / introducer scope. # # 0.9 is the TIP OF DIG-NODE'S CHIA LINE, and that — not 0.10 — is the correct target here (#2072). @@ -542,7 +542,7 @@ dig-keystore = { version = "0.13", features = ["testing"] } # rather than on the normal dependency so the shipped binary never carries the test hook: cargo # unifies the feature into test builds only, since dev-dependencies are not built for `cargo build`. # NC-7 exception: same permanent git pin as the runtime dep above (dig_ecosystem#2647). -dig-gossip = { git = "https://github.com/DIG-Network/dig-gossip", rev = "48a14ed2ed395a41c913d5c8f944e6ea0b747075", default-features = false, features = ["rustls", "relay", "test-util"] } +dig-gossip = { git = "https://github.com/DIG-Network/dig-gossip", rev = "f21f993e7c46d0f506dd200743b11b3b19f4eef6", default-features = false, features = ["rustls", "relay", "test-util"] } # Generates the P-256 leaf key pairs the opcode-222 holdings tests sign with: `public_key_der()` IS the # leaf `SubjectPublicKeyInfo` DER the wire carries as `provider_spki`, and whose SHA-256 is the # announcing peer_id — so a test signer is a real §5.2 identity rather than a stand-in that could hide diff --git a/crates/dig-node-core/src/peer.rs b/crates/dig-node-core/src/peer.rs index ba718cbb..a6be8cee 100644 --- a/crates/dig-node-core/src/peer.rs +++ b/crates/dig-node-core/src/peer.rs @@ -2862,6 +2862,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); @@ -2881,7 +2884,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`] @@ -3213,7 +3216,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 @@ -3221,11 +3224,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); @@ -3248,6 +3257,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). @@ -3260,8 +3270,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"), } @@ -3275,6 +3316,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 @@ -3283,10 +3390,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