diff --git a/Cargo.lock b/Cargo.lock index a9bc3a2b855..8988436e247 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -223,6 +223,18 @@ version = "0.7.7" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "4858a9d740c5007a9069007c3b4e91152d0506f13c1b31dd49051fd537656156" +[[package]] +name = "async-broadcast" +version = "0.7.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "435a87a52755b8f27fcf321ac4f04b2802e337c8c4872923137471ec39c37532" +dependencies = [ + "event-listener", + "event-listener-strategy", + "futures-core", + "pin-project-lite", +] + [[package]] name = "async-channel" version = "2.5.0" @@ -479,6 +491,15 @@ version = "0.6.3" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "230c5f1ca6a325a32553f8640d31ac9b49f2411e901e427570154868b46da4f7" +[[package]] +name = "bit-vec" +version = "0.9.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "b71798fca2c1fe1086445a7258a4bc81e6e49dcd24c8d0dd9a1e57395b603f51" +dependencies = [ + "serde", +] + [[package]] name = "bitflags" version = "1.3.2" @@ -968,6 +989,15 @@ version = "2.5.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "217698eaf96b4a3f0bc4f3662aaa55bdf913cd54d7204591faa790070c6d0853" +[[package]] +name = "crc32c" +version = "0.6.8" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "3a47af21622d091a8f0fb295b88bc886ac74efcc613efc19f5d0b21de5c89e47" +dependencies = [ + "rustc_version", +] + [[package]] name = "crc32fast" version = "1.5.0" @@ -1373,43 +1403,6 @@ dependencies = [ "tracing-subscriber", ] -[[package]] -name = "dtls" -version = "0.17.2" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "01a431e87fc386bd5e02deb554a013f97bc406e47a7d8e97efb7c6366b980e9c" -dependencies = [ - "aes", - "aes-gcm", - "async-trait", - "bytecheck", - "byteorder", - "cbc", - "ccm", - "chacha20poly1305", - "der-parser 9.0.0", - "hmac 0.12.1", - "log", - "p256 0.13.2", - "p384", - "pem", - "portable-atomic", - "rand 0.9.5", - "rand_core 0.6.4", - "rcgen", - "ring", - "rkyv", - "rustls", - "sec1 0.7.3", - "sha1", - "sha2 0.10.9", - "thiserror 1.0.69", - "tokio", - "webrtc-util", - "x25519-dalek 2.0.1", - "x509-parser 0.16.0", -] - [[package]] name = "dtoa" version = "1.0.11" @@ -2341,7 +2334,7 @@ dependencies = [ "libc", "percent-encoding", "pin-project-lite", - "socket2 0.6.5", + "socket2", "system-configuration", "tokio", "tower-service", @@ -2544,27 +2537,6 @@ dependencies = [ "hybrid-array", ] -[[package]] -name = "interceptor" -version = "0.17.2" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "88c11a956a48159f7fe539b8198f12b4db9b709ae5f94385b840db38f97fed74" -dependencies = [ - "async-trait", - "bytes", - "futures", - "log", - "portable-atomic", - "rand 0.9.5", - "rtcp", - "rtp", - "thiserror 1.0.69", - "tokio", - "waitgroup", - "webrtc-srtp", - "webrtc-util", -] - [[package]] name = "interop-tests" version = "0.1.0" @@ -2604,7 +2576,7 @@ version = "0.3.4" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "4d40460c0ce33d6ce4b0630ad68ff63d6661961c48b6dba35e5a4d81cfb48222" dependencies = [ - "socket2 0.6.5", + "socket2", "widestring", "windows-registry", "windows-result", @@ -3170,7 +3142,7 @@ dependencies = [ "libp2p-swarm-test", "rand 0.10.2", "smallvec", - "socket2 0.6.5", + "socket2", "tokio", "tracing", "tracing-subscriber", @@ -3385,7 +3357,7 @@ dependencies = [ "quinn-proto", "rand 0.10.2", "ring", - "socket2 0.6.5", + "socket2", "thiserror 2.0.19", "tokio", "tracing", @@ -3566,7 +3538,7 @@ dependencies = [ "if-watch", "libc", "libp2p-core", - "socket2 0.6.5", + "socket2", "tokio", "tracing", "tracing-subscriber", @@ -3583,7 +3555,7 @@ dependencies = [ "libp2p-identity", "libp2p-swarm", "libp2p-yamux", - "rcgen", + "rcgen 0.13.2", "ring", "rustls", "rustls-webpki", @@ -3622,6 +3594,7 @@ name = "libp2p-webrtc" version = "0.10.0-alpha" dependencies = [ "async-trait", + "bytes", "futures", "futures-timer", "hex", @@ -3633,11 +3606,11 @@ dependencies = [ "multihash", "quickcheck", "rand 0.10.2", - "rcgen", + "rcgen 0.14.8", + "rtc", "stun", "thiserror 2.0.19", "tokio", - "tokio-util", "tracing", "tracing-subscriber", "webrtc", @@ -3697,7 +3670,7 @@ dependencies = [ "libp2p-tcp", "parking_lot", "pin-project-lite", - "rcgen", + "rcgen 0.13.2", "rw-stream-sink", "soketto", "thiserror 2.0.19", @@ -4976,10 +4949,10 @@ dependencies = [ "futures-io", "pin-project-lite", "quinn-proto", - "quinn-udp", + "quinn-udp 0.5.15", "rustc-hash", "rustls", - "socket2 0.6.5", + "socket2", "thiserror 2.0.19", "tokio", "tracing", @@ -5019,11 +4992,24 @@ dependencies = [ "cfg_aliases", "libc", "once_cell", - "socket2 0.6.5", + "socket2", "tracing", "windows-sys 0.61.2", ] +[[package]] +name = "quinn-udp" +version = "0.6.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "76150b617afc75e6e21ac5f39bc196e80b65415ae48d62dbef8e2519d040ce42" +dependencies = [ + "cfg_aliases", + "libc", + "log", + "socket2", + "windows-sys 0.61.2", +] + [[package]] name = "quote" version = "1.0.47" @@ -5210,10 +5196,23 @@ dependencies = [ "ring", "rustls-pki-types", "time", - "x509-parser 0.16.0", "yasna 0.5.2", ] +[[package]] +name = "rcgen" +version = "0.14.8" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "57f6d249aad744e274e682777a50283a225a32705394ee6d5fcc01efa25e4055" +dependencies = [ + "pem", + "ring", + "rustls-pki-types", + "time", + "x509-parser 0.18.1", + "yasna 0.6.0", +] + [[package]] name = "redis" version = "1.4.1" @@ -5230,7 +5229,7 @@ dependencies = [ "percent-encoding", "pin-project-lite", "ryu", - "socket2 0.6.5", + "socket2", "tokio", "tokio-util", "url", @@ -5460,14 +5459,281 @@ dependencies = [ ] [[package]] -name = "rtcp" -version = "0.17.2" +name = "rtc" +version = "0.20.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "adad7f6a501162881032fc84b4bc78ae11f1b30748180b9f5cbe8810bf613aab" +checksum = "ce487200429491d9295789a6dd5ce16f56c4715373c97164df9f544a220c8c64" dependencies = [ "bytes", - "thiserror 1.0.69", - "webrtc-util", + "hex", + "log", + "pem", + "rand 0.10.2", + "rcgen 0.14.8", + "ring", + "rtc-datachannel", + "rtc-dtls", + "rtc-ice", + "rtc-interceptor", + "rtc-mdns", + "rtc-media", + "rtc-rtcp", + "rtc-rtp", + "rtc-sctp", + "rtc-sdp", + "rtc-shared", + "rtc-srtp", + "rtc-stun", + "rtc-turn", + "rustls", + "sansio", + "serde", + "serde_json", + "sha2 0.10.9", + "unicase", + "url", +] + +[[package]] +name = "rtc-datachannel" +version = "0.20.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "bf1815e5d40058b8eb0ed79e89414ecbda1da74a848d741c68e4de5d66aed9c1" +dependencies = [ + "bytes", + "log", + "rtc-sctp", + "rtc-shared", + "sansio", +] + +[[package]] +name = "rtc-dtls" +version = "0.20.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "473e1837315631df4e0aafdf08b0386c77aaa8b1632bd7484bdd3bf7ded65b19" +dependencies = [ + "aes", + "bytecheck", + "byteorder", + "bytes", + "cbc", + "ccm", + "chacha20poly1305", + "der-parser 9.0.0", + "hmac 0.12.1", + "log", + "p256 0.13.2", + "p384", + "pem", + "rand 0.10.2", + "rand_core 0.6.4", + "rcgen 0.14.8", + "ring", + "rkyv", + "rtc-shared", + "rustls", + "sec1 0.7.3", + "sha1", + "sha2 0.10.9", + "subtle", + "x25519-dalek 2.0.1", + "x509-parser 0.16.0", +] + +[[package]] +name = "rtc-ice" +version = "0.20.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "9cbf161bce9556df68a9c2dd781d9cecbb8de09a3acba5655b5a4f6ad80aa06c" +dependencies = [ + "bytes", + "crc", + "log", + "rand 0.10.2", + "rtc-mdns", + "rtc-shared", + "rtc-stun", + "sansio", + "serde", + "url", + "uuid", +] + +[[package]] +name = "rtc-interceptor" +version = "0.20.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "e16672ca0665d05be30a8d51e539df18afe15491b5bf29ed644791d053575a96" +dependencies = [ + "log", + "rand 0.10.2", + "rtc-interceptor-derive", + "rtc-rtcp", + "rtc-rtp", + "rtc-shared", + "sansio", +] + +[[package]] +name = "rtc-interceptor-derive" +version = "0.20.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "4d50521cd5a26f7e829a5bf6ade44c3bec6855716fa7a0f6141b413c186d185d" +dependencies = [ + "proc-macro2", + "quote", + "syn 2.0.119", +] + +[[package]] +name = "rtc-mdns" +version = "0.20.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "52cba0f67ae11d0820c2ce589dd134a6618f3eddcbb1b53ba2f2b56d64da02ac" +dependencies = [ + "bytes", + "log", + "rtc-shared", + "sansio", + "socket2", +] + +[[package]] +name = "rtc-media" +version = "0.20.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "0842037bf13bf9effbe5c2ca1815e9a977db79becc859deb2d43239d606a1935" +dependencies = [ + "byteorder", + "bytes", + "rand 0.10.2", + "rtc-rtp", + "rtc-shared", + "thiserror 2.0.19", +] + +[[package]] +name = "rtc-rtcp" +version = "0.20.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "c46297e46a9aeb640001c73bcfc3df1831cb98bd3a93cd98543d188600af825e" +dependencies = [ + "bytes", + "rtc-shared", +] + +[[package]] +name = "rtc-rtp" +version = "0.20.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "7e5180eabadb76b7bb95c3341fa25232279733fbc9db7665e2abb9aaf6ea7556" +dependencies = [ + "bytes", + "memchr", + "rand 0.10.2", + "rtc-shared", + "serde", +] + +[[package]] +name = "rtc-sctp" +version = "0.20.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "85fff50f44bea4e5c37f9cd72cbe2d746beb62b111c2d4d9443ac3c5dd19d846" +dependencies = [ + "bytes", + "crc32c", + "log", + "rand 0.10.2", + "rtc-shared", + "rustc-hash", + "slab", + "thiserror 2.0.19", +] + +[[package]] +name = "rtc-sdp" +version = "0.20.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "6ef2bf0bc026bf4c524dc857c65b2c646425e810db33c2bb74fe85e121794786" +dependencies = [ + "rand 0.10.2", + "rtc-shared", + "url", +] + +[[package]] +name = "rtc-shared" +version = "0.20.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "542a2b3c560ba295f883fceba918a4f796d3f270b55679162308fa99b900d45f" +dependencies = [ + "aes", + "aes-gcm", + "bitflags 1.3.2", + "bytes", + "nix 0.26.4", + "p256 0.13.2", + "rand 0.10.2", + "rcgen 0.14.8", + "sec1 0.7.3", + "serde", + "substring", + "thiserror 2.0.19", + "url", + "winapi", +] + +[[package]] +name = "rtc-srtp" +version = "0.20.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "33666140469dd1e9effd093824a8aa66ac4c753ad1c9f0123c4acc9b8840c4c8" +dependencies = [ + "aes", + "byteorder", + "bytes", + "ctr", + "hmac 0.12.1", + "ring", + "rtc-rtcp", + "rtc-rtp", + "rtc-shared", + "sha1", + "subtle", +] + +[[package]] +name = "rtc-stun" +version = "0.20.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "46e84f06c69299b98f9fc271dda4b628d3f1694375061b8af1145b47a7e25f05" +dependencies = [ + "base64 0.22.1", + "bytes", + "crc", + "lazy_static", + "md-5", + "rand 0.10.2", + "ring", + "rtc-shared", + "sansio", + "subtle", + "url", +] + +[[package]] +name = "rtc-turn" +version = "0.20.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "6169b99a62428c832423613aadc460d32edb792afd216bde95670d28443bf527" +dependencies = [ + "bytes", + "log", + "rtc-shared", + "rtc-stun", + "sansio", ] [[package]] @@ -5488,21 +5754,6 @@ dependencies = [ "tokio", ] -[[package]] -name = "rtp" -version = "0.17.2" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "149329e78ada26b5e174a4c281a7a7e5c8bda6008adc90daddd6f98fce56db29" -dependencies = [ - "bytes", - "memchr", - "portable-atomic", - "rand 0.9.5", - "serde", - "thiserror 1.0.69", - "webrtc-util", -] - [[package]] name = "rust-embed" version = "8.12.0" @@ -5694,6 +5945,12 @@ dependencies = [ "winapi-util", ] +[[package]] +name = "sansio" +version = "1.0.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "c62751faa8bc286982334a082fe125184a29fc89d17775766e4f891b7d726980" + [[package]] name = "schannel" version = "0.1.29" @@ -5709,18 +5966,6 @@ version = "1.2.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "94143f37725109f92c262ed2cf5e59bce7498c01bcc1502d7b9afe439a4e9f49" -[[package]] -name = "sdp" -version = "0.17.2" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "66b6eecfa5151edef84d544ff3885dff98e5b5fe97586757a4b5118bde51c958" -dependencies = [ - "rand 0.9.5", - "substring", - "thiserror 1.0.69", - "url", -] - [[package]] name = "sec1" version = "0.7.3" @@ -6028,15 +6273,6 @@ version = "1.15.2" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "8ed6a63f02c8539c91a8685a86f4099661ba3da017932f6ebbea6de3f0fa7c90" -[[package]] -name = "smol_str" -version = "0.2.2" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "dd538fb6910ac1099850255cf94a94df6551fbdd602454387d0adb2d1ca6dead" -dependencies = [ - "serde", -] - [[package]] name = "snow" version = "0.10.0" @@ -6054,16 +6290,6 @@ dependencies = [ "subtle", ] -[[package]] -name = "socket2" -version = "0.5.10" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "e22376abed350d73dd1cd119b57ffccad95b4e585a7cda43e286245ce23c0678" -dependencies = [ - "libc", - "windows-sys 0.52.0", -] - [[package]] name = "socket2" version = "0.6.5" @@ -6483,7 +6709,7 @@ dependencies = [ "parking_lot", "pin-project-lite", "signal-hook-registry", - "socket2 0.6.5", + "socket2", "tokio-macros", "windows-sys 0.61.2", ] @@ -6808,27 +7034,6 @@ dependencies = [ "toml", ] -[[package]] -name = "turn" -version = "0.17.2" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "d99249a493335eb44c4d7943a8751b22561a82c9903d8eb2d29780b3927880a7" -dependencies = [ - "async-trait", - "base64 0.22.1", - "futures", - "log", - "md-5", - "portable-atomic", - "rand 0.9.5", - "ring", - "stun", - "thiserror 1.0.69", - "tokio", - "tokio-util", - "webrtc-util", -] - [[package]] name = "typed-path" version = "0.12.3" @@ -6960,15 +7165,6 @@ version = "0.9.5" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "0b928f33d975fc6ad9f86c8f283853ad26bdd5b10b7f1542aa2fa15e2289105a" -[[package]] -name = "waitgroup" -version = "0.1.2" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "d1f50000a783467e6c0200f9d10642f4bc424e39efc1b770203e88b488f79292" -dependencies = [ - "atomic-waker", -] - [[package]] name = "walkdir" version = "2.5.0" @@ -7171,152 +7367,20 @@ dependencies = [ [[package]] name = "webrtc" -version = "0.17.2" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "baaacdf9d96224d7b6e2872ba065578f38775f7634367e3ac2cc87c7271da433" -dependencies = [ - "arc-swap", - "async-trait", - "bytes", - "dtls", - "hex", - "interceptor", - "lazy_static", - "log", - "pem", - "portable-atomic", - "rand 0.9.5", - "rcgen", - "regex", - "ring", - "rtcp", - "rtp", - "sdp", - "serde", - "serde_json", - "sha2 0.10.9", - "smol_str", - "stun", - "thiserror 1.0.69", - "tokio", - "turn", - "unicase", - "url", - "waitgroup", - "webrtc-data", - "webrtc-ice", - "webrtc-mdns", - "webrtc-media", - "webrtc-sctp", - "webrtc-srtp", - "webrtc-util", -] - -[[package]] -name = "webrtc-data" -version = "0.17.2" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "bd470286275809f2fcfcdb1e73ef5f1500be82eff7fe98150ce81b20aad5a2a4" -dependencies = [ - "bytes", - "log", - "portable-atomic", - "thiserror 1.0.69", - "tokio", - "webrtc-sctp", - "webrtc-util", -] - -[[package]] -name = "webrtc-ice" -version = "0.17.2" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "5b7fd30f52e6fda8664779b84b7904b2553b76fee24d9ca665e774ae32b13f53" -dependencies = [ - "arc-swap", - "async-trait", - "crc", - "log", - "portable-atomic", - "rand 0.9.5", - "serde", - "serde_json", - "stun", - "thiserror 1.0.69", - "tokio", - "turn", - "url", - "uuid", - "waitgroup", - "webrtc-mdns", - "webrtc-util", -] - -[[package]] -name = "webrtc-mdns" -version = "0.17.2" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "91ffa0ea00c0fae979aafa5db285fec5eeb360e1ea246ebec529ba4e09f8e03e" -dependencies = [ - "log", - "socket2 0.5.10", - "thiserror 1.0.69", - "tokio", - "webrtc-util", -] - -[[package]] -name = "webrtc-media" -version = "0.17.2" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "26a6c7335bdd03dc023cb9a3bd7966866d969c92c86987b73aa16dfa39c7ca29" -dependencies = [ - "byteorder", - "bytes", - "rand 0.9.5", - "rtp", - "thiserror 1.0.69", -] - -[[package]] -name = "webrtc-sctp" -version = "0.17.2" +version = "0.20.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "1c4b637f0d8eb96d900ac0f79b3060ddd21ca88dfefa467e96e67ab0016f2574" +checksum = "530c3a4621c0a7a6559eae25f34ee7a430e78ecf4a6bfece1fc43a5b71985193" dependencies = [ - "arc-swap", + "async-broadcast", + "async-channel", "async-trait", "bytes", - "crc", - "log", - "portable-atomic", - "rand 0.9.5", - "thiserror 1.0.69", - "tokio", - "webrtc-util", -] - -[[package]] -name = "webrtc-srtp" -version = "0.17.2" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "45d2b667a0b5d04eebcb7cb22fd51b2af3721f418097b496e72de16386c8d0c3" -dependencies = [ - "aead", - "aes", - "aes-gcm", - "byteorder", - "bytes", - "ctr", - "hmac 0.12.1", + "event-listener", + "futures", "log", - "rtcp", - "rtp", - "sha1", - "subtle", - "thiserror 1.0.69", + "quinn-udp 0.6.1", + "rtc", "tokio", - "webrtc-util", ] [[package]] @@ -7661,7 +7725,6 @@ dependencies = [ "lazy_static", "nom", "oid-registry 0.7.1", - "ring", "rusticata-macros", "thiserror 1.0.69", "time", @@ -7679,6 +7742,7 @@ dependencies = [ "lazy_static", "nom", "oid-registry 0.8.1", + "ring", "rusticata-macros", "thiserror 2.0.19", "time", @@ -7744,6 +7808,10 @@ name = "yasna" version = "0.6.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "b5f6765e852b9b4dc8e2a76843e4d64d1cea8e79bcde0b6901aea8e7c7f08282" +dependencies = [ + "bit-vec", + "time", +] [[package]] name = "yoke" diff --git a/transports/webrtc/Cargo.toml b/transports/webrtc/Cargo.toml index 3bc79fe668e..2aee719561d 100644 --- a/transports/webrtc/Cargo.toml +++ b/transports/webrtc/Cargo.toml @@ -12,6 +12,7 @@ categories = ["network-programming", "asynchronous"] [dependencies] async-trait = "0.1" +bytes = { workspace = true } futures = { workspace = true } futures-timer = { workspace = true } hex = "0.4" @@ -22,17 +23,17 @@ libp2p-identity = { workspace = true } libp2p-webrtc-utils = { workspace = true } multihash = { workspace = true } rand = { workspace = true } -rcgen = { workspace = true } +rcgen = "0.14" +rtc = { version = "0.20", optional = true } stun = "0.17" thiserror = { workspace = true } tokio = { workspace = true, features = ["net"], optional = true } -tokio-util = { version = "0.7", features = ["compat"], optional = true } tracing = { workspace = true } -webrtc = { version = "0.17", optional = true } +webrtc = { version = "0.20", default-features = false, features = ["runtime-tokio"], optional = true } [features] -tokio = ["dep:tokio", "dep:tokio-util", "dep:webrtc", "if-watch/tokio"] -pem = ["webrtc?/pem"] +tokio = ["dep:rtc", "dep:tokio", "dep:webrtc", "if-watch/tokio"] +pem = ["rtc?/pem"] [dev-dependencies] libp2p-identity = { workspace = true, features = ["rand"] } diff --git a/transports/webrtc/src/tokio/certificate.rs b/transports/webrtc/src/tokio/certificate.rs index 373062d7693..3538f35830a 100644 --- a/transports/webrtc/src/tokio/certificate.rs +++ b/transports/webrtc/src/tokio/certificate.rs @@ -19,7 +19,7 @@ // DEALINGS IN THE SOFTWARE. use rand::{CryptoRng, Rng}; -use webrtc::peer_connection::certificate::RTCCertificate; +use webrtc::peer_connection::RTCCertificate; use crate::tokio::fingerprint::Fingerprint; @@ -92,7 +92,7 @@ pub struct Error(#[from] Kind); #[derive(thiserror::Error, Debug)] enum Kind { #[error(transparent)] - InvalidPEM(#[from] webrtc::Error), + InvalidPEM(#[from] webrtc::error::Error), } #[cfg(all(test, feature = "pem"))] diff --git a/transports/webrtc/src/tokio/connection.rs b/transports/webrtc/src/tokio/connection.rs index 3009e7ed33c..91bcc4365e7 100644 --- a/transports/webrtc/src/tokio/connection.rs +++ b/transports/webrtc/src/tokio/connection.rs @@ -25,45 +25,26 @@ use std::{ }; use futures::{ - StreamExt, - channel::{ - mpsc, - oneshot::{self, Sender}, - }, - future::BoxFuture, - lock::Mutex as FutMutex, - ready, + StreamExt, channel::mpsc, future::BoxFuture, lock::Mutex as FutMutex, ready, stream::FuturesUnordered, }; use libp2p_core::muxing::{StreamMuxer, StreamMuxerEvent}; use webrtc::{ - data::data_channel::DataChannel as DetachedDataChannel, data_channel::RTCDataChannel, - peer_connection::RTCPeerConnection, + data_channel::{DataChannel, DataChannelEvent}, + peer_connection::{PeerConnection, PeerConnectionEventHandler}, + runtime::Runtime, }; use crate::tokio::{error::Error, stream, stream::Stream}; -/// Maximum number of unprocessed data channels. -/// See [`Connection::poll_inbound`]. const MAX_DATA_CHANNELS_IN_FLIGHT: usize = 10; -/// A WebRTC connection, wrapping [`RTCPeerConnection`] and implementing [`StreamMuxer`] trait. +/// A WebRTC connection implementing libp2p's stream muxer interface. pub struct Connection { - /// [`RTCPeerConnection`] to the remote peer. - /// - /// Uses futures mutex because used in async code (see poll_outbound and poll_close). - peer_conn: Arc>, - - /// Channel onto which incoming data channels are put. - incoming_data_channels_rx: mpsc::Receiver>, - - /// Future, which, once polled, will result in an outbound stream. - outbound_fut: Option, Error>>>, - - /// Future, which, once polled, will result in closing the entire connection. + peer_conn: Arc, + incoming_data_channels_rx: mpsc::Receiver>, + outbound_fut: Option, Error>>>, close_fut: Option>>, - - /// A list of futures, which, once completed, signal that a [`Stream`] has been dropped. drop_listeners: FuturesUnordered, no_drop_listeners_waker: Option, } @@ -71,80 +52,19 @@ pub struct Connection { impl Unpin for Connection {} impl Connection { - /// Creates a new connection. - pub(crate) async fn new(rtc_conn: RTCPeerConnection) -> Self { - let (data_channel_tx, data_channel_rx) = mpsc::channel(MAX_DATA_CHANNELS_IN_FLIGHT); - - Connection::register_incoming_data_channels_handler( - &rtc_conn, - Arc::new(FutMutex::new(data_channel_tx)), - ) - .await; - + pub(crate) fn new( + peer_conn: Arc, + incoming_data_channels_rx: mpsc::Receiver>, + ) -> Self { Self { - peer_conn: Arc::new(FutMutex::new(rtc_conn)), - incoming_data_channels_rx: data_channel_rx, + peer_conn, + incoming_data_channels_rx, outbound_fut: None, close_fut: None, - drop_listeners: FuturesUnordered::default(), + drop_listeners: FuturesUnordered::new(), no_drop_listeners_waker: None, } } - - /// Registers a handler for incoming data channels. - /// - /// NOTE: `mpsc::Sender` is wrapped in `Arc` because cloning a raw sender would make the channel - /// unbounded. "The channel’s capacity is equal to buffer + num-senders. In other words, each - /// sender gets a guaranteed slot in the channel capacity..." - /// See - async fn register_incoming_data_channels_handler( - rtc_conn: &RTCPeerConnection, - tx: Arc>>>, - ) { - rtc_conn.on_data_channel(Box::new(move |data_channel: Arc| { - tracing::debug!(channel=%data_channel.id(), "Incoming data channel"); - - let tx = tx.clone(); - - Box::pin(async move { - data_channel.on_open({ - let data_channel = data_channel.clone(); - Box::new(move || { - tracing::debug!(channel=%data_channel.id(), "Data channel open"); - - Box::pin(async move { - let data_channel = data_channel.clone(); - let id = data_channel.id(); - match data_channel.detach().await { - Ok(detached) => { - let mut tx = tx.lock().await; - if let Err(e) = tx.try_send(detached.clone()) { - tracing::error!(channel=%id, "Can't send data channel: {}", e); - // We're not accepting data channels fast enough => - // close this channel. - // - // Ideally we'd refuse to accept a data channel - // during the negotiation process, but it's not - // possible with the current API. - if let Err(e) = detached.close().await { - tracing::error!( - channel=%id, - "Failed to close data channel: {}", - e - ); - } - } - } - Err(e) => { - tracing::error!(channel=%id, "Can't detach data channel: {}", e); - } - }; - }) - }) - }); - }) - })); - } } impl StreamMuxer for Connection { @@ -155,29 +75,17 @@ impl StreamMuxer for Connection { mut self: Pin<&mut Self>, cx: &mut Context<'_>, ) -> Poll> { - match ready!(self.incoming_data_channels_rx.poll_next_unpin(cx)) { - Some(detached) => { - tracing::trace!(stream=%detached.stream_identifier(), "Incoming stream"); - - let (stream, drop_listener) = Stream::new(detached); - self.drop_listeners.push(drop_listener); - if let Some(waker) = self.no_drop_listeners_waker.take() { - waker.wake() - } - - Poll::Ready(Ok(stream)) - } - None => { - debug_assert!( - false, - "Sender-end of channel should be owned by `RTCPeerConnection`" - ); - - // Return `Pending` without registering a waker: If the channel is - // closed, we don't need to be called anymore. - Poll::Pending - } + let Some(data_channel) = ready!(self.incoming_data_channels_rx.poll_next_unpin(cx)) else { + return Poll::Pending; + }; + + tracing::trace!(channel=%data_channel.id(), "Incoming stream"); + let (stream, drop_listener) = Stream::new(data_channel); + self.drop_listeners.push(drop_listener); + if let Some(waker) = self.no_drop_listeners_waker.take() { + waker.wake(); } + Poll::Ready(Ok(stream)) } fn poll( @@ -187,9 +95,7 @@ impl StreamMuxer for Connection { loop { match ready!(self.drop_listeners.poll_next_unpin(cx)) { Some(Ok(())) => {} - Some(Err(e)) => { - tracing::debug!("a DropListener failed: {e}") - } + Some(Err(err)) => tracing::debug!("a DropListener failed: {err}"), None => { self.no_drop_listeners_waker = Some(cx.waker().clone()); return Poll::Pending; @@ -202,61 +108,37 @@ impl StreamMuxer for Connection { mut self: Pin<&mut Self>, cx: &mut Context<'_>, ) -> Poll> { - let peer_conn = self.peer_conn.clone(); + let peer_conn = Arc::clone(&self.peer_conn); let fut = self.outbound_fut.get_or_insert_with(|| { Box::pin(async move { - let peer_conn = peer_conn.lock().await; - let data_channel = peer_conn.create_data_channel("", None).await?; - - // No need to hold the lock during the DTLS handshake. - drop(peer_conn); - tracing::trace!(channel=%data_channel.id(), "Opening data channel"); - - let (tx, rx) = oneshot::channel::>(); - - // Wait until the data channel is opened and detach it. - register_data_channel_open_handler(data_channel, tx).await; - - // Wait until data channel is opened and ready to use - match rx.await { - Ok(detached) => Ok(detached), - Err(e) => Err(Error::Internal(e.to_string())), - } + await_data_channel_open(data_channel).await }) }); match ready!(fut.as_mut().poll(cx)) { - Ok(detached) => { + Ok(data_channel) => { self.outbound_fut = None; - - tracing::trace!(stream=%detached.stream_identifier(), "Outbound stream"); - - let (stream, drop_listener) = Stream::new(detached); + let (stream, drop_listener) = Stream::new(data_channel); self.drop_listeners.push(drop_listener); if let Some(waker) = self.no_drop_listeners_waker.take() { - waker.wake() + waker.wake(); } - Poll::Ready(Ok(stream)) } - Err(e) => { + Err(err) => { self.outbound_fut = None; - Poll::Ready(Err(e)) + Poll::Ready(Err(err)) } } } fn poll_close(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll> { - tracing::debug!("Closing connection"); - - let peer_conn = self.peer_conn.clone(); + let peer_conn = Arc::clone(&self.peer_conn); let fut = self.close_fut.get_or_insert_with(|| { Box::pin(async move { - let peer_conn = peer_conn.lock().await; peer_conn.close().await?; - Ok(()) }) }); @@ -267,40 +149,70 @@ impl StreamMuxer for Connection { self.close_fut = None; Poll::Ready(Ok(())) } - Err(e) => { + Err(err) => { self.close_fut = None; - Poll::Ready(Err(e)) + Poll::Ready(Err(err)) } } } } -pub(crate) async fn register_data_channel_open_handler( - data_channel: Arc, - data_channel_tx: Sender>, -) { - data_channel.on_open({ - let data_channel = data_channel.clone(); - Box::new(move || { - tracing::debug!(channel=%data_channel.id(), "Data channel open"); +pub(crate) struct ConnectionHandler { + incoming_tx: Arc>>>, + runtime: Arc, +} - Box::pin(async move { - let data_channel = data_channel.clone(); - let id = data_channel.id(); - match data_channel.detach().await { - Ok(detached) => { - if let Err(e) = data_channel_tx.send(detached.clone()) { - tracing::error!(channel=%id, "Can't send data channel: {:?}", e); - if let Err(e) = detached.close().await { - tracing::error!(channel=%id, "Failed to close data channel: {}", e); - } +impl ConnectionHandler { + pub(crate) fn new( + runtime: Arc, + ) -> (Arc, mpsc::Receiver>) { + let (incoming_tx, incoming_rx) = mpsc::channel(MAX_DATA_CHANNELS_IN_FLIGHT); + ( + Arc::new(Self { + incoming_tx: Arc::new(FutMutex::new(incoming_tx)), + runtime, + }), + incoming_rx, + ) + } +} + +#[async_trait::async_trait] +impl PeerConnectionEventHandler for ConnectionHandler { + async fn on_data_channel(&self, data_channel: Arc) { + let incoming_tx = Arc::clone(&self.incoming_tx); + self.runtime.spawn(Box::pin(async move { + let channel_id = data_channel.id(); + match await_data_channel_open(data_channel).await { + Ok(data_channel) => { + let mut tx = incoming_tx.lock().await; + if let Err(err) = tx.try_send(data_channel.clone()) { + tracing::error!(channel=%channel_id, "Can't queue data channel: {err}"); + if let Err(err) = data_channel.close().await { + tracing::error!(channel=%channel_id, "Failed to close data channel: {err}"); } } - Err(e) => { - tracing::error!(channel=%id, "Can't detach data channel: {}", e); - } - }; - }) - }) - }); + } + Err(err) => tracing::debug!(channel=%channel_id, "Data channel failed to open: {err}"), + } + })); + } +} + +#[allow(clippy::result_large_err)] +pub(crate) async fn await_data_channel_open( + data_channel: Arc, +) -> Result, Error> { + loop { + match data_channel.poll().await { + Some(DataChannelEvent::OnOpen) => return Ok(data_channel), + Some(DataChannelEvent::OnError) => { + return Err(Error::Internal("data channel failed to open".into())); + } + Some(DataChannelEvent::OnClosing | DataChannelEvent::OnClose) | None => { + return Err(Error::Internal("data channel closed before opening".into())); + } + Some(_) => {} + } + } } diff --git a/transports/webrtc/src/tokio/error.rs b/transports/webrtc/src/tokio/error.rs index 1b274686ef3..45ecf4541e6 100644 --- a/transports/webrtc/src/tokio/error.rs +++ b/transports/webrtc/src/tokio/error.rs @@ -25,7 +25,7 @@ use thiserror::Error; #[derive(Error, Debug)] pub enum Error { #[error(transparent)] - WebRTC(#[from] webrtc::Error), + WebRTC(#[from] webrtc::error::Error), #[error("IO error")] Io(#[from] std::io::Error), #[error("failed to authenticate peer")] diff --git a/transports/webrtc/src/tokio/fingerprint.rs b/transports/webrtc/src/tokio/fingerprint.rs index c075e486232..da8bab82cc8 100644 --- a/transports/webrtc/src/tokio/fingerprint.rs +++ b/transports/webrtc/src/tokio/fingerprint.rs @@ -18,7 +18,7 @@ // FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER // DEALINGS IN THE SOFTWARE. -use webrtc::dtls_transport::dtls_fingerprint::RTCDtlsFingerprint; +use rtc::peer_connection::transport::RTCDtlsFingerprint; const SHA256: &str = "sha-256"; diff --git a/transports/webrtc/src/tokio/sdp.rs b/transports/webrtc/src/tokio/sdp.rs index f28c5c33105..1c05db78fb5 100644 --- a/transports/webrtc/src/tokio/sdp.rs +++ b/transports/webrtc/src/tokio/sdp.rs @@ -22,7 +22,7 @@ use std::net::SocketAddr; pub(crate) use libp2p_webrtc_utils::sdp::random_ufrag; use libp2p_webrtc_utils::{Fingerprint, sdp::render_description}; -use webrtc::peer_connection::sdp::session_description::RTCSessionDescription; +use webrtc::peer_connection::RTCSessionDescription; /// Creates the SDP answer used by the client. pub(crate) fn answer( diff --git a/transports/webrtc/src/tokio/stream.rs b/transports/webrtc/src/tokio/stream.rs index 9d5a9faf440..4d2202c0fd3 100644 --- a/transports/webrtc/src/tokio/stream.rs +++ b/transports/webrtc/src/tokio/stream.rs @@ -19,44 +19,37 @@ // DEALINGS IN THE SOFTWARE. use std::{ + io, pin::Pin, sync::Arc, task::{Context, Poll}, }; -use futures::prelude::*; -use libp2p_webrtc_utils::MAX_MSG_LEN; -use tokio_util::compat::{Compat, TokioAsyncReadCompatExt}; -use webrtc::data::data_channel::{DataChannel, PollDataChannel}; +use bytes::{Bytes, BytesMut}; +use futures::{AsyncRead, AsyncWrite, future::BoxFuture, ready}; +use webrtc::data_channel::{DataChannel, DataChannelEvent}; /// A substream on top of a WebRTC data channel. -/// -/// To be a proper libp2p substream, we need to implement [`AsyncRead`] and [`AsyncWrite`] as well -/// as support a half-closed state which we do by framing messages in a protobuf envelope. pub struct Stream { - inner: libp2p_webrtc_utils::Stream>, + inner: libp2p_webrtc_utils::Stream, } -pub(crate) type DropListener = libp2p_webrtc_utils::DropListener>; +pub(crate) type DropListener = libp2p_webrtc_utils::DropListener; impl Stream { - /// Returns a new `Substream` and a listener, which will notify the receiver when/if the - /// substream is dropped. - pub(crate) fn new(data_channel: Arc) -> (Self, DropListener) { - let mut data_channel = PollDataChannel::new(data_channel).compat(); - data_channel.get_mut().set_read_buf_capacity(MAX_MSG_LEN); - - let (inner, drop_listener) = libp2p_webrtc_utils::Stream::new(data_channel); - + pub(crate) fn new(data_channel: Arc) -> (Self, DropListener) { + let (inner, drop_listener) = + libp2p_webrtc_utils::Stream::new(DataChannelIo::new(data_channel)); (Self { inner }, drop_listener) } } + impl AsyncRead for Stream { fn poll_read( self: Pin<&mut Self>, cx: &mut Context<'_>, buf: &mut [u8], - ) -> Poll> { + ) -> Poll> { Pin::new(&mut self.get_mut().inner).poll_read(cx, buf) } } @@ -66,15 +59,144 @@ impl AsyncWrite for Stream { self: Pin<&mut Self>, cx: &mut Context<'_>, buf: &[u8], - ) -> Poll> { + ) -> Poll> { Pin::new(&mut self.get_mut().inner).poll_write(cx, buf) } - fn poll_flush(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll> { + fn poll_flush(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll> { Pin::new(&mut self.get_mut().inner).poll_flush(cx) } - fn poll_close(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll> { + fn poll_close(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll> { Pin::new(&mut self.get_mut().inner).poll_close(cx) } } + +/// Presents the message-oriented 0.20 data-channel API as a byte stream. +pub(crate) struct DataChannelIo { + data_channel: Arc, + read_fut: Option>>, + write_fut: Option<(usize, BoxFuture<'static, webrtc::error::Result<()>>)>, + close_fut: Option>>, + read_buffer: Bytes, + read_closed: bool, +} + +impl DataChannelIo { + fn new(data_channel: Arc) -> Self { + Self { + data_channel, + read_fut: None, + write_fut: None, + close_fut: None, + read_buffer: Bytes::new(), + read_closed: false, + } + } + + fn poll_pending_write(&mut self, cx: &mut Context<'_>) -> Poll>> { + let Some((len, fut)) = self.write_fut.as_mut() else { + return Poll::Ready(Ok(None)); + }; + let len = *len; + ready!(fut.as_mut().poll(cx)).map_err(io_error)?; + self.write_fut = None; + Poll::Ready(Ok(Some(len))) + } +} + +impl Clone for DataChannelIo { + fn clone(&self) -> Self { + Self::new(Arc::clone(&self.data_channel)) + } +} + +impl Unpin for DataChannelIo {} + +impl AsyncRead for DataChannelIo { + fn poll_read( + mut self: Pin<&mut Self>, + cx: &mut Context<'_>, + buf: &mut [u8], + ) -> Poll> { + loop { + if !self.read_buffer.is_empty() { + let len = self.read_buffer.len().min(buf.len()); + let data = self.read_buffer.split_to(len); + buf[..len].copy_from_slice(&data); + return Poll::Ready(Ok(len)); + } + if self.read_closed { + return Poll::Ready(Ok(0)); + } + + let data_channel = Arc::clone(&self.data_channel); + let fut = self + .read_fut + .get_or_insert_with(|| Box::pin(async move { data_channel.poll().await })); + let event = ready!(fut.as_mut().poll(cx)); + self.read_fut = None; + + match event { + Some(DataChannelEvent::OnMessage(message)) => { + self.read_buffer = message.data.freeze(); + } + Some(DataChannelEvent::OnClose) | None => self.read_closed = true, + Some(DataChannelEvent::OnError) => { + return Poll::Ready(Err(io::Error::other("WebRTC data channel error"))); + } + Some(_) => {} + } + } + } +} + +impl AsyncWrite for DataChannelIo { + fn poll_write( + mut self: Pin<&mut Self>, + cx: &mut Context<'_>, + buf: &[u8], + ) -> Poll> { + if let Poll::Ready(result) = self.poll_pending_write(cx) { + if let Some(len) = result? { + return Poll::Ready(Ok(len)); + } + } else { + return Poll::Pending; + } + if buf.is_empty() { + return Poll::Ready(Ok(0)); + } + + let data_channel = Arc::clone(&self.data_channel); + let data = BytesMut::from(buf); + let len = data.len(); + self.write_fut = Some((len, Box::pin(async move { data_channel.send(data).await }))); + match self.poll_pending_write(cx) { + Poll::Ready(Ok(Some(len))) => Poll::Ready(Ok(len)), + Poll::Ready(Ok(None)) => unreachable!("write future was just installed"), + Poll::Ready(Err(err)) => Poll::Ready(Err(err)), + Poll::Pending => Poll::Pending, + } + } + + fn poll_flush(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll> { + ready!(self.poll_pending_write(cx))?; + Poll::Ready(Ok(())) + } + + fn poll_close(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll> { + ready!(self.as_mut().poll_flush(cx))?; + let data_channel = Arc::clone(&self.data_channel); + let fut = self + .close_fut + .get_or_insert_with(|| Box::pin(async move { data_channel.close().await })); + ready!(fut.as_mut().poll(cx)).map_err(io_error)?; + self.close_fut = None; + Poll::Ready(Ok(())) + } +} + +fn io_error(error: webrtc::error::Error) -> io::Error { + io::Error::other(error) +} diff --git a/transports/webrtc/src/tokio/transport.rs b/transports/webrtc/src/tokio/transport.rs index b14a3062d0e..e4ce0e9094d 100644 --- a/transports/webrtc/src/tokio/transport.rs +++ b/transports/webrtc/src/tokio/transport.rs @@ -33,7 +33,7 @@ use libp2p_core::{ }; use libp2p_identity as identity; use libp2p_identity::PeerId; -use webrtc::peer_connection::configuration::RTCConfiguration; +use webrtc::peer_connection::{RTCConfiguration, RTCConfigurationBuilder}; use crate::tokio::{ certificate::Certificate, @@ -373,10 +373,9 @@ impl Config { Self { id_keys, - inner: RTCConfiguration { - certificates: vec![certificate.to_rtc_certificate()], - ..RTCConfiguration::default() - }, + inner: RTCConfigurationBuilder::new() + .with_certificates(vec![certificate.to_rtc_certificate()]) + .build(), fingerprint, } } diff --git a/transports/webrtc/src/tokio/udp_mux.rs b/transports/webrtc/src/tokio/udp_mux.rs index 1be3bb822e0..261a4457011 100644 --- a/transports/webrtc/src/tokio/udp_mux.rs +++ b/transports/webrtc/src/tokio/udp_mux.rs @@ -20,34 +20,32 @@ use std::{ collections::{HashMap, HashSet}, - io, - io::ErrorKind, + fmt, + future::Future, + io::{self, ErrorKind, IoSliceMut}, net::SocketAddr, - sync::Arc, + pin::Pin, + sync::{Arc, Mutex}, task::{Context, Poll}, + time::Duration, }; -use async_trait::async_trait; -use futures::{ - StreamExt, - channel::oneshot, - future::{BoxFuture, FutureExt, OptionFuture}, - stream::FuturesUnordered, -}; +use futures::{channel::mpsc, prelude::*, ready}; use stun::{ attributes::ATTR_USERNAME, message::{Message as STUNMessage, is_message as is_stun_message}, }; -use thiserror::Error; use tokio::{io::ReadBuf, net::UdpSocket}; -use webrtc::{ - ice::udp_mux::{UDPMux, UDPMuxConn, UDPMuxConnParams, UDPMuxWriter}, - util::{Conn, Error}, +use webrtc::runtime::{ + AsyncInterval, AsyncTcpListener, AsyncTcpStream, AsyncUdpSocket, JoinHandle, RecvMeta, Runtime, + Transmit, }; use crate::tokio::req_res_chan; -const RECEIVE_MTU: usize = 8192; +const RECEIVE_MTU: usize = 64 * 1024; +const RECEIVE_QUEUE_CAPACITY: usize = 64; +const SEND_QUEUE_CAPACITY: usize = 64; /// A previously unseen address of a remote which has sent us an ICE binding request. #[derive(Debug)] @@ -56,80 +54,62 @@ pub(crate) struct NewAddr { pub(crate) ufrag: String, } -/// An event emitted by [`UDPMuxNewAddr`] when it's polled. +/// An event emitted by [`UDPMuxNewAddr`] when it is polled. #[derive(Debug)] pub(crate) enum UDPMuxEvent { - /// Connection error. UDP mux should be stopped. - Error(std::io::Error), - /// Got a [`NewAddr`] from the socket. + Error(io::Error), NewAddr(NewAddr), } -/// A modified version of [`webrtc::ice::udp_mux::UDPMuxDefault`]. -/// -/// - It has been rewritten to work without locks and channels instead. -/// - It reports previously unseen addresses instead of ignoring them. +#[derive(Debug)] +struct Datagram { + data: Vec, + remote_addr: SocketAddr, +} + +#[derive(Debug)] +struct Route { + incoming: mpsc::Sender, + remote_addrs: HashSet, +} + +/// Demultiplexes one listening UDP socket into one socket-like handle per ICE ufrag. pub(crate) struct UDPMuxNewAddr { udp_sock: UdpSocket, - listen_addr: SocketAddr, - - /// Maps from ufrag to the underlying connection. - conns: HashMap, - - /// Maps from socket address to the underlying connection. - address_map: HashMap, - - /// Set of the new addresses to avoid sending the same address multiple times. + conns: HashMap, + address_map: HashMap, new_addrs: HashSet, - - /// `true` when UDP mux is closed. - is_closed: bool, - - send_buffer: Option<(Vec, SocketAddr, oneshot::Sender>)>, - - close_futures: FuturesUnordered>, - write_future: OptionFuture>, - - close_command: req_res_chan::Receiver<(), Result<(), Error>>, - get_conn_command: req_res_chan::Receiver, Error>>, - remove_conn_command: req_res_chan::Receiver, - registration_command: req_res_chan::Receiver<(UDPMuxConn, SocketAddr), ()>, - send_command: req_res_chan::Receiver<(Vec, SocketAddr), Result>, - + send_buffer: Option<(Vec, SocketAddr)>, + send_command: mpsc::Receiver<(Vec, SocketAddr)>, + get_conn_command: req_res_chan::Receiver>>, udp_mux_handle: Arc, - udp_mux_writer_handle: Arc, } impl UDPMuxNewAddr { - pub(crate) fn listen_on(addr: SocketAddr) -> Result { + pub(crate) fn listen_on(addr: SocketAddr) -> io::Result { let std_sock = std::net::UdpSocket::bind(addr)?; std_sock.set_nonblocking(true)?; - let tokio_socket = UdpSocket::from_std(std_sock)?; - let listen_addr = tokio_socket.local_addr()?; - - let (udp_mux_handle, close_command, get_conn_command, remove_conn_command) = - UdpMuxHandle::new(); - let (udp_mux_writer_handle, registration_command, send_command) = UdpMuxWriterHandle::new(); + let udp_sock = UdpSocket::from_std(std_sock)?; + let listen_addr = udp_sock.local_addr()?; + let (get_conn_sender, get_conn_command) = req_res_chan::new(1); + let (send_sender, send_command) = mpsc::channel(SEND_QUEUE_CAPACITY); + let udp_mux_handle = Arc::new(UdpMuxHandle { + get_conn_sender, + send_sender: Arc::new(Mutex::new(send_sender)), + }); Ok(Self { - udp_sock: tokio_socket, + udp_sock, listen_addr, - conns: HashMap::default(), - address_map: HashMap::default(), - new_addrs: HashSet::default(), - is_closed: false, + conns: HashMap::new(), + address_map: HashMap::new(), + new_addrs: HashSet::new(), send_buffer: None, - close_futures: FuturesUnordered::default(), - write_future: OptionFuture::default(), - close_command, - get_conn_command, - remove_conn_command, - registration_command, send_command, - udp_mux_handle: Arc::new(udp_mux_handle), - udp_mux_writer_handle: Arc::new(udp_mux_writer_handle), + get_conn_command, + udp_mux_handle, }) } @@ -138,442 +118,352 @@ impl UDPMuxNewAddr { } pub(crate) fn udp_mux_handle(&self) -> Arc { - self.udp_mux_handle.clone() + Arc::clone(&self.udp_mux_handle) } - /// Create a muxed connection for a given ufrag. - fn create_muxed_conn(&self, ufrag: &str) -> Result { - let local_addr = self.udp_sock.local_addr()?; - - let params = UDPMuxConnParams { - local_addr, - key: ufrag.into(), - udp_mux: Arc::downgrade( - &(self.udp_mux_writer_handle.clone() as Arc), - ), - }; - - Ok(UDPMuxConn::new(params)) - } - - /// Returns a muxed connection if the `ufrag` from the given STUN message matches an existing - /// connection. - fn conn_from_stun_message( - &self, - buffer: &[u8], - addr: &SocketAddr, - ) -> Option> { - match ufrag_from_stun_message(buffer, true) { - Ok(ufrag) => { - if let Some(conn) = self.conns.get(&ufrag) { - let associated_addrs = conn.get_addresses(); - // This basically ensures only one address is registered per ufrag. - if associated_addrs.is_empty() || associated_addrs.contains(addr) { - return Some(Ok(conn.clone())); - } else { - return Some(Err(ConnQueryError::UfragAlreadyTaken { associated_addrs })); - } - } - None - } - Err(e) => { - tracing::debug!(address=%addr, "{}", e); - None - } - } - } - - /// Reads from the underlying UDP socket and either reports a new address or proxies data to the - /// muxed connection. - pub(crate) fn poll(&mut self, cx: &mut Context) -> Poll { + pub(crate) fn poll(&mut self, cx: &mut Context<'_>) -> Poll { let mut recv_buf = [0u8; RECEIVE_MTU]; loop { - // => Send data to target - match self.send_buffer.take() { - None => { - if let Poll::Ready(Some(((buf, target), response))) = - self.send_command.poll_next_unpin(cx) - { - self.send_buffer = Some((buf, target, response)); - continue; - } - } - Some((buf, target, response)) => { - match self.udp_sock.poll_send_to(cx, &buf, target) { - Poll::Ready(result) => { - let _ = response.send(result.map_err(|e| Error::Io(e.into()))); - continue; - } - Poll::Pending => { - self.send_buffer = Some((buf, target, response)); - } - } - } - } - - // => Register a new connection - if let Poll::Ready(Some(((conn, addr), response))) = - self.registration_command.poll_next_unpin(cx) - { - let key = conn.key(); - - self.address_map - .entry(addr) - .and_modify(|e| { - if e.key() != key { - e.remove_address(&addr); - *e = conn.clone(); - } - }) - .or_insert_with(|| conn.clone()); - - // remove addr from new_addrs once conn is established - self.new_addrs.remove(&addr); - - let _ = response.send(()); + self.remove_closed_connections(); + if let Some((buf, target)) = self.send_buffer.take() { + match self.udp_sock.poll_send_to(cx, &buf, target) { + Poll::Ready(Ok(_)) => continue, + Poll::Ready(Err(err)) => return Poll::Ready(UDPMuxEvent::Error(err)), + Poll::Pending => self.send_buffer = Some((buf, target)), + } + } else if let Poll::Ready(Some((buf, target))) = self.send_command.poll_next_unpin(cx) { + self.send_buffer = Some((buf, target)); continue; } - // => Get connection with the given ufrag if let Poll::Ready(Some((ufrag, response))) = self.get_conn_command.poll_next_unpin(cx) { - if self.is_closed { - let _ = response.send(Err(Error::ErrUseClosedNetworkConn)); - continue; - } - - if let Some(conn) = self.conns.get(&ufrag).cloned() { - let _ = response.send(Ok(Arc::new(conn))); - continue; - } - - let muxed_conn = match self.create_muxed_conn(&ufrag) { - Ok(conn) => conn, - Err(e) => { - let _ = response.send(Err(e)); - continue; - } + let result = if self.conns.contains_key(&ufrag) { + Err(io::Error::new( + ErrorKind::AlreadyExists, + format!("UDP mux connection for ufrag {ufrag} already exists"), + )) + } else { + let (incoming, incoming_rx) = mpsc::channel(RECEIVE_QUEUE_CAPACITY); + let conn = Arc::new(MuxConnection { + ufrag: ufrag.clone(), + incoming: Mutex::new(incoming_rx), + send_sender: Arc::clone(&self.udp_mux_handle.send_sender), + }); + self.conns.insert( + ufrag, + Route { + incoming, + remote_addrs: HashSet::new(), + }, + ); + Ok(conn) }; - let mut close_rx = muxed_conn.close_rx(); - - self.close_futures.push({ - let ufrag = ufrag.clone(); - let udp_mux_handle = self.udp_mux_handle.clone(); - - Box::pin(async move { - let _ = close_rx.changed().await; - udp_mux_handle.remove_conn_by_ufrag(&ufrag).await; - }) - }); - - self.conns.insert(ufrag, muxed_conn.clone()); - - let _ = response.send(Ok(Arc::new(muxed_conn) as Arc)); - + let _ = response.send(result); continue; } - // => Close UDPMux - if let Poll::Ready(Some(((), response))) = self.close_command.poll_next_unpin(cx) { - if self.is_closed { - let _ = response.send(Err(Error::ErrAlreadyClosed)); - continue; - } - - for (_, conn) in self.conns.drain() { - conn.close(); - } - - // NOTE: This is important, we need to drop all instances of `UDPMuxConn` to - // avoid a retain cycle due to the use of [`std::sync::Arc`] on both sides. - self.address_map.clear(); - - // NOTE: This is important, we need to drop all instances of `UDPMuxConn` to - // avoid a retain cycle due to the use of [`std::sync::Arc`] on both sides. - self.new_addrs.clear(); - - let _ = response.send(Ok(())); + let mut read = ReadBuf::new(&mut recv_buf); + match self.udp_sock.poll_recv_from(cx, &mut read) { + Poll::Ready(Ok(remote_addr)) => { + let packet = read.filled(); + let route = self.route_for_packet(packet, remote_addr); - self.is_closed = true; - - continue; - } - - // => Remove connection with the given ufrag - if let Poll::Ready(Some((ufrag, response))) = - self.remove_conn_command.poll_next_unpin(cx) - { - // Pion's ice implementation has both `RemoveConnByFrag` and `RemoveConn`, but since - // `conns` is keyed on `ufrag` their implementation is equivalent. + if let Some(ufrag) = route { + self.new_addrs.remove(&remote_addr); + let Some(route) = self.conns.get_mut(&ufrag) else { + continue; + }; + route.remote_addrs.insert(remote_addr); + self.address_map.insert(remote_addr, ufrag.clone()); + if let Err(err) = route.incoming.try_send(Datagram { + data: packet.to_vec(), + remote_addr, + }) { + tracing::debug!( + address=%remote_addr, + %ufrag, + "Dropping UDP datagram: receive queue unavailable: {err}", + ); + } + continue; + } - if let Some(removed_conn) = self.conns.remove(&ufrag) { - for address in removed_conn.get_addresses() { - self.address_map.remove(&address); + if !self.new_addrs.contains(&remote_addr) { + match ufrag_from_stun_message(packet, false) { + Ok(ufrag) => { + self.new_addrs.insert(remote_addr); + return Poll::Ready(UDPMuxEvent::NewAddr(NewAddr { + addr: remote_addr, + ufrag, + })); + } + Err(err) => tracing::debug!( + address=%remote_addr, + "Unknown address or invalid STUN packet: {err}", + ), + } } + continue; } - - let _ = response.send(()); - - continue; + Poll::Ready(Err(err)) if err.kind() == ErrorKind::TimedOut => continue, + Poll::Ready(Err(err)) if err.kind() == ErrorKind::ConnectionReset => { + tracing::debug!("Connection reset by remote client: {err}"); + continue; + } + Poll::Ready(Err(err)) => return Poll::Ready(UDPMuxEvent::Error(err)), + Poll::Pending => return Poll::Pending, } + } + } - // => Remove closed connections - let _ = self.close_futures.poll_next_unpin(cx); + fn route_for_packet(&self, packet: &[u8], remote_addr: SocketAddr) -> Option { + if let Some(ufrag) = self.address_map.get(&remote_addr) { + return Some(ufrag.clone()); + } + if !is_stun_message(packet) { + return None; + } - // => Write previously received data to local connections - match self.write_future.poll_unpin(cx) { - Poll::Ready(Some(())) => { - self.write_future = OptionFuture::default(); - continue; - } - Poll::Ready(None) => { - // => Read from the socket - let mut read = ReadBuf::new(&mut recv_buf); - - match self.udp_sock.poll_recv_from(cx, &mut read) { - Poll::Ready(Ok(addr)) => { - // Find connection based on previously having seen this source address - let conn = self.address_map.get(&addr); - - let conn = match conn { - // If we couldn't find the connection based on source address, see - // if this is a STUN message and if - // so if we can find the connection based on ufrag. - None if is_stun_message(read.filled()) => { - match self.conn_from_stun_message(read.filled(), &addr) { - Some(Ok(s)) => Some(s), - Some(Err(e)) => { - tracing::debug!(address=%&addr, "Error when querying existing connections: {}", e); - continue; - } - None => None, - } - } - Some(s) => Some(s.to_owned()), - _ => None, - }; - - match conn { - None => { - if !self.new_addrs.contains(&addr) { - match ufrag_from_stun_message(read.filled(), false) { - Ok(ufrag) => { - tracing::trace!( - address=%&addr, - %ufrag, - "Notifying about new address from ufrag", - ); - self.new_addrs.insert(addr); - return Poll::Ready(UDPMuxEvent::NewAddr( - NewAddr { addr, ufrag }, - )); - } - Err(e) => { - tracing::debug!( - address=%&addr, - "Unknown address (non STUN packet: {})", - e - ); - } - } - } - } - Some(conn) => { - let mut packet = vec![0u8; read.filled().len()]; - packet.copy_from_slice(read.filled()); - self.write_future = OptionFuture::from(Some( - async move { - if let Err(err) = conn.write_packet(&packet, addr).await - { - tracing::error!( - address=%addr, - "Failed to write packet: {}", - err, - ); - } - } - .boxed(), - )); - } - } + let ufrag = ufrag_from_stun_message(packet, true).ok()?; + let route = self.conns.get(&ufrag)?; + if route.remote_addrs.is_empty() || route.remote_addrs.contains(&remote_addr) { + Some(ufrag) + } else { + tracing::debug!( + address=%remote_addr, + %ufrag, + "ICE ufrag is already associated with another address", + ); + None + } + } - continue; - } - Poll::Pending => {} - Poll::Ready(Err(err)) if err.kind() == ErrorKind::TimedOut => {} - Poll::Ready(Err(err)) if err.kind() == ErrorKind::ConnectionReset => { - tracing::debug!("ConnectionReset by remote client {err:?}") - } - Poll::Ready(Err(err)) => { - tracing::error!("Could not read udp packet: {}", err); - return Poll::Ready(UDPMuxEvent::Error(err)); - } - } + fn remove_closed_connections(&mut self) { + let closed = self + .conns + .iter() + .filter(|(_, route)| route.incoming.is_closed()) + .map(|(ufrag, _)| ufrag.clone()) + .collect::>(); + + for ufrag in closed { + if let Some(route) = self.conns.remove(&ufrag) { + for addr in route.remote_addrs { + self.address_map.remove(&addr); } - Poll::Pending => {} } - - return Poll::Pending; } } } -/// Handle which utilizes [`req_res_chan`] to transmit commands (e.g. remove connection) from the -/// WebRTC ICE agent to [`UDPMuxNewAddr::poll`]. +/// Handle used by connection upgrades to obtain a socket for one ICE ufrag. pub(crate) struct UdpMuxHandle { - close_sender: req_res_chan::Sender<(), Result<(), Error>>, - get_conn_sender: req_res_chan::Sender, Error>>, - remove_sender: req_res_chan::Sender, + get_conn_sender: req_res_chan::Sender>>, + send_sender: Arc, SocketAddr)>>>, } -impl UdpMuxHandle { - /// Returns a new `UdpMuxHandle` and `close`, `get_conn` and `remove` receivers. - pub(crate) fn new() -> ( - Self, - req_res_chan::Receiver<(), Result<(), Error>>, - req_res_chan::Receiver, Error>>, - req_res_chan::Receiver, - ) { - let (sender1, receiver1) = req_res_chan::new(1); - let (sender2, receiver2) = req_res_chan::new(1); - let (sender3, receiver3) = req_res_chan::new(1); - - let this = Self { - close_sender: sender1, - get_conn_sender: sender2, - remove_sender: sender3, - }; - - (this, receiver1, receiver2, receiver3) +impl fmt::Debug for UdpMuxHandle { + fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { + f.debug_struct("UdpMuxHandle").finish_non_exhaustive() } } -#[async_trait] -impl UDPMux for UdpMuxHandle { - async fn close(&self) -> Result<(), Error> { - self.close_sender - .send(()) +impl UdpMuxHandle { + pub(crate) async fn get_conn(&self, ufrag: &str) -> io::Result> { + self.get_conn_sender + .send(ufrag.to_owned()) .await - .map_err(|e| Error::Io(e.into()))??; + .map_err(|err| io::Error::new(ErrorKind::BrokenPipe, err))? + } +} - Ok(()) +/// A per-ufrag packet queue backed by the listener's shared UDP socket. +pub(crate) struct MuxConnection { + ufrag: String, + incoming: Mutex>, + send_sender: Arc, SocketAddr)>>>, +} + +impl fmt::Debug for MuxConnection { + fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { + f.debug_struct("MuxConnection") + .field("ufrag", &self.ufrag) + .finish_non_exhaustive() } +} - async fn get_conn(self: Arc, ufrag: &str) -> Result, Error> { - let conn = self - .get_conn_sender - .send(ufrag.to_owned()) - .await - .map_err(|e| Error::Io(e.into()))??; +#[derive(Debug)] +struct MuxedUdpSocket { + local_addr: SocketAddr, + conn: Arc, +} + +impl AsyncUdpSocket for MuxedUdpSocket { + fn local_addr(&self) -> io::Result { + Ok(self.local_addr) + } - Ok(conn) + fn poll_send(&self, cx: &mut Context<'_>, transmit: &Transmit<'_>) -> Poll> { + let mut sender = self + .conn + .send_sender + .lock() + .map_err(|_| io::Error::other("UDP mux send queue lock poisoned"))?; + ready!(Pin::new(&mut *sender).poll_ready(cx)) + .map_err(|_| io::Error::new(ErrorKind::BrokenPipe, "UDP mux listener closed"))?; + + let len = transmit.contents.len(); + Pin::new(&mut *sender) + .start_send((transmit.contents.to_vec(), transmit.destination)) + .map_err(|_| io::Error::new(ErrorKind::BrokenPipe, "UDP mux listener closed"))?; + Poll::Ready(Ok(len)) } - async fn remove_conn_by_ufrag(&self, ufrag: &str) { - if let Err(e) = self.remove_sender.send(ufrag.to_owned()).await { - tracing::debug!("Failed to send message through channel: {:?}", e); + fn poll_recv( + &self, + cx: &mut Context<'_>, + bufs: &mut [IoSliceMut<'_>], + meta: &mut [RecvMeta], + ) -> Poll> { + if bufs.is_empty() || meta.is_empty() { + return Poll::Ready(Ok(0)); } + + let mut incoming = self + .conn + .incoming + .lock() + .map_err(|_| io::Error::other("UDP mux receive queue lock poisoned"))?; + let Some(datagram) = ready!(Pin::new(&mut *incoming).poll_next(cx)) else { + return Poll::Ready(Err(io::Error::new( + ErrorKind::BrokenPipe, + "UDP mux listener closed", + ))); + }; + if datagram.data.len() > bufs[0].len() { + return Poll::Ready(Err(io::Error::new( + ErrorKind::InvalidData, + "UDP datagram exceeds receive buffer", + ))); + } + + let len = datagram.data.len(); + bufs[0][..len].copy_from_slice(&datagram.data); + let mut recv_meta = RecvMeta::default(); + recv_meta.addr = datagram.remote_addr; + recv_meta.len = len; + recv_meta.stride = len.max(1); + recv_meta.dst_ip = Some(self.local_addr.ip()); + meta[0] = recv_meta; + Poll::Ready(Ok(1)) } } -/// Handle which utilizes [`req_res_chan`] to transmit commands from [`UDPMuxConn`] connections to -/// [`UDPMuxNewAddr::poll`]. -pub(crate) struct UdpMuxWriterHandle { - registration_channel: req_res_chan::Sender<(UDPMuxConn, SocketAddr), ()>, - send_channel: req_res_chan::Sender<(Vec, SocketAddr), Result>, +/// Delegates runtime services to webrtc's Tokio runtime while replacing its UDP socket. +#[derive(Debug)] +pub(crate) struct MuxRuntime { + inner: Arc, + conn: Arc, } -impl UdpMuxWriterHandle { - /// Returns a new `UdpMuxWriterHandle` and `registration`, `send` receivers. - fn new() -> ( - Self, - req_res_chan::Receiver<(UDPMuxConn, SocketAddr), ()>, - req_res_chan::Receiver<(Vec, SocketAddr), Result>, - ) { - let (sender1, receiver1) = req_res_chan::new(1); - let (sender2, receiver2) = req_res_chan::new(1); - - let this = Self { - registration_channel: sender1, - send_channel: sender2, - }; - - (this, receiver1, receiver2) +impl MuxRuntime { + pub(crate) fn new(conn: Arc) -> io::Result> { + let inner = webrtc::runtime::default_runtime() + .ok_or_else(|| io::Error::other("webrtc Tokio runtime is not enabled"))?; + Ok(Arc::new(Self { inner, conn })) } } -#[async_trait] -impl UDPMuxWriter for UdpMuxWriterHandle { - async fn register_conn_for_address(&self, conn: &UDPMuxConn, addr: SocketAddr) { - match self - .registration_channel - .send((conn.to_owned(), addr)) - .await - { - Ok(()) => {} - Err(e) => { - tracing::debug!("Failed to send message through channel: {:?}", e); - return; - } - } +impl Runtime for MuxRuntime { + fn spawn(&self, future: Pin + Send>>) -> Box { + self.inner.spawn(future) + } - tracing::debug!(address=%addr, connection=%conn.key(), "Registered address for connection"); + fn spawn_reactor( + &self, + reactor_pool_size: usize, + future: Pin + Send>>, + ) -> Box { + self.inner.spawn_reactor(reactor_pool_size, future) } - async fn send_to(&self, buf: &[u8], target: &SocketAddr) -> Result { - let bytes_written = self - .send_channel - .send((buf.to_owned(), target.to_owned())) - .await - .map_err(|e| Error::Io(e.into()))??; + fn wrap_udp_socket(&self, socket: std::net::UdpSocket) -> io::Result> { + let local_addr = socket.local_addr()?; + drop(socket); + Ok(Arc::new(MuxedUdpSocket { + local_addr, + conn: Arc::clone(&self.conn), + })) + } - Ok(bytes_written) + fn wrap_tcp_listener( + &self, + listener: std::net::TcpListener, + ) -> io::Result> { + self.inner.wrap_tcp_listener(listener) } -} -/// Gets the ufrag from the given STUN message or returns an error, if failed to decode or the -/// username attribute is not present. -fn ufrag_from_stun_message(buffer: &[u8], local_ufrag: bool) -> Result { - let (result, message) = { - let mut m = STUNMessage::new(); - - (m.unmarshal_binary(buffer), m) - }; - - if let Err(err) = result { - Err(Error::Other(format!("failed to handle decode ICE: {err}"))) - } else { - let (attr, found) = message.attributes.get(ATTR_USERNAME); - if !found { - return Err(Error::Other("no username attribute in STUN message".into())); - } + fn connect_tcp<'a>( + &'a self, + remote_addr: SocketAddr, + ) -> Pin>> + Send + 'a>> { + self.inner.connect_tcp(remote_addr) + } - match String::from_utf8(attr.value) { - // Per the RFC this shouldn't happen - // https://datatracker.ietf.org/doc/html/rfc5389#section-15.3 - Err(err) => Err(Error::Other(format!( - "failed to decode USERNAME from STUN message as UTF-8: {err}" - ))), - Ok(s) => { - // s is a combination of the local_ufrag and the remote ufrag separated by `:`. - let res = if local_ufrag { - s.split(':').next() - } else { - s.split(':').next_back() - }; - match res { - Some(s) => Ok(s.to_owned()), - None => Err(Error::Other("can't get ufrag from username".into())), - } - } - } + fn resolve_host<'a>( + &'a self, + host: &'a str, + ) -> Pin>> + Send + 'a>> { + self.inner.resolve_host(host) + } + + fn sleep(&self, duration: Duration) -> Pin + Send + 'static>> { + self.inner.sleep(duration) + } + + fn interval(&self, period: Duration) -> Box { + self.inner.interval(period) + } + + fn block_on(&self, future: Pin + '_>>) { + self.inner.block_on(future) + } + + fn yield_now(&self) -> Pin + Send + 'static>> { + self.inner.yield_now() + } + + fn name(&self) -> &'static str { + "tokio-udp-mux" } } -#[derive(Error, Debug)] -enum ConnQueryError { - #[error("ufrag is already taken (associated_addrs={associated_addrs:?})")] - UfragAlreadyTaken { associated_addrs: Vec }, +/// Gets one half of the `local:remote` ICE username from a STUN message. +fn ufrag_from_stun_message(buffer: &[u8], local_ufrag: bool) -> io::Result { + let mut message = STUNMessage::new(); + message + .unmarshal_binary(buffer) + .map_err(|err| io::Error::new(ErrorKind::InvalidData, err))?; + + let (attr, found) = message.attributes.get(ATTR_USERNAME); + if !found { + return Err(io::Error::new( + ErrorKind::InvalidData, + "no username attribute in STUN message", + )); + } + + let username = + String::from_utf8(attr.value).map_err(|err| io::Error::new(ErrorKind::InvalidData, err))?; + let (first, second) = username.split_once(':').ok_or_else(|| { + io::Error::new( + ErrorKind::InvalidData, + "ICE username does not contain two ufrags", + ) + })?; + + Ok(if local_ufrag { first } else { second }.to_owned()) } diff --git a/transports/webrtc/src/tokio/upgrade.rs b/transports/webrtc/src/tokio/upgrade.rs index 5333dda5c94..cd491b8a904 100644 --- a/transports/webrtc/src/tokio/upgrade.rs +++ b/transports/webrtc/src/tokio/upgrade.rs @@ -19,44 +19,54 @@ // DEALINGS IN THE SOFTWARE. use std::{ - net::SocketAddr, - sync::{ - Arc, - atomic::{AtomicBool, Ordering}, - }, - time::Duration, + net::{IpAddr, Ipv4Addr, Ipv6Addr, SocketAddr}, + sync::Arc, + time::{Duration, Instant}, }; -use futures::{channel::oneshot, future::Either}; +use futures::{channel::mpsc, future::Either}; use futures_timer::Delay; use libp2p_identity as identity; use libp2p_identity::PeerId; use libp2p_webrtc_utils::{Fingerprint, noise}; +use rtc::{ + ice::{mdns::MulticastDnsMode, network_type::NetworkType}, + peer_connection::transport::{RTCDtlsFingerprint, RTCDtlsRole}, +}; use webrtc::{ - api::{APIBuilder, setting_engine::SettingEngine}, - data::data_channel::DataChannel, - data_channel::data_channel_init::RTCDataChannelInit, - dtls_transport::dtls_role::DTLSRole, - ice::{network_type::NetworkType, udp_mux::UDPMux, udp_network::UDPNetwork}, - peer_connection::{RTCPeerConnection, configuration::RTCConfiguration}, + data_channel::{DataChannel, RTCDataChannelInit}, + peer_connection::{ + PeerConnection, PeerConnectionBuilder, PeerConnectionEventHandler, RTCConfiguration, + RTCStatsReportEntry, SettingEngine, StatsSelector, + }, + runtime::Runtime, }; -use crate::tokio::{Connection, error::Error, sdp, sdp::random_ufrag, stream::Stream}; +use crate::tokio::{ + Connection, + connection::{ConnectionHandler, await_data_channel_open}, + error::Error, + sdp, + sdp::random_ufrag, + stream::Stream, + udp_mux::{MuxRuntime, UdpMuxHandle}, +}; -/// Creates a new outbound WebRTC connection. #[allow(clippy::result_large_err)] pub(crate) async fn outbound( addr: SocketAddr, config: RTCConfiguration, - udp_mux: Arc, + udp_mux: Arc, client_fingerprint: Fingerprint, server_fingerprint: Fingerprint, id_keys: identity::Keypair, ) -> Result<(PeerId, Connection), Error> { tracing::debug!(address=%addr, "new outbound connection to address"); - let (peer_connection, ufrag) = new_outbound_connection(addr, config, udp_mux).await?; - let noise_channel_open_rx = create_noise_data_channel(&peer_connection).await?; + let ufrag = random_ufrag(); + let (peer_connection, incoming_rx) = + new_peer_connection(addr, config, udp_mux, &ufrag, false).await?; + let noise_channel = create_noise_data_channel(&peer_connection).await?; let offer = peer_connection.create_offer(None).await?; tracing::debug!(offer=%offer.sdp, "created SDP offer for outbound connection"); @@ -64,34 +74,36 @@ pub(crate) async fn outbound( let answer = sdp::answer(addr, server_fingerprint, &ufrag); tracing::debug!(?answer, "calculated SDP answer for outbound connection"); - peer_connection.set_remote_description(answer).await?; // This will start the gathering of ICE candidates. + peer_connection.set_remote_description(answer).await?; - let data_channel = await_noise_data_channel_open(noise_channel_open_rx).await?; + let data_channel = await_noise_data_channel_open(noise_channel).await?; + let (noise_stream, drop_listener) = Stream::new(data_channel); + drop(drop_listener); let peer_id = noise::outbound( id_keys, - data_channel, + noise_stream, server_fingerprint, client_fingerprint, ) .await?; - Ok((peer_id, Connection::new(peer_connection).await)) + Ok((peer_id, Connection::new(peer_connection, incoming_rx))) } -/// Creates a new inbound WebRTC connection. #[allow(clippy::result_large_err)] pub(crate) async fn inbound( addr: SocketAddr, config: RTCConfiguration, - udp_mux: Arc, + udp_mux: Arc, server_fingerprint: Fingerprint, remote_ufrag: String, id_keys: identity::Keypair, ) -> Result<(PeerId, Connection), Error> { tracing::debug!(address=%addr, ufrag=%remote_ufrag, "new inbound connection from address"); - let peer_connection = new_inbound_connection(addr, config, udp_mux, &remote_ufrag).await?; - let noise_channel_open_rx = create_noise_data_channel(&peer_connection).await?; + let (peer_connection, incoming_rx) = + new_peer_connection(addr, config, udp_mux, &remote_ufrag, true).await?; + let noise_channel = create_noise_data_channel(&peer_connection).await?; let offer = sdp::offer(addr, &remote_ufrag); tracing::debug!(?offer, "calculated SDP offer for inbound connection"); @@ -99,157 +111,143 @@ pub(crate) async fn inbound( let answer = peer_connection.create_answer(None).await?; tracing::debug!(?answer, "created SDP answer for inbound connection"); - peer_connection.set_local_description(answer).await?; // This will start the gathering of ICE candidates. + peer_connection.set_local_description(answer).await?; - let data_channel = await_noise_data_channel_open(noise_channel_open_rx).await?; - let client_fingerprint = get_remote_fingerprint(&peer_connection).await; + let data_channel = await_noise_data_channel_open(noise_channel).await?; + let client_fingerprint = remote_fingerprint(peer_connection.as_ref()).await?; + let (noise_stream, drop_listener) = Stream::new(data_channel); + drop(drop_listener); let peer_id = noise::inbound( id_keys, - data_channel, + noise_stream, client_fingerprint, server_fingerprint, ) .await?; - Ok((peer_id, Connection::new(peer_connection).await)) + Ok((peer_id, Connection::new(peer_connection, incoming_rx))) } #[allow(clippy::result_large_err)] -async fn new_outbound_connection( +async fn new_peer_connection( addr: SocketAddr, config: RTCConfiguration, - udp_mux: Arc, -) -> Result<(RTCPeerConnection, String), Error> { - let ufrag = random_ufrag(); - let se = setting_engine(udp_mux, &ufrag, addr); - - let connection = APIBuilder::new() - .with_setting_engine(se) - .build() - .new_peer_connection(config) - .await?; - - Ok((connection, ufrag)) -} - -#[allow(clippy::result_large_err)] -async fn new_inbound_connection( - addr: SocketAddr, - config: RTCConfiguration, - udp_mux: Arc, + udp_mux: Arc, ufrag: &str, -) -> Result { - let mut se = setting_engine(udp_mux, ufrag, addr); - { - se.set_lite(true); - se.disable_certificate_fingerprint_verification(true); - // Act as a DTLS server (one which waits for a connection). - // - // NOTE: removing this seems to break DTLS setup (both sides send `ClientHello` messages, - // but none end up responding). - se.set_answering_dtls_role(DTLSRole::Server)?; + inbound: bool, +) -> Result< + ( + Arc, + mpsc::Receiver>, + ), + Error, +> { + let conn = udp_mux.get_conn(ufrag).await?; + let runtime: Arc = MuxRuntime::new(conn)?; + let (handler, incoming_rx) = ConnectionHandler::new(Arc::clone(&runtime)); + let handler: Arc = handler; + + let mut setting_engine = setting_engine(ufrag, addr); + if inbound { + setting_engine.set_lite(true); + setting_engine.disable_certificate_fingerprint_verification(true); + setting_engine.set_answering_dtls_role(RTCDtlsRole::Server)?; } - let connection = APIBuilder::new() - .with_setting_engine(se) + // The runtime discards this temporary socket and substitutes the per-ufrag mux socket. + let bind_addr = match addr { + SocketAddr::V4(_) => SocketAddr::new(IpAddr::V4(Ipv4Addr::UNSPECIFIED), 0), + SocketAddr::V6(_) => SocketAddr::new(IpAddr::V6(Ipv6Addr::UNSPECIFIED), 0), + }; + let peer_connection = PeerConnectionBuilder::::new() + .with_configuration(config) + .with_setting_engine(setting_engine) + .with_runtime(runtime) + .with_handler(handler) + .with_udp_addrs(vec![bind_addr]) .build() - .new_peer_connection(config) .await?; - Ok(connection) + Ok((Arc::new(peer_connection), incoming_rx)) } -fn setting_engine( - udp_mux: Arc, - ufrag: &str, - addr: SocketAddr, -) -> SettingEngine { - let mut se = SettingEngine::default(); - - // Set both ICE user and password to our fingerprint because that's what the client is - // expecting.. - se.set_ice_credentials(ufrag.to_owned(), ufrag.to_owned()); - - se.set_udp_network(UDPNetwork::Muxed(udp_mux.clone())); - - // Allow detaching data channels. - se.detach_data_channels(); - - // Set the desired network type. - // - // NOTE: if not set, a [`webrtc_ice::agent::Agent`] might pick a wrong local candidate - // (e.g. IPv6 `[::1]` while dialing an IPv4 `10.11.12.13`). - let network_type = match addr { +fn setting_engine(ufrag: &str, addr: SocketAddr) -> SettingEngine { + let mut setting_engine = SettingEngine::default(); + setting_engine.set_ice_credentials(ufrag.to_owned(), ufrag.to_owned()); + setting_engine.set_multicast_dns_mode(MulticastDnsMode::Disabled); + setting_engine.set_network_types(vec![match addr { SocketAddr::V4(_) => NetworkType::Udp4, SocketAddr::V6(_) => NetworkType::Udp6, - }; - se.set_network_types(vec![network_type]); - - // Select only the first address of the local candidates. - // See https://github.com/libp2p/rust-libp2p/pull/5448#discussion_r2017418520. - // TODO: remove when https://github.com/webrtc-rs/webrtc/issues/662 gets addressed. - se.set_ip_filter(Box::new({ - let once = AtomicBool::new(true); - move |_ip| { - if once.load(Ordering::Relaxed) { - once.store(false, Ordering::Relaxed); - return true; - } - false - } - })); - - se -} - -/// Returns the SHA-256 fingerprint of the remote. -async fn get_remote_fingerprint(conn: &RTCPeerConnection) -> Fingerprint { - let cert_bytes = conn.sctp().transport().get_remote_certificate().await; - - Fingerprint::from_certificate(&cert_bytes) + }]); + setting_engine } #[allow(clippy::result_large_err)] async fn create_noise_data_channel( - conn: &RTCPeerConnection, -) -> Result>, Error> { - // NOTE: the data channel w/ `negotiated` flag set to `true` MUST be created on both ends. - let data_channel = conn + connection: &Arc, +) -> Result, Error> { + Ok(connection .create_data_channel( "", Some(RTCDataChannelInit { - negotiated: Some(0), // 0 is reserved for the Noise substream + negotiated: Some(0), ..RTCDataChannelInit::default() }), ) - .await?; - - let (tx, rx) = oneshot::channel::>(); - - // Wait until the data channel is opened and detach it. - crate::tokio::connection::register_data_channel_open_handler(data_channel, tx).await; - - Ok(rx) + .await?) } #[allow(clippy::result_large_err)] async fn await_noise_data_channel_open( - rx: oneshot::Receiver>, -) -> Result { - let channel = match futures::future::select(rx, Delay::new(Duration::from_secs(10))).await { - Either::Left((Ok(channel), _)) => channel, - Either::Left((Err(_), _)) => { - return Err(Error::Internal("failed to open data channel".to_owned())); - } - Either::Right(((), _)) => { - return Err(Error::Internal( - "data channel opening took longer than 10 seconds (see logs)".into(), - )); - } - }; - - let (substream, drop_listener) = Stream::new(channel); - drop(drop_listener); // Don't care about cancelled substreams during initial handshake. + data_channel: Arc, +) -> Result, Error> { + match futures::future::select( + Box::pin(await_data_channel_open(data_channel)), + Delay::new(Duration::from_secs(10)), + ) + .await + { + Either::Left((result, _)) => result, + Either::Right(((), _)) => Err(Error::Internal( + "data channel opening took longer than 10 seconds (see logs)".into(), + )), + } +} - Ok(substream) +#[allow(clippy::result_large_err)] +async fn remote_fingerprint(connection: &dyn PeerConnection) -> Result { + let report = connection + .get_stats(Instant::now(), StatsSelector::None) + .await; + let certificate_id = report + .iter() + .find_map(|entry| match entry { + RTCStatsReportEntry::Transport(transport) + if !transport.remote_certificate_id.is_empty() => + { + Some(transport.remote_certificate_id.clone()) + } + _ => None, + }) + .ok_or_else(|| Error::Internal("remote certificate is missing from WebRTC stats".into()))?; + let fingerprint = report + .iter() + .find_map(|entry| match entry { + RTCStatsReportEntry::Certificate(certificate) + if certificate.stats.id == certificate_id => + { + Some(certificate.fingerprint.clone()) + } + _ => None, + }) + .ok_or_else(|| { + Error::Internal("remote certificate fingerprint is missing from WebRTC stats".into()) + })?; + + crate::tokio::Fingerprint::try_from_rtc_dtls(&RTCDtlsFingerprint { + algorithm: "sha-256".into(), + value: fingerprint, + }) + .map(|fingerprint| fingerprint.into_inner()) + .ok_or_else(|| Error::Internal("invalid remote SHA-256 certificate fingerprint".into())) }