mirror of
https://github.com/rustdesk/rustdesk.git
synced 2026-05-20 20:48:39 +03:00
Compare commits
2 Commits
webrtc
...
dependabot
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
2734d1baa7 | ||
|
|
377547fa11 |
@@ -47,7 +47,7 @@ screencapturekit = ["cpal/screencapturekit"]
|
|||||||
[dependencies]
|
[dependencies]
|
||||||
async-trait = "0.1"
|
async-trait = "0.1"
|
||||||
scrap = { path = "libs/scrap", features = ["wayland"] }
|
scrap = { path = "libs/scrap", features = ["wayland"] }
|
||||||
hbb_common = { path = "libs/hbb_common", features = ["webrtc"] }
|
hbb_common = { path = "libs/hbb_common" }
|
||||||
serde_derive = "1.0"
|
serde_derive = "1.0"
|
||||||
serde = "1.0"
|
serde = "1.0"
|
||||||
serde_json = "1.0"
|
serde_json = "1.0"
|
||||||
|
|||||||
Submodule libs/hbb_common updated: 5a78ec4230...9043c15acc
@@ -276,12 +276,21 @@ impl PipeWireRecorder {
|
|||||||
// see: https://gitlab.freedesktop.org/pipewire/pipewire/-/issues/982
|
// see: https://gitlab.freedesktop.org/pipewire/pipewire/-/issues/982
|
||||||
src.set_property("always-copy", &true)?;
|
src.set_property("always-copy", &true)?;
|
||||||
|
|
||||||
|
// COSMIC/Wayland fix: insert videoconvert between pipewiresrc and appsink.
|
||||||
|
// xdg-desktop-portal-cosmic's modifier negotiation fails when the downstream
|
||||||
|
// format set is too narrow (appsink only accepts BGRx/RGBx), producing
|
||||||
|
// "no more output formats" / not-negotiated (-4). videoconvert accepts any
|
||||||
|
// system-memory video/x-raw format, widening negotiation so the portal can
|
||||||
|
// settle on a format it can deliver via its SHM path.
|
||||||
|
let convert = gst::ElementFactory::make("videoconvert", None)?;
|
||||||
|
|
||||||
let sink = gst::ElementFactory::make("appsink", None)?;
|
let sink = gst::ElementFactory::make("appsink", None)?;
|
||||||
sink.set_property("drop", &true)?;
|
sink.set_property("drop", &true)?;
|
||||||
sink.set_property("max-buffers", &1u32)?;
|
sink.set_property("max-buffers", &1u32)?;
|
||||||
|
|
||||||
pipeline.add_many(&[&src, &sink])?;
|
pipeline.add_many(&[&src, &convert, &sink])?;
|
||||||
src.link(&sink)?;
|
src.link(&convert)?;
|
||||||
|
convert.link(&sink)?;
|
||||||
|
|
||||||
let appsink = sink
|
let appsink = sink
|
||||||
.dynamic_cast::<AppSink>()
|
.dynamic_cast::<AppSink>()
|
||||||
|
|||||||
21
src/cli.rs
21
src/cli.rs
@@ -3,7 +3,7 @@ use async_trait::async_trait;
|
|||||||
use hbb_common::{
|
use hbb_common::{
|
||||||
config::PeerConfig,
|
config::PeerConfig,
|
||||||
config::READ_TIMEOUT,
|
config::READ_TIMEOUT,
|
||||||
futures::StreamExt,
|
futures::{SinkExt, StreamExt},
|
||||||
log,
|
log,
|
||||||
message_proto::*,
|
message_proto::*,
|
||||||
protobuf::Message as _,
|
protobuf::Message as _,
|
||||||
@@ -46,7 +46,6 @@ impl Session {
|
|||||||
false,
|
false,
|
||||||
None,
|
None,
|
||||||
None,
|
None,
|
||||||
None,
|
|
||||||
);
|
);
|
||||||
session
|
session
|
||||||
}
|
}
|
||||||
@@ -54,7 +53,7 @@ impl Session {
|
|||||||
|
|
||||||
#[async_trait]
|
#[async_trait]
|
||||||
impl Interface for Session {
|
impl Interface for Session {
|
||||||
fn get_lch(&self) -> Arc<RwLock<LoginConfigHandler>> {
|
fn get_login_config_handler(&self) -> Arc<RwLock<LoginConfigHandler>> {
|
||||||
return self.lc.clone();
|
return self.lc.clone();
|
||||||
}
|
}
|
||||||
|
|
||||||
@@ -62,20 +61,14 @@ impl Interface for Session {
|
|||||||
match msgtype {
|
match msgtype {
|
||||||
"input-password" => {
|
"input-password" => {
|
||||||
self.sender
|
self.sender
|
||||||
.send(Data::Login((
|
.send(Data::Login((self.password.clone(), true)))
|
||||||
String::new(),
|
|
||||||
String::new(),
|
|
||||||
self.password.clone(),
|
|
||||||
true,
|
|
||||||
)))
|
|
||||||
.ok();
|
.ok();
|
||||||
}
|
}
|
||||||
"re-input-password" => {
|
"re-input-password" => {
|
||||||
log::error!("{}: {}", title, text);
|
log::error!("{}: {}", title, text);
|
||||||
match rpassword::prompt_password("Enter password: ") {
|
match rpassword::prompt_password("Enter password: ") {
|
||||||
Ok(password) => {
|
Ok(password) => {
|
||||||
let login_data =
|
let login_data = Data::Login((password, true));
|
||||||
Data::Login((String::new(), String::new(), password, true));
|
|
||||||
self.sender.send(login_data).ok();
|
self.sender.send(login_data).ok();
|
||||||
}
|
}
|
||||||
Err(e) => {
|
Err(e) => {
|
||||||
@@ -100,8 +93,6 @@ impl Interface for Session {
|
|||||||
self.lc.write().unwrap().handle_peer_info(&pi);
|
self.lc.write().unwrap().handle_peer_info(&pi);
|
||||||
}
|
}
|
||||||
|
|
||||||
fn set_multiple_windows_session(&self, _sessions: Vec<WindowsSession>) {}
|
|
||||||
|
|
||||||
async fn handle_hash(&self, pass: &str, hash: Hash, peer: &mut Stream) {
|
async fn handle_hash(&self, pass: &str, hash: Hash, peer: &mut Stream) {
|
||||||
log::info!(
|
log::info!(
|
||||||
"password={}",
|
"password={}",
|
||||||
@@ -146,8 +137,8 @@ pub async fn connect_test(id: &str, key: String, token: String) {
|
|||||||
Err(err) => {
|
Err(err) => {
|
||||||
log::error!("Failed to connect {}: {}", &id, err);
|
log::error!("Failed to connect {}: {}", &id, err);
|
||||||
}
|
}
|
||||||
Ok(((mut stream, _direct, _secure, _kcp, _typ), direct)) => {
|
Ok((mut stream, direct)) => {
|
||||||
log::info!("direct: {:?}", direct);
|
log::info!("direct: {}", direct);
|
||||||
// rpassword::prompt_password("Input anything to exit").ok();
|
// rpassword::prompt_password("Input anything to exit").ok();
|
||||||
loop {
|
loop {
|
||||||
tokio::select! {
|
tokio::select! {
|
||||||
|
|||||||
247
src/client.rs
247
src/client.rs
@@ -65,12 +65,11 @@ use hbb_common::{
|
|||||||
self,
|
self,
|
||||||
net::UdpSocket,
|
net::UdpSocket,
|
||||||
sync::{
|
sync::{
|
||||||
mpsc::{error::TryRecvError, unbounded_channel, UnboundedReceiver},
|
mpsc::{unbounded_channel, UnboundedReceiver},
|
||||||
oneshot,
|
oneshot,
|
||||||
},
|
},
|
||||||
time::{interval, Duration, Instant},
|
time::{interval, Duration, Instant},
|
||||||
},
|
},
|
||||||
webrtc::WebRTCStream,
|
|
||||||
AddrMangle, ResultType, Stream,
|
AddrMangle, ResultType, Stream,
|
||||||
};
|
};
|
||||||
pub use helper::*;
|
pub use helper::*;
|
||||||
@@ -331,19 +330,6 @@ impl Client {
|
|||||||
} else {
|
} else {
|
||||||
(None, None)
|
(None, None)
|
||||||
};
|
};
|
||||||
let ipv6 = if crate::get_ipv6_punch_enabled() {
|
|
||||||
crate::get_ipv6_socket().await
|
|
||||||
} else {
|
|
||||||
None
|
|
||||||
};
|
|
||||||
let webrtc_offerer =
|
|
||||||
match WebRTCStream::new("", interface.is_force_relay(), CONNECT_TIMEOUT).await {
|
|
||||||
Ok(stream) => Some(stream),
|
|
||||||
Err(err) => {
|
|
||||||
log::warn!("webrtc offerer setup failed: {}", err);
|
|
||||||
None
|
|
||||||
}
|
|
||||||
};
|
|
||||||
let fut = Self::_start_inner(
|
let fut = Self::_start_inner(
|
||||||
peer.to_owned(),
|
peer.to_owned(),
|
||||||
key.to_owned(),
|
key.to_owned(),
|
||||||
@@ -352,8 +338,6 @@ impl Client {
|
|||||||
interface.clone(),
|
interface.clone(),
|
||||||
udp.clone(),
|
udp.clone(),
|
||||||
Some(stop_udp_tx),
|
Some(stop_udp_tx),
|
||||||
ipv6,
|
|
||||||
webrtc_offerer,
|
|
||||||
rendezvous_server.clone(),
|
rendezvous_server.clone(),
|
||||||
servers.clone(),
|
servers.clone(),
|
||||||
contained,
|
contained,
|
||||||
@@ -371,8 +355,6 @@ impl Client {
|
|||||||
interface,
|
interface,
|
||||||
(None, None),
|
(None, None),
|
||||||
None,
|
None,
|
||||||
None,
|
|
||||||
None,
|
|
||||||
rendezvous_server,
|
rendezvous_server,
|
||||||
servers,
|
servers,
|
||||||
contained,
|
contained,
|
||||||
@@ -384,67 +366,6 @@ impl Client {
|
|||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
fn is_expected_webrtc_ice_candidate(ice: &IceCandidate, session_key: &str) -> bool {
|
|
||||||
!session_key.is_empty() && ice.session_key == session_key && !ice.candidate.is_empty()
|
|
||||||
}
|
|
||||||
|
|
||||||
fn spawn_webrtc_ice_bridge(
|
|
||||||
mut socket: Stream,
|
|
||||||
mut local_ice_rx: Option<UnboundedReceiver<String>>,
|
|
||||||
webrtc: WebRTCStream,
|
|
||||||
peer: String,
|
|
||||||
session_key: String,
|
|
||||||
) -> oneshot::Sender<()> {
|
|
||||||
let (stop_tx, mut stop_rx) = oneshot::channel::<()>();
|
|
||||||
tokio::spawn(async move {
|
|
||||||
loop {
|
|
||||||
match stop_rx.try_recv() {
|
|
||||||
Ok(_) | Err(tokio::sync::oneshot::error::TryRecvError::Closed) => break,
|
|
||||||
Err(tokio::sync::oneshot::error::TryRecvError::Empty) => {}
|
|
||||||
}
|
|
||||||
|
|
||||||
if let Some(rx) = local_ice_rx.as_mut() {
|
|
||||||
loop {
|
|
||||||
match rx.try_recv() {
|
|
||||||
Ok(candidate) => {
|
|
||||||
let mut msg = RendezvousMessage::new();
|
|
||||||
msg.set_ice_candidate(IceCandidate {
|
|
||||||
id: peer.clone(),
|
|
||||||
session_key: session_key.clone(),
|
|
||||||
candidate,
|
|
||||||
..Default::default()
|
|
||||||
});
|
|
||||||
if let Err(err) = socket.send(&msg).await {
|
|
||||||
log::warn!("failed to send WebRTC ICE candidate: {}", err);
|
|
||||||
return;
|
|
||||||
}
|
|
||||||
}
|
|
||||||
Err(TryRecvError::Empty) => break,
|
|
||||||
Err(TryRecvError::Disconnected) => {
|
|
||||||
local_ice_rx = None;
|
|
||||||
break;
|
|
||||||
}
|
|
||||||
}
|
|
||||||
}
|
|
||||||
}
|
|
||||||
|
|
||||||
if let Some(msg_in) =
|
|
||||||
crate::get_next_nonkeyexchange_msg(&mut socket, Some(100)).await
|
|
||||||
{
|
|
||||||
if let Some(rendezvous_message::Union::IceCandidate(ice)) = msg_in.union {
|
|
||||||
if Self::is_expected_webrtc_ice_candidate(&ice, &session_key) {
|
|
||||||
if let Err(err) = webrtc.add_remote_ice_candidate(&ice.candidate).await
|
|
||||||
{
|
|
||||||
log::warn!("failed to add WebRTC ICE candidate: {}", err);
|
|
||||||
}
|
|
||||||
}
|
|
||||||
}
|
|
||||||
}
|
|
||||||
}
|
|
||||||
});
|
|
||||||
stop_tx
|
|
||||||
}
|
|
||||||
|
|
||||||
async fn _start_inner(
|
async fn _start_inner(
|
||||||
peer: String,
|
peer: String,
|
||||||
key: String,
|
key: String,
|
||||||
@@ -453,8 +374,6 @@ impl Client {
|
|||||||
interface: impl Interface,
|
interface: impl Interface,
|
||||||
mut udp: (Option<Arc<UdpSocket>>, Option<Arc<Mutex<u16>>>),
|
mut udp: (Option<Arc<UdpSocket>>, Option<Arc<Mutex<u16>>>),
|
||||||
stop_udp_tx: Option<oneshot::Sender<()>>,
|
stop_udp_tx: Option<oneshot::Sender<()>>,
|
||||||
mut ipv6: Option<(Arc<UdpSocket>, bytes::Bytes)>,
|
|
||||||
mut webrtc_offerer: Option<WebRTCStream>,
|
|
||||||
mut rendezvous_server: String,
|
mut rendezvous_server: String,
|
||||||
servers: Vec<String>,
|
servers: Vec<String>,
|
||||||
contained: bool,
|
contained: bool,
|
||||||
@@ -527,20 +446,14 @@ impl Client {
|
|||||||
// Stop UDP NAT test task if still running
|
// Stop UDP NAT test task if still running
|
||||||
stop_udp_tx.map(|tx| tx.send(()));
|
stop_udp_tx.map(|tx| tx.send(()));
|
||||||
let mut msg_out = RendezvousMessage::new();
|
let mut msg_out = RendezvousMessage::new();
|
||||||
let mut ipv6 = ipv6
|
let mut ipv6 = if crate::get_ipv6_punch_enabled() {
|
||||||
.take()
|
if let Some((socket, addr)) = crate::get_ipv6_socket().await {
|
||||||
.map(|(socket, addr)| (Some(socket), Some(addr)))
|
(Some(socket), Some(addr))
|
||||||
.unwrap_or((None, None));
|
} else {
|
||||||
let webrtc_sdp_offer = if let Some(webrtc) = webrtc_offerer.as_ref() {
|
(None, None)
|
||||||
match webrtc.get_local_endpoint().await {
|
|
||||||
Ok(endpoint) => endpoint,
|
|
||||||
Err(err) => {
|
|
||||||
log::warn!("failed to read local WebRTC offer: {}", err);
|
|
||||||
String::new()
|
|
||||||
}
|
|
||||||
}
|
}
|
||||||
} else {
|
} else {
|
||||||
String::new()
|
(None, None)
|
||||||
};
|
};
|
||||||
let udp_nat_port = udp.1.map(|x| *x.lock().unwrap()).unwrap_or(0);
|
let udp_nat_port = udp.1.map(|x| *x.lock().unwrap()).unwrap_or(0);
|
||||||
let punch_type = if udp_nat_port > 0 { "UDP" } else { "TCP" };
|
let punch_type = if udp_nat_port > 0 { "UDP" } else { "TCP" };
|
||||||
@@ -554,16 +467,9 @@ impl Client {
|
|||||||
udp_port: udp_nat_port as _,
|
udp_port: udp_nat_port as _,
|
||||||
force_relay: interface.is_force_relay(),
|
force_relay: interface.is_force_relay(),
|
||||||
socket_addr_v6: ipv6.1.unwrap_or_default(),
|
socket_addr_v6: ipv6.1.unwrap_or_default(),
|
||||||
webrtc_sdp_offer: webrtc_sdp_offer.clone(),
|
|
||||||
..Default::default()
|
..Default::default()
|
||||||
});
|
});
|
||||||
let webrtc_session_key = webrtc_offerer
|
for i in 1..=3 {
|
||||||
.as_ref()
|
|
||||||
.map(|webrtc| webrtc.session_key().to_owned())
|
|
||||||
.unwrap_or_default();
|
|
||||||
let mut webrtc_sdp_answer = String::new();
|
|
||||||
let mut pending_webrtc_ice = Vec::<String>::new();
|
|
||||||
'punch_attempts: for i in 1..=3 {
|
|
||||||
log::info!(
|
log::info!(
|
||||||
"#{} {} punch attempt with {}, id: {}",
|
"#{} {} punch attempt with {}, id: {}",
|
||||||
i,
|
i,
|
||||||
@@ -573,20 +479,9 @@ impl Client {
|
|||||||
);
|
);
|
||||||
socket.send(&msg_out).await?;
|
socket.send(&msg_out).await?;
|
||||||
// below timeout should not bigger than hbbs's connection timeout.
|
// below timeout should not bigger than hbbs's connection timeout.
|
||||||
let attempt_deadline = Instant::now() + Duration::from_millis((i * 3000) as u64);
|
if let Some(msg_in) =
|
||||||
loop {
|
crate::get_next_nonkeyexchange_msg(&mut socket, Some(i * 3000)).await
|
||||||
let remaining = attempt_deadline.saturating_duration_since(Instant::now());
|
{
|
||||||
if remaining.is_zero() {
|
|
||||||
break;
|
|
||||||
}
|
|
||||||
let timeout_ms = remaining
|
|
||||||
.as_millis()
|
|
||||||
.clamp(1, u64::MAX as u128) as u64;
|
|
||||||
let Some(msg_in) =
|
|
||||||
crate::get_next_nonkeyexchange_msg(&mut socket, Some(timeout_ms)).await
|
|
||||||
else {
|
|
||||||
break;
|
|
||||||
};
|
|
||||||
match msg_in.union {
|
match msg_in.union {
|
||||||
Some(rendezvous_message::Union::PunchHoleResponse(ph)) => {
|
Some(rendezvous_message::Union::PunchHoleResponse(ph)) => {
|
||||||
if ph.socket_addr.is_empty() {
|
if ph.socket_addr.is_empty() {
|
||||||
@@ -615,7 +510,6 @@ impl Client {
|
|||||||
relay_server = ph.relay_server;
|
relay_server = ph.relay_server;
|
||||||
peer_addr = AddrMangle::decode(&ph.socket_addr);
|
peer_addr = AddrMangle::decode(&ph.socket_addr);
|
||||||
feedback = ph.feedback;
|
feedback = ph.feedback;
|
||||||
webrtc_sdp_answer = ph.webrtc_sdp_answer;
|
|
||||||
let s = udp.0.take();
|
let s = udp.0.take();
|
||||||
if ph.is_udp && s.is_some() {
|
if ph.is_udp && s.is_some() {
|
||||||
if let Some(s) = s {
|
if let Some(s) = s {
|
||||||
@@ -634,7 +528,7 @@ impl Client {
|
|||||||
}
|
}
|
||||||
}
|
}
|
||||||
log::info!("{} Hole Punched {} = {}", punch_type, peer, peer_addr);
|
log::info!("{} Hole Punched {} = {}", punch_type, peer, peer_addr);
|
||||||
break 'punch_attempts;
|
break;
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
Some(rendezvous_message::Union::RelayResponse(rr)) => {
|
Some(rendezvous_message::Union::RelayResponse(rr)) => {
|
||||||
@@ -655,38 +549,6 @@ impl Client {
|
|||||||
}
|
}
|
||||||
}
|
}
|
||||||
signed_id_pk = rr.pk().into();
|
signed_id_pk = rr.pk().into();
|
||||||
let mut webrtc_bridge_stop = None;
|
|
||||||
let mut webrtc_for_connect = None;
|
|
||||||
if !rr.webrtc_sdp_answer.is_empty() {
|
|
||||||
if let Some(webrtc) = webrtc_offerer.take() {
|
|
||||||
if let Err(err) =
|
|
||||||
webrtc.set_remote_endpoint(&rr.webrtc_sdp_answer).await
|
|
||||||
{
|
|
||||||
log::warn!("failed to set WebRTC relay answer: {}", err);
|
|
||||||
} else {
|
|
||||||
for candidate in pending_webrtc_ice.drain(..) {
|
|
||||||
if let Err(err) =
|
|
||||||
webrtc.add_remote_ice_candidate(&candidate).await
|
|
||||||
{
|
|
||||||
log::warn!(
|
|
||||||
"failed to add buffered WebRTC ICE candidate: {}",
|
|
||||||
err
|
|
||||||
);
|
|
||||||
}
|
|
||||||
}
|
|
||||||
let session_key = webrtc.session_key().to_owned();
|
|
||||||
let local_ice_rx = webrtc.take_local_ice_rx();
|
|
||||||
webrtc_bridge_stop = Some(Self::spawn_webrtc_ice_bridge(
|
|
||||||
socket,
|
|
||||||
local_ice_rx,
|
|
||||||
webrtc.clone(),
|
|
||||||
peer.clone(),
|
|
||||||
session_key,
|
|
||||||
));
|
|
||||||
webrtc_for_connect = Some(webrtc);
|
|
||||||
}
|
|
||||||
}
|
|
||||||
}
|
|
||||||
let fut = Self::create_relay(
|
let fut = Self::create_relay(
|
||||||
&peer,
|
&peer,
|
||||||
rr.uuid,
|
rr.uuid,
|
||||||
@@ -702,81 +564,30 @@ impl Client {
|
|||||||
}
|
}
|
||||||
.boxed(),
|
.boxed(),
|
||||||
);
|
);
|
||||||
if let Some(mut webrtc) = webrtc_for_connect {
|
|
||||||
connect_futures.push(
|
|
||||||
async move {
|
|
||||||
webrtc.wait_connected(CONNECT_TIMEOUT).await?;
|
|
||||||
Ok((Stream::WebRTC(webrtc), None, "WebRTC"))
|
|
||||||
}
|
|
||||||
.boxed(),
|
|
||||||
);
|
|
||||||
}
|
|
||||||
// Run all connection attempts concurrently, return the first successful one
|
// Run all connection attempts concurrently, return the first successful one
|
||||||
let (conn, kcp, typ) = match select_ok(connect_futures).await {
|
let (conn, kcp, typ) = match select_ok(connect_futures).await {
|
||||||
Ok(conn) => (Ok(conn.0 .0), conn.0 .1, conn.0 .2),
|
Ok(conn) => (Ok(conn.0 .0), conn.0 .1, conn.0 .2),
|
||||||
|
|
||||||
Err(e) => (Err(e), None, ""),
|
Err(e) => (Err(e), None, ""),
|
||||||
};
|
};
|
||||||
if let Some(stop) = webrtc_bridge_stop {
|
|
||||||
let _ = stop.send(());
|
|
||||||
}
|
|
||||||
let mut conn = conn?;
|
let mut conn = conn?;
|
||||||
feedback = rr.feedback;
|
feedback = rr.feedback;
|
||||||
log::info!("{:?} used to establish {typ} connection", start.elapsed());
|
log::info!("{:?} used to establish {typ} connection", start.elapsed());
|
||||||
let pk =
|
let pk =
|
||||||
Self::secure_connection(&peer, signed_id_pk, &key, &mut conn).await?;
|
Self::secure_connection(&peer, signed_id_pk, &key, &mut conn).await?;
|
||||||
return Ok((
|
return Ok((
|
||||||
(conn, typ == "IPv6" || typ == "WebRTC", pk, kcp, typ),
|
(conn, typ == "IPv6", pk, kcp, typ),
|
||||||
(feedback, rendezvous_server),
|
(feedback, rendezvous_server),
|
||||||
false,
|
false,
|
||||||
));
|
));
|
||||||
}
|
}
|
||||||
Some(rendezvous_message::Union::IceCandidate(ice)) => {
|
|
||||||
if Self::is_expected_webrtc_ice_candidate(&ice, &webrtc_session_key) {
|
|
||||||
pending_webrtc_ice.push(ice.candidate);
|
|
||||||
} else {
|
|
||||||
log::debug!(
|
|
||||||
"dropping ICE candidate for unexpected WebRTC session key {}",
|
|
||||||
ice.session_key,
|
|
||||||
);
|
|
||||||
}
|
|
||||||
}
|
|
||||||
_ => {
|
_ => {
|
||||||
log::error!("Unexpected protobuf msg received: {:?}", msg_in);
|
log::error!("Unexpected protobuf msg received: {:?}", msg_in);
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
let mut webrtc_bridge_stop = None;
|
drop(socket);
|
||||||
let mut webrtc_for_connect = None;
|
|
||||||
if !webrtc_sdp_answer.is_empty() {
|
|
||||||
if let Some(webrtc) = webrtc_offerer.take() {
|
|
||||||
if let Err(err) = webrtc.set_remote_endpoint(&webrtc_sdp_answer).await {
|
|
||||||
log::warn!("failed to set WebRTC answer: {}", err);
|
|
||||||
drop(socket);
|
|
||||||
} else {
|
|
||||||
for candidate in pending_webrtc_ice.drain(..) {
|
|
||||||
if let Err(err) = webrtc.add_remote_ice_candidate(&candidate).await {
|
|
||||||
log::warn!("failed to add buffered WebRTC ICE candidate: {}", err);
|
|
||||||
}
|
|
||||||
}
|
|
||||||
let session_key = webrtc.session_key().to_owned();
|
|
||||||
let local_ice_rx = webrtc.take_local_ice_rx();
|
|
||||||
webrtc_bridge_stop = Some(Self::spawn_webrtc_ice_bridge(
|
|
||||||
socket,
|
|
||||||
local_ice_rx,
|
|
||||||
webrtc.clone(),
|
|
||||||
peer.clone(),
|
|
||||||
session_key,
|
|
||||||
));
|
|
||||||
webrtc_for_connect = Some(webrtc);
|
|
||||||
}
|
|
||||||
} else {
|
|
||||||
drop(socket);
|
|
||||||
}
|
|
||||||
} else {
|
|
||||||
drop(socket);
|
|
||||||
}
|
|
||||||
if peer_addr.port() == 0 {
|
if peer_addr.port() == 0 {
|
||||||
bail!("Failed to connect via rendezvous server");
|
bail!("Failed to connect via rendezvous server");
|
||||||
}
|
}
|
||||||
@@ -810,8 +621,6 @@ impl Client {
|
|||||||
interface,
|
interface,
|
||||||
udp.0,
|
udp.0,
|
||||||
ipv6.0,
|
ipv6.0,
|
||||||
webrtc_for_connect,
|
|
||||||
webrtc_bridge_stop,
|
|
||||||
punch_type,
|
punch_type,
|
||||||
)
|
)
|
||||||
.await?,
|
.await?,
|
||||||
@@ -838,8 +647,6 @@ impl Client {
|
|||||||
interface: impl Interface,
|
interface: impl Interface,
|
||||||
udp_socket_nat: Option<Arc<UdpSocket>>,
|
udp_socket_nat: Option<Arc<UdpSocket>>,
|
||||||
udp_socket_v6: Option<Arc<UdpSocket>>,
|
udp_socket_v6: Option<Arc<UdpSocket>>,
|
||||||
webrtc_offerer: Option<WebRTCStream>,
|
|
||||||
webrtc_bridge_stop: Option<oneshot::Sender<()>>,
|
|
||||||
punch_type: &str,
|
punch_type: &str,
|
||||||
) -> ResultType<(
|
) -> ResultType<(
|
||||||
Stream,
|
Stream,
|
||||||
@@ -898,23 +705,11 @@ impl Client {
|
|||||||
if let Some(udp_socket_v6) = udp_socket_v6 {
|
if let Some(udp_socket_v6) = udp_socket_v6 {
|
||||||
connect_futures.push(udp_nat_connect(udp_socket_v6, "IPv6", connect_timeout).boxed());
|
connect_futures.push(udp_nat_connect(udp_socket_v6, "IPv6", connect_timeout).boxed());
|
||||||
}
|
}
|
||||||
if let Some(mut webrtc) = webrtc_offerer {
|
|
||||||
connect_futures.push(
|
|
||||||
async move {
|
|
||||||
webrtc.wait_connected(connect_timeout).await?;
|
|
||||||
Ok((Stream::WebRTC(webrtc), None, "WebRTC"))
|
|
||||||
}
|
|
||||||
.boxed(),
|
|
||||||
);
|
|
||||||
}
|
|
||||||
// Run all connection attempts concurrently, return the first successful one
|
// Run all connection attempts concurrently, return the first successful one
|
||||||
let (mut conn, kcp, mut typ) = match select_ok(connect_futures).await {
|
let (mut conn, kcp, mut typ) = match select_ok(connect_futures).await {
|
||||||
Ok(conn) => (Ok(conn.0 .0), conn.0 .1, conn.0 .2),
|
Ok(conn) => (Ok(conn.0 .0), conn.0 .1, conn.0 .2),
|
||||||
Err(e) => (Err(e), None, ""),
|
Err(e) => (Err(e), None, ""),
|
||||||
};
|
};
|
||||||
if let Some(stop) = webrtc_bridge_stop {
|
|
||||||
let _ = stop.send(());
|
|
||||||
}
|
|
||||||
|
|
||||||
let mut direct = !conn.is_err();
|
let mut direct = !conn.is_err();
|
||||||
if interface.is_force_relay() || conn.is_err() {
|
if interface.is_force_relay() || conn.is_err() {
|
||||||
@@ -4325,22 +4120,8 @@ pub mod peer_online {
|
|||||||
|
|
||||||
#[cfg(test)]
|
#[cfg(test)]
|
||||||
mod tests {
|
mod tests {
|
||||||
use crate::client::Client;
|
|
||||||
use hbb_common::rendezvous_proto::IceCandidate;
|
|
||||||
use hbb_common::tokio;
|
use hbb_common::tokio;
|
||||||
|
|
||||||
#[test]
|
|
||||||
fn accepts_webrtc_ice_by_session_key_only() {
|
|
||||||
let ice = IceCandidate {
|
|
||||||
session_key: "session-a".to_owned(),
|
|
||||||
candidate: "candidate-json".to_owned(),
|
|
||||||
..Default::default()
|
|
||||||
};
|
|
||||||
|
|
||||||
assert!(Client::is_expected_webrtc_ice_candidate(&ice, "session-a"));
|
|
||||||
assert!(!Client::is_expected_webrtc_ice_candidate(&ice, "session-b"));
|
|
||||||
}
|
|
||||||
|
|
||||||
#[tokio::test]
|
#[tokio::test]
|
||||||
async fn test_query_onlines() {
|
async fn test_query_onlines() {
|
||||||
super::query_online_states(
|
super::query_online_states(
|
||||||
|
|||||||
81
src/main.rs
81
src/main.rs
@@ -38,68 +38,49 @@ fn main() {
|
|||||||
if !common::global_init() {
|
if !common::global_init() {
|
||||||
return;
|
return;
|
||||||
}
|
}
|
||||||
use clap::{Arg, ArgAction, Command};
|
use clap::App;
|
||||||
use hbb_common::log;
|
use hbb_common::log;
|
||||||
let matches = Command::new("rustdesk")
|
let args = format!(
|
||||||
|
"-p, --port-forward=[PORT-FORWARD-OPTIONS] 'Format: remote-id:local-port:remote-port[:remote-host]'
|
||||||
|
-c, --connect=[REMOTE_ID] 'test only'
|
||||||
|
-k, --key=[KEY] ''
|
||||||
|
-s, --server=[] 'Start server'",
|
||||||
|
);
|
||||||
|
let matches = App::new("rustdesk")
|
||||||
.version(crate::VERSION)
|
.version(crate::VERSION)
|
||||||
.author("Purslane Ltd<info@rustdesk.com>")
|
.author("Purslane Ltd<info@rustdesk.com>")
|
||||||
.about("RustDesk command line tool")
|
.about("RustDesk command line tool")
|
||||||
.arg(
|
.args_from_usage(&args)
|
||||||
Arg::new("port-forward")
|
|
||||||
.short('p')
|
|
||||||
.long("port-forward")
|
|
||||||
.value_name("PORT-FORWARD-OPTIONS")
|
|
||||||
.help("Format: remote-id:local-port:remote-port[:remote-host]"),
|
|
||||||
)
|
|
||||||
.arg(
|
|
||||||
Arg::new("connect")
|
|
||||||
.short('c')
|
|
||||||
.long("connect")
|
|
||||||
.value_name("REMOTE_ID")
|
|
||||||
.help("test only"),
|
|
||||||
)
|
|
||||||
.arg(Arg::new("key").short('k').long("key").value_name("KEY"))
|
|
||||||
.arg(
|
|
||||||
Arg::new("server")
|
|
||||||
.short('s')
|
|
||||||
.long("server")
|
|
||||||
.action(ArgAction::SetTrue)
|
|
||||||
.help("Start server"),
|
|
||||||
)
|
|
||||||
.get_matches();
|
.get_matches();
|
||||||
use hbb_common::{config::LocalConfig, env_logger::*};
|
use hbb_common::{config::LocalConfig, env_logger::*};
|
||||||
init_from_env(Env::default().filter_or(DEFAULT_FILTER_ENV, "info"));
|
init_from_env(Env::default().filter_or(DEFAULT_FILTER_ENV, "info"));
|
||||||
if let Some(p) = matches.get_one::<String>("port-forward") {
|
if let Some(p) = matches.value_of("port-forward") {
|
||||||
let options: Vec<String> = p.split(':').map(|x| x.to_owned()).collect();
|
let options: Vec<String> = p.split(":").map(|x| x.to_owned()).collect();
|
||||||
if options.len() < 3 {
|
if options.len() < 3 {
|
||||||
log::error!("Wrong port-forward options");
|
log::error!("Wrong port-forward options");
|
||||||
return;
|
return;
|
||||||
}
|
}
|
||||||
let port = match options[1].parse::<i32>() {
|
let mut port = 0;
|
||||||
Ok(v) => v,
|
if let Ok(v) = options[1].parse::<i32>() {
|
||||||
Err(_) => {
|
port = v;
|
||||||
log::error!("Wrong local-port");
|
} else {
|
||||||
return;
|
log::error!("Wrong local-port");
|
||||||
}
|
return;
|
||||||
};
|
}
|
||||||
let remote_port = match options[2].parse::<i32>() {
|
let mut remote_port = 0;
|
||||||
Ok(v) => v,
|
if let Ok(v) = options[2].parse::<i32>() {
|
||||||
Err(_) => {
|
remote_port = v;
|
||||||
log::error!("Wrong remote-port");
|
} else {
|
||||||
return;
|
log::error!("Wrong remote-port");
|
||||||
}
|
return;
|
||||||
};
|
}
|
||||||
let mut remote_host = "localhost".to_owned();
|
let mut remote_host = "localhost".to_owned();
|
||||||
if options.len() > 3 {
|
if options.len() > 3 {
|
||||||
remote_host = options[3].clone();
|
remote_host = options[3].clone();
|
||||||
}
|
}
|
||||||
common::test_rendezvous_server();
|
common::test_rendezvous_server();
|
||||||
common::test_nat_type();
|
common::test_nat_type();
|
||||||
let key = matches
|
let key = matches.value_of("key").unwrap_or("").to_owned();
|
||||||
.get_one::<String>("key")
|
|
||||||
.map(String::as_str)
|
|
||||||
.unwrap_or("")
|
|
||||||
.to_owned();
|
|
||||||
let token = LocalConfig::get_option("access_token");
|
let token = LocalConfig::get_option("access_token");
|
||||||
cli::start_one_port_forward(
|
cli::start_one_port_forward(
|
||||||
options[0].clone(),
|
options[0].clone(),
|
||||||
@@ -109,17 +90,13 @@ fn main() {
|
|||||||
key,
|
key,
|
||||||
token,
|
token,
|
||||||
);
|
);
|
||||||
} else if let Some(p) = matches.get_one::<String>("connect") {
|
} else if let Some(p) = matches.value_of("connect") {
|
||||||
common::test_rendezvous_server();
|
common::test_rendezvous_server();
|
||||||
common::test_nat_type();
|
common::test_nat_type();
|
||||||
let key = matches
|
let key = matches.value_of("key").unwrap_or("").to_owned();
|
||||||
.get_one::<String>("key")
|
|
||||||
.map(String::as_str)
|
|
||||||
.unwrap_or("")
|
|
||||||
.to_owned();
|
|
||||||
let token = LocalConfig::get_option("access_token");
|
let token = LocalConfig::get_option("access_token");
|
||||||
cli::connect_test(p, key, token);
|
cli::connect_test(p, key, token);
|
||||||
} else if matches.get_flag("server") {
|
} else if let Some(p) = matches.value_of("server") {
|
||||||
log::info!("id={}", hbb_common::config::Config::get_id());
|
log::info!("id={}", hbb_common::config::Config::get_id());
|
||||||
crate::start_server(true, false);
|
crate::start_server(true, false);
|
||||||
}
|
}
|
||||||
|
|||||||
@@ -1,5 +1,4 @@
|
|||||||
use std::{
|
use std::{
|
||||||
collections::HashMap,
|
|
||||||
net::SocketAddr,
|
net::SocketAddr,
|
||||||
sync::{
|
sync::{
|
||||||
atomic::{AtomicBool, Ordering},
|
atomic::{AtomicBool, Ordering},
|
||||||
@@ -22,13 +21,8 @@ use hbb_common::{
|
|||||||
rendezvous_proto::*,
|
rendezvous_proto::*,
|
||||||
sleep,
|
sleep,
|
||||||
socket_client::{self, connect_tcp, is_ipv4, new_direct_udp_for, new_udp_for},
|
socket_client::{self, connect_tcp, is_ipv4, new_direct_udp_for, new_udp_for},
|
||||||
tokio::{
|
tokio::{self, select, sync::Mutex, time::interval},
|
||||||
self, select,
|
|
||||||
sync::{mpsc, Mutex},
|
|
||||||
time::interval,
|
|
||||||
},
|
|
||||||
udp::FramedSocket,
|
udp::FramedSocket,
|
||||||
webrtc::WebRTCStream,
|
|
||||||
AddrMangle, IntoTargetAddr, ResultType, Stream, TargetAddr,
|
AddrMangle, IntoTargetAddr, ResultType, Stream, TargetAddr,
|
||||||
};
|
};
|
||||||
|
|
||||||
@@ -38,13 +32,11 @@ use crate::{
|
|||||||
};
|
};
|
||||||
|
|
||||||
type Message = RendezvousMessage;
|
type Message = RendezvousMessage;
|
||||||
type RendezvousSender = mpsc::UnboundedSender<Message>;
|
|
||||||
|
|
||||||
lazy_static::lazy_static! {
|
lazy_static::lazy_static! {
|
||||||
static ref SOLVING_PK_MISMATCH: Mutex<String> = Default::default();
|
static ref SOLVING_PK_MISMATCH: Mutex<String> = Default::default();
|
||||||
static ref LAST_MSG: Mutex<(SocketAddr, Instant)> = Mutex::new((SocketAddr::new([0; 4].into(), 0), Instant::now()));
|
static ref LAST_MSG: Mutex<(SocketAddr, Instant)> = Mutex::new((SocketAddr::new([0; 4].into(), 0), Instant::now()));
|
||||||
static ref LAST_RELAY_MSG: Mutex<(SocketAddr, Instant)> = Mutex::new((SocketAddr::new([0; 4].into(), 0), Instant::now()));
|
static ref LAST_RELAY_MSG: Mutex<(SocketAddr, Instant)> = Mutex::new((SocketAddr::new([0; 4].into(), 0), Instant::now()));
|
||||||
static ref WEBRTC_ICE_TXS: Mutex<HashMap<String, mpsc::UnboundedSender<String>>> = Default::default();
|
|
||||||
}
|
}
|
||||||
static SHOULD_EXIT: AtomicBool = AtomicBool::new(false);
|
static SHOULD_EXIT: AtomicBool = AtomicBool::new(false);
|
||||||
static MANUAL_RESTARTED: AtomicBool = AtomicBool::new(false);
|
static MANUAL_RESTARTED: AtomicBool = AtomicBool::new(false);
|
||||||
@@ -80,7 +72,6 @@ pub struct RendezvousMediator {
|
|||||||
host: String,
|
host: String,
|
||||||
host_prefix: String,
|
host_prefix: String,
|
||||||
keep_alive: i32,
|
keep_alive: i32,
|
||||||
rz_sender: RendezvousSender,
|
|
||||||
}
|
}
|
||||||
|
|
||||||
impl RendezvousMediator {
|
impl RendezvousMediator {
|
||||||
@@ -191,13 +182,11 @@ impl RendezvousMediator {
|
|||||||
let host = check_port(&host, RENDEZVOUS_PORT);
|
let host = check_port(&host, RENDEZVOUS_PORT);
|
||||||
log::info!("start udp: {host}");
|
log::info!("start udp: {host}");
|
||||||
let (mut socket, mut addr) = new_udp_for(&host, CONNECT_TIMEOUT).await?;
|
let (mut socket, mut addr) = new_udp_for(&host, CONNECT_TIMEOUT).await?;
|
||||||
let (rz_sender, mut rz_out_rx) = mpsc::unbounded_channel::<Message>();
|
|
||||||
let mut rz = Self {
|
let mut rz = Self {
|
||||||
addr: addr.clone(),
|
addr: addr.clone(),
|
||||||
host: host.clone(),
|
host: host.clone(),
|
||||||
host_prefix: Self::get_host_prefix(&host),
|
host_prefix: Self::get_host_prefix(&host),
|
||||||
keep_alive: crate::DEFAULT_KEEP_ALIVE,
|
keep_alive: crate::DEFAULT_KEEP_ALIVE,
|
||||||
rz_sender,
|
|
||||||
};
|
};
|
||||||
|
|
||||||
let mut timer = crate::rustdesk_interval(interval(crate::TIMER_OUT));
|
let mut timer = crate::rustdesk_interval(interval(crate::TIMER_OUT));
|
||||||
@@ -257,9 +246,6 @@ impl RendezvousMediator {
|
|||||||
},
|
},
|
||||||
}
|
}
|
||||||
},
|
},
|
||||||
Some(msg_out) = rz_out_rx.recv() => {
|
|
||||||
Sink::Framed(&mut socket, &addr).send(&msg_out).await?;
|
|
||||||
},
|
|
||||||
_ = timer.tick() => {
|
_ = timer.tick() => {
|
||||||
if SHOULD_EXIT.load(Ordering::SeqCst) {
|
if SHOULD_EXIT.load(Ordering::SeqCst) {
|
||||||
break;
|
break;
|
||||||
@@ -381,17 +367,6 @@ impl RendezvousMediator {
|
|||||||
allow_err!(rz.handle_intranet(fla, server).await);
|
allow_err!(rz.handle_intranet(fla, server).await);
|
||||||
});
|
});
|
||||||
}
|
}
|
||||||
Some(rendezvous_message::Union::IceCandidate(ice)) => {
|
|
||||||
let tx = WEBRTC_ICE_TXS.lock().await.get(&ice.session_key).cloned();
|
|
||||||
if let Some(tx) = tx {
|
|
||||||
let _ = tx.send(ice.candidate);
|
|
||||||
} else {
|
|
||||||
log::debug!(
|
|
||||||
"dropping ICE candidate for unknown WebRTC session key {}",
|
|
||||||
ice.session_key
|
|
||||||
);
|
|
||||||
}
|
|
||||||
}
|
|
||||||
Some(rendezvous_message::Union::ConfigureUpdate(cu)) => {
|
Some(rendezvous_message::Union::ConfigureUpdate(cu)) => {
|
||||||
let v0 = Config::get_rendezvous_servers();
|
let v0 = Config::get_rendezvous_servers();
|
||||||
Config::set_option(
|
Config::set_option(
|
||||||
@@ -414,13 +389,11 @@ impl RendezvousMediator {
|
|||||||
let mut conn = connect_tcp(host.clone(), CONNECT_TIMEOUT).await?;
|
let mut conn = connect_tcp(host.clone(), CONNECT_TIMEOUT).await?;
|
||||||
let key = crate::get_key(true).await;
|
let key = crate::get_key(true).await;
|
||||||
crate::secure_tcp(&mut conn, &key).await?;
|
crate::secure_tcp(&mut conn, &key).await?;
|
||||||
let (rz_sender, mut rz_out_rx) = mpsc::unbounded_channel::<Message>();
|
|
||||||
let mut rz = Self {
|
let mut rz = Self {
|
||||||
addr: conn.local_addr().into_target_addr()?,
|
addr: conn.local_addr().into_target_addr()?,
|
||||||
host: host.clone(),
|
host: host.clone(),
|
||||||
host_prefix: Self::get_host_prefix(&host),
|
host_prefix: Self::get_host_prefix(&host),
|
||||||
keep_alive: crate::DEFAULT_KEEP_ALIVE,
|
keep_alive: crate::DEFAULT_KEEP_ALIVE,
|
||||||
rz_sender,
|
|
||||||
};
|
};
|
||||||
let mut timer = crate::rustdesk_interval(interval(crate::TIMER_OUT));
|
let mut timer = crate::rustdesk_interval(interval(crate::TIMER_OUT));
|
||||||
let mut last_register_sent: Option<Instant> = None;
|
let mut last_register_sent: Option<Instant> = None;
|
||||||
@@ -448,9 +421,6 @@ impl RendezvousMediator {
|
|||||||
let msg = Message::parse_from_bytes(&bytes)?;
|
let msg = Message::parse_from_bytes(&bytes)?;
|
||||||
rz.handle_resp(msg.union, Sink::Stream(&mut conn), &server, &mut update_latency).await?
|
rz.handle_resp(msg.union, Sink::Stream(&mut conn), &server, &mut update_latency).await?
|
||||||
}
|
}
|
||||||
Some(msg_out) = rz_out_rx.recv() => {
|
|
||||||
Sink::Stream(&mut conn).send(&msg_out).await?;
|
|
||||||
}
|
|
||||||
_ = timer.tick() => {
|
_ = timer.tick() => {
|
||||||
if SHOULD_EXIT.load(Ordering::SeqCst) {
|
if SHOULD_EXIT.load(Ordering::SeqCst) {
|
||||||
break;
|
break;
|
||||||
@@ -502,7 +472,6 @@ impl RendezvousMediator {
|
|||||||
rr.secure,
|
rr.secure,
|
||||||
false,
|
false,
|
||||||
Default::default(),
|
Default::default(),
|
||||||
String::new(),
|
|
||||||
rr.control_permissions.clone().into_option(),
|
rr.control_permissions.clone().into_option(),
|
||||||
)
|
)
|
||||||
.await
|
.await
|
||||||
@@ -517,7 +486,6 @@ impl RendezvousMediator {
|
|||||||
secure: bool,
|
secure: bool,
|
||||||
initiate: bool,
|
initiate: bool,
|
||||||
socket_addr_v6: bytes::Bytes,
|
socket_addr_v6: bytes::Bytes,
|
||||||
webrtc_sdp_answer: String,
|
|
||||||
control_permissions: Option<ControlPermissions>,
|
control_permissions: Option<ControlPermissions>,
|
||||||
) -> ResultType<()> {
|
) -> ResultType<()> {
|
||||||
let peer_addr = AddrMangle::decode(&socket_addr);
|
let peer_addr = AddrMangle::decode(&socket_addr);
|
||||||
@@ -536,7 +504,6 @@ impl RendezvousMediator {
|
|||||||
socket_addr: socket_addr.into(),
|
socket_addr: socket_addr.into(),
|
||||||
version: crate::VERSION.to_owned(),
|
version: crate::VERSION.to_owned(),
|
||||||
socket_addr_v6,
|
socket_addr_v6,
|
||||||
webrtc_sdp_answer,
|
|
||||||
..Default::default()
|
..Default::default()
|
||||||
};
|
};
|
||||||
if initiate {
|
if initiate {
|
||||||
@@ -604,7 +571,6 @@ impl RendezvousMediator {
|
|||||||
true,
|
true,
|
||||||
true,
|
true,
|
||||||
socket_addr_v6,
|
socket_addr_v6,
|
||||||
String::new(),
|
|
||||||
fla.control_permissions.into_option(),
|
fla.control_permissions.into_option(),
|
||||||
)
|
)
|
||||||
.await
|
.await
|
||||||
@@ -647,81 +613,6 @@ impl RendezvousMediator {
|
|||||||
Ok(())
|
Ok(())
|
||||||
}
|
}
|
||||||
|
|
||||||
async fn spawn_webrtc_answerer(
|
|
||||||
&self,
|
|
||||||
ph: &PunchHole,
|
|
||||||
force_relay: bool,
|
|
||||||
server: ServerPtr,
|
|
||||||
peer_addr: SocketAddr,
|
|
||||||
control_permissions: Option<ControlPermissions>,
|
|
||||||
) -> ResultType<String> {
|
|
||||||
let mut stream =
|
|
||||||
WebRTCStream::new(&ph.webrtc_sdp_offer, force_relay, CONNECT_TIMEOUT).await?;
|
|
||||||
let answer = stream.get_local_endpoint().await?;
|
|
||||||
let session_key = stream.session_key().to_owned();
|
|
||||||
let return_route = ph.socket_addr.clone();
|
|
||||||
|
|
||||||
let (remote_ice_tx, mut remote_ice_rx) = mpsc::unbounded_channel::<String>();
|
|
||||||
WEBRTC_ICE_TXS
|
|
||||||
.lock()
|
|
||||||
.await
|
|
||||||
.insert(session_key.clone(), remote_ice_tx);
|
|
||||||
|
|
||||||
let stream_for_remote_ice = stream.clone();
|
|
||||||
tokio::spawn(async move {
|
|
||||||
while let Some(candidate) = remote_ice_rx.recv().await {
|
|
||||||
if let Err(err) = stream_for_remote_ice.add_remote_ice_candidate(&candidate).await
|
|
||||||
{
|
|
||||||
log::warn!("failed to add remote WebRTC ICE candidate: {}", err);
|
|
||||||
}
|
|
||||||
}
|
|
||||||
});
|
|
||||||
|
|
||||||
if let Some(mut local_ice_rx) = stream.take_local_ice_rx() {
|
|
||||||
let sender = self.rz_sender.clone();
|
|
||||||
let socket_addr = return_route.clone();
|
|
||||||
let session_key_for_ice = session_key.clone();
|
|
||||||
tokio::spawn(async move {
|
|
||||||
while let Some(candidate) = local_ice_rx.recv().await {
|
|
||||||
let mut msg = Message::new();
|
|
||||||
msg.set_ice_candidate(IceCandidate {
|
|
||||||
socket_addr: socket_addr.clone(),
|
|
||||||
session_key: session_key_for_ice.clone(),
|
|
||||||
candidate,
|
|
||||||
..Default::default()
|
|
||||||
});
|
|
||||||
let _ = sender.send(msg);
|
|
||||||
}
|
|
||||||
});
|
|
||||||
}
|
|
||||||
|
|
||||||
let session_key_for_cleanup = session_key.clone();
|
|
||||||
tokio::spawn(async move {
|
|
||||||
let result = stream.wait_connected(CONNECT_TIMEOUT).await;
|
|
||||||
WEBRTC_ICE_TXS
|
|
||||||
.lock()
|
|
||||||
.await
|
|
||||||
.remove(&session_key_for_cleanup);
|
|
||||||
if let Err(err) = result {
|
|
||||||
log::warn!("webrtc wait_connected failed: {}", err);
|
|
||||||
return;
|
|
||||||
}
|
|
||||||
if let Err(err) = crate::server::create_tcp_connection(
|
|
||||||
server,
|
|
||||||
Stream::WebRTC(stream),
|
|
||||||
peer_addr,
|
|
||||||
true,
|
|
||||||
control_permissions,
|
|
||||||
)
|
|
||||||
.await
|
|
||||||
{
|
|
||||||
log::warn!("failed to create WebRTC server connection: {}", err);
|
|
||||||
}
|
|
||||||
});
|
|
||||||
|
|
||||||
Ok(answer)
|
|
||||||
}
|
|
||||||
|
|
||||||
async fn handle_punch_hole(&self, ph: PunchHole, server: ServerPtr) -> ResultType<()> {
|
async fn handle_punch_hole(&self, ph: PunchHole, server: ServerPtr) -> ResultType<()> {
|
||||||
let mut peer_addr = AddrMangle::decode(&ph.socket_addr);
|
let mut peer_addr = AddrMangle::decode(&ph.socket_addr);
|
||||||
let last = *LAST_MSG.lock().await;
|
let last = *LAST_MSG.lock().await;
|
||||||
@@ -733,23 +624,7 @@ impl RendezvousMediator {
|
|||||||
let peer_addr_v6 = hbb_common::AddrMangle::decode(&ph.socket_addr_v6);
|
let peer_addr_v6 = hbb_common::AddrMangle::decode(&ph.socket_addr_v6);
|
||||||
let relay = use_ws() || Config::is_proxy() || ph.force_relay;
|
let relay = use_ws() || Config::is_proxy() || ph.force_relay;
|
||||||
let mut socket_addr_v6 = Default::default();
|
let mut socket_addr_v6 = Default::default();
|
||||||
let control_permissions = ph.control_permissions.clone().into_option();
|
let control_permissions = ph.control_permissions.into_option();
|
||||||
let webrtc_sdp_answer = if !ph.webrtc_sdp_offer.is_empty() {
|
|
||||||
self.spawn_webrtc_answerer(
|
|
||||||
&ph,
|
|
||||||
relay,
|
|
||||||
server.clone(),
|
|
||||||
peer_addr,
|
|
||||||
control_permissions.clone(),
|
|
||||||
)
|
|
||||||
.await
|
|
||||||
.unwrap_or_else(|err| {
|
|
||||||
log::warn!("failed to create WebRTC answer: {}", err);
|
|
||||||
String::new()
|
|
||||||
})
|
|
||||||
} else {
|
|
||||||
String::new()
|
|
||||||
};
|
|
||||||
if peer_addr_v6.port() > 0 && !relay {
|
if peer_addr_v6.port() > 0 && !relay {
|
||||||
socket_addr_v6 = start_ipv6(
|
socket_addr_v6 = start_ipv6(
|
||||||
peer_addr_v6,
|
peer_addr_v6,
|
||||||
@@ -776,7 +651,6 @@ impl RendezvousMediator {
|
|||||||
true,
|
true,
|
||||||
true,
|
true,
|
||||||
socket_addr_v6.clone(),
|
socket_addr_v6.clone(),
|
||||||
webrtc_sdp_answer.clone(),
|
|
||||||
control_permissions,
|
control_permissions,
|
||||||
)
|
)
|
||||||
.await;
|
.await;
|
||||||
@@ -790,7 +664,6 @@ impl RendezvousMediator {
|
|||||||
nat_type: nat_type.into(),
|
nat_type: nat_type.into(),
|
||||||
version: crate::VERSION.to_owned(),
|
version: crate::VERSION.to_owned(),
|
||||||
socket_addr_v6,
|
socket_addr_v6,
|
||||||
webrtc_sdp_answer,
|
|
||||||
..Default::default()
|
..Default::default()
|
||||||
};
|
};
|
||||||
if ph.udp_port > 0 {
|
if ph.udp_port > 0 {
|
||||||
|
|||||||
Reference in New Issue
Block a user