Port native reference implementation to WASM compatible state
This commit is contained in:
mfahampshire
Generated
+521
-3
@@ -536,6 +536,19 @@ dependencies = [
|
||||
"tungstenite 0.21.0",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "asynchronous-codec"
|
||||
version = "0.7.0"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "a860072022177f903e59730004fb5dc13db9275b79bb2aef7ba8ce831956c233"
|
||||
dependencies = [
|
||||
"bytes",
|
||||
"futures-sink",
|
||||
"futures-util",
|
||||
"memchr",
|
||||
"pin-project-lite",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "atoi"
|
||||
version = "2.0.0"
|
||||
@@ -703,12 +716,28 @@ dependencies = [
|
||||
"url",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "base-x"
|
||||
version = "0.2.11"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "4cbbc9d0964165b47557570cce6c952866c2678457aca742aafc9fb771d30270"
|
||||
|
||||
[[package]]
|
||||
name = "base16ct"
|
||||
version = "0.2.0"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "4c7f02d4ea65f2c1853089ffd8d2787bdbc63de2f0d29dedbcf8ccdfa0ccd4cf"
|
||||
|
||||
[[package]]
|
||||
name = "base256emoji"
|
||||
version = "1.0.2"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "b5e9430d9a245a77c92176e649af6e275f20839a48389859d1661e9a128d077c"
|
||||
dependencies = [
|
||||
"const-str 0.4.3",
|
||||
"match-lookup",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "base64"
|
||||
version = "0.13.1"
|
||||
@@ -1391,6 +1420,12 @@ version = "0.9.6"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "c2459377285ad874054d797f3ccebf984978aa39129f6eafde5cdc8315b612f8"
|
||||
|
||||
[[package]]
|
||||
name = "const-str"
|
||||
version = "0.4.3"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "2f421161cb492475f1661ddc9815a745a1c894592070661180fdec3d4872e9c3"
|
||||
|
||||
[[package]]
|
||||
name = "const-str"
|
||||
version = "0.5.7"
|
||||
@@ -1459,6 +1494,15 @@ dependencies = [
|
||||
"rand 0.9.2",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "core2"
|
||||
version = "0.4.0"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "b49ba7ef1ad6107f8824dbe97de947cbaac53c44e7f9756a1fba0d37c1eec505"
|
||||
dependencies = [
|
||||
"memchr",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "cosmos-sdk-proto"
|
||||
version = "0.27.0"
|
||||
@@ -2134,6 +2178,26 @@ version = "2.9.0"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "2a2330da5de22e8a3cb63252ce2abb30116bf5265e89c0e01bc17015ce30a476"
|
||||
|
||||
[[package]]
|
||||
name = "data-encoding-macro"
|
||||
version = "0.1.18"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "47ce6c96ea0102f01122a185683611bd5ac8d99e62bc59dd12e6bda344ee673d"
|
||||
dependencies = [
|
||||
"data-encoding",
|
||||
"data-encoding-macro-internal",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "data-encoding-macro-internal"
|
||||
version = "0.1.16"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "8d162beedaa69905488a8da94f5ac3edb4dd4788b732fadb7bd120b2625c1976"
|
||||
dependencies = [
|
||||
"data-encoding",
|
||||
"syn 2.0.114",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "defguard_boringtun"
|
||||
version = "0.6.3"
|
||||
@@ -2905,6 +2969,16 @@ dependencies = [
|
||||
"futures-util",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "futures-bounded"
|
||||
version = "0.2.4"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "91f328e7fb845fc832912fb6a34f40cf6d1888c92f974d1893a54e97b5ff542e"
|
||||
dependencies = [
|
||||
"futures-timer",
|
||||
"futures-util",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "futures-channel"
|
||||
version = "0.3.31"
|
||||
@@ -2930,6 +3004,7 @@ dependencies = [
|
||||
"futures-core",
|
||||
"futures-task",
|
||||
"futures-util",
|
||||
"num_cpus",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
@@ -2987,6 +3062,27 @@ version = "0.3.31"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "f90f7dce0722e95104fcb095585910c0977252f286e354b5e3bd38902cd99988"
|
||||
|
||||
[[package]]
|
||||
name = "futures-ticker"
|
||||
version = "0.0.3"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "9763058047f713632a52e916cc7f6a4b3fc6e9fc1ff8c5b1dc49e5a89041682e"
|
||||
dependencies = [
|
||||
"futures",
|
||||
"futures-timer",
|
||||
"instant",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "futures-timer"
|
||||
version = "3.0.3"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "f288b0a4f20f9a56b5d1da57e2227c661b7b16168e2f72365f57b63326e29b24"
|
||||
dependencies = [
|
||||
"gloo-timers 0.2.6",
|
||||
"send_wrapper 0.4.0",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "futures-util"
|
||||
version = "0.3.31"
|
||||
@@ -3096,6 +3192,18 @@ dependencies = [
|
||||
"web-sys",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "gloo-timers"
|
||||
version = "0.2.6"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "9b995a66bb87bebce9a0f4a95aed01daca4872c050bfcb21653361c03bc35e5c"
|
||||
dependencies = [
|
||||
"futures-channel",
|
||||
"futures-core",
|
||||
"js-sys",
|
||||
"wasm-bindgen",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "gloo-timers"
|
||||
version = "0.3.0"
|
||||
@@ -3383,6 +3491,12 @@ version = "0.3.4"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "7ebdb29d2ea9ed0083cd8cece49bbd968021bd99b0849edb4a9a7ee0fdf6a4e0"
|
||||
|
||||
[[package]]
|
||||
name = "hex_fmt"
|
||||
version = "0.3.0"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "b07f60793ff0a4d9cef0f18e63b5357e06209987153a64648c972c1e5aff336f"
|
||||
|
||||
[[package]]
|
||||
name = "hickory-proto"
|
||||
version = "0.25.2"
|
||||
@@ -4679,6 +4793,215 @@ version = "0.2.15"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "f9fbbcab51052fe104eb5e5d351cf728d30a5be1fe14d9be8a3b097481fb97de"
|
||||
|
||||
[[package]]
|
||||
name = "libp2p"
|
||||
version = "0.54.1"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "bbbe80f9c7e00526cd6b838075b9c171919404a4732cb2fa8ece0a093223bfc4"
|
||||
dependencies = [
|
||||
"bytes",
|
||||
"either",
|
||||
"futures",
|
||||
"futures-timer",
|
||||
"getrandom 0.2.16",
|
||||
"libp2p-allow-block-list",
|
||||
"libp2p-connection-limits",
|
||||
"libp2p-core",
|
||||
"libp2p-gossipsub",
|
||||
"libp2p-identify",
|
||||
"libp2p-identity",
|
||||
"libp2p-metrics",
|
||||
"libp2p-ping",
|
||||
"libp2p-swarm",
|
||||
"multiaddr",
|
||||
"pin-project",
|
||||
"rw-stream-sink",
|
||||
"thiserror 1.0.69",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "libp2p-allow-block-list"
|
||||
version = "0.4.0"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "d1027ccf8d70320ed77e984f273bc8ce952f623762cb9bf2d126df73caef8041"
|
||||
dependencies = [
|
||||
"libp2p-core",
|
||||
"libp2p-identity",
|
||||
"libp2p-swarm",
|
||||
"void",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "libp2p-connection-limits"
|
||||
version = "0.4.0"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "8d003540ee8baef0d254f7b6bfd79bac3ddf774662ca0abf69186d517ef82ad8"
|
||||
dependencies = [
|
||||
"libp2p-core",
|
||||
"libp2p-identity",
|
||||
"libp2p-swarm",
|
||||
"void",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "libp2p-core"
|
||||
version = "0.42.0"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "a61f26c83ed111104cd820fe9bc3aaabbac5f1652a1d213ed6e900b7918a1298"
|
||||
dependencies = [
|
||||
"either",
|
||||
"fnv",
|
||||
"futures",
|
||||
"futures-timer",
|
||||
"libp2p-identity",
|
||||
"multiaddr",
|
||||
"multihash",
|
||||
"multistream-select",
|
||||
"once_cell",
|
||||
"parking_lot",
|
||||
"pin-project",
|
||||
"quick-protobuf",
|
||||
"rand 0.8.5",
|
||||
"rw-stream-sink",
|
||||
"smallvec",
|
||||
"thiserror 1.0.69",
|
||||
"tracing",
|
||||
"unsigned-varint 0.8.0",
|
||||
"void",
|
||||
"web-time",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "libp2p-gossipsub"
|
||||
version = "0.47.0"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "b4e830fdf24ac8c444c12415903174d506e1e077fbe3875c404a78c5935a8543"
|
||||
dependencies = [
|
||||
"asynchronous-codec",
|
||||
"base64 0.22.1",
|
||||
"byteorder",
|
||||
"bytes",
|
||||
"either",
|
||||
"fnv",
|
||||
"futures",
|
||||
"futures-ticker",
|
||||
"getrandom 0.2.16",
|
||||
"hex_fmt",
|
||||
"libp2p-core",
|
||||
"libp2p-identity",
|
||||
"libp2p-swarm",
|
||||
"prometheus-client",
|
||||
"quick-protobuf",
|
||||
"quick-protobuf-codec",
|
||||
"rand 0.8.5",
|
||||
"regex",
|
||||
"sha2 0.10.9",
|
||||
"smallvec",
|
||||
"tracing",
|
||||
"void",
|
||||
"web-time",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "libp2p-identify"
|
||||
version = "0.45.0"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "1711b004a273be4f30202778856368683bd9a83c4c7dcc8f848847606831a4e3"
|
||||
dependencies = [
|
||||
"asynchronous-codec",
|
||||
"either",
|
||||
"futures",
|
||||
"futures-bounded",
|
||||
"futures-timer",
|
||||
"libp2p-core",
|
||||
"libp2p-identity",
|
||||
"libp2p-swarm",
|
||||
"lru",
|
||||
"quick-protobuf",
|
||||
"quick-protobuf-codec",
|
||||
"smallvec",
|
||||
"thiserror 1.0.69",
|
||||
"tracing",
|
||||
"void",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "libp2p-identity"
|
||||
version = "0.2.13"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "f0c7892c221730ba55f7196e98b0b8ba5e04b4155651736036628e9f73ed6fc3"
|
||||
dependencies = [
|
||||
"bs58",
|
||||
"ed25519-dalek",
|
||||
"hkdf",
|
||||
"multihash",
|
||||
"quick-protobuf",
|
||||
"rand 0.8.5",
|
||||
"sha2 0.10.9",
|
||||
"thiserror 2.0.17",
|
||||
"tracing",
|
||||
"zeroize",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "libp2p-metrics"
|
||||
version = "0.15.0"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "77ebafa94a717c8442d8db8d3ae5d1c6a15e30f2d347e0cd31d057ca72e42566"
|
||||
dependencies = [
|
||||
"futures",
|
||||
"libp2p-core",
|
||||
"libp2p-identify",
|
||||
"libp2p-identity",
|
||||
"libp2p-ping",
|
||||
"libp2p-swarm",
|
||||
"pin-project",
|
||||
"prometheus-client",
|
||||
"web-time",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "libp2p-ping"
|
||||
version = "0.45.0"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "005a34420359223b974ee344457095f027e51346e992d1e0dcd35173f4cdd422"
|
||||
dependencies = [
|
||||
"either",
|
||||
"futures",
|
||||
"futures-timer",
|
||||
"libp2p-core",
|
||||
"libp2p-identity",
|
||||
"libp2p-swarm",
|
||||
"rand 0.8.5",
|
||||
"tracing",
|
||||
"void",
|
||||
"web-time",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "libp2p-swarm"
|
||||
version = "0.45.1"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "d7dd6741793d2c1fb2088f67f82cf07261f25272ebe3c0b0c311e0c6b50e851a"
|
||||
dependencies = [
|
||||
"either",
|
||||
"fnv",
|
||||
"futures",
|
||||
"futures-timer",
|
||||
"getrandom 0.2.16",
|
||||
"libp2p-core",
|
||||
"libp2p-identity",
|
||||
"lru",
|
||||
"multistream-select",
|
||||
"once_cell",
|
||||
"rand 0.8.5",
|
||||
"smallvec",
|
||||
"tracing",
|
||||
"void",
|
||||
"wasm-bindgen-futures",
|
||||
"web-time",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "libredox"
|
||||
version = "0.1.12"
|
||||
@@ -4758,6 +5081,15 @@ version = "0.4.29"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "5e5032e24019045c762d3c0f28f5b6b8bbf38563a65908389bf7978758920897"
|
||||
|
||||
[[package]]
|
||||
name = "lru"
|
||||
version = "0.12.5"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "234cf4f4a04dc1f57e24b96cc0cd600cf2af460d4161ac5ecdd0af8e1f3b2a38"
|
||||
dependencies = [
|
||||
"hashbrown 0.15.5",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "lru-slab"
|
||||
version = "0.1.2"
|
||||
@@ -4835,6 +5167,17 @@ dependencies = [
|
||||
"web_atoms",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "match-lookup"
|
||||
version = "0.1.2"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "757aee279b8bdbb9f9e676796fd459e4207a1f986e87886700abf589f5abf771"
|
||||
dependencies = [
|
||||
"proc-macro2",
|
||||
"quote",
|
||||
"syn 2.0.114",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "match_token"
|
||||
version = "0.35.0"
|
||||
@@ -5035,6 +5378,60 @@ dependencies = [
|
||||
"version_check",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "multiaddr"
|
||||
version = "0.18.2"
|
||||
source = "git+https://github.com/mfahampshire/rust-multiaddr?branch=nym-protocol#b9966d6eedd09236ee76f157f985d04e02f8b880"
|
||||
dependencies = [
|
||||
"arrayref",
|
||||
"byteorder",
|
||||
"data-encoding",
|
||||
"libp2p-identity",
|
||||
"multibase",
|
||||
"multihash",
|
||||
"percent-encoding",
|
||||
"serde",
|
||||
"static_assertions",
|
||||
"unsigned-varint 0.8.0",
|
||||
"url",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "multibase"
|
||||
version = "0.9.2"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "8694bb4835f452b0e3bb06dbebb1d6fc5385b6ca1caf2e55fd165c042390ec77"
|
||||
dependencies = [
|
||||
"base-x",
|
||||
"base256emoji",
|
||||
"data-encoding",
|
||||
"data-encoding-macro",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "multihash"
|
||||
version = "0.19.3"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "6b430e7953c29dd6a09afc29ff0bb69c6e306329ee6794700aee27b76a1aea8d"
|
||||
dependencies = [
|
||||
"core2",
|
||||
"unsigned-varint 0.8.0",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "multistream-select"
|
||||
version = "0.13.0"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "ea0df8e5eec2298a62b326ee4f0d7fe1a6b90a09dfcf9df37b38f947a8c42f19"
|
||||
dependencies = [
|
||||
"bytes",
|
||||
"futures",
|
||||
"log",
|
||||
"pin-project",
|
||||
"smallvec",
|
||||
"unsigned-varint 0.7.2",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "netlink-packet-core"
|
||||
version = "0.8.1"
|
||||
@@ -5522,7 +5919,7 @@ dependencies = [
|
||||
"clap",
|
||||
"clap_complete",
|
||||
"clap_complete_fig",
|
||||
"const-str",
|
||||
"const-str 0.5.7",
|
||||
"log",
|
||||
"opentelemetry",
|
||||
"opentelemetry-jaeger",
|
||||
@@ -5693,7 +6090,7 @@ dependencies = [
|
||||
"clap",
|
||||
"comfy-table",
|
||||
"futures",
|
||||
"gloo-timers",
|
||||
"gloo-timers 0.3.0",
|
||||
"http-body-util",
|
||||
"humantime",
|
||||
"hyper 1.8.1",
|
||||
@@ -5802,7 +6199,7 @@ version = "1.4.1"
|
||||
dependencies = [
|
||||
"anyhow",
|
||||
"futures",
|
||||
"gloo-timers",
|
||||
"gloo-timers 0.3.0",
|
||||
"js-sys",
|
||||
"nym-bin-common",
|
||||
"nym-gateway-requests",
|
||||
@@ -6837,6 +7234,38 @@ dependencies = [
|
||||
"thiserror 2.0.17",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "nym-libp2p-wasm"
|
||||
version = "0.1.0"
|
||||
dependencies = [
|
||||
"futures",
|
||||
"getrandom 0.2.16",
|
||||
"gloo-timers 0.3.0",
|
||||
"hex",
|
||||
"js-sys",
|
||||
"libp2p",
|
||||
"libp2p-identify",
|
||||
"libp2p-identity",
|
||||
"libp2p-ping",
|
||||
"libp2p-swarm",
|
||||
"log",
|
||||
"nym-client-wasm",
|
||||
"nym-sphinx-addressing",
|
||||
"nym-sphinx-anonymous-replies",
|
||||
"nym-wasm-client-core",
|
||||
"nym-wasm-utils",
|
||||
"parking_lot",
|
||||
"send_wrapper 0.6.0",
|
||||
"serde",
|
||||
"serde-wasm-bindgen 0.6.5",
|
||||
"thiserror 2.0.17",
|
||||
"tokio_with_wasm",
|
||||
"wasm-bindgen",
|
||||
"wasm-bindgen-futures",
|
||||
"wasm-bindgen-test",
|
||||
"web-sys",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "nym-lp"
|
||||
version = "0.1.0"
|
||||
@@ -9382,6 +9811,29 @@ dependencies = [
|
||||
"thiserror 2.0.17",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "prometheus-client"
|
||||
version = "0.22.3"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "504ee9ff529add891127c4827eb481bd69dc0ebc72e9a682e187db4caa60c3ca"
|
||||
dependencies = [
|
||||
"dtoa",
|
||||
"itoa",
|
||||
"parking_lot",
|
||||
"prometheus-client-derive-encode",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "prometheus-client-derive-encode"
|
||||
version = "0.4.2"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "440f724eba9f6996b75d63681b0a92b06947f1457076d503a4d2e2c8f56442b8"
|
||||
dependencies = [
|
||||
"proc-macro2",
|
||||
"quote",
|
||||
"syn 2.0.114",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "prost"
|
||||
version = "0.13.5"
|
||||
@@ -9465,6 +9917,28 @@ version = "2.0.1"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "a993555f31e5a609f617c12db6250dedcac1b0a85076912c436e6fc9b2c8e6a3"
|
||||
|
||||
[[package]]
|
||||
name = "quick-protobuf"
|
||||
version = "0.8.1"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "9d6da84cc204722a989e01ba2f6e1e276e190f22263d0cb6ce8526fcdb0d2e1f"
|
||||
dependencies = [
|
||||
"byteorder",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "quick-protobuf-codec"
|
||||
version = "0.3.1"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "15a0580ab32b169745d7a39db2ba969226ca16738931be152a3209b409de2474"
|
||||
dependencies = [
|
||||
"asynchronous-codec",
|
||||
"bytes",
|
||||
"quick-protobuf",
|
||||
"thiserror 1.0.69",
|
||||
"unsigned-varint 0.8.0",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "quinn"
|
||||
version = "0.11.9"
|
||||
@@ -10106,6 +10580,17 @@ version = "1.0.22"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "b39cdef0fa800fc44525c84ccb54a029961a8215f9619753635a9c0d2538d46d"
|
||||
|
||||
[[package]]
|
||||
name = "rw-stream-sink"
|
||||
version = "0.4.0"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "d8c9026ff5d2f23da5e45bbc283f156383001bfb09c4e44256d02c1a685fe9a1"
|
||||
dependencies = [
|
||||
"futures",
|
||||
"pin-project",
|
||||
"static_assertions",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "ryu"
|
||||
version = "1.0.22"
|
||||
@@ -10308,6 +10793,21 @@ version = "0.7.0"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "388a1df253eca08550bef6c72392cfe7c30914bf41df5269b68cbd6ff8f570a3"
|
||||
|
||||
[[package]]
|
||||
name = "send_wrapper"
|
||||
version = "0.4.0"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "f638d531eccd6e23b980caf34876660d38e265409d8e99b397ab71eb3612fad0"
|
||||
|
||||
[[package]]
|
||||
name = "send_wrapper"
|
||||
version = "0.6.0"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "cd0b0ec5f1c1ca621c432a25813d8d60c88abe6d3e08a3eb9cf37d97a0fe3d73"
|
||||
dependencies = [
|
||||
"futures-core",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "serde"
|
||||
version = "1.0.228"
|
||||
@@ -12675,6 +13175,18 @@ version = "0.2.11"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "673aac59facbab8a9007c7f6108d11f63b603f7cabff99fabf650fea5c32b861"
|
||||
|
||||
[[package]]
|
||||
name = "unsigned-varint"
|
||||
version = "0.7.2"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "6889a77d49f1f013504cec6bf97a2c730394adedaeb1deb5ea08949a50541105"
|
||||
|
||||
[[package]]
|
||||
name = "unsigned-varint"
|
||||
version = "0.8.0"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "eb066959b24b5196ae73cb057f45598450d2c5f71460e98c49b738086eff9c06"
|
||||
|
||||
[[package]]
|
||||
name = "untrusted"
|
||||
version = "0.9.0"
|
||||
@@ -12898,6 +13410,12 @@ version = "0.9.5"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "0b928f33d975fc6ad9f86c8f283853ad26bdd5b10b7f1542aa2fa15e2289105a"
|
||||
|
||||
[[package]]
|
||||
name = "void"
|
||||
version = "1.0.2"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "6a02e4885ed3bc0f2de90ea6dd45ebcbb66dacffe03547fadbb0eeae2770887d"
|
||||
|
||||
[[package]]
|
||||
name = "vt100"
|
||||
version = "0.16.2"
|
||||
|
||||
@@ -172,6 +172,7 @@ members = [
|
||||
# "wasm/full-nym-wasm", # If we uncomment this again, remember to also uncomment the profile settings below
|
||||
"wasm/mix-fetch",
|
||||
"wasm/node-tester",
|
||||
"wasm/libp2p-nym",
|
||||
"wasm/zknym-lib",
|
||||
"nym-gateway-probe",
|
||||
"integration-tests", "common/nym-lp-transport", "common/nym-kkt-ciphersuite",
|
||||
@@ -543,6 +544,11 @@ web-sys = "0.3.76"
|
||||
#[patch.crates-io]
|
||||
#sphinx-packet = { path = "../sphinx" }
|
||||
|
||||
# Patch multiaddr with Protocol::Nym support for libp2p-nym transport
|
||||
[patch.crates-io.multiaddr]
|
||||
git = "https://github.com/mfahampshire/rust-multiaddr"
|
||||
branch = "nym-protocol"
|
||||
|
||||
# Profile settings for individual crates
|
||||
|
||||
# Compile-time verified queries do quite a bit of work at compile time. Incremental
|
||||
|
||||
@@ -0,0 +1,62 @@
|
||||
[package]
|
||||
name = "nym-libp2p-wasm"
|
||||
version = "0.1.0"
|
||||
edition = "2021"
|
||||
description = "libp2p transport over Nym mixnet for WASM/browser environments"
|
||||
rust-version = "1.85"
|
||||
publish = false
|
||||
|
||||
[lib]
|
||||
crate-type = ["cdylib", "rlib"]
|
||||
|
||||
[dependencies]
|
||||
# Async
|
||||
futures = { workspace = true }
|
||||
|
||||
# libp2p - minimal WASM-compatible features
|
||||
libp2p = { version = "0.54.1", default-features = false, features = [
|
||||
"wasm-bindgen",
|
||||
"ping",
|
||||
"identify",
|
||||
] }
|
||||
libp2p-identity = { version = "0.2.10", features = ["ed25519"] }
|
||||
libp2p-ping = "0.45.0"
|
||||
libp2p-identify = "0.45.0"
|
||||
libp2p-swarm = { version = "0.45.1", features = ["wasm-bindgen"] }
|
||||
|
||||
# Nym dependencies
|
||||
nym-client-wasm = { path = "../client", default-features = false }
|
||||
nym-wasm-client-core = { path = "../../common/wasm/client-core" }
|
||||
nym-wasm-utils = { path = "../../common/wasm/utils" }
|
||||
nym-sphinx-addressing = { path = "../../common/nymsphinx/addressing" }
|
||||
nym-sphinx-anonymous-replies = { path = "../../common/nymsphinx/anonymous-replies" }
|
||||
|
||||
# WASM essentials
|
||||
wasm-bindgen = { workspace = true }
|
||||
wasm-bindgen-futures = { workspace = true }
|
||||
js-sys = { workspace = true }
|
||||
gloo-timers = { workspace = true, features = ["futures"] }
|
||||
getrandom = { workspace = true, features = ["js"] }
|
||||
|
||||
# Tokio for WASM (provides sync primitives)
|
||||
tokio_with_wasm = { workspace = true, features = ["full"] }
|
||||
|
||||
# Utilities
|
||||
parking_lot = { workspace = true }
|
||||
send_wrapper = { version = "0.6", features = ["futures"] }
|
||||
thiserror = { workspace = true }
|
||||
log = { workspace = true }
|
||||
hex = { workspace = true }
|
||||
serde = { workspace = true, features = ["derive"] }
|
||||
serde-wasm-bindgen = { workspace = true }
|
||||
|
||||
[dev-dependencies]
|
||||
wasm-bindgen-test = { workspace = true }
|
||||
web-sys = { workspace = true, features = ["console"] }
|
||||
|
||||
[features]
|
||||
default = ["console_error_panic_hook"]
|
||||
console_error_panic_hook = ["nym-wasm-client-core/console_error_panic_hook"]
|
||||
|
||||
[package.metadata.wasm-pack.profile.release]
|
||||
wasm-opt = false
|
||||
@@ -0,0 +1,48 @@
|
||||
# nym-libp2p-wasm
|
||||
|
||||
This crate provides a libp2p `Transport` implementation that uses the Nym mixnet for p2p communication in browser environments.
|
||||
|
||||
## Features
|
||||
|
||||
- libp2p `Transport` trait implementation
|
||||
- Stream multiplexing via `StreamMuxer`
|
||||
- Message ordering over the unordered mixnet
|
||||
- Anonymous replies using SURBs
|
||||
- Browser compatible
|
||||
|
||||
## Building
|
||||
|
||||
```sh
|
||||
# Check
|
||||
cargo check -p nym-libp2p-wasm --target wasm32-unknown-unknown
|
||||
cargo build -p nym-libp2p-wasm --target wasm32-unknown-unknown
|
||||
|
||||
# Build with wasm-pack (for browser):
|
||||
cd wasm/libp2p-nym
|
||||
wasm-pack build --target web
|
||||
```
|
||||
|
||||
## Testing
|
||||
|
||||
Run WASM tests with wasm-pack:
|
||||
```sh
|
||||
cd wasm/libp2p-nym
|
||||
|
||||
# Browser tests (headless)
|
||||
wasm-pack test --headless --firefox
|
||||
wasm-pack test --headless --chrome
|
||||
```
|
||||
|
||||
## Modules
|
||||
|
||||
| Module | Description |
|
||||
|--------|-------------|
|
||||
| `lib.rs` | Crate root, re-exports public API and JS bindings |
|
||||
| `client.rs` | Client initialization, connects to Mixnet |
|
||||
| `transport.rs` | libp2p `Transport` trait implementation |
|
||||
| `connection.rs` | libp2p `StreamMuxer` implementation |
|
||||
| `substream.rs` | `AsyncRead`/`AsyncWrite` implementation |
|
||||
| `mixnet.rs` | Bridges Nym client to async channels |
|
||||
| `queue.rs` | Message ordering (mixnet doesn't guarantee order) |
|
||||
| `message.rs` | Wire protocol for connections/substreams |
|
||||
| `error.rs` | Error types |
|
||||
@@ -0,0 +1,104 @@
|
||||
// Copyright 2024 Nym Technologies SA
|
||||
// SPDX-License-Identifier: Apache-2.0
|
||||
|
||||
//! Client initialization for the libp2p transport.
|
||||
//!
|
||||
//! This module provides a simple way to create a Nym client configured
|
||||
//! for use with the libp2p transport.
|
||||
|
||||
use nym_client_wasm::client::NymClientBuilder;
|
||||
use nym_client_wasm::config::{ClientConfig, ClientConfigOpts};
|
||||
use nym_client_wasm::stream::NymClientStream;
|
||||
use nym_sphinx_addressing::clients::Recipient;
|
||||
use nym_wasm_utils::console_log;
|
||||
use std::str::FromStr;
|
||||
|
||||
use crate::error::Error;
|
||||
|
||||
// Type alias to help with inference
|
||||
type WasmClientError = nym_client_wasm::error::WasmClientError;
|
||||
|
||||
/// Result of creating a transport client.
|
||||
pub struct TransportClient {
|
||||
/// Our Nym address
|
||||
pub self_address: Recipient,
|
||||
/// The client stream for sending/receiving messages
|
||||
pub stream: NymClientStream,
|
||||
}
|
||||
|
||||
/// Options for creating a transport client.
|
||||
#[derive(Default)]
|
||||
pub struct TransportClientOpts {
|
||||
/// Optional Nym API URL override
|
||||
pub nym_api_url: Option<String>,
|
||||
/// Force TLS connections to gateways (required for browser environments)
|
||||
pub force_tls: bool,
|
||||
/// Client ID for storage namespace (avoids conflicts with other clients)
|
||||
pub client_id: Option<String>,
|
||||
}
|
||||
|
||||
/// Create a Nym client configured for libp2p transport use.
|
||||
///
|
||||
/// This connects to the Nym network and returns a `NymClientStream`
|
||||
/// that implements `Stream` for receiving messages.
|
||||
///
|
||||
/// # Example
|
||||
/// ```ignore
|
||||
/// use nym_libp2p_wasm::{create_transport_client_async, TransportClientOpts, NymTransport};
|
||||
/// use libp2p_identity::Keypair;
|
||||
/// use futures::StreamExt;
|
||||
///
|
||||
/// // Create the transport client (connects to Nym network)
|
||||
/// let opts = TransportClientOpts { force_tls: true, ..Default::default() };
|
||||
/// let result = create_transport_client_async(opts).await?;
|
||||
///
|
||||
/// // Use the stream directly
|
||||
/// while let Some(msg) = result.stream.next().await {
|
||||
/// println!("Received: {:?}", msg);
|
||||
/// }
|
||||
/// ```
|
||||
pub async fn create_transport_client_async(
|
||||
opts: TransportClientOpts,
|
||||
) -> Result<TransportClient, Error> {
|
||||
let client_id = opts
|
||||
.client_id
|
||||
.unwrap_or_else(|| "libp2p-transport".to_string());
|
||||
console_log!(
|
||||
"Creating transport client (id={}, force_tls={})...",
|
||||
client_id,
|
||||
opts.force_tls
|
||||
);
|
||||
|
||||
// Create config with client ID for storage namespace isolation
|
||||
let config_opts = ClientConfigOpts {
|
||||
id: Some(client_id),
|
||||
nym_api: opts.nym_api_url,
|
||||
nyxd: None,
|
||||
debug: None,
|
||||
};
|
||||
let config: ClientConfig = ClientConfig::new(config_opts)
|
||||
.map_err(|e: WasmClientError| Error::ClientCreationFailed(e.to_string()))?;
|
||||
|
||||
// Create builder with a dummy handler (we won't use JS callbacks)
|
||||
let dummy_handler = js_sys::Function::new_no_args("");
|
||||
let builder = NymClientBuilder::new(config, dummy_handler, opts.force_tls, None, None);
|
||||
|
||||
// Start client for transport use - returns (NymClient, ClientOutput)
|
||||
let (client, client_output) = builder
|
||||
.start_client_for_transport()
|
||||
.await
|
||||
.map_err(|e: WasmClientError| Error::ClientCreationFailed(e.to_string()))?;
|
||||
|
||||
let address_str = client.self_address();
|
||||
let self_address = Recipient::from_str(&address_str).map_err(Error::InvalidRecipientBytes)?;
|
||||
|
||||
console_log!("Transport client ready at: {}", address_str);
|
||||
|
||||
// Wrap in NymClientStream for easy async usage
|
||||
let stream = NymClientStream::new(client, client_output);
|
||||
|
||||
Ok(TransportClient {
|
||||
self_address,
|
||||
stream,
|
||||
})
|
||||
}
|
||||
@@ -0,0 +1,367 @@
|
||||
// Copyright 2024 Nym Technologies SA
|
||||
// SPDX-License-Identifier: Apache-2.0
|
||||
|
||||
//! Connection (StreamMuxer) implementation for multiplexing substreams over Nym.
|
||||
|
||||
use futures::channel::{
|
||||
mpsc::{unbounded, UnboundedReceiver, UnboundedSender},
|
||||
oneshot,
|
||||
};
|
||||
use futures::StreamExt;
|
||||
use libp2p::core::{muxing::StreamMuxerEvent, PeerId, StreamMuxer};
|
||||
use log::debug;
|
||||
use nym_sphinx_addressing::clients::Recipient;
|
||||
use nym_sphinx_anonymous_replies::requests::AnonymousSenderTag;
|
||||
use std::{
|
||||
collections::{HashMap, HashSet},
|
||||
pin::Pin,
|
||||
sync::{
|
||||
atomic::{AtomicU64, Ordering},
|
||||
Arc,
|
||||
},
|
||||
task::{Context, Poll, Waker},
|
||||
};
|
||||
|
||||
use super::error::Error;
|
||||
use super::message::{
|
||||
ConnectionId, Message, OutboundMessage, SubstreamId, SubstreamMessage, SubstreamMessageType,
|
||||
TransportMessage,
|
||||
};
|
||||
use super::substream::Substream;
|
||||
|
||||
/// Connection represents the result of a connection setup process.
|
||||
/// It implements `StreamMuxer` and thus has stream multiplexing built in.
|
||||
#[derive(Debug)]
|
||||
pub struct Connection {
|
||||
pub(crate) peer_id: PeerId,
|
||||
/// This will be Some(Recipient) for dialing connections since the outbound conn knows the nym/ multiaddr of the recipient, whereas receivers of connection requests will reply with SURBs
|
||||
pub(crate) remote_recipient: Option<Recipient>,
|
||||
pub(crate) id: ConnectionId,
|
||||
|
||||
/// receive inbound messages from the `InnerConnection`
|
||||
pub(crate) inbound_rx: UnboundedReceiver<SubstreamMessage>,
|
||||
|
||||
/// substream ID -> outbound pending substream exists
|
||||
/// the key is deleted when the response is received, or the request times out
|
||||
pending_substreams: HashSet<SubstreamId>,
|
||||
|
||||
/// substream ID -> substream's inbound_tx channel
|
||||
substream_inbound_txs: HashMap<SubstreamId, UnboundedSender<Vec<u8>>>,
|
||||
|
||||
/// substream ID -> substream's close_tx channel
|
||||
substream_close_txs: HashMap<SubstreamId, oneshot::Sender<()>>,
|
||||
|
||||
/// send messages to the mixnet
|
||||
/// used for sending `SubstreamMessageType::OpenRequest` messages
|
||||
/// also passed to each substream so they can write to the mixnet
|
||||
pub(crate) mixnet_outbound_tx: UnboundedSender<OutboundMessage>,
|
||||
|
||||
/// sender_tag for SURB replies to incoming messages
|
||||
pub(crate) sender_tag: Option<AnonymousSenderTag>,
|
||||
|
||||
/// inbound substream open requests; used in poll_inbound
|
||||
inbound_open_tx: UnboundedSender<Substream>,
|
||||
inbound_open_rx: UnboundedReceiver<Substream>,
|
||||
|
||||
/// closed substream IDs; used in poll_close
|
||||
close_tx: UnboundedSender<SubstreamId>,
|
||||
close_rx: UnboundedReceiver<SubstreamId>,
|
||||
|
||||
/// message nonce contains the next nonce that should be used when
|
||||
/// sending a message over the connection
|
||||
pub(crate) message_nonce: Arc<AtomicU64>,
|
||||
|
||||
waker: Option<Waker>,
|
||||
}
|
||||
|
||||
impl Connection {
|
||||
pub(crate) fn new_with_sender_tag(
|
||||
peer_id: PeerId,
|
||||
remote_recipient: Option<Recipient>,
|
||||
id: ConnectionId,
|
||||
inbound_rx: UnboundedReceiver<SubstreamMessage>,
|
||||
mixnet_outbound_tx: UnboundedSender<OutboundMessage>,
|
||||
sender_tag: Option<AnonymousSenderTag>,
|
||||
) -> Self {
|
||||
let (inbound_open_tx, inbound_open_rx) = unbounded();
|
||||
let (close_tx, close_rx) = unbounded();
|
||||
|
||||
Connection {
|
||||
peer_id,
|
||||
remote_recipient,
|
||||
id,
|
||||
inbound_rx,
|
||||
pending_substreams: HashSet::new(),
|
||||
substream_inbound_txs: HashMap::new(),
|
||||
substream_close_txs: HashMap::new(),
|
||||
mixnet_outbound_tx,
|
||||
sender_tag,
|
||||
inbound_open_tx,
|
||||
inbound_open_rx,
|
||||
close_tx,
|
||||
close_rx,
|
||||
message_nonce: Arc::new(AtomicU64::new(1)),
|
||||
waker: None,
|
||||
}
|
||||
}
|
||||
|
||||
/// Returns the remote peer's libp2p PeerId.
|
||||
pub fn peer_id(&self) -> PeerId {
|
||||
self.peer_id
|
||||
}
|
||||
|
||||
/// Returns the remote peer's Nym address, if known.
|
||||
///
|
||||
/// This is `Some(Recipient)` for connections we initiated (we know who we dialed),
|
||||
/// and `None` for incoming connections (we use SURBs to reply, never learning their address).
|
||||
///
|
||||
/// **Privacy property**: For incoming connections, this is always `None` - the listener
|
||||
/// never learns the dialer's Nym address.
|
||||
pub fn remote_nym_address(&self) -> Option<Recipient> {
|
||||
self.remote_recipient
|
||||
}
|
||||
|
||||
/// Returns true if this connection uses anonymous replies (SURBs).
|
||||
///
|
||||
/// This is true for incoming connections where we don't know the remote's address.
|
||||
pub fn uses_anonymous_replies(&self) -> bool {
|
||||
self.sender_tag.is_some()
|
||||
}
|
||||
|
||||
fn new_outbound_substream(&mut self) -> Result<Substream, Error> {
|
||||
debug!("new_outbound_substream called");
|
||||
let substream_id = SubstreamId::generate();
|
||||
debug!("Generated substream_id: {:?}", substream_id);
|
||||
let nonce = self.message_nonce.fetch_add(1, Ordering::SeqCst);
|
||||
debug!("Using nonce {}", nonce);
|
||||
debug!("Connection sender_tag: {:?}", self.sender_tag);
|
||||
debug!(
|
||||
"About to send with sender_tag: {:?}",
|
||||
self.sender_tag.is_some()
|
||||
);
|
||||
|
||||
let outbound_msg = OutboundMessage {
|
||||
recipient: self.remote_recipient, // Some(Recipient) for dialer, None for receiver
|
||||
message: Message::TransportMessage(TransportMessage {
|
||||
nonce,
|
||||
id: self.id.clone(),
|
||||
message: SubstreamMessage {
|
||||
substream_id: substream_id.clone(),
|
||||
message_type: SubstreamMessageType::OpenRequest,
|
||||
},
|
||||
}),
|
||||
sender_tag: self.sender_tag.clone(), // None for dialer, Some(sender_tag) for receiver
|
||||
};
|
||||
|
||||
debug!("Sending OpenRequest for substream: {:?}", substream_id);
|
||||
// Send the outbound message
|
||||
self.mixnet_outbound_tx
|
||||
.unbounded_send(outbound_msg)
|
||||
.map_err(|e| {
|
||||
debug!("Failed to send outbound message: {}", e);
|
||||
Error::OutboundSendFailure(e.to_string())
|
||||
})?;
|
||||
|
||||
debug!("Creating substream");
|
||||
// track pending outbound substreams
|
||||
let res = self.new_substream(substream_id.clone());
|
||||
if res.is_ok() {
|
||||
debug!("Adding to pending_substreams");
|
||||
self.pending_substreams.insert(substream_id);
|
||||
} else {
|
||||
debug!("Failed to create substream: {:?}", res);
|
||||
}
|
||||
res
|
||||
}
|
||||
|
||||
// creates a new substream instance with the given ID.
|
||||
fn new_substream(&mut self, id: SubstreamId) -> Result<Substream, Error> {
|
||||
// check we don't already have a substream with this ID
|
||||
if self.substream_inbound_txs.contains_key(&id) {
|
||||
return Err(Error::SubstreamIdExists(id));
|
||||
}
|
||||
|
||||
let (inbound_tx, inbound_rx) = unbounded::<Vec<u8>>();
|
||||
let (close_tx, close_rx) = oneshot::channel::<()>();
|
||||
self.substream_inbound_txs.insert(id.clone(), inbound_tx);
|
||||
self.substream_close_txs.insert(id.clone(), close_tx);
|
||||
|
||||
if let Some(waker) = self.waker.take() {
|
||||
waker.wake();
|
||||
}
|
||||
|
||||
Ok(Substream::new_with_sender_tag(
|
||||
self.remote_recipient,
|
||||
self.id.clone(),
|
||||
id,
|
||||
inbound_rx,
|
||||
self.mixnet_outbound_tx.clone(),
|
||||
close_rx,
|
||||
self.message_nonce.clone(),
|
||||
self.sender_tag.clone(), // Pass the connection's SURB directly
|
||||
))
|
||||
}
|
||||
|
||||
fn handle_close(&mut self, substream_id: SubstreamId) -> Result<(), Error> {
|
||||
if self.substream_inbound_txs.remove(&substream_id).is_none() {
|
||||
return Err(Error::SubstreamIdDoesNotExist(substream_id));
|
||||
}
|
||||
|
||||
// notify substream that it's closed
|
||||
let close_tx = self.substream_close_txs.remove(&substream_id);
|
||||
if let Some(tx) = close_tx {
|
||||
let _ = tx.send(());
|
||||
}
|
||||
|
||||
// notify poll_close that the substream is closed
|
||||
self.close_tx
|
||||
.unbounded_send(substream_id)
|
||||
.map_err(|e| Error::InboundSendFailure(e.to_string()))
|
||||
}
|
||||
}
|
||||
|
||||
impl StreamMuxer for Connection {
|
||||
type Substream = Substream;
|
||||
type Error = Error;
|
||||
|
||||
fn poll_inbound(
|
||||
mut self: Pin<&mut Self>,
|
||||
cx: &mut Context<'_>,
|
||||
) -> Poll<Result<Self::Substream, Self::Error>> {
|
||||
if let Poll::Ready(Some(substream)) = self.inbound_open_rx.poll_next_unpin(cx) {
|
||||
return Poll::Ready(Ok(substream));
|
||||
}
|
||||
|
||||
Poll::Pending
|
||||
}
|
||||
|
||||
fn poll_outbound(
|
||||
mut self: Pin<&mut Self>,
|
||||
_cx: &mut Context<'_>,
|
||||
) -> Poll<Result<Self::Substream, Self::Error>> {
|
||||
debug!("poll_outbound called");
|
||||
let result = self.new_outbound_substream();
|
||||
debug!("poll_outbound result: {:?}", result.is_ok());
|
||||
Poll::Ready(result)
|
||||
}
|
||||
|
||||
fn poll_close(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Result<(), Self::Error>> {
|
||||
if let Poll::Ready(Some(_)) = self.close_rx.poll_next_unpin(cx) {
|
||||
return Poll::Ready(Ok(()));
|
||||
}
|
||||
|
||||
Poll::Pending
|
||||
}
|
||||
|
||||
fn poll(
|
||||
mut self: Pin<&mut Self>,
|
||||
cx: &mut Context<'_>,
|
||||
) -> Poll<Result<StreamMuxerEvent, Self::Error>> {
|
||||
while let Poll::Ready(Some(msg)) = self.inbound_rx.poll_next_unpin(cx) {
|
||||
debug!(
|
||||
"Connection poll received message type: {:?} for substream: {:?}",
|
||||
msg.message_type, msg.substream_id
|
||||
);
|
||||
match msg.message_type {
|
||||
SubstreamMessageType::OpenRequest => {
|
||||
debug!(
|
||||
"Processing OpenRequest for substream: {:?}",
|
||||
msg.substream_id
|
||||
);
|
||||
|
||||
if self.remote_recipient.is_none() {
|
||||
debug!("Listener received OpenRequest - correct");
|
||||
} else {
|
||||
debug!("Dialer received OpenRequest - something is not right here");
|
||||
}
|
||||
|
||||
// create a new substream with the given ID
|
||||
let substream = self.new_substream(msg.substream_id.clone())?;
|
||||
let nonce = self.message_nonce.fetch_add(1, Ordering::SeqCst);
|
||||
|
||||
debug!("About to send OpenResponse with nonce: {}", nonce);
|
||||
debug!("Using sender_tag: {:?}", self.sender_tag);
|
||||
|
||||
// send the response to the remote peer
|
||||
let response_msg = OutboundMessage {
|
||||
recipient: self.remote_recipient,
|
||||
message: Message::TransportMessage(TransportMessage {
|
||||
nonce,
|
||||
id: self.id.clone(),
|
||||
message: SubstreamMessage {
|
||||
substream_id: msg.substream_id.clone(),
|
||||
message_type: SubstreamMessageType::OpenResponse,
|
||||
},
|
||||
}),
|
||||
sender_tag: self.sender_tag.clone(),
|
||||
};
|
||||
|
||||
debug!("Created OutboundMessage: {:?}", response_msg);
|
||||
|
||||
self.mixnet_outbound_tx
|
||||
.unbounded_send(response_msg)
|
||||
.map_err(|e| {
|
||||
debug!("FAILED to send OpenResponse: {}", e);
|
||||
Error::OutboundSendFailure(e.to_string())
|
||||
})?;
|
||||
debug!("Queued OpenResponse for mixnet");
|
||||
|
||||
// send the substream to our own channel to be returned in poll_inbound
|
||||
self.inbound_open_tx
|
||||
.unbounded_send(substream)
|
||||
.map_err(|e| Error::InboundSendFailure(e.to_string()))?;
|
||||
|
||||
debug!("new inbound substream: {:?}", &msg.substream_id);
|
||||
}
|
||||
SubstreamMessageType::OpenResponse => {
|
||||
debug!(
|
||||
"Processing OpenResponse for substream: {:?}",
|
||||
msg.substream_id
|
||||
);
|
||||
if !self.pending_substreams.remove(&msg.substream_id) {
|
||||
debug!(
|
||||
"SubstreamMessageType::OpenResponse no substream pending for ID: {:?}",
|
||||
&msg.substream_id
|
||||
);
|
||||
}
|
||||
}
|
||||
SubstreamMessageType::Close => {
|
||||
debug!("Processing Close for substream: {:?}", msg.substream_id);
|
||||
self.handle_close(msg.substream_id)?;
|
||||
}
|
||||
SubstreamMessageType::Data(data) => {
|
||||
debug!("Processing Data: {:?}", &data);
|
||||
let inbound_tx = self
|
||||
.substream_inbound_txs
|
||||
.get_mut(&msg.substream_id)
|
||||
.expect("must have a substream channel for substream");
|
||||
|
||||
// NOTE: this ignores channel closed errors, which is fine because the substream
|
||||
// might have been closed/dropped
|
||||
let _ = inbound_tx.unbounded_send(data);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
self.waker = Some(cx.waker().clone());
|
||||
Poll::Pending
|
||||
}
|
||||
}
|
||||
|
||||
/// PendingConnection represents a connection that's been initiated, but not completed.
|
||||
pub(crate) struct PendingConnection {
|
||||
pub(crate) remote_recipient: Recipient,
|
||||
pub(crate) connection_tx: oneshot::Sender<Connection>,
|
||||
}
|
||||
|
||||
impl PendingConnection {
|
||||
pub(crate) fn new(
|
||||
remote_recipient: Recipient,
|
||||
connection_tx: oneshot::Sender<Connection>,
|
||||
) -> Self {
|
||||
PendingConnection {
|
||||
remote_recipient,
|
||||
connection_tx,
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,67 @@
|
||||
// Copyright 2024 Nym Technologies SA
|
||||
// SPDX-License-Identifier: Apache-2.0
|
||||
|
||||
//! Error types for the Nym libp2p WASM transport.
|
||||
|
||||
use libp2p::core::multiaddr;
|
||||
use nym_sphinx_addressing::clients::RecipientFormattingError;
|
||||
|
||||
use super::message::SubstreamId;
|
||||
|
||||
#[derive(Debug, thiserror::Error)]
|
||||
pub enum Error {
|
||||
#[error("unimplemented")]
|
||||
Unimplemented,
|
||||
#[error("failed to format multiaddress from nym address")]
|
||||
FailedToFormatMultiaddr(#[from] multiaddr::Error),
|
||||
#[error("unexpected protocol in multiaddress")]
|
||||
InvalidProtocolForMultiaddr,
|
||||
#[error("failed to decode message")]
|
||||
InvalidMessageBytes,
|
||||
#[error("no connection found for ConnectionResponse")]
|
||||
NoConnectionForResponse,
|
||||
#[error("received ConnectionResponse but connection was already established")]
|
||||
ConnectionAlreadyEstablished,
|
||||
#[error("cannot handle connection request; already have connection with given ID")]
|
||||
ConnectionIDExists,
|
||||
#[error("no connection found for TransportMessage")]
|
||||
NoConnectionForTransportMessage,
|
||||
#[error("failed to decode ConnectionMessage; too short")]
|
||||
ConnectionMessageBytesTooShort,
|
||||
#[error("failed to decode ConnectionMessage; no peer ID")]
|
||||
ConnectionMessageBytesNoPeerId,
|
||||
#[error("invalid peer ID bytes")]
|
||||
InvalidPeerIdBytes,
|
||||
#[error("invalid recipient bytes")]
|
||||
InvalidRecipientBytes(#[from] RecipientFormattingError),
|
||||
#[error("failed to decode TransportMessage; too short")]
|
||||
TransportMessageBytesTooShort,
|
||||
#[error("failed to decode TransportMessage; invalid nonce")]
|
||||
InvalidNonce,
|
||||
#[error("invalid substream ID")]
|
||||
InvalidSubstreamMessageBytes,
|
||||
#[error("invalid substream message type byte")]
|
||||
InvalidSubstreamMessageType,
|
||||
#[error("substream with given ID already exists")]
|
||||
SubstreamIdExists(SubstreamId),
|
||||
#[error("no substream found for given ID")]
|
||||
SubstreamIdDoesNotExist(SubstreamId),
|
||||
#[error("recv error: channel closed")]
|
||||
OneshotRecvFailure,
|
||||
#[error("recv error: channel closed")]
|
||||
RecvFailure,
|
||||
#[error("outbound send error")]
|
||||
OutboundSendFailure(String),
|
||||
#[error("inbound send error")]
|
||||
InboundSendFailure(String),
|
||||
#[error("failed to send new connection; receiver dropped")]
|
||||
ConnectionSendFailure,
|
||||
#[error("failed to send initial TransportEvent::NewAddress")]
|
||||
SendErrorTransportEvent,
|
||||
#[error("dial timed out")]
|
||||
DialTimeout,
|
||||
#[error("mixnet client error: {0}")]
|
||||
MixnetClientError(String),
|
||||
#[error("failed to create client: {0}")]
|
||||
ClientCreationFailed(String),
|
||||
}
|
||||
@@ -0,0 +1,124 @@
|
||||
// Copyright 2024 Nym Technologies SA
|
||||
// SPDX-License-Identifier: Apache-2.0
|
||||
|
||||
//! libp2p transport over Nym mixnet for WASM/browser environments.
|
||||
//!
|
||||
//! This crate provides a libp2p Transport implementation that uses the Nym mixnet
|
||||
//! for privacy-preserving peer-to-peer communication in browser environments.
|
||||
//!
|
||||
//! # Features
|
||||
//!
|
||||
//! - Full libp2p Transport trait implementation
|
||||
//! - Stream multiplexing via StreamMuxer
|
||||
//! - Message ordering over the unordered mixnet
|
||||
//! - WASM/browser compatible (uses futures channels, gloo_timers)
|
||||
//!
|
||||
//! # Example
|
||||
//!
|
||||
//! ```ignore
|
||||
//! use nym_libp2p_wasm::{create_transport_client_async, NymTransport};
|
||||
//! use libp2p_identity::Keypair;
|
||||
//!
|
||||
//! async fn example() {
|
||||
//! // Create the transport client (connects to Nym network)
|
||||
//! let result = create_transport_client_async(None).await.unwrap();
|
||||
//!
|
||||
//! // Create the transport
|
||||
//! let keypair = Keypair::generate_ed25519();
|
||||
//! let transport = NymTransport::new(
|
||||
//! result.self_address,
|
||||
//! result.stream,
|
||||
//! keypair,
|
||||
//! ).await.unwrap();
|
||||
//! // Use transport with libp2p Swarm...
|
||||
//! }
|
||||
//! ```
|
||||
|
||||
#[cfg(target_arch = "wasm32")]
|
||||
pub mod client;
|
||||
#[cfg(target_arch = "wasm32")]
|
||||
pub(crate) mod connection;
|
||||
#[cfg(target_arch = "wasm32")]
|
||||
pub mod error;
|
||||
#[cfg(target_arch = "wasm32")]
|
||||
pub(crate) mod message;
|
||||
#[cfg(target_arch = "wasm32")]
|
||||
pub(crate) mod mixnet;
|
||||
#[cfg(target_arch = "wasm32")]
|
||||
pub(crate) mod queue;
|
||||
#[cfg(target_arch = "wasm32")]
|
||||
pub mod substream;
|
||||
#[cfg(target_arch = "wasm32")]
|
||||
pub mod transport;
|
||||
|
||||
// Re-exports for convenience
|
||||
#[cfg(target_arch = "wasm32")]
|
||||
pub use client::{create_transport_client_async, TransportClient, TransportClientOpts};
|
||||
#[cfg(target_arch = "wasm32")]
|
||||
pub use connection::Connection;
|
||||
#[cfg(target_arch = "wasm32")]
|
||||
pub use error::Error;
|
||||
#[cfg(target_arch = "wasm32")]
|
||||
pub use nym_sphinx_addressing::clients::Recipient;
|
||||
#[cfg(target_arch = "wasm32")]
|
||||
pub use substream::Substream;
|
||||
pub use transport::{nym_address_to_multiaddress, NymTransport, Upgrade};
|
||||
|
||||
/// The default timeout in seconds for the transport upgrade/handshake.
|
||||
#[cfg(target_arch = "wasm32")]
|
||||
const DEFAULT_HANDSHAKE_TIMEOUT_SECS: u64 = 30;
|
||||
|
||||
// WASM-bindgen exports for JavaScript
|
||||
#[cfg(target_arch = "wasm32")]
|
||||
mod js_api {
|
||||
use js_sys::Promise;
|
||||
use libp2p_identity::Keypair;
|
||||
use nym_wasm_utils::console_log;
|
||||
use wasm_bindgen::prelude::*;
|
||||
use wasm_bindgen_futures::future_to_promise;
|
||||
|
||||
use crate::client::{create_transport_client_async, TransportClientOpts};
|
||||
use crate::transport::NymTransport;
|
||||
|
||||
/// Initialize and test the libp2p-nym transport.
|
||||
/// Returns a Promise that resolves with the Nym address on success.
|
||||
///
|
||||
/// @param force_tls - If true, forces TLS connections to gateways (required for browsers)
|
||||
/// @param client_id - Optional client ID for storage namespace (defaults to "libp2p-transport")
|
||||
#[wasm_bindgen(js_name = "testTransport")]
|
||||
pub fn test_transport(
|
||||
nym_api_url: Option<String>,
|
||||
force_tls: Option<bool>,
|
||||
client_id: Option<String>,
|
||||
) -> Promise {
|
||||
future_to_promise(async move {
|
||||
let opts = TransportClientOpts {
|
||||
nym_api_url,
|
||||
force_tls: force_tls.unwrap_or(true), // Default to true for browser safety
|
||||
client_id, // Uses "libp2p-transport" if None
|
||||
};
|
||||
console_log!("testTransport: force_tls={}", opts.force_tls);
|
||||
|
||||
let result = create_transport_client_async(opts)
|
||||
.await
|
||||
.map_err(|e| JsValue::from_str(&e.to_string()))?;
|
||||
|
||||
let address = result.self_address.to_string();
|
||||
console_log!("Client connected! Address: {}", address);
|
||||
|
||||
// Create a libp2p keypair
|
||||
let keypair = Keypair::generate_ed25519();
|
||||
console_log!("Generated libp2p keypair");
|
||||
|
||||
// Create the transport
|
||||
let _transport = NymTransport::new(result.self_address, result.stream, keypair)
|
||||
.await
|
||||
.map_err(|e| JsValue::from_str(&e.to_string()))?;
|
||||
|
||||
console_log!("Transport created successfully!");
|
||||
console_log!("Listening on: /nym/{}", address);
|
||||
|
||||
Ok(JsValue::from_str(&address))
|
||||
})
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,298 @@
|
||||
// Copyright 2024 Nym Technologies SA
|
||||
// SPDX-License-Identifier: Apache-2.0
|
||||
|
||||
//! Wire protocol message types for the Nym libp2p transport.
|
||||
|
||||
use libp2p::core::PeerId;
|
||||
use nym_sphinx_addressing::clients::Recipient;
|
||||
use nym_sphinx_anonymous_replies::requests::AnonymousSenderTag;
|
||||
use std::fmt::{Debug, Formatter};
|
||||
|
||||
use super::error::Error;
|
||||
|
||||
const CONNECTION_ID_LENGTH: usize = 32;
|
||||
const SUBSTREAM_ID_LENGTH: usize = 32;
|
||||
|
||||
const NONCE_BYTES_LEN: usize = 8; // length of u64
|
||||
const MIN_CONNECTION_MESSAGE_LEN: usize = CONNECTION_ID_LENGTH + NONCE_BYTES_LEN;
|
||||
|
||||
/// Generate random bytes using getrandom (works in WASM with js feature)
|
||||
fn generate_random_bytes<const N: usize>() -> [u8; N] {
|
||||
let mut bytes = [0u8; N];
|
||||
getrandom::getrandom(&mut bytes).expect("getrandom failed");
|
||||
bytes
|
||||
}
|
||||
|
||||
/// ConnectionId is a unique, randomly-generated per-connection ID that's used to
|
||||
/// identify which connection a message belongs to.
|
||||
#[derive(Clone, Default, Eq, Hash, PartialEq)]
|
||||
pub(crate) struct ConnectionId([u8; 32]);
|
||||
|
||||
impl ConnectionId {
|
||||
pub(crate) fn generate() -> Self {
|
||||
ConnectionId(generate_random_bytes())
|
||||
}
|
||||
|
||||
fn from_bytes(bytes: &[u8]) -> Self {
|
||||
let mut id = [0u8; 32];
|
||||
id[..].copy_from_slice(&bytes[0..CONNECTION_ID_LENGTH]);
|
||||
ConnectionId(id)
|
||||
}
|
||||
}
|
||||
|
||||
impl Debug for ConnectionId {
|
||||
fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result {
|
||||
write!(f, "{}", hex::encode(self.0))
|
||||
}
|
||||
}
|
||||
|
||||
/// SubstreamId is a unique, randomly-generated per-substream ID that's used to
|
||||
/// identify which substream a message belongs to.
|
||||
#[derive(Clone, Default, Eq, Hash, PartialEq)]
|
||||
pub struct SubstreamId(pub(crate) [u8; 32]);
|
||||
|
||||
impl SubstreamId {
|
||||
pub(crate) fn generate() -> Self {
|
||||
SubstreamId(generate_random_bytes())
|
||||
}
|
||||
|
||||
fn from_bytes(bytes: &[u8]) -> Self {
|
||||
let mut id = [0u8; 32];
|
||||
id[..].copy_from_slice(&bytes[0..SUBSTREAM_ID_LENGTH]);
|
||||
SubstreamId(id)
|
||||
}
|
||||
}
|
||||
|
||||
impl Debug for SubstreamId {
|
||||
fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result {
|
||||
write!(f, "{}", hex::encode(self.0))
|
||||
}
|
||||
}
|
||||
|
||||
#[derive(Debug)]
|
||||
#[allow(clippy::enum_variant_names)]
|
||||
pub(crate) enum Message {
|
||||
ConnectionRequest(ConnectionMessage),
|
||||
ConnectionResponse(ConnectionMessage),
|
||||
TransportMessage(TransportMessage),
|
||||
}
|
||||
|
||||
/// ConnectionMessage is exchanged to open a new connection.
|
||||
#[derive(Debug)]
|
||||
pub(crate) struct ConnectionMessage {
|
||||
pub(crate) peer_id: PeerId,
|
||||
pub(crate) id: ConnectionId,
|
||||
}
|
||||
|
||||
/// TransportMessage is sent over a connection after establishment.
|
||||
#[derive(Debug, Clone)]
|
||||
pub(crate) struct TransportMessage {
|
||||
/// increments by 1 for every TransportMessage sent over a connection.
|
||||
/// required for ordering, since Nym does not guarantee ordering.
|
||||
/// ConnectionMessages do not need nonces, as we know that they will
|
||||
/// be the first messages sent over a connection.
|
||||
/// the first TransportMessage sent over a connection will have nonce 1.
|
||||
pub(crate) nonce: u64,
|
||||
pub(crate) message: SubstreamMessage,
|
||||
pub(crate) id: ConnectionId,
|
||||
}
|
||||
|
||||
impl Message {
|
||||
fn try_from_bytes(bytes: Vec<u8>) -> Result<Self, Error> {
|
||||
if bytes.len() < 2 {
|
||||
return Err(Error::InvalidMessageBytes);
|
||||
}
|
||||
|
||||
Ok(match bytes[0] {
|
||||
0 => Message::ConnectionRequest(ConnectionMessage::try_from_bytes(&bytes[1..])?),
|
||||
1 => Message::ConnectionResponse(ConnectionMessage::try_from_bytes(&bytes[1..])?),
|
||||
2 => Message::TransportMessage(TransportMessage::try_from_bytes(&bytes[1..])?),
|
||||
_ => return Err(Error::InvalidMessageBytes),
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
impl ConnectionMessage {
|
||||
fn to_bytes(&self) -> Vec<u8> {
|
||||
let mut bytes = self.id.0.to_vec();
|
||||
bytes.append(&mut self.peer_id.to_bytes());
|
||||
bytes
|
||||
}
|
||||
|
||||
fn try_from_bytes(bytes: &[u8]) -> Result<Self, Error> {
|
||||
if bytes.len() < CONNECTION_ID_LENGTH + 1 {
|
||||
return Err(Error::ConnectionMessageBytesTooShort);
|
||||
}
|
||||
|
||||
let id = ConnectionId::from_bytes(&bytes[0..CONNECTION_ID_LENGTH]);
|
||||
|
||||
let peer_id = PeerId::from_bytes(&bytes[CONNECTION_ID_LENGTH..])
|
||||
.map_err(|_| Error::InvalidPeerIdBytes)?;
|
||||
Ok(ConnectionMessage { peer_id, id })
|
||||
}
|
||||
}
|
||||
|
||||
impl TransportMessage {
|
||||
fn to_bytes(&self) -> Vec<u8> {
|
||||
let mut bytes = self.nonce.to_be_bytes().to_vec();
|
||||
bytes.extend_from_slice(self.id.0.as_ref());
|
||||
bytes.extend_from_slice(&self.message.to_bytes());
|
||||
bytes
|
||||
}
|
||||
|
||||
fn try_from_bytes(bytes: &[u8]) -> Result<Self, Error> {
|
||||
if bytes.len() < MIN_CONNECTION_MESSAGE_LEN + 1 {
|
||||
return Err(Error::TransportMessageBytesTooShort);
|
||||
}
|
||||
|
||||
let nonce = u64::from_be_bytes(
|
||||
bytes[0..NONCE_BYTES_LEN]
|
||||
.to_vec()
|
||||
.try_into()
|
||||
.map_err(|_| Error::InvalidNonce)?,
|
||||
);
|
||||
let id = ConnectionId::from_bytes(&bytes[NONCE_BYTES_LEN..MIN_CONNECTION_MESSAGE_LEN]);
|
||||
let message = SubstreamMessage::try_from_bytes(&bytes[MIN_CONNECTION_MESSAGE_LEN..])?;
|
||||
Ok(TransportMessage { nonce, message, id })
|
||||
}
|
||||
}
|
||||
|
||||
impl Ord for TransportMessage {
|
||||
fn cmp(&self, other: &Self) -> std::cmp::Ordering {
|
||||
self.nonce.cmp(&other.nonce)
|
||||
}
|
||||
}
|
||||
|
||||
impl std::cmp::PartialOrd for TransportMessage {
|
||||
fn partial_cmp(&self, other: &Self) -> Option<std::cmp::Ordering> {
|
||||
Some(self.cmp(other))
|
||||
}
|
||||
}
|
||||
|
||||
impl std::cmp::Eq for TransportMessage {}
|
||||
|
||||
impl std::cmp::PartialEq for TransportMessage {
|
||||
fn eq(&self, other: &Self) -> bool {
|
||||
self.nonce == other.nonce
|
||||
}
|
||||
}
|
||||
|
||||
#[derive(Debug, Clone)]
|
||||
pub(crate) enum SubstreamMessageType {
|
||||
OpenRequest,
|
||||
OpenResponse,
|
||||
Close,
|
||||
Data(Vec<u8>),
|
||||
}
|
||||
|
||||
impl SubstreamMessageType {
|
||||
fn to_u8(&self) -> u8 {
|
||||
match self {
|
||||
SubstreamMessageType::OpenRequest => 0,
|
||||
SubstreamMessageType::OpenResponse => 1,
|
||||
SubstreamMessageType::Close => 2,
|
||||
SubstreamMessageType::Data(_) => 3,
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
/// SubstreamMessage is a message sent over a substream.
|
||||
#[derive(Debug, Clone)]
|
||||
pub(crate) struct SubstreamMessage {
|
||||
pub(crate) substream_id: SubstreamId,
|
||||
pub(crate) message_type: SubstreamMessageType,
|
||||
}
|
||||
|
||||
impl SubstreamMessage {
|
||||
pub(crate) fn new_with_data(substream_id: SubstreamId, message: Vec<u8>) -> Self {
|
||||
SubstreamMessage {
|
||||
substream_id,
|
||||
message_type: SubstreamMessageType::Data(message),
|
||||
}
|
||||
}
|
||||
|
||||
pub(crate) fn new_close(substream_id: SubstreamId) -> Self {
|
||||
SubstreamMessage {
|
||||
substream_id,
|
||||
message_type: SubstreamMessageType::Close,
|
||||
}
|
||||
}
|
||||
|
||||
pub(crate) fn to_bytes(&self) -> Vec<u8> {
|
||||
let mut bytes = self.substream_id.0.clone().to_vec();
|
||||
bytes.push(self.message_type.to_u8());
|
||||
if let SubstreamMessageType::Data(message) = &self.message_type {
|
||||
bytes.extend_from_slice(message);
|
||||
}
|
||||
bytes
|
||||
}
|
||||
|
||||
pub(crate) fn try_from_bytes(bytes: &[u8]) -> Result<Self, Error> {
|
||||
if bytes.len() < SUBSTREAM_ID_LENGTH + 1 {
|
||||
return Err(Error::InvalidSubstreamMessageBytes);
|
||||
}
|
||||
|
||||
let substream_id = SubstreamId::from_bytes(&bytes[0..SUBSTREAM_ID_LENGTH]);
|
||||
let message_type = match bytes[SUBSTREAM_ID_LENGTH] {
|
||||
0 => SubstreamMessageType::OpenRequest,
|
||||
1 => SubstreamMessageType::OpenResponse,
|
||||
2 => SubstreamMessageType::Close,
|
||||
3 => {
|
||||
if bytes.len() < SUBSTREAM_ID_LENGTH + 2 {
|
||||
return Err(Error::InvalidSubstreamMessageBytes);
|
||||
}
|
||||
SubstreamMessageType::Data(bytes[SUBSTREAM_ID_LENGTH + 1..].to_vec())
|
||||
}
|
||||
_ => return Err(Error::InvalidSubstreamMessageType),
|
||||
};
|
||||
|
||||
Ok(SubstreamMessage {
|
||||
substream_id,
|
||||
message_type,
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
impl Message {
|
||||
pub(crate) fn to_bytes(&self) -> Vec<u8> {
|
||||
match self {
|
||||
Message::ConnectionRequest(msg) => {
|
||||
let mut bytes = 0_u8.to_be_bytes().to_vec();
|
||||
bytes.append(&mut msg.to_bytes());
|
||||
bytes
|
||||
}
|
||||
Message::ConnectionResponse(msg) => {
|
||||
let mut bytes = 1_u8.to_be_bytes().to_vec();
|
||||
bytes.append(&mut msg.to_bytes());
|
||||
bytes
|
||||
}
|
||||
Message::TransportMessage(msg) => {
|
||||
let mut bytes = 2_u8.to_be_bytes().to_vec();
|
||||
bytes.append(&mut msg.to_bytes());
|
||||
bytes
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
/// InboundMessage represents an inbound mixnet message.
|
||||
pub(crate) struct InboundMessage(pub(crate) Message, pub(crate) Option<AnonymousSenderTag>);
|
||||
|
||||
/// OutboundMessage represents an outbound mixnet message.
|
||||
#[derive(Debug)]
|
||||
pub(crate) struct OutboundMessage {
|
||||
pub(crate) message: Message,
|
||||
pub(crate) recipient: Option<Recipient>,
|
||||
pub(crate) sender_tag: Option<AnonymousSenderTag>,
|
||||
}
|
||||
|
||||
pub(crate) fn parse_message_data(
|
||||
data: &[u8],
|
||||
sender_tag: Option<AnonymousSenderTag>,
|
||||
) -> Result<InboundMessage, Error> {
|
||||
if data.len() < 2 {
|
||||
return Err(Error::InvalidMessageBytes);
|
||||
}
|
||||
let msg = Message::try_from_bytes(data.to_vec())?;
|
||||
Ok(InboundMessage(msg, sender_tag))
|
||||
}
|
||||
@@ -0,0 +1,200 @@
|
||||
// Copyright 2024 Nym Technologies SA
|
||||
// SPDX-License-Identifier: Apache-2.0
|
||||
|
||||
//! Mixnet bridge for WASM - adapts the Nym client stream to channel-based async.
|
||||
//!
|
||||
//! This module bridges the NymClientStream to the channel-based architecture
|
||||
//! used by the libp2p transport.
|
||||
|
||||
use futures::channel::mpsc::{unbounded, UnboundedReceiver, UnboundedSender};
|
||||
use futures::{SinkExt, StreamExt};
|
||||
use log::debug;
|
||||
use nym_client_wasm::stream::NymClientStream;
|
||||
use nym_sphinx_addressing::clients::Recipient;
|
||||
use nym_wasm_client_core::client::inbound_messages::InputMessage;
|
||||
use nym_wasm_client_core::nym_task::connections::TransmissionLane;
|
||||
use std::sync::Arc;
|
||||
use tokio_with_wasm::sync::RwLock;
|
||||
use wasm_bindgen_futures::spawn_local;
|
||||
|
||||
use super::error::Error;
|
||||
use super::message::*;
|
||||
|
||||
/// Default number of reply SURBs to attach when sending anonymous messages.
|
||||
const DEFAULT_REPLY_SURBS: u32 = 10;
|
||||
|
||||
/// Initialize the mixnet bridge for libp2p transport.
|
||||
///
|
||||
/// This function bridges the NymClientStream to the channel-based async pattern
|
||||
/// used by the libp2p transport. It spawns background tasks to:
|
||||
/// 1. Forward outbound messages from the transport to the mixnet client
|
||||
/// 2. Forward inbound messages from the mixnet client to the transport
|
||||
pub(crate) async fn initialize_mixnet(
|
||||
self_address: Recipient,
|
||||
stream: NymClientStream,
|
||||
notify_inbound_tx: Option<UnboundedSender<()>>,
|
||||
) -> Result<
|
||||
(
|
||||
Recipient,
|
||||
UnboundedReceiver<InboundMessage>,
|
||||
UnboundedSender<OutboundMessage>,
|
||||
),
|
||||
Error,
|
||||
> {
|
||||
// Channel for inbound messages from the mixnet to the transport
|
||||
let (inbound_tx, inbound_rx) = unbounded::<InboundMessage>();
|
||||
|
||||
// Channel for outbound messages from the transport to the mixnet
|
||||
let (outbound_tx, outbound_rx) = unbounded::<OutboundMessage>();
|
||||
|
||||
// Get client_input for sending before we move stream
|
||||
let client_input = stream.client_input();
|
||||
|
||||
// Wrap stream in Arc<RwLock> so we can share it
|
||||
let stream = Arc::new(RwLock::new(stream));
|
||||
|
||||
// Spawn the outbound message handler
|
||||
spawn_local(run_outbound_loop(client_input, outbound_rx));
|
||||
|
||||
// Spawn the inbound message handler
|
||||
spawn_local(run_inbound_loop(stream, inbound_tx, notify_inbound_tx));
|
||||
|
||||
Ok((self_address, inbound_rx, outbound_tx))
|
||||
}
|
||||
|
||||
/// Background task that forwards outbound messages from the transport to the mixnet.
|
||||
async fn run_outbound_loop(
|
||||
client_input: Arc<RwLock<nym_wasm_client_core::client::base_client::ClientInput>>,
|
||||
mut outbound_rx: UnboundedReceiver<OutboundMessage>,
|
||||
) {
|
||||
while let Some(message) = outbound_rx.next().await {
|
||||
if let Err(e) = send_outbound_message(&client_input, message).await {
|
||||
debug!("Failed to send outbound message: {:?}", e);
|
||||
}
|
||||
}
|
||||
debug!("Outbound message loop ended");
|
||||
}
|
||||
|
||||
/// Send a single outbound message via the mixnet client.
|
||||
async fn send_outbound_message(
|
||||
client_input: &Arc<RwLock<nym_wasm_client_core::client::base_client::ClientInput>>,
|
||||
message: OutboundMessage,
|
||||
) -> Result<(), Error> {
|
||||
log_outbound_message(&message);
|
||||
|
||||
let data = message.message.to_bytes();
|
||||
|
||||
let input_msg = match (&message.recipient, &message.sender_tag) {
|
||||
// Reply using SURB (anonymous reply)
|
||||
(_, Some(sender_tag)) => {
|
||||
debug!(
|
||||
"Sending reply to sender_tag {:?}",
|
||||
sender_tag.to_base58_string()
|
||||
);
|
||||
InputMessage::new_reply(sender_tag.clone(), data, TransmissionLane::General, None)
|
||||
}
|
||||
// Regular message with recipient, include SURBs for reply capability
|
||||
(Some(recipient), None) => {
|
||||
debug!("Sending anonymous message to recipient {}", recipient);
|
||||
InputMessage::new_anonymous(
|
||||
*recipient,
|
||||
data,
|
||||
DEFAULT_REPLY_SURBS,
|
||||
TransmissionLane::General,
|
||||
None,
|
||||
)
|
||||
}
|
||||
// No recipient or sender_tag - cannot route
|
||||
(None, None) => {
|
||||
debug!("No recipient or sender_tag provided, cannot route message");
|
||||
return Err(Error::OutboundSendFailure(
|
||||
"No recipient or sender_tag provided".to_string(),
|
||||
));
|
||||
}
|
||||
};
|
||||
|
||||
// Send via the client input
|
||||
let mut client = client_input.write().await;
|
||||
client
|
||||
.input_sender
|
||||
.send(input_msg)
|
||||
.await
|
||||
.map_err(|_| Error::OutboundSendFailure("InputMessageReceiver stopped".to_string()))
|
||||
}
|
||||
|
||||
/// Log outbound message details for debugging.
|
||||
fn log_outbound_message(message: &OutboundMessage) {
|
||||
match &message.message {
|
||||
Message::TransportMessage(tm) => match &tm.message.message_type {
|
||||
SubstreamMessageType::OpenResponse => {
|
||||
debug!(
|
||||
"Outbound OpenResponse: nonce={}, substream={:?}",
|
||||
tm.nonce, tm.message.substream_id
|
||||
);
|
||||
}
|
||||
SubstreamMessageType::OpenRequest => {
|
||||
debug!(
|
||||
"Outbound OpenRequest: nonce={}, substream={:?}",
|
||||
tm.nonce, tm.message.substream_id
|
||||
);
|
||||
}
|
||||
SubstreamMessageType::Data(_) => {
|
||||
debug!(
|
||||
"Outbound Data: nonce={}, substream={:?}",
|
||||
tm.nonce, tm.message.substream_id
|
||||
);
|
||||
}
|
||||
SubstreamMessageType::Close => {
|
||||
debug!(
|
||||
"Outbound Close: nonce={}, substream={:?}",
|
||||
tm.nonce, tm.message.substream_id
|
||||
);
|
||||
}
|
||||
},
|
||||
Message::ConnectionRequest(_) => debug!("Outbound ConnectionRequest"),
|
||||
Message::ConnectionResponse(_) => debug!("Outbound ConnectionResponse"),
|
||||
}
|
||||
}
|
||||
|
||||
/// Background task that forwards inbound messages from the mixnet to the transport.
|
||||
async fn run_inbound_loop(
|
||||
stream: Arc<RwLock<NymClientStream>>,
|
||||
inbound_tx: UnboundedSender<InboundMessage>,
|
||||
notify_inbound_tx: Option<UnboundedSender<()>>,
|
||||
) {
|
||||
loop {
|
||||
// Lock the stream to poll for next message
|
||||
let msg = {
|
||||
let mut stream_guard = stream.write().await;
|
||||
stream_guard.next().await
|
||||
};
|
||||
|
||||
match msg {
|
||||
Some(reconstructed_msg) => {
|
||||
// Notify if requested
|
||||
if let Some(ref notify_tx) = notify_inbound_tx {
|
||||
let _ = notify_tx.unbounded_send(());
|
||||
}
|
||||
|
||||
let (message_bytes, sender_tag) = reconstructed_msg.into_inner();
|
||||
match parse_message_data(&message_bytes, sender_tag) {
|
||||
Ok(data) => {
|
||||
if inbound_tx.unbounded_send(data).is_err() {
|
||||
debug!("Inbound channel closed");
|
||||
break;
|
||||
}
|
||||
}
|
||||
Err(e) => {
|
||||
debug!("Failed to parse inbound message: {:?}", e);
|
||||
}
|
||||
}
|
||||
}
|
||||
None => {
|
||||
debug!("Inbound stream ended");
|
||||
break;
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
debug!("Inbound message loop ended");
|
||||
}
|
||||
@@ -0,0 +1,144 @@
|
||||
// Copyright 2024 Nym Technologies SA
|
||||
// SPDX-License-Identifier: Apache-2.0
|
||||
|
||||
//! Message queue for ordering messages received from the mixnet.
|
||||
//! The Nym mixnet does not guarantee message ordering, so we need to
|
||||
//! reorder messages based on their nonce before processing.
|
||||
|
||||
use log::{debug, warn};
|
||||
use std::collections::BTreeSet;
|
||||
|
||||
use super::message::TransportMessage;
|
||||
|
||||
/// MessageQueue is a queue of messages, ordered by nonce, that we've
|
||||
/// received but are not yet able to process because we're waiting for
|
||||
/// a message with the next expected nonce first.
|
||||
/// This is required because Nym does not guarantee any sort of message
|
||||
/// ordering, only delivery.
|
||||
/// TODO: is there a DOS vector here where a malicious peer sends us
|
||||
/// messages only with nonce higher than the next expected nonce?
|
||||
pub(crate) struct MessageQueue {
|
||||
/// nonce of the next message we expect to receive on the
|
||||
/// connection.
|
||||
/// any messages with a nonce greater than this are pushed into
|
||||
/// the queue.
|
||||
/// if we get a message with a nonce equal to this, then we
|
||||
/// immediately handle it in the transport and increment the nonce.
|
||||
next_expected_nonce: u64,
|
||||
|
||||
/// the actual queue of messages, ordered by nonce.
|
||||
/// the head of the queue's nonce is always greater
|
||||
/// than the next expected nonce.
|
||||
queue: BTreeSet<TransportMessage>,
|
||||
}
|
||||
|
||||
impl MessageQueue {
|
||||
pub(crate) fn new() -> Self {
|
||||
MessageQueue {
|
||||
next_expected_nonce: 0,
|
||||
queue: BTreeSet::new(),
|
||||
}
|
||||
}
|
||||
|
||||
pub(crate) fn print_nonces(&self) {
|
||||
let nonces = self.queue.iter().map(|msg| msg.nonce).collect::<Vec<_>>();
|
||||
debug!("MessageQueue: {:?}", nonces);
|
||||
}
|
||||
|
||||
/// sets the next expected nonce to 1, indicating that we've received
|
||||
/// a ConnectionRequest or ConnectionResponse.
|
||||
pub(crate) fn set_connection_message_received(&mut self) {
|
||||
if self.next_expected_nonce != 0 {
|
||||
panic!("connection message received twice");
|
||||
}
|
||||
|
||||
self.next_expected_nonce = self.next_expected_nonce.wrapping_add(1);
|
||||
}
|
||||
|
||||
/// tries to push a message into the queue.
|
||||
/// if the message has the next expected nonce, then the message is returned,
|
||||
/// and should be processed by the caller.
|
||||
/// in that case, the internal queue's next expected nonce is incremented.
|
||||
pub(crate) fn try_push(&mut self, msg: TransportMessage) -> Option<TransportMessage> {
|
||||
if msg.nonce == self.next_expected_nonce {
|
||||
self.next_expected_nonce = self.next_expected_nonce.wrapping_add(1);
|
||||
Some(msg)
|
||||
} else {
|
||||
if msg.nonce < self.next_expected_nonce {
|
||||
// this shouldn't happen normally, only if the other node
|
||||
// is not following the protocol
|
||||
warn!("received a message with a nonce that is too low");
|
||||
return None;
|
||||
}
|
||||
|
||||
if !self.queue.insert(msg) {
|
||||
// this shouldn't happen normally, only if the other node
|
||||
// is not following the protocol
|
||||
warn!("received a message with a duplicate nonce");
|
||||
return None;
|
||||
}
|
||||
|
||||
None
|
||||
}
|
||||
}
|
||||
|
||||
pub(crate) fn pop(&mut self) -> Option<TransportMessage> {
|
||||
let head = self.queue.first()?;
|
||||
|
||||
if head.nonce == self.next_expected_nonce {
|
||||
self.next_expected_nonce = self.next_expected_nonce.wrapping_add(1);
|
||||
Some(self.queue.pop_first().unwrap())
|
||||
} else {
|
||||
None
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
#[cfg(test)]
|
||||
mod test {
|
||||
use super::super::message::{ConnectionId, SubstreamId, SubstreamMessage};
|
||||
use super::*;
|
||||
|
||||
impl TransportMessage {
|
||||
fn new(nonce: u64, message: SubstreamMessage, id: ConnectionId) -> Self {
|
||||
TransportMessage { nonce, message, id }
|
||||
}
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn test_message_queue() {
|
||||
let mut queue = MessageQueue::new();
|
||||
|
||||
let test_substream_message =
|
||||
SubstreamMessage::new_with_data(SubstreamId::generate(), vec![1, 2, 3]);
|
||||
let connection_id = ConnectionId::generate();
|
||||
|
||||
let msg1 = TransportMessage::new(1, test_substream_message.clone(), connection_id.clone());
|
||||
let msg2 = TransportMessage::new(2, test_substream_message.clone(), connection_id.clone());
|
||||
let msg3 = TransportMessage::new(3, test_substream_message.clone(), connection_id.clone());
|
||||
|
||||
assert_eq!(queue.try_push(msg1.clone()), None);
|
||||
assert_eq!(queue.try_push(msg3.clone()), None);
|
||||
assert_eq!(queue.try_push(msg2.clone()), None);
|
||||
|
||||
assert_eq!(queue.pop(), None);
|
||||
|
||||
// set expected nonce to 1
|
||||
queue.set_connection_message_received();
|
||||
assert_eq!(queue.pop(), Some(msg1));
|
||||
|
||||
let msg4 = TransportMessage::new(4, test_substream_message.clone(), connection_id.clone());
|
||||
assert_eq!(queue.try_push(msg4.clone()), None);
|
||||
|
||||
assert_eq!(queue.pop(), Some(msg2));
|
||||
assert_eq!(queue.pop(), Some(msg3));
|
||||
assert_eq!(queue.pop(), Some(msg4));
|
||||
assert_eq!(queue.pop(), None);
|
||||
assert_eq!(queue.next_expected_nonce, 5);
|
||||
|
||||
// should just return the message and increment nonce when message nonce = next expected nonce
|
||||
let msg5 = TransportMessage::new(5, test_substream_message, connection_id);
|
||||
assert_eq!(queue.try_push(msg5.clone()), Some(msg5));
|
||||
assert_eq!(queue.next_expected_nonce, 6);
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,254 @@
|
||||
// Copyright 2024 Nym Technologies SA
|
||||
// SPDX-License-Identifier: Apache-2.0
|
||||
|
||||
//! Substream implementation providing AsyncRead + AsyncWrite over Nym mixnet.
|
||||
|
||||
use super::message::{
|
||||
ConnectionId, Message, OutboundMessage, SubstreamId, SubstreamMessage, TransportMessage,
|
||||
};
|
||||
use futures::{
|
||||
channel::{mpsc::UnboundedReceiver, oneshot::Receiver},
|
||||
io::{Error as IoError, ErrorKind},
|
||||
AsyncRead, AsyncWrite, FutureExt, StreamExt,
|
||||
};
|
||||
use log::debug;
|
||||
use nym_sphinx_addressing::clients::Recipient;
|
||||
use nym_sphinx_anonymous_replies::requests::AnonymousSenderTag;
|
||||
use parking_lot::Mutex;
|
||||
use std::{
|
||||
pin::Pin,
|
||||
sync::{
|
||||
atomic::{AtomicU64, Ordering},
|
||||
Arc,
|
||||
},
|
||||
task::{Context, Poll},
|
||||
};
|
||||
|
||||
// Re-export UnboundedSender for use in other modules
|
||||
pub(crate) use futures::channel::mpsc::UnboundedSender;
|
||||
|
||||
#[derive(Debug)]
|
||||
pub struct Substream {
|
||||
remote_recipient: Option<Recipient>,
|
||||
connection_id: ConnectionId,
|
||||
pub(crate) substream_id: SubstreamId,
|
||||
|
||||
/// inbound messages; inbound_tx is in the corresponding Connection
|
||||
pub(crate) inbound_rx: UnboundedReceiver<Vec<u8>>,
|
||||
|
||||
/// outbound messages; go directly to the mixnet
|
||||
outbound_tx: UnboundedSender<OutboundMessage>,
|
||||
|
||||
sender_tag: Option<AnonymousSenderTag>,
|
||||
|
||||
/// used to signal when the substream is closed
|
||||
close_rx: Receiver<()>,
|
||||
closed: Mutex<bool>,
|
||||
|
||||
// buffer of data that's been written to the stream,
|
||||
// but not yet read by the application.
|
||||
unread_data: Mutex<Vec<u8>>,
|
||||
|
||||
message_nonce: Arc<AtomicU64>,
|
||||
}
|
||||
|
||||
impl Substream {
|
||||
pub(crate) fn new_with_sender_tag(
|
||||
remote_recipient: Option<Recipient>,
|
||||
connection_id: ConnectionId,
|
||||
substream_id: SubstreamId,
|
||||
inbound_rx: UnboundedReceiver<Vec<u8>>,
|
||||
outbound_tx: UnboundedSender<OutboundMessage>,
|
||||
close_rx: Receiver<()>,
|
||||
message_nonce: Arc<AtomicU64>,
|
||||
sender_tag: Option<AnonymousSenderTag>,
|
||||
) -> Self {
|
||||
Substream {
|
||||
remote_recipient,
|
||||
connection_id,
|
||||
substream_id,
|
||||
inbound_rx,
|
||||
outbound_tx,
|
||||
sender_tag,
|
||||
close_rx,
|
||||
closed: Mutex::new(false),
|
||||
unread_data: Mutex::new(vec![]),
|
||||
message_nonce,
|
||||
}
|
||||
}
|
||||
|
||||
pub(crate) fn new(
|
||||
remote_recipient: Option<Recipient>,
|
||||
connection_id: ConnectionId,
|
||||
substream_id: SubstreamId,
|
||||
inbound_rx: UnboundedReceiver<Vec<u8>>,
|
||||
outbound_tx: UnboundedSender<OutboundMessage>,
|
||||
close_rx: Receiver<()>,
|
||||
message_nonce: Arc<AtomicU64>,
|
||||
) -> Self {
|
||||
Self::new_with_sender_tag(
|
||||
remote_recipient,
|
||||
connection_id,
|
||||
substream_id,
|
||||
inbound_rx,
|
||||
outbound_tx,
|
||||
close_rx,
|
||||
message_nonce,
|
||||
None,
|
||||
)
|
||||
}
|
||||
|
||||
fn check_closed(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Result<(), IoError> {
|
||||
let closed_err = IoError::new(ErrorKind::Other, "stream closed");
|
||||
|
||||
// Poll the close receiver to check if close was signaled
|
||||
let received_closed = self.close_rx.poll_unpin(cx);
|
||||
|
||||
let mut closed = self.closed.lock();
|
||||
if *closed {
|
||||
return Err(closed_err);
|
||||
}
|
||||
|
||||
if let Poll::Ready(Ok(())) = received_closed {
|
||||
*closed = true;
|
||||
return Err(closed_err);
|
||||
}
|
||||
|
||||
Ok(())
|
||||
}
|
||||
}
|
||||
|
||||
impl AsyncRead for Substream {
|
||||
fn poll_read(
|
||||
mut self: Pin<&mut Self>,
|
||||
cx: &mut Context<'_>,
|
||||
buf: &mut [u8],
|
||||
) -> Poll<Result<usize, IoError>> {
|
||||
let closed_result = self.as_mut().check_closed(cx);
|
||||
if let Err(e) = closed_result {
|
||||
return Poll::Ready(Err(e));
|
||||
}
|
||||
|
||||
let inbound_rx_data = self.inbound_rx.poll_next_unpin(cx);
|
||||
|
||||
// first, write any previously unread data to the buf
|
||||
let mut unread_data = self.unread_data.lock();
|
||||
let filled_len = if !unread_data.is_empty() {
|
||||
let unread_len = unread_data.len();
|
||||
let buf_len = buf.len();
|
||||
let copy_len = std::cmp::min(unread_len, buf_len);
|
||||
buf[..copy_len].copy_from_slice(&unread_data[..copy_len]);
|
||||
*unread_data = unread_data[copy_len..].to_vec();
|
||||
copy_len
|
||||
} else {
|
||||
0
|
||||
};
|
||||
|
||||
if let Poll::Ready(Some(data)) = inbound_rx_data {
|
||||
if filled_len == buf.len() {
|
||||
// we've filled the buffer, so we'll have to save the rest for later
|
||||
let mut new = vec![];
|
||||
new.extend(unread_data.drain(..));
|
||||
new.extend(data.iter());
|
||||
*unread_data = new;
|
||||
return Poll::Ready(Ok(filled_len));
|
||||
}
|
||||
|
||||
// otherwise, there's still room in the buffer, so we'll copy the rest of the data
|
||||
let remaining_len = buf.len() - filled_len;
|
||||
let data_len = data.len();
|
||||
|
||||
// we have more data than buffer room remaining, save the extra for later
|
||||
if remaining_len < data_len {
|
||||
unread_data.extend_from_slice(&data[remaining_len..]);
|
||||
}
|
||||
|
||||
let copied = std::cmp::min(remaining_len, data_len);
|
||||
buf[filled_len..filled_len + copied].copy_from_slice(&data[..copied]);
|
||||
debug!("poll_read copied {} bytes", copied);
|
||||
return Poll::Ready(Ok(copied));
|
||||
}
|
||||
|
||||
if filled_len > 0 {
|
||||
debug!("poll_read copied {} bytes", filled_len);
|
||||
return Poll::Ready(Ok(filled_len));
|
||||
}
|
||||
|
||||
Poll::Pending
|
||||
}
|
||||
}
|
||||
|
||||
impl AsyncWrite for Substream {
|
||||
fn poll_write(
|
||||
mut self: Pin<&mut Self>,
|
||||
cx: &mut Context<'_>,
|
||||
buf: &[u8],
|
||||
) -> Poll<Result<usize, IoError>> {
|
||||
if let Err(e) = self.as_mut().check_closed(cx) {
|
||||
return Poll::Ready(Err(e));
|
||||
}
|
||||
|
||||
let nonce = self.message_nonce.fetch_add(1, Ordering::SeqCst);
|
||||
|
||||
self.outbound_tx
|
||||
.unbounded_send(OutboundMessage {
|
||||
recipient: self.remote_recipient,
|
||||
message: Message::TransportMessage(TransportMessage {
|
||||
nonce,
|
||||
id: self.connection_id.clone(),
|
||||
message: SubstreamMessage::new_with_data(
|
||||
self.substream_id.clone(),
|
||||
buf.to_vec(),
|
||||
),
|
||||
}),
|
||||
sender_tag: self.sender_tag.clone(),
|
||||
})
|
||||
.map_err(|e| {
|
||||
IoError::new(
|
||||
ErrorKind::Other,
|
||||
format!("poll_write outbound_tx error: {}", e),
|
||||
)
|
||||
})?;
|
||||
|
||||
Poll::Ready(Ok(buf.len()))
|
||||
}
|
||||
|
||||
fn poll_close(self: Pin<&mut Self>, _cx: &mut Context<'_>) -> Poll<Result<(), IoError>> {
|
||||
let nonce = self.message_nonce.fetch_add(1, Ordering::SeqCst);
|
||||
|
||||
let mut closed = self.closed.lock();
|
||||
if *closed {
|
||||
return Poll::Ready(Err(IoError::new(ErrorKind::Other, "stream closed")));
|
||||
}
|
||||
|
||||
*closed = true;
|
||||
|
||||
// send a close message to the mixnet
|
||||
self.outbound_tx
|
||||
.unbounded_send(OutboundMessage {
|
||||
recipient: self.remote_recipient,
|
||||
message: Message::TransportMessage(TransportMessage {
|
||||
nonce,
|
||||
id: self.connection_id.clone(),
|
||||
message: SubstreamMessage::new_close(self.substream_id.clone()),
|
||||
}),
|
||||
sender_tag: self.sender_tag.clone(),
|
||||
})
|
||||
.map_err(|e| {
|
||||
IoError::new(
|
||||
ErrorKind::Other,
|
||||
format!("poll_close outbound_rx error: {}", e),
|
||||
)
|
||||
})?;
|
||||
|
||||
Poll::Ready(Ok(()))
|
||||
}
|
||||
|
||||
fn poll_flush(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Result<(), IoError>> {
|
||||
if let Err(e) = self.check_closed(cx) {
|
||||
return Poll::Ready(Err(e));
|
||||
}
|
||||
|
||||
Poll::Ready(Ok(()))
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,607 @@
|
||||
// Copyright 2024 Nym Technologies SA
|
||||
// SPDX-License-Identifier: Apache-2.0
|
||||
|
||||
//! NymTransport implementation of libp2p Transport trait for WASM.
|
||||
|
||||
use futures::channel::mpsc::{unbounded, UnboundedReceiver, UnboundedSender};
|
||||
use futures::channel::oneshot;
|
||||
use futures::prelude::*;
|
||||
use gloo_timers::future::TimeoutFuture;
|
||||
use libp2p::core::{
|
||||
multiaddr::Multiaddr,
|
||||
transport::{DialOpts, ListenerId, TransportError, TransportEvent},
|
||||
Transport,
|
||||
};
|
||||
use libp2p_identity::{Keypair, PeerId};
|
||||
use log::{debug, info};
|
||||
use nym_client_wasm::stream::NymClientStream;
|
||||
use nym_sphinx_addressing::clients::Recipient;
|
||||
use send_wrapper::SendWrapper;
|
||||
use std::{
|
||||
collections::HashMap,
|
||||
pin::Pin,
|
||||
str::FromStr,
|
||||
task::{Context, Poll, Waker},
|
||||
time::Duration,
|
||||
};
|
||||
|
||||
use super::connection::{Connection, PendingConnection};
|
||||
use super::error::Error;
|
||||
use super::message::{
|
||||
ConnectionId, ConnectionMessage, InboundMessage, Message, OutboundMessage, SubstreamMessage,
|
||||
TransportMessage,
|
||||
};
|
||||
use super::mixnet::initialize_mixnet;
|
||||
use super::queue::MessageQueue;
|
||||
use super::DEFAULT_HANDSHAKE_TIMEOUT_SECS;
|
||||
use nym_sphinx_anonymous_replies::requests::AnonymousSenderTag;
|
||||
|
||||
/// InboundTransportEvent represents an inbound event from the mixnet.
|
||||
pub enum InboundTransportEvent {
|
||||
ConnectionRequest(Upgrade),
|
||||
ConnectionResponse,
|
||||
TransportMessage,
|
||||
}
|
||||
|
||||
/// NymTransport implements the Transport trait using the Nym mixnet.
|
||||
pub struct NymTransport {
|
||||
/// our Nym address
|
||||
self_address: Recipient,
|
||||
pub(crate) listen_addr: Multiaddr,
|
||||
pub(crate) listener_id: ListenerId,
|
||||
|
||||
/// our libp2p keypair; currently not really used
|
||||
keypair: Keypair,
|
||||
|
||||
/// established connections -> channel which sends messages received from
|
||||
/// the mixnet to the corresponding Connection
|
||||
connections: HashMap<ConnectionId, UnboundedSender<SubstreamMessage>>,
|
||||
|
||||
/// outbound pending dials
|
||||
pending_dials: HashMap<ConnectionId, PendingConnection>,
|
||||
|
||||
/// connection message queues
|
||||
message_queues: HashMap<ConnectionId, MessageQueue>,
|
||||
|
||||
/// inbound mixnet messages
|
||||
inbound_rx: UnboundedReceiver<InboundMessage>,
|
||||
|
||||
/// outbound mixnet messages
|
||||
outbound_tx: UnboundedSender<OutboundMessage>,
|
||||
|
||||
/// inbound messages for Transport.poll()
|
||||
poll_rx: UnboundedReceiver<TransportEvent<Upgrade, Error>>,
|
||||
|
||||
/// outbound messages to Transport.poll()
|
||||
poll_tx: UnboundedSender<TransportEvent<Upgrade, Error>>,
|
||||
|
||||
waker: Option<Waker>,
|
||||
|
||||
/// Timeout for the [`Upgrade`] future (in milliseconds for WASM).
|
||||
handshake_timeout_ms: u32,
|
||||
}
|
||||
|
||||
impl NymTransport {
|
||||
/// Create a new NymTransport from a NymClientStream.
|
||||
///
|
||||
/// # Example
|
||||
/// ```ignore
|
||||
/// use nym_libp2p_wasm::{create_transport_client_async, NymTransport};
|
||||
/// use libp2p_identity::Keypair;
|
||||
///
|
||||
/// // Create the transport client
|
||||
/// let result = create_transport_client_async(None).await?;
|
||||
///
|
||||
/// // Create the transport
|
||||
/// let keypair = Keypair::generate_ed25519();
|
||||
/// let transport = NymTransport::new(
|
||||
/// result.self_address,
|
||||
/// result.stream,
|
||||
/// keypair,
|
||||
/// ).await?;
|
||||
/// ```
|
||||
pub async fn new(
|
||||
self_address: Recipient,
|
||||
stream: NymClientStream,
|
||||
keypair: Keypair,
|
||||
) -> Result<Self, Error> {
|
||||
Self::new_with_options(self_address, stream, keypair, None, None).await
|
||||
}
|
||||
|
||||
/// Create a new NymTransport with a custom timeout.
|
||||
pub async fn new_with_timeout(
|
||||
self_address: Recipient,
|
||||
stream: NymClientStream,
|
||||
keypair: Keypair,
|
||||
timeout: Duration,
|
||||
) -> Result<Self, Error> {
|
||||
Self::new_with_options(self_address, stream, keypair, None, Some(timeout)).await
|
||||
}
|
||||
|
||||
/// Add timeout to transport and return self.
|
||||
#[allow(dead_code)]
|
||||
pub fn with_timeout(mut self, timeout: Duration) -> Self {
|
||||
self.handshake_timeout_ms = timeout.as_millis() as u32;
|
||||
self
|
||||
}
|
||||
|
||||
async fn new_with_options(
|
||||
self_address: Recipient,
|
||||
stream: NymClientStream,
|
||||
keypair: Keypair,
|
||||
notify_inbound_tx: Option<UnboundedSender<()>>,
|
||||
timeout: Option<Duration>,
|
||||
) -> Result<Self, Error> {
|
||||
let (self_address, inbound_rx, outbound_tx) =
|
||||
initialize_mixnet(self_address, stream, notify_inbound_tx).await?;
|
||||
let listen_addr = nym_address_to_multiaddress(self_address)?;
|
||||
let listener_id = ListenerId::next();
|
||||
|
||||
let (poll_tx, poll_rx) = unbounded::<TransportEvent<Upgrade, Error>>();
|
||||
|
||||
poll_tx
|
||||
.unbounded_send(TransportEvent::NewAddress {
|
||||
listener_id,
|
||||
listen_addr: listen_addr.clone(),
|
||||
})
|
||||
.map_err(|_| Error::SendErrorTransportEvent)?;
|
||||
|
||||
let handshake_timeout_ms = timeout
|
||||
.map(|t| t.as_millis() as u32)
|
||||
.unwrap_or((DEFAULT_HANDSHAKE_TIMEOUT_SECS * 1000) as u32);
|
||||
|
||||
Ok(Self {
|
||||
self_address,
|
||||
listen_addr,
|
||||
listener_id,
|
||||
keypair,
|
||||
connections: HashMap::new(),
|
||||
pending_dials: HashMap::new(),
|
||||
message_queues: HashMap::new(),
|
||||
inbound_rx,
|
||||
outbound_tx,
|
||||
poll_rx,
|
||||
poll_tx,
|
||||
waker: None,
|
||||
handshake_timeout_ms,
|
||||
})
|
||||
}
|
||||
|
||||
pub(crate) fn peer_id(&self) -> PeerId {
|
||||
PeerId::from_public_key(&self.keypair.public())
|
||||
}
|
||||
|
||||
fn handle_message_queue_on_connection_initiation(
|
||||
&mut self,
|
||||
id: &ConnectionId,
|
||||
) -> Result<(), Error> {
|
||||
debug!("handle_message_queue_on_connection_initiation");
|
||||
let Some(inbound_tx) = self.connections.get(id) else {
|
||||
// this should not happen
|
||||
return Err(Error::NoConnectionForTransportMessage);
|
||||
};
|
||||
|
||||
match self.message_queues.get_mut(id) {
|
||||
Some(queue) => {
|
||||
// update expected nonce
|
||||
queue.set_connection_message_received();
|
||||
|
||||
// push pending inbound some messages in this case
|
||||
while let Some(msg) = queue.pop() {
|
||||
debug!(
|
||||
"popped queued message with nonce {} for connection",
|
||||
msg.nonce
|
||||
);
|
||||
inbound_tx
|
||||
.unbounded_send(msg.message.clone())
|
||||
.map_err(|e| Error::InboundSendFailure(e.to_string()))?;
|
||||
}
|
||||
}
|
||||
None => {
|
||||
// no queue exists for this connection, create one
|
||||
let queue = MessageQueue::new();
|
||||
self.message_queues.insert(id.clone(), queue);
|
||||
let queue = self.message_queues.get_mut(id).unwrap();
|
||||
queue.set_connection_message_received();
|
||||
}
|
||||
};
|
||||
|
||||
debug!("returning from handle_message_queue_on_connection_initiation");
|
||||
Ok(())
|
||||
}
|
||||
|
||||
// handle_connection_response resolves the pending connection corresponding to the response
|
||||
// (if there is one) into a Connection.
|
||||
fn handle_connection_response(
|
||||
&mut self,
|
||||
msg: &ConnectionMessage,
|
||||
sender_tag: Option<AnonymousSenderTag>,
|
||||
) -> Result<(), Error> {
|
||||
if self.connections.contains_key(&msg.id) {
|
||||
return Err(Error::ConnectionAlreadyEstablished);
|
||||
}
|
||||
|
||||
if let Some(pending_conn) = self.pending_dials.remove(&msg.id) {
|
||||
// Create connection with sender_tag
|
||||
let (conn, conn_tx) = self.create_connection_types(
|
||||
msg.peer_id,
|
||||
Some(pending_conn.remote_recipient), // Dialer knows recipient
|
||||
msg.id.clone(),
|
||||
sender_tag,
|
||||
);
|
||||
|
||||
self.connections.insert(msg.id.clone(), conn_tx);
|
||||
self.handle_message_queue_on_connection_initiation(&msg.id)?;
|
||||
|
||||
pending_conn
|
||||
.connection_tx
|
||||
.send(conn)
|
||||
.map_err(|_| Error::ConnectionSendFailure)?;
|
||||
|
||||
if let Some(waker) = self.waker.take() {
|
||||
waker.wake();
|
||||
}
|
||||
|
||||
Ok(())
|
||||
} else {
|
||||
Err(Error::NoConnectionForResponse)
|
||||
}
|
||||
}
|
||||
|
||||
/// handle_connection_request handles an incoming connection request, sends back a
|
||||
/// connection response, and finally completes the upgrade into a Connection.
|
||||
fn handle_connection_request(
|
||||
&mut self,
|
||||
msg: &ConnectionMessage,
|
||||
sender_tag: Option<AnonymousSenderTag>,
|
||||
) -> Result<Connection, Error> {
|
||||
// ensure we don't already have a conn with the same id
|
||||
if self.connections.contains_key(&msg.id) {
|
||||
return Err(Error::ConnectionIDExists);
|
||||
}
|
||||
|
||||
// Create connection with sender_tag
|
||||
let (conn, conn_tx) = self.create_connection_types(
|
||||
msg.peer_id,
|
||||
None, // Receiver doesn't know dialer address
|
||||
msg.id.clone(),
|
||||
sender_tag.clone(),
|
||||
);
|
||||
|
||||
info!("Created connection: {:?}", conn);
|
||||
|
||||
self.connections.insert(msg.id.clone(), conn_tx);
|
||||
info!("Current active connections: {}", self.connections.len());
|
||||
|
||||
self.handle_message_queue_on_connection_initiation(&msg.id)?;
|
||||
|
||||
let resp = ConnectionMessage {
|
||||
peer_id: self.peer_id(),
|
||||
id: msg.id.clone(),
|
||||
};
|
||||
|
||||
// Send response using sender_tag if available
|
||||
self.outbound_tx
|
||||
.unbounded_send(OutboundMessage {
|
||||
message: Message::ConnectionResponse(resp),
|
||||
recipient: None,
|
||||
sender_tag,
|
||||
})
|
||||
.map_err(|e| Error::OutboundSendFailure(e.to_string()))?;
|
||||
|
||||
debug!(
|
||||
"Sent ConnectionResponse with sender_tag: {:?}",
|
||||
sender_tag.is_some()
|
||||
);
|
||||
if let Some(waker) = self.waker.take() {
|
||||
waker.wake();
|
||||
}
|
||||
|
||||
Ok(conn)
|
||||
}
|
||||
|
||||
fn handle_transport_message(&mut self, msg: TransportMessage) -> Result<(), Error> {
|
||||
let queue = match self.message_queues.get_mut(&msg.id) {
|
||||
Some(queue) => queue,
|
||||
None => {
|
||||
// no queue exists for this connection, create one
|
||||
let queue = MessageQueue::new();
|
||||
self.message_queues.insert(msg.id.clone(), queue);
|
||||
self.message_queues.get_mut(&msg.id).unwrap()
|
||||
}
|
||||
};
|
||||
|
||||
queue.print_nonces();
|
||||
|
||||
let nonce = msg.nonce;
|
||||
let Some(msg) = queue.try_push(msg) else {
|
||||
// don't push the message yet, it's been queued
|
||||
debug!("message with nonce {} queued for connection", nonce);
|
||||
return Ok(());
|
||||
};
|
||||
|
||||
let Some(inbound_tx) = self.connections.get(&msg.id) else {
|
||||
return Err(Error::NoConnectionForTransportMessage);
|
||||
};
|
||||
|
||||
// send original message
|
||||
debug!(
|
||||
"sending original message with nonce {} for connection",
|
||||
nonce
|
||||
);
|
||||
inbound_tx
|
||||
.unbounded_send(msg.message.clone())
|
||||
.map_err(|e| Error::InboundSendFailure(e.to_string()))?;
|
||||
|
||||
// try to pop queued messages and send them on inbound channel
|
||||
while let Some(msg) = queue.pop() {
|
||||
debug!(
|
||||
"popped queued message with nonce {} for connection",
|
||||
msg.nonce
|
||||
);
|
||||
inbound_tx
|
||||
.unbounded_send(msg.message.clone())
|
||||
.map_err(|e| Error::InboundSendFailure(e.to_string()))?;
|
||||
}
|
||||
|
||||
if let Some(waker) = self.waker.clone().take() {
|
||||
waker.wake();
|
||||
}
|
||||
|
||||
Ok(())
|
||||
}
|
||||
|
||||
fn create_connection_types(
|
||||
&self,
|
||||
remote_peer_id: PeerId,
|
||||
remote_recipient: Option<Recipient>,
|
||||
id: ConnectionId,
|
||||
sender_tag: Option<AnonymousSenderTag>,
|
||||
) -> (Connection, UnboundedSender<SubstreamMessage>) {
|
||||
let (inbound_tx, inbound_rx) = unbounded::<SubstreamMessage>();
|
||||
|
||||
let conn = Connection::new_with_sender_tag(
|
||||
remote_peer_id,
|
||||
remote_recipient,
|
||||
id,
|
||||
inbound_rx,
|
||||
self.outbound_tx.clone(),
|
||||
sender_tag,
|
||||
);
|
||||
|
||||
(conn, inbound_tx)
|
||||
}
|
||||
|
||||
/// handle_inbound handles an inbound message from the mixnet, received via self.inbound_rx.
|
||||
fn handle_inbound(
|
||||
&mut self,
|
||||
msg: Message,
|
||||
sender_tag: Option<AnonymousSenderTag>,
|
||||
) -> Result<InboundTransportEvent, Error> {
|
||||
match msg {
|
||||
Message::ConnectionRequest(inner) => {
|
||||
debug!("got inbound connection request {:?}", inner);
|
||||
match self.handle_connection_request(&inner, sender_tag) {
|
||||
Ok(conn) => {
|
||||
let (connection_tx, connection_rx) =
|
||||
oneshot::channel::<(PeerId, Connection)>();
|
||||
let upgrade = Upgrade::new(connection_rx);
|
||||
connection_tx
|
||||
.send((inner.peer_id, conn))
|
||||
.map_err(|_| Error::ConnectionSendFailure)?;
|
||||
Ok(InboundTransportEvent::ConnectionRequest(upgrade))
|
||||
}
|
||||
Err(e) => Err(e),
|
||||
}
|
||||
}
|
||||
Message::ConnectionResponse(msg) => {
|
||||
debug!("got inbound connection response {:?}", msg);
|
||||
self.handle_connection_response(&msg, sender_tag)
|
||||
.map(|_| InboundTransportEvent::ConnectionResponse)
|
||||
}
|
||||
Message::TransportMessage(msg) => {
|
||||
debug!(
|
||||
"Transport received TransportMessage: nonce={}, substream={:?}, msg_type={:?}",
|
||||
msg.nonce, msg.message.substream_id, msg.message.message_type
|
||||
);
|
||||
self.handle_transport_message(msg)
|
||||
.map(|_| InboundTransportEvent::TransportMessage)
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
/// Upgrade represents a transport listener upgrade.
|
||||
/// Note: we immediately upgrade a connection request to a connection,
|
||||
/// so this only contains a channel for receiving that connection.
|
||||
pub struct Upgrade {
|
||||
connection_rx: oneshot::Receiver<(PeerId, Connection)>,
|
||||
}
|
||||
|
||||
impl Upgrade {
|
||||
fn new(connection_rx: oneshot::Receiver<(PeerId, Connection)>) -> Upgrade {
|
||||
Upgrade { connection_rx }
|
||||
}
|
||||
}
|
||||
|
||||
impl Future for Upgrade {
|
||||
type Output = Result<(PeerId, Connection), Error>;
|
||||
|
||||
// poll checks if the upgrade has turned into a connection yet
|
||||
fn poll(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
|
||||
match self.connection_rx.poll_unpin(cx) {
|
||||
Poll::Ready(Ok(conn)) => Poll::Ready(Ok(conn)),
|
||||
Poll::Ready(Err(_)) => Poll::Ready(Err(Error::RecvFailure)),
|
||||
Poll::Pending => Poll::Pending,
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
impl Transport for NymTransport {
|
||||
type Output = (PeerId, Connection);
|
||||
type Error = Error;
|
||||
type ListenerUpgrade = Upgrade;
|
||||
// Use SendWrapper to make the future Send for libp2p's SwarmBuilder
|
||||
// This is safe in WASM's single-threaded environment
|
||||
type Dial = futures::future::BoxFuture<'static, Result<Self::Output, Self::Error>>;
|
||||
|
||||
fn listen_on(
|
||||
&mut self,
|
||||
_listener_id: ListenerId,
|
||||
_multi_addr: libp2p::Multiaddr,
|
||||
) -> Result<(), TransportError<Self::Error>> {
|
||||
info!("called listen_on, this is currently just a dummy function - client starts listening on new()");
|
||||
Ok(())
|
||||
}
|
||||
|
||||
fn remove_listener(&mut self, id: ListenerId) -> bool {
|
||||
if self.listener_id != id {
|
||||
return false;
|
||||
}
|
||||
|
||||
let _ = self.poll_tx.unbounded_send(TransportEvent::ListenerClosed {
|
||||
listener_id: id,
|
||||
reason: Ok(()),
|
||||
});
|
||||
true
|
||||
}
|
||||
|
||||
fn dial(
|
||||
&mut self,
|
||||
addr: Multiaddr,
|
||||
_dial_opts: DialOpts,
|
||||
) -> Result<Self::Dial, TransportError<Self::Error>> {
|
||||
debug!("dialing {}", addr);
|
||||
|
||||
let id = ConnectionId::generate();
|
||||
|
||||
// create remote recipient address
|
||||
let recipient = multiaddress_to_nym_address(addr).map_err(TransportError::Other)?;
|
||||
|
||||
// create pending conn structs and store
|
||||
let (connection_tx, connection_rx) = oneshot::channel::<Connection>();
|
||||
|
||||
let inner_pending_conn = PendingConnection::new(recipient, connection_tx);
|
||||
self.pending_dials.insert(id.clone(), inner_pending_conn);
|
||||
|
||||
let local_key = Keypair::generate_ed25519();
|
||||
let connection_peer_id = PeerId::from(local_key.public());
|
||||
|
||||
// put ConnectionRequest message into outbound message channel
|
||||
let msg = ConnectionMessage {
|
||||
peer_id: connection_peer_id,
|
||||
id,
|
||||
};
|
||||
|
||||
let outbound_tx = self.outbound_tx.clone();
|
||||
|
||||
let mut waker = self.waker.clone();
|
||||
let handshake_timeout_ms = self.handshake_timeout_ms;
|
||||
|
||||
// Wrap in SendWrapper to satisfy Send bounds for SwarmBuilder
|
||||
// This is safe because WASM is single-threaded
|
||||
Ok(SendWrapper::new(async move {
|
||||
outbound_tx
|
||||
.unbounded_send(OutboundMessage {
|
||||
message: Message::ConnectionRequest(msg),
|
||||
recipient: Some(recipient),
|
||||
sender_tag: None,
|
||||
})
|
||||
.map_err(|e| Error::OutboundSendFailure(e.to_string()))?;
|
||||
|
||||
debug!("sent outbound ConnectionRequest");
|
||||
if let Some(waker) = waker.take() {
|
||||
waker.wake();
|
||||
};
|
||||
|
||||
// Use gloo_timers for WASM-compatible timeout
|
||||
let timeout_future = TimeoutFuture::new(handshake_timeout_ms);
|
||||
|
||||
futures::select! {
|
||||
conn_result = connection_rx.fuse() => {
|
||||
match conn_result {
|
||||
Ok(conn) => Ok((conn.peer_id, conn)),
|
||||
Err(_) => Err(Error::RecvFailure),
|
||||
}
|
||||
}
|
||||
_ = timeout_future.fuse() => {
|
||||
Err(Error::DialTimeout)
|
||||
}
|
||||
}
|
||||
})
|
||||
.boxed())
|
||||
}
|
||||
|
||||
fn poll(
|
||||
mut self: Pin<&mut Self>,
|
||||
cx: &mut Context<'_>,
|
||||
) -> Poll<TransportEvent<Self::ListenerUpgrade, Self::Error>> {
|
||||
// new addresses + listener close events
|
||||
if let Poll::Ready(Some(res)) = self.poll_rx.poll_next_unpin(cx) {
|
||||
return Poll::Ready(res);
|
||||
}
|
||||
|
||||
// check for and handle inbound messages
|
||||
while let Poll::Ready(Some(msg)) = self.inbound_rx.poll_next_unpin(cx) {
|
||||
debug!(
|
||||
"TRANSPORT: Received inbound message type: {:?}",
|
||||
match &msg.0 {
|
||||
Message::ConnectionRequest(_) => "ConnectionRequest",
|
||||
Message::ConnectionResponse(_) => "ConnectionResponse",
|
||||
Message::TransportMessage(_) => "TransportMessage",
|
||||
}
|
||||
);
|
||||
|
||||
match self.handle_inbound(msg.0, msg.1) {
|
||||
Ok(event) => match event {
|
||||
InboundTransportEvent::ConnectionRequest(upgrade) => {
|
||||
info!("InboundTransportEvent::ConnectionRequest");
|
||||
return Poll::Ready(TransportEvent::Incoming {
|
||||
listener_id: self.listener_id,
|
||||
upgrade,
|
||||
local_addr: self.listen_addr.clone(),
|
||||
send_back_addr: self.listen_addr.clone(),
|
||||
});
|
||||
}
|
||||
InboundTransportEvent::ConnectionResponse => {
|
||||
info!("InboundTransportEvent::ConnectionResponse");
|
||||
}
|
||||
InboundTransportEvent::TransportMessage => {
|
||||
debug!("InboundTransportEvent::TransportMessage");
|
||||
}
|
||||
},
|
||||
Err(e) => {
|
||||
return Poll::Ready(TransportEvent::ListenerError {
|
||||
listener_id: self.listener_id,
|
||||
error: e,
|
||||
});
|
||||
}
|
||||
};
|
||||
}
|
||||
|
||||
self.waker = Some(cx.waker().clone());
|
||||
Poll::Pending
|
||||
}
|
||||
}
|
||||
|
||||
/// Convert a Nym Recipient address to a libp2p Multiaddr.
|
||||
///
|
||||
/// Format: `/nym/<base58-encoded-nym-address>`
|
||||
pub fn nym_address_to_multiaddress(addr: Recipient) -> Result<Multiaddr, Error> {
|
||||
// Create a multiaddr using the Nym protocol
|
||||
// Format: /nym/<base58-encoded-nym-address>
|
||||
// This requires the ChainSafe multiaddr fork with Protocol::Nym support
|
||||
Multiaddr::from_str(&format!("/nym/{}", addr)).map_err(Error::FailedToFormatMultiaddr)
|
||||
}
|
||||
|
||||
fn multiaddress_to_nym_address(multiaddr: Multiaddr) -> Result<Recipient, Error> {
|
||||
// Parse the Nym address from the multiaddr
|
||||
// We expect format: /nym/<nym-address>
|
||||
let addr_str = multiaddr.to_string();
|
||||
|
||||
if let Some(nym_addr) = addr_str.strip_prefix("/nym/") {
|
||||
return Recipient::from_str(nym_addr).map_err(Error::InvalidRecipientBytes);
|
||||
}
|
||||
|
||||
Err(Error::InvalidProtocolForMultiaddr)
|
||||
}
|
||||
Reference in New Issue
Block a user