diff --git a/src/rendezvous_mediator.rs b/src/rendezvous_mediator.rs index ed951314d..1feeb07c9 100644 --- a/src/rendezvous_mediator.rs +++ b/src/rendezvous_mediator.rs @@ -997,7 +997,8 @@ impl RendezvousMediator { let socket = connect_tcp(&*self.host, CONNECT_TIMEOUT).await?; let local_addr = socket.local_addr(); // key important here for punch hole to tell my gateway incoming peer is safe. - // it can not be async here, because local_addr can not be reused, we must close the connection before use it again. + // Awaited rather than spawned so the mapping exists before `PunchHoleSent` goes out; + // `local_addr` itself is shared, not exclusive - every socket here sets SO_REUSEADDR. allow_err!(socket_client::connect_tcp_local(peer_addr, Some(local_addr), 30).await); socket }; @@ -1005,7 +1006,10 @@ impl RendezvousMediator { msg_out.set_punch_hole_sent(msg_punch); let bytes = msg_out.write_to_bytes()?; socket.send_raw(bytes).await?; - crate::accept_connection(server.clone(), socket, peer_addr, true, meta).await; + let local_addr = socket.local_addr(); + // The listener inside takes this address over, so the mediator's socket goes first. + drop(socket); + punch_tcp_until_connected(server, peer_addr, local_addr, meta).await; Ok(()) } @@ -1264,6 +1268,173 @@ async fn udp_nat_listen( Ok(()) } +/// Where the repeats start, and the factor they slow by. The controller's SYN arrives once, at an +/// instant we are never told, so the mapping has to be held open across the whole of its dial - +/// and `Client::connect` sets that to `CONNECT_TIMEOUT` outright whenever neither side is +/// symmetric, which is the only case that reaches a TCP punch at all. No wall clock of our own can +/// name that instant, so the repeats run for as long as the accept below is still willing to take +/// a connection, and back off instead: dense while the controller is most likely to be dialing, +/// sparse afterwards, which is `punch_udp`'s shape for the same reason. +const PUNCH_INTERVAL: f32 = 0.15; +const PUNCH_BACKOFF: f32 = 1.5; +const PUNCH_MAX_INTERVAL: f32 = 2.0; +/// Comfortably past any round trip, because a punch is only cancel-safe while it is still in +/// SYN_SENT. Once the controller's SYN crosses it the socket is half way through a handshake, and +/// dropping it there resets the connection the controller is opening - one the listener below +/// would otherwise have taken. +/// +/// Waiting also costs nothing, because the two gateways that lose us the mapping need opposite +/// things and a long timeout serves both. One that answers with RST kills the attempt at once, +/// leaving nothing to retransmit, and only a fresh attempt can punch again - which the loop makes. +/// One that drops the SYN in silence leaves the socket in SYN_SENT, where it holds the mapping +/// open and the kernel re-sends the SYN on its own; a shorter timeout would only take that away +/// and buy attempts that are not needed. +const PUNCH_TIMEOUT: u64 = 3000; + +/// The punch above leaves before hbbs has told the controller where to dial, so it is never in +/// flight at the same time as the controller's SYN: it opens our NAT, meets nothing, and a gateway +/// that answers it with RST takes the mapping down with it - leaving the listener below waiting on +/// a hole that no longer exists. Punching again across the window in which the controller dials +/// rebuilds it, and once the controller sits in SYN_SENT one of those punches meets its SYN and +/// completes as a simultaneous open: a second way in, which a single punch never had. +async fn punch_tcp_until_connected( + server: ServerPtr, + peer_addr: SocketAddr, + local_addr: SocketAddr, + meta: ConnectionMeta, +) { + use hbb_common::tcp::new_listener; + // The same clock the sleeps and the connect timeouts below run on, so a paused timer stops + // the loop with them instead of spinning against a wall clock that never advances. + use hbb_common::tokio::time::Instant; + + // Not fatal on its own - the punch below can still meet the controller's SYN without it, and + // that half is the one a listener the OS refused to bind could not have covered anyway. + let listener = match new_listener(local_addr, true).await { + Ok(listener) => { + log::info!("Server listening on: {local_addr}"); + Some(listener) + } + Err(err) => { + log::warn!("Failed to listen on {local_addr} after punching: {err}"); + None + } + }; + // Bounds both halves: the punch keeps the mapping open only while the accept is still + // willing to take a connection through it. + let until = Instant::now() + Duration::from_millis(CONNECT_TIMEOUT); + let punch = async { + let mut interval = PUNCH_INTERVAL; + let mut round = 0; + loop { + sleep(interval).await; + interval = (interval * PUNCH_BACKOFF).min(PUNCH_MAX_INTERVAL); + // The deadline decides whether another punch starts, never how long one already in + // flight may take: cutting that one short is the mid-handshake reset PUNCH_TIMEOUT + // exists to avoid. + if Instant::now() >= until { + log::debug!("No punch to {peer_addr} was met in {round} rounds"); + return None; + } + let tm = Instant::now(); + match socket_client::connect_tcp_local(peer_addr, Some(local_addr), PUNCH_TIMEOUT).await + { + // The controller's SYN crossed this punch, so the stream is the connection it + // dialed, not a spare one: dropping it would reset that connection. + Ok(stream) => return Some(stream), + // How long it took says which it was: a gateway that answers with RST fails this + // in well under a round trip, where one that drops it in silence runs to the + // timeout. The error itself cannot tell them apart - `connect_tcp_local` keeps + // no errno. + Err(err) => { + log::debug!( + "Punch {round} to {peer_addr} failed in {:?}: {err}", + tm.elapsed() + ); + round += 1; + } + } + } + }; + let Some(listener) = listener else { + if let Some(stream) = punch.await { + serve_punched(server, stream, peer_addr, meta).await; + } + return; + }; + // Accepting in a loop, not once: a stray connection - or a transient `accept` error, which + // the punches make likelier by resetting queued ones - must not spend the whole window the + // controller still has to arrive in. + let accept = async { + loop { + let left = until.saturating_duration_since(Instant::now()).as_millis() as u64; + if left == 0 { + log::info!("Nothing connected to the hole punched to {peer_addr}"); + return None; + } + match hbb_common::timeout(left, listener.accept()).await { + // A first filter only, and a weak one where a carrier NAT shares an address + // between subscribers: what actually keeps `meta`'s control permissions from + // being handed to a second peer is that exactly one connection is ever served. + Ok(Ok(accepted)) if accepted.1.ip() == peer_addr.ip() => return Some(accepted), + Ok(Ok((_, addr))) => { + log::warn!("Ignoring {addr} on the hole punched to {peer_addr}") + } + Ok(Err(err)) => log::warn!("Failed to accept from {peer_addr}: {err}"), + Err(_) => return None, + } + } + }; + // Only the accept races the punch. Racing `accept_connection` instead would race the whole + // session it goes on to run, so a punch landing mid-session would tear that session down. + // + // Whichever arrives first is the one connection this request produces. Serving the loser too + // would give a second peer the control permissions hbbs granted for this one controller, and + // no test on the connection itself can tell the two apart before `create_tcp_connection` has + // spoken to it - so the invariant is kept here, by there being no second serve. + let punched = select! { + // The accept is the path that cannot reach anything but the controller, so it wins a tie. + biased; + Some((stream, addr)) = accept => { + return accept_punched_connection(server, stream, addr, meta).await; + } + Some(stream) = punch => stream, + else => return, + }; + serve_punched(server, punched, peer_addr, meta).await; +} + +async fn serve_punched( + server: ServerPtr, + stream: Stream, + peer_addr: SocketAddr, + meta: ConnectionMeta, +) { + log::info!("Punched tcp hole to {peer_addr}, connected on the punch itself"); + allow_err!(crate::server::create_tcp_connection(server, stream, peer_addr, true, meta).await); +} + +/// The accept half of `accept_connection`, kept here because only the accept may race the punch. +async fn accept_punched_connection( + server: ServerPtr, + stream: tokio::net::TcpStream, + addr: SocketAddr, + meta: ConnectionMeta, +) { + use crate::server::create_tcp_connection; + + stream.set_nodelay(true).ok(); + match stream.local_addr() { + Ok(stream_addr) => { + let stream = Stream::from(stream, stream_addr); + if let Err(err) = create_tcp_connection(server, stream, addr, true, meta).await { + log::warn!("Failed to serve the connection from {addr}: {err}"); + } + } + Err(err) => log::warn!("Failed to read the address accepted from {addr}: {err}"), + } +} + // When config is not yet synced from root, register_pk may have already been sent with a new generated pk. // After config sync completes, the pk may change. This struct detects pk changes and triggers // a re-registration by setting key_confirmed to false. @@ -1291,7 +1462,25 @@ impl Drop for CheckIfResendPk { #[cfg(test)] mod tests { - use super::{mpsc, IceRoute, ICE_DEDUP_WINDOW, MAX_PENDING_REMOTE_ICE}; + use super::{mpsc, socket_client, tokio, IceRoute, ICE_DEDUP_WINDOW, MAX_PENDING_REMOTE_ICE}; + use hbb_common::tcp::new_listener; + use std::net::SocketAddr; + + // A SOCKS proxy makes `connect_tcp_local` dial the proxy and ignore the local address, so + // nothing these two assert can hold. Read once, from the same global config production reads. + fn proxied() -> bool { + hbb_common::config::Config::get_socks().is_some() + } + + /// Both held while their addresses are read, so the pair cannot be the same port - which + /// `SO_REUSEPORT` would let bind twice rather than refuse, leaving the tests degenerate. + async fn free_loopback_pair() -> (SocketAddr, SocketAddr) { + let (a, b) = ( + tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap(), + tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap(), + ); + (a.local_addr().unwrap(), b.local_addr().unwrap()) + } fn queue(route: &mut IceRoute, candidate: &str) -> bool { route.queue(candidate.to_owned()) @@ -1357,4 +1546,95 @@ mod tests { assert!(queue(&mut route, &recent)); assert!(rx.try_recv().is_err()); } + + // The second way in that the repeat punch opens: a punch reaching a peer already in SYN_SENT + // is answered by that socket rather than reset, and the two ends come up on one connection. + // A punch that misses the crossing is reset outright here, loopback having no NAT to absorb + // it and no round trip to hide behind - so a single punch lands only by luck, and repeating + // is what makes it land at all. That is the premise of the repeat, asserted directly. + #[tokio::test(flavor = "multi_thread", worker_threads = 2)] + async fn a_punch_that_meets_the_peers_syn_connects_both_ends() { + // The crossing needs both connects genuinely in flight at once. Loopback answers a SYN to + // a port nobody is listening on with an instant RST, so on one CPU the first connect runs + // to completion before the second is scheduled and no round can ever cross - a property of + // the box, which this test cannot tell apart from a broken punch. + if proxied() || std::thread::available_parallelism().map_or(true, |cpus| cpus.get() < 2) { + return; + } + for _ in 0..32 { + let (a, b) = free_loopback_pair().await; + // Held for the whole crossing, because production always has one here and the design + // rests on which of the two the kernel hands the connection to: the punch and the + // peer's SYN share a four-tuple exactly, the listener only matches the address, and + // the punch has to win that or every crossing would be swallowed as a plain accept. + let listener = new_listener(a, true).await.unwrap(); + let to_b = tokio::spawn(socket_client::connect_tcp_local(b, Some(a), 3000)); + let to_a = tokio::spawn(socket_client::connect_tcp_local(a, Some(b), 3000)); + let (at_a, at_b) = tokio::join!(to_b, to_a); + let (Ok(Ok(mut at_a)), Ok(Ok(mut at_b))) = (at_a, at_b) else { + continue; + }; + at_a.send_bytes(bytes::Bytes::from_static(b"punch")) + .await + .unwrap(); + let got = at_b.next_timeout(3000).await.unwrap().unwrap(); + assert_eq!(&got[..], b"punch", "both ends must share one connection"); + assert!( + hbb_common::timeout(200, listener.accept()).await.is_err(), + "the crossing must reach the punch, not be accepted as an inbound connection" + ); + return; + } + panic!("no punch met the peer's SYN in 32 rounds on a machine that can cross them"); + } + + // The punch binds the address the listener already holds, so it has to go through the same + // `connect_tcp_local` production uses - a punch built by hand here would still pass if + // `new_socket` ever stopped setting the reuse flags, while every real punch failed to bind. + // The peer's view of the source port is what proves the bind took: a fallback to an ephemeral + // one would connect just as happily. + #[tokio::test(flavor = "multi_thread", worker_threads = 2)] + async fn a_punch_binds_the_address_the_listener_holds() { + if proxied() { + return; + } + // `free_loopback_addr` hands back a port it no longer holds, so another process can take + // it in between; retry rather than fail for something the punch had no part in. + for _ in 0..8 { + let (local, peer_addr) = free_loopback_pair().await; + let (Ok(listener), Ok(peer)) = ( + new_listener(local, true).await, + new_listener(peer_addr, true).await, + ) else { + continue; + }; + let punch = tokio::spawn(socket_client::connect_tcp_local( + peer_addr, + Some(local), + 1500, + )); + let (_peer_side, seen_as) = hbb_common::timeout(3000, peer.accept()) + .await + .expect("the punch must reach the peer") + .unwrap(); + assert_eq!( + seen_as.port(), + local.port(), + "the punch must leave from the address the listener holds, not an ephemeral one" + ); + // Held, not asserted and dropped: the coexistence below is only exercised while this + // socket is still on the address, which is the state production spends its window in. + let _punched = punch.await.unwrap().expect("the punch must connect"); + + let dialed = tokio::spawn(tokio::net::TcpStream::connect(local)); + let accepted = hbb_common::timeout(3000, listener.accept()).await; + assert!( + matches!(accepted, Ok(Ok(_))), + "the listener must still take connections while a punch shares its address: {accepted:?}" + ); + assert!(dialed.await.unwrap().is_ok()); + return; + } + panic!("could not hold two free loopback addresses in 8 tries"); + } }