mirror of
https://github.com/rustdesk/rustdesk.git
synced 2026-09-15 00:41:01 +03:00
tcp: repeat the punch across the controller's dial window
The single punch 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 waiting on a hole that no longer exists. Punch again while the controller may still be dialing, and race those punches against the accept. That is two ways in where there was one: the mapping is rebuilt if a RST took it, and once the controller sits in SYN_SENT one of the punches meets its SYN and completes as a simultaneous open - which a punch sent before the controller had been told anything never could. The crossing reaches the punch rather than the listener because the two sockets share the address but only the punch matches the four-tuple, which the tests now pin down. There is no instant to aim at. `Client::connect` gives the controller's dial CONNECT_TIMEOUT outright whenever neither side is symmetric, and that is the only case reaching a TCP punch at all, so the window is nothing this side can name in advance. The repeats therefore run for exactly as long as the accept is still willing to take a connection through the hole, 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. 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 its timeout exists to avoid, and it would take away the retransmits a silently dropped punch keeps the mapping open with, which are what stands in for repeats on a gateway that never answers. Only the accept races the punch, never `accept_connection`: that one does not return until the session it goes on to run has ended, so racing it would tear a live session down. Whichever arrives first is the one connection the request produces. `meta` carries the control permissions hbbs granted for this one controller, so serving the loser as well would hand them to a second peer - and nothing about a connection tells the two apart before `create_tcp_connection` has spoken to it, least of all its address, which a carrier NAT shares between subscribers. The address is checked all the same, to keep a stray out of the race at all, but what holds the invariant is that there is no second serve. Those permissions are a ceiling and not a grant either way: `Connection` gates every message on `authorized`, and latches the login scope of the first request it accepts, so a peer that reached the hole still arrives with nothing. The accept loops rather than taking a single connection, so that neither a stray nor a transient accept error spends the window the controller still has to arrive in. libp2p's DCUtR reaches the same place by having both peers dial at one instant agreed over the relay. Nothing we send reaches the controller directly, so we cover its dial window rather than name an instant inside it. Co-Authored-By: Claude Opus 5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_019aokqJuhjvB3kijXtAg5Ns
This commit is contained in:
rustdesk
@@ -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");
|
||||
}
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user