Merge pull request #579 from rustdesk/webrtc

Webrtc

Cargo.toml

@@ -52,6 +52,7 @@ thiserror = "1.0"
 httparse = "1.10"
 base64 = "0.22"
 url = "2.5"
+percent-encoding = "2"
 sha2 = "0.10"
 whoami = "1.5"
 
@@ -68,7 +69,11 @@ rustls-pki-types = "1.11"
 rustls-native-certs = "0.8"
 webpki-roots = "1.0.4"
 async-recursion = "1.1"
-webrtc = { version = "0.14.0", optional = true }
+# Pinned to 0.13: webrtc >=0.14 pulls sdp 0.10 / webrtc-util 0.12, which use
+# usize::is_multiple_of (needs rustc >=1.87), while rustdesk CI builds with Rust 1.75
+# (sciter i128 ABI pin, flutter-build.yml). Bump only after CI's Rust moves past 1.87,
+# and work through the upgrade checklist at the top of src/webrtc.rs first.
+webrtc = { version = "0.13.0", optional = true }
 libloading = "0.8"
 
 [target.'cfg(not(any(target_os = "android", target_os = "ios")))'.dependencies]
@@ -81,7 +86,7 @@ protobuf-codegen = { version = "3.7" }
 
 [dev-dependencies]
 clap = "4.5.51"
-webrtc = "0.14.0"
+webrtc = "0.13.0" # keep in lockstep with [dependencies] webrtc (rustc 1.75 pin, see above)
 
 [target.'cfg(target_os = "windows")'.dependencies]
 winapi = { version = "0.3", features = [

examples/webrtc_echo.rs

@@ -0,0 +1,80 @@
+// Temporary e2e helper for the web client: answers a browser offer read from
+// stdin, bridges ICE candidates over stdin/stdout as JSON lines, then echoes
+// every message it receives.
+extern crate hbb_common;
+
+#[cfg(feature = "webrtc")]
+#[hbb_common::tokio::main]
+async fn main() -> hbb_common::anyhow::Result<()> {
+    use hbb_common::{
+        anyhow::anyhow,
+        serde_json::{self, json, Value},
+        tokio,
+        webrtc::WebRTCStream,
+    };
+    use std::io::BufRead;
+
+    let mut offer = String::new();
+    std::io::stdin().lock().read_line(&mut offer)?;
+    if offer.trim().is_empty() {
+        return Err(anyhow!("no offer"));
+    }
+    let mut stream = WebRTCStream::new(offer.trim(), false, 30000).await?;
+    let out = json!({
+        "answer": stream.local_endpoint(),
+        "session_key": stream.session_key(),
+        "local_fingerprint": stream.local_dtls_fingerprint().await?,
+        "remote_fingerprint": stream.remote_dtls_fingerprint().await?,
+    });
+    println!("{}", out);
+    let mut ice_rx = stream
+        .take_local_ice_rx()
+        .ok_or_else(|| anyhow!("no ice rx"))?;
+    tokio::spawn(async move {
+        while let Some(c) = ice_rx.recv().await {
+            println!("{}", json!({ "candidate": c }));
+        }
+    });
+    let (tx, mut rx) = tokio::sync::mpsc::unbounded_channel::<String>();
+    std::thread::spawn(move || {
+        let stdin = std::io::stdin();
+        for line in stdin.lock().lines().flatten() {
+            if tx.send(line).is_err() {
+                break;
+            }
+        }
+    });
+    let for_ice = stream.clone();
+    tokio::spawn(async move {
+        while let Some(line) = rx.recv().await {
+            let Ok(v) = serde_json::from_str::<Value>(&line) else {
+                continue;
+            };
+            if let Some(c) = v.get("candidate").and_then(Value::as_str) {
+                if let Err(e) = for_ice.add_remote_ice_candidate(c).await {
+                    eprintln!("add_remote_ice_candidate failed: {e}");
+                }
+            }
+        }
+    });
+    stream.wait_connected(30000).await?;
+    println!(
+        "{}",
+        json!({
+            "connected": true,
+            "relayed": stream.is_relayed().await,
+            "remote_ipv6": stream.is_remote_ipv6().await,
+        })
+    );
+    while let Some(res) = stream.next().await {
+        let bytes = res?;
+        let len = bytes.len();
+        stream.send_bytes(bytes.freeze()).await?;
+        eprintln!("echoed {len} bytes");
+    }
+    stream.close().await;
+    Ok(())
+}
+
+#[cfg(not(feature = "webrtc"))]
+fn main() {}

protos/message.proto

@@ -38,6 +38,10 @@ message VideoFrame {
 message IdPk {
   string id = 1;
   bytes pk = 2;
+  // DTLS certificate fingerprint of the signer's WebRTC endpoint, signed together with id/pk so
+  // a WebRTC peer's DTLS channel can be bound to its verified identity (defeats a rendezvous/relay
+  // that swaps SDP fingerprints). Empty for non-WebRTC handshakes.
+  string dtls_fingerprint = 3;
 }
 
 message DisplayInfo {

protos/rendezvous.proto

@@ -29,6 +29,10 @@ message PunchHoleRequest {
   int32 upnp_port = 9;
   bytes socket_addr_v6 = 10;
   string switch_code = 11;
+  // The offer's envelope declares its own ICE transport policy (`ice_policy` key inside
+  // the webrtc:// payload): under force_relay it tells the peer whether the relay is
+  // transport-forced (WebSocket — answer may use full ICE) or policy (Relay-only + TURN).
+  string webrtc_sdp_offer = 12;
 }
 
 message ControlPermissions {
@@ -64,6 +68,7 @@ message PunchHole {
   bytes socket_addr_v6 = 7;
   ControlPermissions control_permissions = 8;
   ControlledContext controlled_context = 9;
+  string webrtc_sdp_offer = 10;
 }
 
 message TestNatRequest {
@@ -90,6 +95,7 @@ message PunchHoleSent {
   string version = 5;
   int32 upnp_port = 6;
   bytes socket_addr_v6 = 7;
+  string webrtc_sdp_answer = 8;
 }
 
 message RegisterPk {
@@ -135,6 +141,7 @@ message PunchHoleResponse {
   bool is_udp = 9;
   int32 upnp_port = 10;
   bytes socket_addr_v6 = 11;
+  string webrtc_sdp_answer = 12;
 }
 
 message ConfigUpdate {
@@ -169,6 +176,7 @@ message RelayResponse {
   int32 feedback = 9;
   bytes socket_addr_v6 = 10;
   int32 upnp_port = 11;
+  string webrtc_sdp_answer = 12;
 }
 
 message SoftwareUpdate { string url = 1; }
@@ -240,6 +248,13 @@ message HttpProxyResponse {
   string error = 4;
 }
 
+message IceCandidate {
+  string id = 1;
+  bytes socket_addr = 2;
+  string session_key = 3;
+  string candidate = 4;
+}
+
 message RendezvousMessage {
   oneof union {
     RegisterPeer register_peer = 6;
@@ -265,5 +280,6 @@ message RendezvousMessage {
     HealthCheck hc = 26;
     HttpProxyRequest http_proxy_request = 27;
     HttpProxyResponse http_proxy_response = 28;
+    IceCandidate ice_candidate = 29;
   }
 }

src/bytes_codec.rs

@@ -4,6 +4,8 @@ use tokio_util::codec::{Decoder, Encoder};
 
 // Bound speculative allocation from untrusted frame headers.
 const MAX_PREALLOCATED_PAYLOAD_LEN: usize = 256 * 1024;
+/// Largest payload representable by the four-byte RustDesk frame header.
+pub const MAX_FRAME_LENGTH: usize = 0x3FFF_FFFF;
 
 #[derive(Debug, Clone, Copy)]
 pub struct BytesCodec {
@@ -132,7 +134,7 @@ impl Encoder<Bytes> for BytesCodec {
             let h = (data.len() << 2) as u32 | 0x2;
             buf.put_u16_le((h & 0xFFFF) as u16);
             buf.put_u8((h >> 16) as u8);
-        } else if data.len() <= 0x3FFFFFFF {
+        } else if data.len() <= MAX_FRAME_LENGTH {
             buf.put_u32_le((data.len() << 2) as u32 | 0x3);
         } else {
             return Err(io::Error::new(io::ErrorKind::InvalidInput, "Overflow"));
@@ -290,7 +292,7 @@ mod tests {
     fn decode_large_frame_header_caps_preallocation() {
         let mut codec = BytesCodec::new();
         let mut buf = BytesMut::new();
-        let n = 0x3FFFFFFFusize;
+        let n = MAX_FRAME_LENGTH;
         const MAX_REASONABLE_CAPACITY: usize = MAX_PREALLOCATED_PAYLOAD_LEN * 4;
 
         buf.put_u32_le((n << 2) as u32 | 0x3);

src/config.rs

@@ -2994,9 +2994,13 @@ pub mod keys {
         "allow-command-line-settings-when-settings-disabled";
 
     // Connection punch-through / port-forward options
+    pub const OPTION_ENABLE_TCP_PUNCH: &str = "enable-tcp-punch";
     pub const OPTION_ENABLE_UDP_PUNCH: &str = "enable-udp-punch";
     pub const OPTION_ENABLE_IPV6_PUNCH: &str = "enable-ipv6-punch";
     pub const OPTION_ENABLE_PORT_FORWARD_MUX: &str = "enable-port-forward-mux";
+    pub const OPTION_ENABLE_WEBRTC: &str = "enable-webrtc";
+    pub const OPTION_ALLOW_KCP_CC: &str = "allow-kcp-congestion-control";
+    pub const OPTION_ALLOW_WEBRTC_CC: &str = "allow-webrtc-congestion-control";
     pub const OPTION_HIDE_USERNAME_ON_CARD: &str = "hide-username-on-card";
     pub const OPTION_HIDE_HELP_CARDS: &str = "hide-help-cards";
     pub const OPTION_DEFAULT_CONNECT_PASSWORD: &str = "default-connect-password";
@@ -3133,9 +3137,11 @@ pub mod keys {
         OPTION_ALLOW_AUTO_RECORD_OUTGOING,
         OPTION_HIDE_RECORDING_BUTTON,
         OPTION_VIDEO_SAVE_DIRECTORY,
+        OPTION_ENABLE_TCP_PUNCH,
         OPTION_ENABLE_UDP_PUNCH,
         OPTION_ENABLE_IPV6_PUNCH,
         OPTION_ENABLE_PORT_FORWARD_MUX,
+        OPTION_ENABLE_WEBRTC,
         OPTION_TOUCH_MODE,
         OPTION_SHOW_VIRTUAL_MOUSE,
         OPTION_SHOW_VIRTUAL_JOYSTICK,
@@ -3202,6 +3208,8 @@ pub mod keys {
         OPTION_ALLOW_INSECURE_TLS_FALLBACK,
         OPTION_KEEP_AWAKE_DURING_INCOMING_SESSIONS,
         OPTION_ALLOW_AUTO_UPDATE,
+        OPTION_ALLOW_KCP_CC,
+        OPTION_ALLOW_WEBRTC_CC,
     ];
 
     // BUILDIN_SETTINGS

src/lib.rs

@@ -57,6 +57,7 @@ pub use toml;
 pub use uuid;
 pub mod fingerprint;
 pub use flexi_logger;
+pub mod log_throttle;
 pub mod stream;
 pub mod websocket;
 #[cfg(feature = "webrtc")]
@@ -429,10 +430,15 @@ pub fn init_log(_is_async: bool, _name: &str) -> Option<flexi_logger::LoggerHand
     #[allow(unused_mut)]
     let mut logger_holder: Option<flexi_logger::LoggerHandle> = None;
     INIT.call_once(|| {
+        // webrtc-rs reports the racing design's normal outcome at warn: cancelling the transport
+        // that lost closes every gathered candidate one line at a time, and trickle means the
+        // agent always checks before it holds a pair. Both read as faults beside a connection
+        // that succeeded. agent_gather keeps its warn level - a STUN server it could not reach
+        // is the one upstream signal that explains a session which never connected.
         #[cfg(debug_assertions)]
         {
             use env_logger::*;
-            init_from_env(Env::default().filter_or(DEFAULT_FILTER_ENV, "info,reqwest=warn,rustls=warn,webrtc-sctp=warn,webrtc=warn"));
+            init_from_env(Env::default().filter_or(DEFAULT_FILTER_ENV, "info,reqwest=warn,rustls=warn,webrtc-sctp=warn,webrtc=warn,webrtc_ice::agent::agent_internal=error,webrtc::peer_connection=error"));
         }
         #[cfg(not(debug_assertions))]
         {
@@ -447,7 +453,7 @@ pub fn init_log(_is_async: bool, _name: &str) -> Option<flexi_logger::LoggerHand
                 path.push(_name);
             }
             use flexi_logger::*;
-            if let Ok(x) = Logger::try_with_env_or_str("debug,reqwest=warn,rustls=warn,webrtc-sctp=warn,webrtc=warn") {
+            if let Ok(x) = Logger::try_with_env_or_str("debug,reqwest=warn,rustls=warn,webrtc-sctp=warn,webrtc=warn,webrtc_ice::agent::agent_internal=error,webrtc::peer_connection=error") {
                 logger_holder = x
                     .log_to_file(FileSpec::default().directory(path))
                     .write_mode(if _is_async {
@@ -457,8 +463,18 @@ pub fn init_log(_is_async: bool, _name: &str) -> Option<flexi_logger::LoggerHand
                     })
                     .format(opt_format)
                     .rotate(
-                        Criterion::Age(Age::Day),
+                        // Size as well as age: rotating only daily lets one day's file grow
+                        // without limit, so whoever can drive a hot log site — a peer sending
+                        // malformed packets, a socket erroring in a retry loop — decides how
+                        // much disk this uses. Bounding it here covers every call site at once.
+                        Criterion::AgeOrSize(Age::Day, 16 * 1024 * 1024),
                         Naming::Timestamps,
+                        // 90 files is ~90 days for any machine that stays under the
+                        // size criterion — i.e. every ordinary install, on every platform this
+                        // ships to. Raising it to protect the flood case would have bought little
+                        // (a count cannot outrun a flood; only the rate limits at the log sites
+                        // can) at the price of multiplying steady-state retention and disk for
+                        // everyone.
                         Cleanup::KeepLogFiles(90),
                     )
                     .start()

src/log_throttle.rs

@@ -0,0 +1,154 @@
+use std::sync::Mutex;
+use std::time::{Duration, Instant};
+
+/// Collapses a log site whose call rate is set by someone else — a peer's message rate, or a
+/// retry loop — into at most one line per interval.
+///
+/// Debug output is written to the log file, so a site that fires per received packet lets a
+/// peer decide how much a machine writes to disk. Dropping the line entirely instead would
+/// hide real faults, so keep one line per interval and carry the count of everything
+/// suppressed since the last one.
+///
+/// Prefer the [`throttled_log!`](crate::throttled_log) macro, which declares the static for
+/// you. Reach for this type directly only when the count belongs somewhere other than the end
+/// of the line, or when the decision drives more than a log call.
+///
+/// Declare one per site (they do not share counts):
+///
+/// ```ignore
+/// static DROPPED_ICE: LogThrottle = LogThrottle::new(Duration::from_secs(60));
+///
+/// if let Some(n) = DROPPED_ICE.due() {
+///     log::debug!("dropped {n} ICE candidate(s) with no route");
+/// }
+/// ```
+pub struct LogThrottle {
+    interval: Duration,
+    state: Mutex<ThrottleState>,
+}
+
+struct ThrottleState {
+    suppressed: u64,
+    last: Option<Instant>,
+}
+
+impl LogThrottle {
+    pub const fn new(interval: Duration) -> Self {
+        Self {
+            interval,
+            state: Mutex::new(ThrottleState {
+                suppressed: 0,
+                last: None,
+            }),
+        }
+    }
+
+    /// Record one occurrence. Returns the number of occurrences to report (including this one)
+    /// when a line is due, or `None` while still inside the interval.
+    ///
+    /// The first occurrence after a quiet period always reports, so an isolated fault is not
+    /// delayed by the interval.
+    pub fn due(&self) -> Option<u64> {
+        // A poisoned lock only means some other thread panicked while holding it; the guarded
+        // data is two plain counters that are still usable, and going silent for the rest of
+        // the process would be worse than a stale count.
+        let mut state = self
+            .state
+            .lock()
+            .unwrap_or_else(|poisoned| poisoned.into_inner());
+        state.suppressed += 1;
+        // `map_or(true, ..)` rather than clippy's preferred `is_none_or`: that was stabilized in
+        // Rust 1.82 and this crate builds on the 1.75 pinned by CI.
+        #[allow(clippy::unnecessary_map_or)]
+        let due = state
+            .last
+            .map_or(true, |last| last.elapsed() >= self.interval);
+        if !due {
+            return None;
+        }
+        state.last = Some(Instant::now());
+        Some(std::mem::replace(&mut state.suppressed, 0))
+    }
+}
+
+/// Log at most one line per interval from this call site, suffixed with the number of
+/// occurrences it stands for.
+///
+/// Each expansion declares its own hidden static, so two sites never share a count and
+/// adding one is a single line:
+///
+/// ```ignore
+/// throttled_log!(Duration::from_secs(5), warn, "rejected ipc peer {peer_pid:?}");
+/// ```
+///
+/// An isolated event logs unchanged; a burst collapses to `... (x47)`. The count includes
+/// the occurrence being reported, so it reads as a total rather than as "and N more".
+#[macro_export]
+macro_rules! throttled_log {
+    ($interval:expr, $level:ident, $($arg:tt)+) => {{
+        static THROTTLE: $crate::log_throttle::LogThrottle =
+            $crate::log_throttle::LogThrottle::new($interval);
+        if let Some(n) = THROTTLE.due() {
+            if n > 1 {
+                $crate::log::$level!("{} (x{})", format_args!($($arg)+), n);
+            } else {
+                $crate::log::$level!("{}", format_args!($($arg)+));
+            }
+        }
+    }};
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    // Two directions of one socket need two throttles: an ICMP error on a connected socket is
+    // reported once and cleared, so the steady state alternates (send succeeds, the next recv
+    // reports it) and anything shared between them is reset by the succeeding side every cycle.
+    #[test]
+    fn separate_throttles_do_not_reset_each_other() {
+        let send = LogThrottle::new(Duration::from_secs(60));
+        let recv = LogThrottle::new(Duration::from_secs(60));
+        assert_eq!(recv.due(), Some(1));
+        for _ in 0..1_000 {
+            // The send side succeeding must not hand the recv side a fresh emit slot.
+            assert_eq!(recv.due(), None);
+        }
+        assert_eq!(
+            send.due(),
+            Some(1),
+            "the other direction keeps its own slot"
+        );
+    }
+
+    #[test]
+    fn first_call_reports_immediately() {
+        let t = LogThrottle::new(Duration::from_secs(60));
+        assert_eq!(t.due(), Some(1));
+    }
+
+    #[test]
+    fn calls_inside_the_interval_are_counted_not_reported() {
+        let t = LogThrottle::new(Duration::from_secs(60));
+        assert_eq!(t.due(), Some(1));
+        for _ in 0..100 {
+            assert_eq!(t.due(), None);
+        }
+    }
+
+    #[test]
+    fn the_next_due_line_carries_everything_suppressed() {
+        let t = LogThrottle::new(Duration::ZERO);
+        assert_eq!(t.due(), Some(1));
+        // A zero interval is always due, so each call reports exactly itself.
+        assert_eq!(t.due(), Some(1));
+
+        let t = LogThrottle::new(Duration::from_millis(30));
+        assert_eq!(t.due(), Some(1));
+        assert_eq!(t.due(), None);
+        assert_eq!(t.due(), None);
+        std::thread::sleep(Duration::from_millis(40));
+        // The two suppressed calls plus this one.
+        assert_eq!(t.due(), Some(3));
+    }
+}

src/stream.rs

@@ -74,6 +74,87 @@ impl Stream {
         }
     }
 
+    /// Whether this is a WebRTC transport. Used to enforce the WebRTC-only DTLS fingerprint
+    /// binding (fail closed) in the secure handshake.
+    #[inline]
+    pub fn is_webrtc(&self) -> bool {
+        match self {
+            #[cfg(feature = "webrtc")]
+            Stream::WebRTC(_) => true,
+            #[allow(unreachable_patterns)]
+            _ => false,
+        }
+    }
+
+    /// Close the underlying WebRTC peer connection if this is a WebRTC stream; no-op otherwise.
+    /// A WebRTC pc is kept alive by the global session cache and its cleanup handler only fires on
+    /// a terminal ICE state, so it must be closed explicitly at session end. TCP/WebSocket streams
+    /// release their resources on drop and need nothing here.
+    ///
+    /// Deliberately not `async`: most callers sit in a `select!` arm or a future the UI can
+    /// abandon, and an awaited close that loses that race is unretryable — `close()` latches
+    /// `is_closed` before its first await, so every later attempt early-returns while the state
+    /// handler that would evict the session never runs. With no await point here there is
+    /// nothing to cancel; the teardown runs to completion on the WebRTC I/O runtime
+    /// (see `webrtc::WEBRTC_RT`) that owns the pc's sockets and pump tasks, so it reaches the
+    /// wire even after a caller's own runtime dies on return.
+    #[inline]
+    pub fn close_webrtc(&self) {
+        match self {
+            #[cfg(feature = "webrtc")]
+            Stream::WebRTC(s) => s.close_detached(),
+            #[allow(unreachable_patterns)]
+            _ => {}
+        }
+    }
+
+    /// Whether an established WebRTC transport runs through a TURN relay (used for the UI's
+    /// direct/relayed flag). `None` for non-WebRTC transports, and for a non-relay-policy pc
+    /// before ICE selects a pair; a Relay-policy pc answers `Some(true)` straight away — see
+    /// `WebRTCStream::is_relayed`.
+    #[inline]
+    pub async fn webrtc_relayed(&self) -> Option<bool> {
+        match self {
+            #[cfg(feature = "webrtc")]
+            Stream::WebRTC(s) => s.is_relayed().await,
+            #[allow(unreachable_patterns)]
+            _ => None,
+        }
+    }
+
+    /// Whether an established WebRTC transport reaches the peer over IPv6 (used to name the
+    /// transport in the UI). `None` for non-WebRTC transports and before ICE selects a pair —
+    /// every other transport already carries the family in the label it was raced under.
+    #[inline]
+    pub async fn webrtc_remote_ipv6(&self) -> Option<bool> {
+        match self {
+            #[cfg(feature = "webrtc")]
+            Stream::WebRTC(s) => s.is_remote_ipv6().await,
+            #[allow(unreachable_patterns)]
+            _ => None,
+        }
+    }
+
+    /// DTLS certificate fingerprint for a WebRTC stream (`local`=true for this endpoint's own
+    /// cert, false for the peer's), used to bind the channel to the signed peer identity.
+    /// Returns None for non-WebRTC transports, which authenticate via the secretbox key exchange.
+    #[inline]
+    #[cfg_attr(not(feature = "webrtc"), allow(unused_variables))]
+    pub async fn dtls_fingerprint(&self, local: bool) -> Option<String> {
+        match self {
+            #[cfg(feature = "webrtc")]
+            Stream::WebRTC(s) => {
+                if local {
+                    s.local_dtls_fingerprint().await.ok()
+                } else {
+                    s.remote_dtls_fingerprint().await.ok()
+                }
+            }
+            #[allow(unreachable_patterns)]
+            _ => None,
+        }
+    }
+
     #[inline]
     pub async fn next_timeout(
         &mut self,
@@ -137,13 +218,21 @@ impl Stream {
     pub fn from(stream: TcpStream, stream_addr: SocketAddr) -> Self {
         Self::Tcp(tcp::FramedStream::from(stream, stream_addr))
     }
+}
 
-    #[inline]
-    #[cfg(feature = "webrtc")]
-    pub fn get_webrtc_stream(&self) -> Option<webrtc::WebRTCStream> {
-        match self {
-            Self::WebRTC(s) => Some(s.clone()),
-            _ => None,
-        }
+/// Owning the stream owns the transport, WebRTC included.
+///
+/// A peer connection outlives its handle — the session cache holds a clone, and the handler that
+/// evicts it only fires on a terminal ICE state — so it has to be closed explicitly. Doing that
+/// at each exit path made it an obligation every `return`, `break` and `?` had to remember, and
+/// the long-lived side never did: `server::connection` ends its ~15 exits by dropping the stream.
+/// Nothing warned, because a missed close leaks silently and only under WebRTC.
+///
+/// `Stream` is not `Clone`, so dropping it really is the end of the transport and there is no
+/// second owner to surprise. Explicit `close_webrtc()` calls remain valid — they close sooner
+/// than scope end — but they are now an optimization rather than the thing correctness rests on.
+impl Drop for Stream {
+    fn drop(&mut self) {
+        self.close_webrtc();
     }
 }