From 8e6ceddc66cd9da3f3bd6b8a31e70b3edd771635 Mon Sep 17 00:00:00 2001 From: mfahampshire Date: Mon, 9 Feb 2026 12:13:27 +0000 Subject: [PATCH] Port native reference implementation to WASM compatible state --- Cargo.lock | 524 +++++++++++++++++++++++++- Cargo.toml | 6 + wasm/libp2p-nym/Cargo.toml | 62 +++ wasm/libp2p-nym/README.md | 48 +++ wasm/libp2p-nym/src/client.rs | 104 +++++ wasm/libp2p-nym/src/connection.rs | 367 ++++++++++++++++++ wasm/libp2p-nym/src/error.rs | 67 ++++ wasm/libp2p-nym/src/lib.rs | 124 ++++++ wasm/libp2p-nym/src/message.rs | 298 +++++++++++++++ wasm/libp2p-nym/src/mixnet.rs | 200 ++++++++++ wasm/libp2p-nym/src/queue.rs | 144 +++++++ wasm/libp2p-nym/src/substream.rs | 254 +++++++++++++ wasm/libp2p-nym/src/transport.rs | 607 ++++++++++++++++++++++++++++++ 13 files changed, 2802 insertions(+), 3 deletions(-) create mode 100644 wasm/libp2p-nym/Cargo.toml create mode 100644 wasm/libp2p-nym/README.md create mode 100644 wasm/libp2p-nym/src/client.rs create mode 100644 wasm/libp2p-nym/src/connection.rs create mode 100644 wasm/libp2p-nym/src/error.rs create mode 100644 wasm/libp2p-nym/src/lib.rs create mode 100644 wasm/libp2p-nym/src/message.rs create mode 100644 wasm/libp2p-nym/src/mixnet.rs create mode 100644 wasm/libp2p-nym/src/queue.rs create mode 100644 wasm/libp2p-nym/src/substream.rs create mode 100644 wasm/libp2p-nym/src/transport.rs diff --git a/Cargo.lock b/Cargo.lock index b2786d4974..9a1ed9a538 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -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" diff --git a/Cargo.toml b/Cargo.toml index a9a240b504..0875cd1bcb 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -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 diff --git a/wasm/libp2p-nym/Cargo.toml b/wasm/libp2p-nym/Cargo.toml new file mode 100644 index 0000000000..d58533e4a5 --- /dev/null +++ b/wasm/libp2p-nym/Cargo.toml @@ -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 diff --git a/wasm/libp2p-nym/README.md b/wasm/libp2p-nym/README.md new file mode 100644 index 0000000000..d49fb5cc1b --- /dev/null +++ b/wasm/libp2p-nym/README.md @@ -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 | diff --git a/wasm/libp2p-nym/src/client.rs b/wasm/libp2p-nym/src/client.rs new file mode 100644 index 0000000000..59e22aca64 --- /dev/null +++ b/wasm/libp2p-nym/src/client.rs @@ -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, + /// 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, +} + +/// 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 { + 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, + }) +} diff --git a/wasm/libp2p-nym/src/connection.rs b/wasm/libp2p-nym/src/connection.rs new file mode 100644 index 0000000000..5828be4d8c --- /dev/null +++ b/wasm/libp2p-nym/src/connection.rs @@ -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, + pub(crate) id: ConnectionId, + + /// receive inbound messages from the `InnerConnection` + pub(crate) inbound_rx: UnboundedReceiver, + + /// substream ID -> outbound pending substream exists + /// the key is deleted when the response is received, or the request times out + pending_substreams: HashSet, + + /// substream ID -> substream's inbound_tx channel + substream_inbound_txs: HashMap>>, + + /// substream ID -> substream's close_tx channel + substream_close_txs: HashMap>, + + /// 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, + + /// sender_tag for SURB replies to incoming messages + pub(crate) sender_tag: Option, + + /// inbound substream open requests; used in poll_inbound + inbound_open_tx: UnboundedSender, + inbound_open_rx: UnboundedReceiver, + + /// closed substream IDs; used in poll_close + close_tx: UnboundedSender, + close_rx: UnboundedReceiver, + + /// message nonce contains the next nonce that should be used when + /// sending a message over the connection + pub(crate) message_nonce: Arc, + + waker: Option, +} + +impl Connection { + pub(crate) fn new_with_sender_tag( + peer_id: PeerId, + remote_recipient: Option, + id: ConnectionId, + inbound_rx: UnboundedReceiver, + mixnet_outbound_tx: UnboundedSender, + sender_tag: Option, + ) -> 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 { + 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 { + 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 { + // 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::>(); + 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> { + 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> { + 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> { + 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> { + 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, +} + +impl PendingConnection { + pub(crate) fn new( + remote_recipient: Recipient, + connection_tx: oneshot::Sender, + ) -> Self { + PendingConnection { + remote_recipient, + connection_tx, + } + } +} diff --git a/wasm/libp2p-nym/src/error.rs b/wasm/libp2p-nym/src/error.rs new file mode 100644 index 0000000000..2eb63cdb55 --- /dev/null +++ b/wasm/libp2p-nym/src/error.rs @@ -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), +} diff --git a/wasm/libp2p-nym/src/lib.rs b/wasm/libp2p-nym/src/lib.rs new file mode 100644 index 0000000000..0dbb3e26b3 --- /dev/null +++ b/wasm/libp2p-nym/src/lib.rs @@ -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, + force_tls: Option, + client_id: Option, + ) -> 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)) + }) + } +} diff --git a/wasm/libp2p-nym/src/message.rs b/wasm/libp2p-nym/src/message.rs new file mode 100644 index 0000000000..b52e0aaa02 --- /dev/null +++ b/wasm/libp2p-nym/src/message.rs @@ -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() -> [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) -> Result { + 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 { + let mut bytes = self.id.0.to_vec(); + bytes.append(&mut self.peer_id.to_bytes()); + bytes + } + + fn try_from_bytes(bytes: &[u8]) -> Result { + 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 { + 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 { + 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 { + 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), +} + +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) -> 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 { + 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 { + 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 { + 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); + +/// OutboundMessage represents an outbound mixnet message. +#[derive(Debug)] +pub(crate) struct OutboundMessage { + pub(crate) message: Message, + pub(crate) recipient: Option, + pub(crate) sender_tag: Option, +} + +pub(crate) fn parse_message_data( + data: &[u8], + sender_tag: Option, +) -> Result { + if data.len() < 2 { + return Err(Error::InvalidMessageBytes); + } + let msg = Message::try_from_bytes(data.to_vec())?; + Ok(InboundMessage(msg, sender_tag)) +} diff --git a/wasm/libp2p-nym/src/mixnet.rs b/wasm/libp2p-nym/src/mixnet.rs new file mode 100644 index 0000000000..0e9c11a5b9 --- /dev/null +++ b/wasm/libp2p-nym/src/mixnet.rs @@ -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>, +) -> Result< + ( + Recipient, + UnboundedReceiver, + UnboundedSender, + ), + Error, +> { + // Channel for inbound messages from the mixnet to the transport + let (inbound_tx, inbound_rx) = unbounded::(); + + // Channel for outbound messages from the transport to the mixnet + let (outbound_tx, outbound_rx) = unbounded::(); + + // Get client_input for sending before we move stream + let client_input = stream.client_input(); + + // Wrap stream in Arc 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>, + mut outbound_rx: UnboundedReceiver, +) { + 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>, + 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>, + inbound_tx: UnboundedSender, + notify_inbound_tx: Option>, +) { + 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"); +} diff --git a/wasm/libp2p-nym/src/queue.rs b/wasm/libp2p-nym/src/queue.rs new file mode 100644 index 0000000000..c1c88a002e --- /dev/null +++ b/wasm/libp2p-nym/src/queue.rs @@ -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, +} + +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::>(); + 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 { + 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 { + 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); + } +} diff --git a/wasm/libp2p-nym/src/substream.rs b/wasm/libp2p-nym/src/substream.rs new file mode 100644 index 0000000000..6353adff28 --- /dev/null +++ b/wasm/libp2p-nym/src/substream.rs @@ -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, + connection_id: ConnectionId, + pub(crate) substream_id: SubstreamId, + + /// inbound messages; inbound_tx is in the corresponding Connection + pub(crate) inbound_rx: UnboundedReceiver>, + + /// outbound messages; go directly to the mixnet + outbound_tx: UnboundedSender, + + sender_tag: Option, + + /// used to signal when the substream is closed + close_rx: Receiver<()>, + closed: Mutex, + + // buffer of data that's been written to the stream, + // but not yet read by the application. + unread_data: Mutex>, + + message_nonce: Arc, +} + +impl Substream { + pub(crate) fn new_with_sender_tag( + remote_recipient: Option, + connection_id: ConnectionId, + substream_id: SubstreamId, + inbound_rx: UnboundedReceiver>, + outbound_tx: UnboundedSender, + close_rx: Receiver<()>, + message_nonce: Arc, + sender_tag: Option, + ) -> 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, + connection_id: ConnectionId, + substream_id: SubstreamId, + inbound_rx: UnboundedReceiver>, + outbound_tx: UnboundedSender, + close_rx: Receiver<()>, + message_nonce: Arc, + ) -> 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> { + 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> { + 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> { + 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> { + if let Err(e) = self.check_closed(cx) { + return Poll::Ready(Err(e)); + } + + Poll::Ready(Ok(())) + } +} diff --git a/wasm/libp2p-nym/src/transport.rs b/wasm/libp2p-nym/src/transport.rs new file mode 100644 index 0000000000..75bfc5022e --- /dev/null +++ b/wasm/libp2p-nym/src/transport.rs @@ -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>, + + /// outbound pending dials + pending_dials: HashMap, + + /// connection message queues + message_queues: HashMap, + + /// inbound mixnet messages + inbound_rx: UnboundedReceiver, + + /// outbound mixnet messages + outbound_tx: UnboundedSender, + + /// inbound messages for Transport.poll() + poll_rx: UnboundedReceiver>, + + /// outbound messages to Transport.poll() + poll_tx: UnboundedSender>, + + waker: Option, + + /// 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::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::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>, + timeout: Option, + ) -> Result { + 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::>(); + + 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, + ) -> 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, + ) -> Result { + // 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, + id: ConnectionId, + sender_tag: Option, + ) -> (Connection, UnboundedSender) { + let (inbound_tx, inbound_rx) = unbounded::(); + + 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, + ) -> Result { + 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 { + 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>; + + fn listen_on( + &mut self, + _listener_id: ListenerId, + _multi_addr: libp2p::Multiaddr, + ) -> Result<(), TransportError> { + 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> { + 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::(); + + 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> { + // 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/` +pub fn nym_address_to_multiaddress(addr: Recipient) -> Result { + // Create a multiaddr using the Nym protocol + // Format: /nym/ + // 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 { + // Parse the Nym address from the multiaddr + // We expect format: /nym/ + 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) +}