diff --git a/Cargo.lock b/Cargo.lock index 21d0bba..bab4b38 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -45,6 +45,18 @@ version = "1.0.104" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "330a5ed07fa54e4702c9d6c4174f74427fc0ef6e214bbd677ae50a5099946470" +[[package]] +name = "argon2" +version = "0.5.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "3c3610892ee6e0cbce8ae2700349fcf8f98adb0dbfbee85aec3c9179d29cc072" +dependencies = [ + "base64ct", + "blake2", + "cpufeatures 0.2.17", + "password-hash", +] + [[package]] name = "asn1-rs" version = "0.7.2" @@ -57,7 +69,7 @@ dependencies = [ "nom", "num-traits", "rusticata-macros", - "thiserror 2.0.19", + "thiserror 2.0.20", "time", ] @@ -86,9 +98,9 @@ dependencies = [ [[package]] name = "async-trait" -version = "0.1.91" +version = "0.1.92" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "ae36dc4177970ef04fde5178d3e2429882def40e57a451f919c098f72baa6cec" +checksum = "82f6aeea286b8eb4dd3431a1be1b59d290ace00f5bfd8e2a159bc2a05e2c1667" dependencies = [ "proc-macro2", "quote", @@ -170,6 +182,15 @@ version = "2.13.1" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "b588b76d00fde79687d7646a9b5bdf3cc0f655e0bbd080335a95d7e96f3587da" +[[package]] +name = "blake2" +version = "0.10.6" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "46502ad458c9a52b69d4d4d32775c788b7a1b85e8bc9d482d92250fc0e3f8efe" +dependencies = [ + "digest 0.10.7", +] + [[package]] name = "block-buffer" version = "0.10.4" @@ -217,9 +238,9 @@ checksum = "fc652a48c352aef3ea3aed32080501cf3ef6ed5da78602a020c991775b0aff04" [[package]] name = "cc" -version = "1.4.1" +version = "1.4.3" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "9066c49992464636f92905fa096ec58baaa4d57ec19a5c096c68d3e25ef3d136" +checksum = "509591b7bcd67f4ef775afad7662703b4935daaa6ec0e5605cfb1090b32a2b6d" dependencies = [ "find-msvc-tools", "jobserver", @@ -504,7 +525,7 @@ dependencies = [ "core-foundation", "jni 0.21.1", "libc", - "thiserror 2.0.19", + "thiserror 2.0.20", "wasm-bindgen", "web-sys", "windows-sys 0.59.0", @@ -622,11 +643,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "39cab71617ae0d63f51a36d69f866391735b51691dbda63cf6f96d042b63efeb" dependencies = [ "libc", -<<<<<<< HEAD "windows-sys 0.61.2", -======= - "windows-sys 0.52.0", ->>>>>>> e0dd9a085fd88168e714af0d6af88769af3aa644 ] [[package]] @@ -661,9 +678,9 @@ checksum = "64cd1e32ddd350061ae6edb1b082d7c54915b5c672c389143b9a63403a109f24" [[package]] name = "find-msvc-tools" -version = "0.1.10" +version = "0.1.11" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "26b73573e6edcd2af0cdf47bd6cb58f0b3839491263c314eaad1ccf24430e1de" +checksum = "d45db016d36b838f563236e9193d0ee6ce38f3f68b6c94e914b4929c96bbb890" [[package]] name = "fixedbitset" @@ -700,9 +717,9 @@ checksum = "42703706b716c37f96a77aea830392ad231f44c9e9a67872fa5548707e11b11c" [[package]] name = "futures" -version = "0.3.33" +version = "0.3.34" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "a88cf1f829d945f548cf8fec32c61b1f202b6d93b45848602fc02af4b12ad218" +checksum = "9a31d2a3fbaaeb2af2368bbdd904aa8e812d3c04a1ee10d3171f52d556e5d0a3" dependencies = [ "futures-channel", "futures-core", @@ -715,9 +732,9 @@ dependencies = [ [[package]] name = "futures-channel" -version = "0.3.33" +version = "0.3.34" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "262590f4fe6afeb0bc83be1daa64e52657fe185690a958af7f3ad0e92085c5ae" +checksum = "b1f9e3d69d39e4862ffed03ed071a76f9a13ba1d9109d355b0f0aa6b15e393c4" dependencies = [ "futures-core", "futures-sink", @@ -725,15 +742,15 @@ dependencies = [ [[package]] name = "futures-core" -version = "0.3.33" +version = "0.3.34" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "2cd50c473c80f6d7c3670a752354b8e569b1a7cbfdc0419ec88e5edad85e0dc7" +checksum = "92d699e522242e69e3003b94ecc1f960f3a5e015aa7c5d7486e65ad01dd94f5e" [[package]] name = "futures-executor" -version = "0.3.33" +version = "0.3.34" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "6754879cc9f2c66f88c6e5c35344bb0bdb0708b0352b1201815667c7eabc7458" +checksum = "031b47cf1a3c6cc8bc2fc76cd437f521619387907d469316e7c0bc278f1f5432" dependencies = [ "futures-core", "futures-task", @@ -742,38 +759,38 @@ dependencies = [ [[package]] name = "futures-io" -version = "0.3.33" +version = "0.3.34" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "4577ecaa3c4f96589d473f679a71b596316f6641bc350038b962a5daf0085d7a" +checksum = "53c0fa8157de1303bfffdaa1cc2a673bfffb60102f76b0ef4441659124373fed" [[package]] name = "futures-macro" -version = "0.3.33" +version = "0.3.34" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "2d6d3cde68c518367be28956066ddfef33813991b77a55005a69dae04bf3b10b" +checksum = "9fb9654ba8355388abeb8dcb4fc62f511300867002afc858860463bdd9fe0c44" dependencies = [ "proc-macro2", "quote", - "syn 2.0.119", + "syn 3.0.3", ] [[package]] name = "futures-sink" -version = "0.3.33" +version = "0.3.34" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "e34418ac499d6305c2fb5ad0ed2f6ac998c5f8ca209b4510f7f94242c647e307" +checksum = "1944426bf7d03f1d14f708785e4b33efd750b36d48a157b836b3efc15ede8e1d" [[package]] name = "futures-task" -version = "0.3.33" +version = "0.3.34" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "b231ed28831efb4a61a08580c4bc233ec56bc009f4cd8f52da2c3cb97df0c109" +checksum = "cd417de3d1d015fc3bfd2b1ea46dfc7bab72ef86f1cc7cc9c78e728b34a6d1fd" [[package]] name = "futures-util" -version = "0.3.33" +version = "0.3.34" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "a77a90a256fce34da66415271e30f94ee91c57b04b8a2c042d9cf3220179deaa" +checksum = "0d50a92467f8ba5dd6e3ee5d4bd04d73ab2e4e1c44474a0674821dfce14b79bc" dependencies = [ "futures-channel", "futures-core", @@ -837,9 +854,9 @@ dependencies = [ [[package]] name = "h2" -version = "0.4.15" +version = "0.4.16" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "6cb093c84e8bd9b188d4c4a8cb6579fc016968d14c99882163cd3ff402a4f155" +checksum = "a9f37a958b41b3b19ee2707c06439c0e9e547e847223eb791ecb0cb821c65e27" dependencies = [ "atomic-waker", "bytes", @@ -989,9 +1006,9 @@ dependencies = [ [[package]] name = "http-body-util" -version = "0.1.4" +version = "0.1.5" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "e9f41fd6a08e4d4ec69df65976da761afd5ad5e58a9d4acb46bd1c953a9e3ff2" +checksum = "23169fe34a5fbcdd3f3862e78fb9b6fccd5f02a6dc6f732547005d45631ce71c" dependencies = [ "bytes", "futures-core", @@ -1085,9 +1102,9 @@ dependencies = [ [[package]] name = "icu_collections" -version = "2.2.0" +version = "2.3.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "2984d1cd16c883d7935b9e07e44071dca8d917fd52ecc02c04d5fa0b5a3f191c" +checksum = "fa68d21081c4a05d5a901a1c62add574c77048b6a1c67be3b50ce0b60d4ca513" dependencies = [ "displaydoc", "potential_utf", @@ -1099,9 +1116,9 @@ dependencies = [ [[package]] name = "icu_locale_core" -version = "2.2.0" +version = "2.3.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "92219b62b3e2b4d88ac5119f8904c10f8f61bf7e95b640d25ba3075e6cac2c29" +checksum = "d56e28588da92eee5c3201a6eff33fabdd49b62269c8938d4ff050ce4d900deb" dependencies = [ "displaydoc", "litemap", @@ -1112,9 +1129,9 @@ dependencies = [ [[package]] name = "icu_normalizer" -version = "2.2.0" +version = "2.3.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "c56e5ee99d6e3d33bd91c5d85458b6005a22140021cc324cea84dd0e72cff3b4" +checksum = "12f9cf5f235641ed274641dd81c3f28d870e276763d0797aeeab72317b1c646f" dependencies = [ "icu_collections", "icu_normalizer_data", @@ -1126,16 +1143,17 @@ dependencies = [ [[package]] name = "icu_normalizer_data" -version = "2.2.0" +version = "2.3.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "da3be0ae77ea334f4da67c12f149704f19f81d1adf7c51cf482943e84a2bad38" +checksum = "1563da1ed3e0b3bf3d74c9b85917ac9c56464d2f57242270c09c9e752f8021a0" [[package]] name = "icu_properties" -version = "2.2.0" +version = "2.3.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "bee3b67d0ea5c2cca5003417989af8996f8604e34fb9ddf96208a033901e70de" +checksum = "7e7ca276ad3145661a65914e6daf131ca5120cd3dcee8f8f3214b8875184a148" dependencies = [ + "displaydoc", "icu_collections", "icu_locale_core", "icu_properties_data", @@ -1146,15 +1164,15 @@ dependencies = [ [[package]] name = "icu_properties_data" -version = "2.2.0" +version = "2.3.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "8e2bbb201e0c04f7b4b3e14382af113e17ba4f63e2c9d2ee626b720cbce54a14" +checksum = "e590f038c1464a96894fd6d10127e90a8be4509f56ff7ecef851b15cee0b7caa" [[package]] name = "icu_provider" -version = "2.2.0" +version = "2.3.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "139c4cf31c8b5f33d7e199446eff9c1e02decfc2f0eec2c8d71f65befa45b421" +checksum = "92a7ed671a6aad807a8651a2e1782a6598fda9ce5185dd8158549e95a91c6428" dependencies = [ "displaydoc", "icu_locale_core", @@ -1286,7 +1304,7 @@ dependencies = [ "jni-sys 0.4.1", "log", "simd_cesu8", - "thiserror 2.0.19", + "thiserror 2.0.20", "walkdir", "windows-link", ] @@ -1344,9 +1362,9 @@ dependencies = [ [[package]] name = "js-sys" -version = "0.3.103" +version = "0.3.104" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "53b44bfcdb3f8d5837a46dae1ca9660a837176eee74a28b229bc626816589102" +checksum = "0e0c1080212aad755ea003d18543e8768dd432c48819efd73a7bf1e39b7a5a3a" dependencies = [ "cfg-if", "futures-util", @@ -1419,9 +1437,9 @@ checksum = "32a66949e030da00e8c7d4434b251670a91556f4144941d37452769c25d58a53" [[package]] name = "litemap" -version = "0.8.2" +version = "0.8.3" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "92daf443525c4cce67b150400bc2316076100ce0b3686209eb8cf3c31612e6f0" +checksum = "47d9d19d1d6efa0109d2f65ff4c85cddd50bd572e5a00127ab10987290bcefae" [[package]] name = "livekit-api" @@ -1450,7 +1468,7 @@ dependencies = [ "serde_json", "sha2 0.10.9", "signature 2.2.0", - "thiserror 2.0.19", + "thiserror 2.0.20", "tokio", "tokio-rustls", "tokio-tungstenite", @@ -1468,15 +1486,9 @@ dependencies = [ [[package]] name = "livekit-protocol" -<<<<<<< HEAD version = "0.7.10" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "4d26880e94e2f9bab298445e7d86a3794453d211a12ddbd051bd9991a343f9ff" -======= -version = "0.7.12" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "526f22bddf409e5f15449d55cf341647d7c16a1f43df9db9b05be106e4208e1c" ->>>>>>> e0dd9a085fd88168e714af0d6af88769af3aa644 dependencies = [ "pbjson", "pbjson-types", @@ -1602,8 +1614,8 @@ dependencies = [ [[package]] name = "mtp" -version = "0.2.0" -source = "git+https://git.methanium.net/Methanium/mtp.git#b067614a684eb1856bc5db7b3fd82148c036ce6b" +version = "0.3.0" +source = "git+https://git.methanium.net/Methanium/mtp.git#b331b9f6a3943d0331d8fdcb7c696d6c2bc5e8b8" dependencies = [ "mtp-client", "mtp-codec", @@ -1618,8 +1630,8 @@ dependencies = [ [[package]] name = "mtp-client" -version = "0.2.0" -source = "git+https://git.methanium.net/Methanium/mtp.git#b067614a684eb1856bc5db7b3fd82148c036ce6b" +version = "0.3.0" +source = "git+https://git.methanium.net/Methanium/mtp.git#b331b9f6a3943d0331d8fdcb7c696d6c2bc5e8b8" dependencies = [ "mtp-codec", "mtp-common", @@ -1631,8 +1643,8 @@ dependencies = [ [[package]] name = "mtp-codec" -version = "0.2.0" -source = "git+https://git.methanium.net/Methanium/mtp.git#b067614a684eb1856bc5db7b3fd82148c036ce6b" +version = "0.3.0" +source = "git+https://git.methanium.net/Methanium/mtp.git#b331b9f6a3943d0331d8fdcb7c696d6c2bc5e8b8" dependencies = [ "base64 0.23.1", "byteorder", @@ -1640,25 +1652,27 @@ dependencies = [ "mtp-crypto", "mtp-type-map", "rand 0.10.2", + "thiserror 2.0.20", ] [[package]] name = "mtp-common" -version = "0.2.0" -source = "git+https://git.methanium.net/Methanium/mtp.git#b067614a684eb1856bc5db7b3fd82148c036ce6b" +version = "0.3.0" +source = "git+https://git.methanium.net/Methanium/mtp.git#b331b9f6a3943d0331d8fdcb7c696d6c2bc5e8b8" dependencies = [ "quinn", "rustls", - "thiserror 2.0.19", + "thiserror 2.0.20", "wtransport", ] [[package]] name = "mtp-crypto" -version = "0.2.0" -source = "git+https://git.methanium.net/Methanium/mtp.git#b067614a684eb1856bc5db7b3fd82148c036ce6b" +version = "0.3.0" +source = "git+https://git.methanium.net/Methanium/mtp.git#b331b9f6a3943d0331d8fdcb7c696d6c2bc5e8b8" dependencies = [ - "base64 0.23.1", + "argon2", + "base64 0.22.1", "chacha20poly1305", "ed25519-dalek", "getrandom 0.4.3", @@ -1666,36 +1680,37 @@ dependencies = [ "ml-dsa", "mlkem-tls", "rand 0.10.2", - "rand_core 0.10.1", + "rand_core 0.6.4", "rustls", "serde", "sha2 0.11.0", - "thiserror 2.0.19", + "thiserror 1.0.69", "tokio", "zeroize", ] [[package]] name = "mtp-files" -version = "0.2.0" -source = "git+https://git.methanium.net/Methanium/mtp.git#b067614a684eb1856bc5db7b3fd82148c036ce6b" +version = "0.3.0" +source = "git+https://git.methanium.net/Methanium/mtp.git#b331b9f6a3943d0331d8fdcb7c696d6c2bc5e8b8" dependencies = [ "mtp-crypto", "rand 0.10.2", - "thiserror 2.0.19", + "thiserror 2.0.20", "zeroize", ] [[package]] name = "mtp-host" -version = "0.2.0" -source = "git+https://git.methanium.net/Methanium/mtp.git#b067614a684eb1856bc5db7b3fd82148c036ce6b" +version = "0.3.0" +source = "git+https://git.methanium.net/Methanium/mtp.git#b331b9f6a3943d0331d8fdcb7c696d6c2bc5e8b8" dependencies = [ "mtp-codec", "mtp-common", "mtp-crypto", "mtp-transport", "rand 0.10.2", + "thiserror 2.0.20", "tokio", "tracing", "wtransport", @@ -1703,13 +1718,14 @@ dependencies = [ [[package]] name = "mtp-transport" -version = "0.2.0" -source = "git+https://git.methanium.net/Methanium/mtp.git#b067614a684eb1856bc5db7b3fd82148c036ce6b" +version = "0.3.0" +source = "git+https://git.methanium.net/Methanium/mtp.git#b331b9f6a3943d0331d8fdcb7c696d6c2bc5e8b8" dependencies = [ "async-trait", "mtp-codec", "mtp-common", "mtp-crypto", + "rand 0.10.2", "rcgen", "rustls", "rustls-native-certs", @@ -1717,12 +1733,13 @@ dependencies = [ "tokio", "tracing", "wtransport", + "zeroize", ] [[package]] name = "mtp-type-map" -version = "0.2.0" -source = "git+https://git.methanium.net/Methanium/mtp.git#b067614a684eb1856bc5db7b3fd82148c036ce6b" +version = "0.3.0" +source = "git+https://git.methanium.net/Methanium/mtp.git#b331b9f6a3943d0331d8fdcb7c696d6c2bc5e8b8" dependencies = [ "serde", "serde_yaml", @@ -1730,8 +1747,8 @@ dependencies = [ [[package]] name = "mtp-webserver" -version = "0.2.0" -source = "git+https://git.methanium.net/Methanium/mtp.git#b067614a684eb1856bc5db7b3fd82148c036ce6b" +version = "0.3.0" +source = "git+https://git.methanium.net/Methanium/mtp.git#b331b9f6a3943d0331d8fdcb7c696d6c2bc5e8b8" dependencies = [ "async-trait", "bytes", @@ -1748,9 +1765,8 @@ dependencies = [ "mtp-host", "mtp-transport", "quinn", - "rand 0.10.2", "rustls", - "thiserror 2.0.19", + "thiserror 2.0.20", "tokio", "tokio-rustls", "tokio-stream", @@ -1803,9 +1819,9 @@ checksum = "521739c6d2bac4aa25192232afe6841231376b2b26d4d9fae5ecf8ca5772e441" [[package]] name = "num-integer" -version = "0.1.46" +version = "0.1.47" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "7969661fd2958a5cb096e56c8e1ad0444ac2bbcd0061bd28660485a44879858f" +checksum = "7ce2d95d4b3734dc35aa2f45e1aa22cd416814592a4f9d9205e11affd5b8e10b" dependencies = [ "num-traits", ] @@ -1998,7 +2014,7 @@ name = "omikron" version = "0.1.0" dependencies = [ "ansi_term", - "base64 0.22.1", + "base64 0.23.1", "dashmap", "dotenv", "livekit-api", @@ -2012,7 +2028,7 @@ dependencies = [ "serde_json", "strum", "strum_macros", - "thiserror 2.0.19", + "thiserror 2.0.20", "tokio", "trust-dns-resolver", "uuid", @@ -2075,6 +2091,17 @@ dependencies = [ "windows-link", ] +[[package]] +name = "password-hash" +version = "0.5.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "346f04948ba92c43e8469c1ee6736c7563d71012b17d40745260fe106aac2166" +dependencies = [ + "base64ct", + "rand_core 0.6.4", + "subtle", +] + [[package]] name = "pbjson" version = "0.6.0" @@ -2165,9 +2192,9 @@ dependencies = [ [[package]] name = "pkg-config" -version = "0.3.33" +version = "0.3.34" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "19f132c84eca552bf34cab8ec81f1c1dcc229b811638f9d283dceabe58c5569e" +checksum = "f6b464fbc74e149a392436b17d523f769e057cb6877f6a5c4618bc6f11800548" [[package]] name = "poly1305" @@ -2182,15 +2209,15 @@ dependencies = [ [[package]] name = "portable-atomic" -version = "1.14.0" +version = "1.15.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "3d20d5497ef88037a52ff98267d066e7f11fcc5e99bbfbd58a42336193aacec3" +checksum = "05c8b63e8d9609db387f0324918f81d68fe27748f084ef092fb35954d0539a85" [[package]] name = "potential_utf" -version = "0.1.5" +version = "0.1.6" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "0103b1cef7ec0cf76490e969665504990193874ea05c85ff9bab8b911d0a0564" +checksum = "d83eb9bc6d8e5cf568e7a1101d60ee05e81ed50ea106026f3d18deeb046d7661" dependencies = [ "zerovec", ] @@ -2297,7 +2324,7 @@ dependencies = [ "rustc-hash", "rustls", "socket2", - "thiserror 2.0.19", + "thiserror 2.0.20", "tokio", "tracing", "web-time", @@ -2305,9 +2332,9 @@ dependencies = [ [[package]] name = "quinn-proto" -version = "0.11.16" +version = "0.11.17" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "2f4bfc015262b9df63c8845072ce59068853ff5872180c2ce2f13038b970e560" +checksum = "04759210543be93709136e28212294a659ef5001836ff4eab4d663e4529bba83" dependencies = [ "aws-lc-rs", "bytes", @@ -2322,7 +2349,7 @@ dependencies = [ "rustls-pki-types", "rustls-platform-verifier", "slab", - "thiserror 2.0.19", + "thiserror 2.0.20", "tinyvec", "tracing", "web-time", @@ -2339,11 +2366,7 @@ dependencies = [ "once_cell", "socket2", "tracing", -<<<<<<< HEAD "windows-sys 0.61.2", -======= - "windows-sys 0.52.0", ->>>>>>> e0dd9a085fd88168e714af0d6af88769af3aa644 ] [[package]] @@ -2454,9 +2477,9 @@ dependencies = [ [[package]] name = "rcgen" -version = "0.14.8" +version = "0.14.9" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "57f6d249aad744e274e682777a50283a225a32705394ee6d5fcc01efa25e4055" +checksum = "091e7a8e7d86e6feb87a27ce8e2cba29d49eff9507afeebefab7eeb2ca667fb4" dependencies = [ "aws-lc-rs", "pem", @@ -2597,11 +2620,7 @@ dependencies = [ "errno", "libc", "linux-raw-sys", -<<<<<<< HEAD "windows-sys 0.61.2", -======= - "windows-sys 0.52.0", ->>>>>>> e0dd9a085fd88168e714af0d6af88769af3aa644 ] [[package]] @@ -2660,11 +2679,7 @@ dependencies = [ "security-framework", "security-framework-sys", "webpki-root-certs", -<<<<<<< HEAD "windows-sys 0.61.2", -======= - "windows-sys 0.52.0", ->>>>>>> e0dd9a085fd88168e714af0d6af88769af3aa644 ] [[package]] @@ -2675,9 +2690,9 @@ checksum = "f87165f0995f63a9fbeea62b64d10b4d9d8e78ec6d7d51fb2125fda7bb36788f" [[package]] name = "rustls-webpki" -version = "0.103.13" +version = "0.103.14" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "61c429a8649f110dddef65e2a5ad240f747e85f7758a6bccc7e5777bd33f756e" +checksum = "0527518605e68109d875e248ea259b6758801cf165e4b2c2733ae3b51f12535a" dependencies = [ "aws-lc-rs", "ring", @@ -3049,11 +3064,7 @@ dependencies = [ "getrandom 0.4.3", "once_cell", "rustix", -<<<<<<< HEAD "windows-sys 0.61.2", -======= - "windows-sys 0.52.0", ->>>>>>> e0dd9a085fd88168e714af0d6af88769af3aa644 ] [[package]] @@ -3067,11 +3078,11 @@ dependencies = [ [[package]] name = "thiserror" -version = "2.0.19" +version = "2.0.20" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "09a43598840e33d5b0331f38c5e30d13bb11c11210a4b58f0d9b18a5a5eefcd9" +checksum = "ec86235f5fcc2a73650310756d2ac5b138a5780bbbdfae3eeccec992c435ba4f" dependencies = [ - "thiserror-impl 2.0.19", + "thiserror-impl 2.0.20", ] [[package]] @@ -3087,9 +3098,9 @@ dependencies = [ [[package]] name = "thiserror-impl" -version = "2.0.19" +version = "2.0.20" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "43cbfe0cf76104d42a574802844187e84a305e531ed54455f11fbde0f10541cd" +checksum = "bc04cd3e1236dd4a98afca4569f2deb3f120e5422a4023be2cb683f8486292af" dependencies = [ "proc-macro2", "quote", @@ -3128,9 +3139,9 @@ dependencies = [ [[package]] name = "tinystr" -version = "0.8.3" +version = "0.8.4" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "c8323304221c2a851516f22236c5722a72eaa19749016521d6dff0824447d96d" +checksum = "b1e27c91459209c2986af3dcf603a5a74a4368754ce37414f59acc971167f643" dependencies = [ "displaydoc", "zerovec", @@ -3374,7 +3385,7 @@ dependencies = [ "rustls", "rustls-pki-types", "sha1", - "thiserror 2.0.19", + "thiserror 2.0.20", "url", ] @@ -3453,9 +3464,9 @@ checksum = "b6c140620e7ffbb22c2dee59cafe6084a59b5ffc27a8859a5f0d494b5d52b6be" [[package]] name = "uuid" -version = "1.24.0" +version = "1.24.1" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "bf3923a6f5c4c6382e0b653c4117f48d631ea17f38ed86e2a828e6f7412f5239" +checksum = "2cefc03fd367c0c6d4305de1b312cf00248c4114f4a0418ce6a6af769e3b0bd9" dependencies = [ "getrandom 0.4.3", "js-sys", @@ -3504,9 +3515,9 @@ dependencies = [ [[package]] name = "wasm-bindgen" -version = "0.2.126" +version = "0.2.127" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "4b067c0c11094aef6b7a801c1e34a26affafdf3d051dba08456b868789aaf9a4" +checksum = "1b70935747edd64d89de3efa29d73789b806c15798f8e7dca4d8ac356b50ce70" dependencies = [ "cfg-if", "once_cell", @@ -3517,9 +3528,9 @@ dependencies = [ [[package]] name = "wasm-bindgen-futures" -version = "0.4.76" +version = "0.4.77" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "c62df1340f32221cb9c54d6a27b030e3dba64361d4a95bed55f9aacb44da291d" +checksum = "6b7777d5cc23d0e91404e53ce2d5e8ec7acae3026b16233dba62cd3246457950" dependencies = [ "js-sys", "wasm-bindgen", @@ -3527,9 +3538,9 @@ dependencies = [ [[package]] name = "wasm-bindgen-macro" -version = "0.2.126" +version = "0.2.127" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "167ce5e579f6bcf889c4f7175a8a5a585de84e8ff93976ce393efa5f2837aab1" +checksum = "77775f8f3f7217702089053b94958f8f54061a3f663417df76e19cbdcca29bc1" dependencies = [ "quote", "wasm-bindgen-macro-support", @@ -3537,9 +3548,9 @@ dependencies = [ [[package]] name = "wasm-bindgen-macro-support" -version = "0.2.126" +version = "0.2.127" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "f3997c7839262f4ef12cf90b818d6340c18e80f263f1a94bf157d0ec4420380e" +checksum = "e11d33f857dc2fb11b8bc75aee111aa9cbeb12cd9f25efd3d4c2a3dd4e235284" dependencies = [ "bumpalo", "proc-macro2", @@ -3550,18 +3561,18 @@ dependencies = [ [[package]] name = "wasm-bindgen-shared" -version = "0.2.126" +version = "0.2.127" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "dc1b4cb0cc549fcf58d7dfc081778139b3d283a081644e833e84682ad71cea24" +checksum = "7ef64dbcc55df09c7e5a46182d181c2cfa3e925f3da937ea764728b4bbb9dcbf" dependencies = [ "unicode-ident", ] [[package]] name = "web-sys" -version = "0.3.103" +version = "0.3.104" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "8622dcb61c0bcc9fffa6938bed81210af2da9a7e4a1a834b2e37a59b6dfb6141" +checksum = "c435338968042f4f59a557f690a253676d47ce13ceb55d70100e7facf6620a30" dependencies = [ "js-sys", "wasm-bindgen", @@ -3614,11 +3625,7 @@ version = "0.1.11" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "c2a7b1c03c876122aa43f3020e6c3c3ee5c05081c9a00739faf7503aeba10d22" dependencies = [ -<<<<<<< HEAD "windows-sys 0.61.2", -======= - "windows-sys 0.52.0", ->>>>>>> e0dd9a085fd88168e714af0d6af88769af3aa644 ] [[package]] @@ -3827,15 +3834,15 @@ checksum = "1ebf944e87a7c253233ad6766e082e3cd714b5d03812acc24c318f549614536e" [[package]] name = "writeable" -version = "0.6.3" +version = "0.6.4" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "1ffae5123b2d3fc086436f8834ae3ab053a283cfac8fe0a0b8eaae044768a4c4" +checksum = "3ad82d2a33cdc9674dc7465672f271e096168fcdbe0f799d9e6db8c5892679dc" [[package]] name = "wtransport" -version = "0.7.1" +version = "0.7.2" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "ea4aacf790813ee1956751491800537f4e04af7557b7b370501ccbfbc85963e4" +checksum = "b4273ce3157a3262a68665f8d3f20a0ac0c5b8a69ffd67f05ae986832ebec036" dependencies = [ "bytes", "pem", @@ -3846,7 +3853,7 @@ dependencies = [ "rustls-pki-types", "sha2 0.11.0", "socket2", - "thiserror 2.0.19", + "thiserror 2.0.20", "time", "tokio", "tracing", @@ -3857,13 +3864,13 @@ dependencies = [ [[package]] name = "wtransport-proto" -version = "0.7.1" +version = "0.7.2" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "d5867c629e4252f7439d82315923daaf27f4fa442410d51b78ab93ef4c432a11" +checksum = "aad9059572c7dbd6901ccef37f3b7321678cd708dcf58a64b1921dbeab7bfede" dependencies = [ "httlib-huffman", "octets", - "thiserror 2.0.19", + "thiserror 2.0.20", "url", ] @@ -3894,7 +3901,7 @@ dependencies = [ "oid-registry", "ring", "rusticata-macros", - "thiserror 2.0.19", + "thiserror 2.0.20", "time", ] @@ -3994,9 +4001,9 @@ dependencies = [ [[package]] name = "zerotrie" -version = "0.2.4" +version = "0.2.5" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "0f9152d31db0792fa83f70fb2f83148effb5c1f5b8c7686c3459e361d9bc20bf" +checksum = "4ea269c3bd32f0a32c321907a2ae912ba6f4649bb0fc764a15627e99a7095a3f" dependencies = [ "displaydoc", "yoke", @@ -4005,9 +4012,9 @@ dependencies = [ [[package]] name = "zerovec" -version = "0.11.6" +version = "0.11.7" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "90f911cbc359ab6af17377d242225f4d75119aec87ea711a880987b18cd7b239" +checksum = "94b5c6b5976d66c1d703c4fd17d3f5e43c8cedaacf604961b171adc7130896d8" dependencies = [ "yoke", "zerofrom", @@ -4016,13 +4023,13 @@ dependencies = [ [[package]] name = "zerovec-derive" -version = "0.11.3" +version = "0.11.5" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "625dc425cab0dca6dc3c3319506e6593dcb08a9f387ea3b284dbd52a92c40555" +checksum = "9f212a141d820099d57ffafb9569be9617a6f27d3dc881fbee8fb56642f917a9" dependencies = [ "proc-macro2", "quote", - "syn 2.0.119", + "syn 3.0.3", ] [[package]] diff --git a/Cargo.toml b/Cargo.toml index 371ac2d..01536ee 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -4,11 +4,12 @@ version = "0.1.0" edition = "2024" [dependencies] -mtp = { git = "https://git.methanium.net/Methanium/mtp.git", features = [ +mtp = { git = "https://git.methanium.net/Methanium/mtp.git", features = [ "web-server", "client", "crypto", "files", + "raw", ] } mtp-transport = { git = "https://git.methanium.net/Methanium/mtp.git" } diff --git a/src/anonymous_clients/anonymous_client_connection.rs b/src/anonymous_clients/anonymous_client_connection.rs index 2ec1b78..be04ab5 100644 --- a/src/anonymous_clients/anonymous_client_connection.rs +++ b/src/anonymous_clients/anonymous_client_connection.rs @@ -8,7 +8,9 @@ use uuid::Uuid; use crate::anonymous_clients::anonymous_manager::{self, generate_username}; use crate::app_state::AppState; use crate::calls::call_group::call_invite_secret_from_cv; -use crate::rho::connection::{GeneralConnection, MtpReceiver, MtpSender}; +use crate::rho::connection::{ + GeneralConnection, MtpReceiver, MtpSender, MtpValueCompat, OptionalDataValueCompat, +}; use crate::util::data_type_id; use crate::util::logger::PrintType; use crate::{log_cv_in, log_cv_out, log_out}; @@ -108,6 +110,12 @@ impl AnonymousClientConnection { let _permit = permit; log_cv_in!(PrintType::Client, &cv); + if cv.is_type(CommunicationType::Relay) { + self.send_error_response(&cv.get_id(), CommunicationType::ErrorNotAuthenticated) + .await; + return; + } + if cv.is_type(CommunicationType::Identification) { let call_id = Uuid::parse_str(cv.get_data(DataType::CallId).as_str().unwrap_or("")) .unwrap_or(Uuid::new_v4()); @@ -153,19 +161,31 @@ impl AnonymousClientConnection { let mut json_invitee = Vec::new(); let _ = json_invitee.push(( DataType::UserId, - call_invitee_cv.get_data(DataType::UserId).clone(), + call_invitee_cv + .get_data(DataType::UserId) + .cloned() + .unwrap_or(DataValue::Null), )); let _ = json_invitee.push(( DataType::Username, - call_invitee_cv.get_data(DataType::Username).clone(), + call_invitee_cv + .get_data(DataType::Username) + .cloned() + .unwrap_or(DataValue::Null), )); let _ = json_invitee.push(( DataType::Display, - call_invitee_cv.get_data(DataType::Display).clone(), + call_invitee_cv + .get_data(DataType::Display) + .cloned() + .unwrap_or(DataValue::Null), )); let _ = json_invitee.push(( DataType::Avatar, - call_invitee_cv.get_data(DataType::Avatar).clone(), + call_invitee_cv + .get_data(DataType::Avatar) + .cloned() + .unwrap_or(DataValue::Null), )); let _ = invited.push(DataValue::Container( @@ -223,9 +243,7 @@ impl AnonymousClientConnection { // Presence is account-scoped and anonymous sessions have no // persisted account preference to change. - if cv.is_type(CommunicationType::SetUserState) - || cv.is_type(CommunicationType::ClientChanged) - { + if cv.is_type(CommunicationType::ClientChanged) { self.send_error_response(&cv.get_id(), CommunicationType::ErrorNoUserId) .await; return; @@ -333,7 +351,7 @@ impl AnonymousClientConnection { } let call_id = match cv.get_data(DataType::CallId) { - DataValue::Str(id_str) => match Uuid::parse_str(&id_str.to_string()) { + Some(DataValue::Str(id_str)) => match Uuid::parse_str(id_str) { Ok(id) => id, Err(_) => { self.send_error_response(&cv.get_id(), CommunicationType::ErrorInvalidCallId) @@ -443,7 +461,7 @@ impl AnonymousClientConnection { let user_id = self.get_user_id(); let call_id = match cv.get_data(DataType::CallId) { - DataValue::Str(id_str) => match Uuid::parse_str(&id_str.to_string()) { + Some(DataValue::Str(id_str)) => match Uuid::parse_str(id_str) { Ok(id) => id, Err(_) => { self.send_error_response(&cv.get_id(), CommunicationType::ErrorInvalidCallId) diff --git a/src/calls/call_group.rs b/src/calls/call_group.rs index ef03b0f..6e29763 100755 --- a/src/calls/call_group.rs +++ b/src/calls/call_group.rs @@ -8,6 +8,7 @@ use uuid::Uuid; use crate::{ calls::{call_util::LiveKitService, caller::Caller, error::CallError}, omega::omega_connection::OmegaConnection, + rho::connection::OptionalDataValueCompat, util::data_type_id, }; @@ -95,7 +96,7 @@ impl CallSecretEnvelope { } pub fn call_invite_secret_from_cv(cv: &CommunicationValue) -> Option { - CallSecretEnvelope::from_data_value(cv.get_data(DataType::CallSecret)) + CallSecretEnvelope::from_data_value(cv.get_data(DataType::CallSecret)?) } impl CallGroup { @@ -348,8 +349,11 @@ mod tests { receiver_secret.clone().to_data_value(), ); + let Some(call_secret) = cv.get_data(DataType::CallSecret) else { + panic!("call invite did not contain a call secret"); + }; assert_eq!( - CallSecretEnvelope::from_data_value(cv.get_data(DataType::CallSecret)), + CallSecretEnvelope::from_data_value(call_secret), Some(receiver_secret) ); } diff --git a/src/omega/omega_connection.rs b/src/omega/omega_connection.rs index 97f025b..f8d505f 100644 --- a/src/omega/omega_connection.rs +++ b/src/omega/omega_connection.rs @@ -2,6 +2,8 @@ use super::capabilities::PeerCapabilities; use crate::{ config::Config, log_cv_in, log_cv_out, log_err, log_in, log_out, + rho::connection::{MtpValueCompat, OptionalDataValueCompat}, + rho::relay_router, rho::rho_manager::RhoManager, util::{data_type_id, logger::PrintType}, }; @@ -597,6 +599,37 @@ impl OmegaConnection { continue; } + if cv.is_type(CommunicationType::Relay) { + let destination_iota = cv.receiver().unwrap_or_default(); + let request_id = cv.get_id(); + let response = match relay_router::route_from_omega(&self.rho, cv).await { + Ok(()) => CommunicationValue::new(CommunicationType::Success) + .with_id(request_id), + Err(error) => { + log_err!( + self.omikron_id as i64, + PrintType::Omega, + "Relay routing failed for destination Iota {}: {}", + destination_iota, + error + ); + CommunicationValue::new( + relay_router::error_response_type(&error), + ) + .with_id(request_id) + } + }; + if let Err(send_error) = self.send_message_result(&response).await { + log_err!( + self.omikron_id as i64, + PrintType::Omega, + "Relay routing response failed: {}", + send_error + ); + } + continue; + } + let msg_id = cv.get_id(); if let Some((_, task)) = self.waiting_tasks.remove(&msg_id) { if (task.task)(self.clone(), cv.clone()) { @@ -605,7 +638,7 @@ impl OmegaConnection { } if cv.is_type(CommunicationType::IotaUserData) { - if let DataValue::Array(users) = cv.get_data(DataType::UserIds) { + if let Some(DataValue::Array(users)) = cv.get_data(DataType::UserIds) { let mut user_ids = Vec::new(); for value in users { if let DataValue::SignedNumber(user_id) = value { diff --git a/src/rho/app_connection.rs b/src/rho/app_connection.rs index bb71c00..7e8778a 100644 --- a/src/rho/app_connection.rs +++ b/src/rho/app_connection.rs @@ -1,9 +1,12 @@ use crate::anonymous_clients::anonymous_manager; use crate::app_state::AppState; -use crate::rho::connection::{GeneralConnection, MtpReceiver, MtpSender}; +use crate::rho::connection::{ + GeneralConnection, MtpReceiver, MtpSender, MtpValueCompat, OptionalDataValueCompat, +}; +use crate::rho::relay_router::{self, RelaySource}; use crate::rho::rho_connection::RhoConnection; use crate::util::logger::PrintType; -use crate::{log_cv_in, log_cv_out, log_err, log_in, log_out}; +use crate::{log_cv_in, log_cv_out, log_err, log_out}; use mtp::codec::{CommunicationType, CommunicationValue, DataType, DataValue}; use std::sync::Arc; use std::time::Duration; @@ -88,6 +91,59 @@ impl AppConnection { tokio::spawn(async move { log_cv_in!(PrintType::App, cv); + if cv.is_type(CommunicationType::Relay) { + let cv = relay_router::ensure_relay_frame_id(cv); + let request_id = cv.get_id(); + let next_hop = cv.receiver().unwrap_or_default(); + let result = match self.get_rho_connection().await { + Some(rho) => { + relay_router::route_relay( + &self.state, + RelaySource::Client { + iota_id: rho.get_iota_id().await, + }, + cv, + ) + .await + } + None => Err(relay_router::RelayRouteError::DestinationIotaNotLocal), + }; + let response = match result { + Ok(()) => { + CommunicationValue::new(CommunicationType::Success).with_id(request_id) + } + Err(error) => { + log_err!( + self.user_id as i64, + PrintType::App, + "Relay routing failed for destination Iota {}: {}", + next_hop, + error + ); + CommunicationValue::new(relay_router::error_response_type(&error)) + .with_id(request_id) + } + }; + self.send_message(&response).await; + return; + } + + if cv.is_type(CommunicationType::Success) { + if let Some(rho) = self.get_rho_connection().await { + rho.forward_relay_ack(self.user_id, cv.get_id()).await; + } + return; + } + + if matches!( + relay_router::message_security_class(&cv), + relay_router::MessageSecurityClass::RelayOnly + ) { + self.send_error_response(cv.get_id(), CommunicationType::ErrorInvalidData) + .await; + return; + } + if cv.is_type(CommunicationType::GetUserData) { if let Some(anonymous) = { if let Some(user_id) = cv.get_data(DataType::UserId).as_number() { @@ -135,8 +191,14 @@ impl AppConnection { .await; return; } - // Forward other messages to Iota - self.forward_to_iota(cv).await; + log_err!( + self.user_id as i64, + PrintType::App, + "Rejected unsupported communication type {}", + cv.get_type() + ); + self.send_error_response(cv.get_id(), CommunicationType::ErrorInvalidData) + .await; }); } async fn handle_omega_forward(self: Arc, cv: CommunicationValue) { @@ -154,123 +216,6 @@ impl AppConnection { }); } - /// Forward message to Iota - async fn forward_to_iota(self: Arc, cv: CommunicationValue) { - let sender_user_id = self.get_user_id().await; - let msg_id = cv.get_id(); - let msg_type = cv.get_type(); - - log_in!( - sender_user_id as i64, - PrintType::App, - "Forwarding app->iota: sender={} type={:?} id={} receiver={}", - sender_user_id, - msg_type, - msg_id, - cv.get_receiver() - ); - - if cv.is_type(CommunicationType::AddConversation) - && cv.get_data(DataType::ChatPartnerId).as_number().is_none() - { - let chat_partner_name = cv - .get_data(DataType::ChatPartnerName) - .as_str() - .unwrap_or("") - .to_string(); - - if anonymous_manager::get_anonymous_user_by_name(chat_partner_name.to_string()) - .await - .is_some() - { - self.send_error_response(cv.get_id(), CommunicationType::ErrorAnonymous) - .await; - return; - } - - let load_uuid_response = self - .state - .omega - .clone() - .await_response( - &CommunicationValue::new(CommunicationType::GetUserData) - .with_id(cv.clone().get_id()) - .add_typed_default( - DataType::Username, - DataValue::Str(chat_partner_name.clone()), - ), - Some(Duration::from_secs(20)), - ) - .await; - let chat_partner_id = { - if let Ok(load_uuid_response) = load_uuid_response { - load_uuid_response.get_data(DataType::UserId).clone() - } else { - DataValue::Null - } - }; - - if let Some(rho_conn) = self.get_rho_connection().await { - let iota_id = rho_conn.get_iota_id().await; - log_in!( - sender_user_id as i64, - PrintType::App, - "Resolved rho for add_conversation: sender={} -> iota_id={} id={}", - sender_user_id, - iota_id, - msg_id - ); - - let updated_cv = cv - .with_sender(sender_user_id as u64) - .add_typed_default(DataType::ChatPartnerId, chat_partner_id); - rho_conn.message_to_iota(updated_cv).await; - } else { - log_err!( - sender_user_id as i64, - PrintType::App, - "No rho/iota mapping found for add_conversation sender={} type={:?} id={}", - sender_user_id, - msg_type, - msg_id - ); - } - return; - } - - if let Some(rho_conn) = self.get_rho_connection().await { - let iota_id = rho_conn.get_iota_id().await; - log_in!( - sender_user_id as i64, - PrintType::App, - "Resolved rho for forward: sender={} -> iota_id={} type={:?} id={}", - sender_user_id, - iota_id, - msg_type, - msg_id - ); - - let updated_cv = cv.with_sender(sender_user_id as u64); - rho_conn.message_to_iota(updated_cv).await; - } else { - log_err!( - sender_user_id as i64, - PrintType::App, - "No rho/iota mapping found for sender={} type={:?} id={}", - sender_user_id, - msg_type, - msg_id - ); - let error_cv = CommunicationValue::new(CommunicationType::ErrorNoIota) - .with_id(msg_id) - .add_typed_default( - DataType::UserId, - DataValue::SignedNumber(sender_user_id.into()), - ); - self.send_message(&error_cv).await; - } - } - /// Send error response async fn send_error_response(self: Arc, message_id: u32, error_type: CommunicationType) { let error = CommunicationValue::new(error_type).with_id(message_id); diff --git a/src/rho/client_connection.rs b/src/rho/client_connection.rs index 0a6f30f..ba83728 100644 --- a/src/rho/client_connection.rs +++ b/src/rho/client_connection.rs @@ -2,10 +2,13 @@ use crate::anonymous_clients::anonymous_manager; use crate::app_state::AppState; use crate::calls::call_group::call_invite_secret_from_cv; use crate::data::user::UserStatus; -use crate::rho::connection::{GeneralConnection, MtpReceiver, MtpSender}; +use crate::rho::connection::{ + GeneralConnection, MtpReceiver, MtpSender, MtpValueCompat, OptionalDataValueCompat, +}; +use crate::rho::relay_router::{self, RelaySource}; use crate::rho::rho_connection::RhoConnection; use crate::util::logger::PrintType; -use crate::{log_cv_in, log_cv_out, log_err, log_in, log_out}; +use crate::{log_cv_in, log_cv_out, log_err, log_out}; use mtp::codec::{CommunicationType, CommunicationValue, DataType, DataValue}; use std::str::FromStr; use std::sync::Arc; @@ -97,6 +100,16 @@ impl ClientConnection { let _ = self.sender.send(&cv).await; } + pub async fn send_relay(&self, cv: &CommunicationValue) -> Result<(), String> { + if !*self.is_open.read().await { + return Err("client connection is closed".to_string()); + } + self.sender + .send(cv) + .await + .map_err(|error| error.to_string()) + } + /// Handle incoming message from client pub async fn handle_message(self: Arc, cv: CommunicationValue) { let Ok(permit) = self.message_slots.clone().acquire_owned().await else { @@ -108,8 +121,56 @@ impl ClientConnection { let mut cv = cv; - if cv.is_type(CommunicationType::SetUserState) { - self.handle_set_user_state(cv).await; + if cv.is_type(CommunicationType::Relay) { + cv = relay_router::ensure_relay_frame_id(cv); + let request_id = cv.get_id(); + let next_hop = cv.receiver().unwrap_or_default(); + let result = match self.get_rho_connection().await { + Some(rho) => { + relay_router::route_relay( + &self.state, + RelaySource::Client { + iota_id: rho.get_iota_id().await, + }, + cv, + ) + .await + } + None => Err(relay_router::RelayRouteError::DestinationIotaNotLocal), + }; + let response = match result { + Ok(()) => { + CommunicationValue::new(CommunicationType::Success).with_id(request_id) + } + Err(error) => { + log_err!( + self.user_id as i64, + PrintType::Client, + "Relay routing failed for destination Iota {}: {}", + next_hop, + error + ); + CommunicationValue::new(relay_router::error_response_type(&error)) + .with_id(request_id) + } + }; + self.send_message(&response).await; + return; + } + + if cv.is_type(CommunicationType::Success) { + if let Some(rho) = self.get_rho_connection().await { + rho.forward_relay_ack(self.user_id, cv.get_id()).await; + } + return; + } + + if matches!( + relay_router::message_security_class(&cv), + relay_router::MessageSecurityClass::RelayOnly + ) { + self.send_error_response(cv.get_id(), CommunicationType::ErrorInvalidData) + .await; return; } @@ -120,11 +181,13 @@ impl ClientConnection { && cv.get_data_opt(DataType::UserState).is_some() { self.handle_set_user_state( - CommunicationValue::new(CommunicationType::SetUserState) + CommunicationValue::new(CommunicationType::ClientChanged) .with_id(cv.get_id()) .add_typed_default( DataType::UserState, - cv.get_data(DataType::UserState).clone(), + cv.get_data(DataType::UserState) + .cloned() + .unwrap_or(DataValue::Null), ), ) .await; @@ -216,10 +279,12 @@ impl ClientConnection { && cv.get_data_opt(DataType::OnlineStatus).is_some() { let mut profile_request = cv.clone(); - let preference = profile_request.remove_data(DataType::OnlineStatus); - let state_request = CommunicationValue::new(CommunicationType::SetUserState) + let preference = profile_request + .remove_data(DataType::OnlineStatus) + .unwrap_or(DataValue::Null); + let state_request = CommunicationValue::new(CommunicationType::ClientChanged) .with_id(cv.get_id()) - .add_typed_default(DataType::UserState, preference.unwrap()); + .add_typed_default(DataType::UserState, preference); let state_response = match self.request_set_user_state(state_request).await { Ok(response) => response, Err(error_type) => { @@ -346,8 +411,14 @@ impl ClientConnection { ); } - // Forward other messages to Iota - self.forward_to_iota(cv).await; + log_err!( + self.user_id as i64, + PrintType::Client, + "Rejected unsupported communication type {}", + cv.get_type() + ); + self.send_error_response(cv.get_id(), CommunicationType::ErrorInvalidData) + .await; }); } async fn handle_omega_forward(self: Arc, cv: CommunicationValue) { @@ -384,52 +455,25 @@ impl ClientConnection { else { return Err(CommunicationType::ErrorInvalidData); }; - let Some(rho) = self.get_rho_connection().await else { + let Some(_rho) = self.get_rho_connection().await else { return Err(CommunicationType::ErrorNoIota); }; - let request = if self.state.omega.supports_set_user_state().await { - CommunicationValue::new(CommunicationType::SetUserState) - .with_id(cv.get_id()) - .with_sender(self.user_id) - .add_typed_default( - DataType::UserId, - DataValue::SignedNumber(self.user_id as i128), - ) - .add_typed_default( - DataType::SessionId, - DataValue::SignedNumber(self.session_id as i128), - ) - .add_typed_default( - DataType::IotaId, - DataValue::SignedNumber(rho.get_iota_id().await as i128), - ) - .add_typed_default(DataType::UserState, DataValue::Str(state.to_string())) - } else { - // Legacy Omega accepts the original ClientChanged request. It is - // intentionally limited to the authenticated user's ID and a - // writable preference, with no derived connectivity state. - CommunicationValue::new(CommunicationType::ClientChanged) - .with_id(cv.get_id()) - .with_sender(self.user_id) - .add_typed_default( - DataType::UserId, - DataValue::SignedNumber(self.user_id as i128), - ) - .add_typed_default(DataType::UserState, DataValue::Str(state.to_string())) - }; - if !self.state.omega.supports_set_user_state().await { - self.state - .omega - .send_message_result(&request) - .await - .map_err(|_| CommunicationType::ErrorInternal)?; - return Ok(CommunicationValue::new(CommunicationType::Success) - .with_id(cv.get_id()) - .add_typed_default(DataType::UserState, DataValue::Str(state.to_string()))); - } - self.await_omega_response(request) + let request = CommunicationValue::new(CommunicationType::ClientChanged) + .with_id(cv.get_id()) + .with_sender(self.user_id) + .add_typed_default( + DataType::UserId, + DataValue::SignedNumber(self.user_id as i128), + ) + .add_typed_default(DataType::UserState, DataValue::Str(state.to_string())); + self.state + .omega + .send_message_result(&request) .await - .map_err(|_| CommunicationType::ErrorInternal) + .map_err(|_| CommunicationType::ErrorInternal)?; + return Ok(CommunicationValue::new(CommunicationType::Success) + .with_id(cv.get_id()) + .add_typed_default(DataType::UserState, DataValue::Str(state.to_string()))); } async fn handle_set_user_state(self: Arc, cv: CommunicationValue) { @@ -449,7 +493,7 @@ impl ClientConnection { } let call_id = match cv.get_data(DataType::CallId) { - DataValue::Str(id_str) => match Uuid::parse_str(id_str.as_str()) { + Some(DataValue::Str(id_str)) => match Uuid::parse_str(id_str) { Ok(id) => id, Err(_) => { self.send_error_response(cv.get_id(), CommunicationType::ErrorInvalidCallId) @@ -559,7 +603,7 @@ impl ClientConnection { let user_id = self.get_user_id().await; let call_id = match cv.get_data(DataType::CallId) { - DataValue::Str(id_str) => match Uuid::parse_str(id_str.as_str()) { + Some(DataValue::Str(id_str)) => match Uuid::parse_str(id_str) { Ok(id) => id, Err(_) => { self.send_error_response(cv.get_id(), CommunicationType::ErrorInvalidCallId) @@ -600,7 +644,7 @@ impl ClientConnection { let user_id = self.get_user_id().await; let call_id = match cv.get_data(DataType::CallId) { - DataValue::Str(id_str) => match Uuid::parse_str(id_str.as_str()) { + Some(DataValue::Str(id_str)) => match Uuid::parse_str(id_str) { Ok(id) => id, Err(_) => { self.send_error_response(cv.get_id(), CommunicationType::ErrorInvalidCallId) @@ -750,7 +794,10 @@ impl ClientConnection { let resolver = match TokioAsyncResolver::tokio_from_system_conf() { Ok(r) => r, Err(_) => { - let path_data = cv.get_data(DataType::Path).clone(); + let path_data = cv + .get_data(DataType::Path) + .cloned() + .unwrap_or(DataValue::Null); let error_cv = CommunicationValue::new(CommunicationType::ErrorNotFound) .with_id(cv.get_id()) .add_typed_default(DataType::Path, path_data); @@ -786,14 +833,20 @@ impl ClientConnection { return; } - let path_data = cv.get_data(DataType::Path).clone(); + let path_data = cv + .get_data(DataType::Path) + .cloned() + .unwrap_or(DataValue::Null); let error_cv = CommunicationValue::new(CommunicationType::ErrorNotFound) .with_id(cv.get_id()) .add_typed_default(DataType::Path, path_data); self.send_message(&error_cv).await; } Err(_) => { - let path_data = cv.get_data(DataType::Path).clone(); + let path_data = cv + .get_data(DataType::Path) + .cloned() + .unwrap_or(DataValue::Null); let error_cv = CommunicationValue::new(CommunicationType::ErrorNotFound) .with_id(cv.get_id()) .add_typed_default(DataType::Path, path_data); @@ -804,133 +857,16 @@ impl ClientConnection { return; } - let path_data = cv.get_data(DataType::Path).clone(); + let path_data = cv + .get_data(DataType::Path) + .cloned() + .unwrap_or(DataValue::Null); let error_cv = CommunicationValue::new(CommunicationType::ErrorNotFound) .with_id(cv.get_id()) .add_typed_default(DataType::Path, path_data); self.send_message(&error_cv).await; } - /// Forward message to Iota - async fn forward_to_iota(self: Arc, cv: CommunicationValue) { - let sender_user_id = self.get_user_id().await; - let msg_id = cv.get_id(); - let msg_type = cv.get_type(); - - log_in!( - sender_user_id as i64, - PrintType::Client, - "Forwarding client->iota: sender={} type={:?} id={} receiver={}", - sender_user_id, - msg_type, - msg_id, - cv.get_receiver() - ); - - if cv.is_type(CommunicationType::AddConversation) - && cv - .get_data(DataType::ChatPartnerId) - .as_signed_number() - .is_none() - { - let chat_partner_name = cv - .get_data(DataType::ChatPartnerName) - .as_str() - .unwrap_or("") - .to_string(); - - if anonymous_manager::get_anonymous_user_by_name(chat_partner_name.to_string()) - .await - .is_some() - { - self.send_error_response(cv.get_id(), CommunicationType::ErrorAnonymous) - .await; - return; - } - - let load_uuid_response = self - .state - .omega - .clone() - .await_response( - &CommunicationValue::new(CommunicationType::GetUserData) - .with_id(cv.clone().get_id()) - .add_typed_default( - DataType::Username, - DataValue::Str(chat_partner_name.clone()), - ), - Some(Duration::from_secs(20)), - ) - .await; - let chat_partner_id = { - if let Ok(load_uuid_response) = load_uuid_response { - load_uuid_response.get_data(DataType::UserId).clone() - } else { - DataValue::Null - } - }; - - if let Some(rho_conn) = self.get_rho_connection().await { - let iota_id = rho_conn.get_iota_id().await; - log_in!( - sender_user_id as i64, - PrintType::Client, - "Resolved rho for add_conversation: sender={} -> iota_id={} id={}", - sender_user_id, - iota_id, - msg_id - ); - - let updated_cv = cv - .with_sender(sender_user_id as u64) - .add_typed_default(DataType::ChatPartnerId, chat_partner_id); - rho_conn.message_to_iota(updated_cv).await; - } else { - log_err!( - sender_user_id as i64, - PrintType::Client, - "No rho/iota mapping found for add_conversation sender={} type={:?} id={}", - sender_user_id, - msg_type, - msg_id - ); - } - return; - } - - if let Some(rho_conn) = self.get_rho_connection().await { - let iota_id = rho_conn.get_iota_id().await; - log_in!( - sender_user_id as i64, - PrintType::Client, - "Resolved rho for forward: sender={} -> iota_id={} type={:?} id={}", - sender_user_id, - iota_id, - msg_type, - msg_id - ); - - let updated_cv = cv.with_sender(sender_user_id as u64); - rho_conn.message_to_iota(updated_cv).await; - } else { - log_err!( - sender_user_id as i64, - PrintType::Client, - "No rho/iota mapping found for sender={} type={:?} id={}", - sender_user_id, - msg_type, - msg_id - ); - let error_cv = CommunicationValue::new(CommunicationType::ErrorNoIota) - .with_id(msg_id) - .add_typed_default( - DataType::UserId, - DataValue::SignedNumber(sender_user_id.into()), - ); - self.send_message(&error_cv).await; - } - } - /// Send error response async fn send_error_response(self: Arc, message_id: u32, error_type: CommunicationType) { let error = CommunicationValue::new(error_type).with_id(message_id); diff --git a/src/rho/connection.rs b/src/rho/connection.rs index cede813..8109f36 100755 --- a/src/rho/connection.rs +++ b/src/rho/connection.rs @@ -20,6 +20,66 @@ use mtp::webserver::{WebMTPConnection, WebMtpReceiver, WebMtpSender}; pub type MtpSender = WebMtpSender; pub type MtpReceiver = WebMtpReceiver; +/* + * MTP 0.3 exposes absent frame fields and data entries as Options. These + * adapters keep legacy control handlers explicit while Relay code uses the + * native optional accessors directly. + */ +pub(crate) trait MtpValueCompat { + fn get_id(&self) -> u32; + fn get_sender(&self) -> u64; + fn get_receiver(&self) -> u64; + fn get_data_opt(&self, data_type: DataType) -> Option<&DataValue>; +} + +impl MtpValueCompat for CommunicationValue { + fn get_id(&self) -> u32 { + self.id().unwrap_or_default() + } + + fn get_sender(&self) -> u64 { + self.sender().unwrap_or_default() + } + + fn get_receiver(&self) -> u64 { + self.receiver().unwrap_or_default() + } + + fn get_data_opt(&self, data_type: DataType) -> Option<&DataValue> { + self.get_data(data_type) + } +} + +pub(crate) trait OptionalDataValueCompat { + fn as_bool(&self) -> Option; + fn as_number(&self) -> Option; + fn as_signed_number(&self) -> Option; + fn as_str(&self) -> Option<&str>; + fn as_bytes(&self) -> Option>; +} + +impl OptionalDataValueCompat for Option<&DataValue> { + fn as_bool(&self) -> Option { + self.and_then(|value| value.as_bool()) + } + + fn as_number(&self) -> Option { + self.and_then(|value| value.as_number()) + } + + fn as_signed_number(&self) -> Option { + self.and_then(|value| value.as_signed_number()) + } + + fn as_str(&self) -> Option<&str> { + self.and_then(|value| value.as_str()) + } + + fn as_bytes(&self) -> Option> { + self.and_then(|value| value.as_bytes()) + } +} + /* * How a connection identified itself during the mtp handshake driven by * `server.rs` ("iota" / "client" authenticated logins, "anonymous" @@ -194,7 +254,7 @@ impl GeneralConnection { return None; }; - let DataValue::SignedNumber(iota_id) = user_data_cv.get_data(DataType::IotaId) else { + let Some(DataValue::SignedNumber(iota_id)) = user_data_cv.get_data(DataType::IotaId) else { return None; }; let rho = self @@ -264,7 +324,7 @@ impl GeneralConnection { else { return; }; - let DataValue::Array(users) = response.get_data(DataType::UserIds) else { + let Some(DataValue::Array(users)) = response.get_data(DataType::UserIds) else { return; }; let user_ids = users diff --git a/src/rho/iota_connection.rs b/src/rho/iota_connection.rs index 747a394..de4f02a 100755 --- a/src/rho/iota_connection.rs +++ b/src/rho/iota_connection.rs @@ -5,7 +5,10 @@ use crate::log_cv_out; use crate::log_err; use crate::log_in; use crate::log_out; -use crate::rho::connection::{GeneralConnection, MtpReceiver, MtpSender}; +use crate::rho::connection::{ + GeneralConnection, MtpReceiver, MtpSender, MtpValueCompat, OptionalDataValueCompat, +}; +use crate::rho::relay_router::{self, RelaySource}; use crate::util::data_type_id; use crate::util::logger::PrintType; use dashmap::DashMap; @@ -38,7 +41,7 @@ fn contact_snapshot(value: &CommunicationValue) -> Option<(i64, i64, Vec)> .as_signed_number() .and_then(|id| i64::try_from(id).ok()) .filter(|id| *id > 0)?; - let DataValue::Array(values) = value.get_data(DataType::UserIds) else { + let Some(DataValue::Array(values)) = value.get_data(DataType::UserIds) else { return None; }; @@ -76,7 +79,6 @@ pub struct IotaConnection { pub waiting_tasks: DashMap, CommunicationValue) -> bool + Send + Sync>>, pub rho_connection: Arc>>>, - pending_chat_secrets: DashMap>, message_slots: Arc, } @@ -92,7 +94,6 @@ impl IotaConnection { iota_id: iota_id, client_version: general.client_version.read().await.clone(), waiting_tasks: DashMap::new(), - pending_chat_secrets: DashMap::new(), message_slots: Arc::new(tokio::sync::Semaphore::new(32)), }) } @@ -150,10 +151,6 @@ impl IotaConnection { let user_ids_i64: Vec = user_ids.into_iter().map(|u| u as i64).collect(); rho_conn.set_user_ids(user_ids_i64).await; } - - for user_id in self.get_user_ids().await { - self.flush_pending_chat_secrets(user_id).await; - } } pub async fn add_user_id(&self, user_id: u64) { @@ -170,33 +167,9 @@ impl IotaConnection { if let Some(rho_conn) = self.get_rho_connection().await { rho_conn.add_user_id(user_id as i64).await; } - - self.flush_pending_chat_secrets(user_id).await; } } - async fn flush_pending_chat_secrets(&self, user_id: u64) { - let Some((_, messages)) = self.pending_chat_secrets.remove(&user_id) else { - return; - }; - - for message in messages { - self.send_message(&message).await; - } - } - - fn store_pending_chat_secret(&self, cv: CommunicationValue) { - let receiver_id = cv.get_receiver(); - if receiver_id == 0 || !cv.is_type(CommunicationType::SetChatSecret) { - return; - } - - self.pending_chat_secrets - .entry(receiver_id) - .or_default() - .push(cv); - } - /// Set the RhoConnection reference pub async fn set_rho_connection(&self, rho_connection: Arc) { let mut rho_ref = self.rho_connection.write().await; @@ -249,12 +222,58 @@ impl IotaConnection { } } + pub async fn send_relay(&self, cv: &CommunicationValue) -> Result<(), String> { + self.sender + .send(cv) + .await + .map_err(|error| error.to_string()) + } + /// Handle incoming message from Iota pub async fn handle_message(self: Arc, cv: CommunicationValue) { let Ok(permit) = self.message_slots.clone().acquire_owned().await else { return; }; let _permit = permit; + if cv.is_type(CommunicationType::Relay) { + let cv = relay_router::ensure_relay_frame_id(cv); + let request_id = cv.get_id(); + let next_hop = cv.receiver().unwrap_or_default(); + let response = match relay_router::route_relay( + &self.state, + RelaySource::Iota { + iota_id: self.iota_id, + }, + cv, + ) + .await + { + Ok(()) => CommunicationValue::new(CommunicationType::Success).with_id(request_id), + Err(error) => { + log_err!( + self.iota_id as i64, + PrintType::Iota, + "Relay routing failed for destination {}: {}", + next_hop, + error + ); + CommunicationValue::new(relay_router::error_response_type(&error)) + .with_id(request_id) + } + }; + self.send_message(&response).await; + return; + } + + if matches!( + crate::rho::relay_router::message_security_class(&cv), + crate::rho::relay_router::MessageSecurityClass::RelayOnly + ) { + self.send_error_response(cv.get_id(), CommunicationType::ErrorInvalidData, None) + .await; + return; + } + let msg_id = cv.get_id(); if let Some((_, task)) = self.waiting_tasks.remove(&msg_id) { if (task)(self.clone(), cv.clone()) { @@ -304,16 +323,6 @@ impl IotaConnection { return; } - // Handle forwarding to other Iotas or clients - let receiver_id = cv.get_receiver(); - if (receiver_id != 0 && !self.get_user_ids().await.contains(&(receiver_id as u64))) - || cv.is_type(CommunicationType::MessageOtherIota) - || cv.is_type(CommunicationType::SendChat) - { - self.handle_forward_message(cv).await; - return; - } - if cv.is_type(CommunicationType::CompleteRegisterUser) { // Registration carries the authenticated Iota ID separately so // Omega can bind the allocation lease to this connection rather @@ -462,7 +471,14 @@ impl IotaConnection { self.handle_omega_forward_without_sender(cv).await; return; } - self.forward_to_client(cv).await; + log_err!( + self.iota_id as i64, + PrintType::Iota, + "Rejected unsupported communication type {}", + cv.get_type() + ); + self.send_error_response(cv.get_id(), CommunicationType::ErrorInvalidData, None) + .await; } #[allow(dead_code)] @@ -542,63 +558,6 @@ impl IotaConnection { } } } - /// Handle message forwarding to other Iotas - async fn handle_forward_message(&self, cv: CommunicationValue) { - let receiver_id = cv.get_receiver(); - let sender_id = cv.get_sender(); - let my_user_ids = self.get_user_ids().await; - - log_in!( - self.iota_id as i64, - PrintType::Iota, - "Authority check: sender_id={} receiver_id={} iota_user_ids={:?} msg_type={:?} msg_id={}", - sender_id, - receiver_id, - my_user_ids, - cv.get_type(), - cv.get_id() - ); - - if my_user_ids.contains(&(sender_id as u64)) { - if let Some(target_rho) = self.state.rho.get_for_user(receiver_id as i64).await { - target_rho.message_to_iota(cv).await; - } else { - if cv.is_type(CommunicationType::SetChatSecret) { - self.store_pending_chat_secret(cv.clone()); - let success = CommunicationValue::new(CommunicationType::Success) - .with_id(cv.get_id()) - .with_sender(cv.get_sender()) - .with_receiver(cv.get_sender()); - self.send_message(&success).await; - return; - } - - let error = CommunicationValue::new(CommunicationType::ErrorNoIota) - .with_id(cv.get_id()) - .with_sender(cv.get_sender()); - self.send_message(&error).await; - } - } else { - log_err!( - self.iota_id as i64, - PrintType::Iota, - "Rejected client->iota forward: sender_id={} is not authorized for this iota. Known users={:?}", - sender_id, - my_user_ids - ); - - self.send_message( - &CommunicationValue::new(CommunicationType::ErrorInvalidUserId).add_typed_default( - DataType::ErrorType, - DataValue::Str( - "You are sending to another User without authority.".to_string(), - ), - ), - ) - .await; - } - } - /// Handle GET_CHATS message async fn handle_get_chats(&self, cv: CommunicationValue) { let user_id = cv.get_sender(); @@ -633,7 +592,7 @@ impl IotaConnection { // Presence interest is the complete contact set, independent of // whether this account currently participates in a call. - if let DataValue::Array(users) = cv.get_data(DataType::UserIds) { + if let Some(DataValue::Array(users)) = cv.get_data(DataType::UserIds) { for user in users { match user { DataValue::SignedNumber(id) => interested_ids.push(*id as i64), @@ -733,13 +692,13 @@ impl IotaConnection { // ============================ let enriched_contacts = if empty { match cv.get_data(DataType::UserIds) { - DataValue::Array(arr) => DataValue::Array(arr.clone()), + Some(DataValue::Array(arr)) => DataValue::Array(arr.clone()), _ => DataValue::Array(vec![]), } } else { let mut enriched: Vec = Vec::new(); - if let DataValue::Array(users) = cv.get_data(DataType::UserIds) { + if let Some(DataValue::Array(users)) = cv.get_data(DataType::UserIds) { for user_val in users { if let DataValue::Container(entries) = user_val { let mut user_map: BTreeMap = @@ -806,7 +765,8 @@ impl IotaConnection { let mut output = response.clone(); let user_id = response.get_receiver(); - for (key, value) in response.iter_typed_data() { + let typed_data: Vec<_> = response.iter_typed_data().collect(); + for (key, value) in typed_data { if key == Some(DataType::Contacts) { if let Some(contacts) = value.as_array() { let (contacts, global_calls) = self diff --git a/src/rho/mod.rs b/src/rho/mod.rs index af516b1..a3bba4f 100644 --- a/src/rho/mod.rs +++ b/src/rho/mod.rs @@ -2,6 +2,7 @@ pub mod app_connection; pub mod client_connection; pub mod connection; pub mod iota_connection; +pub mod relay_router; pub mod rho_connection; pub mod rho_manager; pub mod server; diff --git a/src/rho/relay_router.rs b/src/rho/relay_router.rs new file mode 100644 index 0000000..0e0ffaa --- /dev/null +++ b/src/rho/relay_router.rs @@ -0,0 +1,409 @@ +use std::{ + sync::{ + Arc, + atomic::{AtomicU32, Ordering}, + }, + time::Duration, +}; + +use mtp::codec::{CommunicationType, CommunicationValue, RelayError, forward_relay_frame}; +use thiserror::Error; + +use crate::{ + app_state::AppState, omega::omega_connection::OmegaConnection, rho::rho_manager::RhoManager, +}; + +const TARGET_KIND_MASK: u64 = 0xC000_0000_0000_0000; +const TARGET_ID_MASK: u64 = (1_u64 << 48) - 1; +const USER_TARGET_KIND: u64 = 0x4000_0000_0000_0000; +const IOTA_TARGET_KIND: u64 = 0x8000_0000_0000_0000; +static NEXT_RELAY_FRAME_ID: AtomicU32 = AtomicU32::new(1); + +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub enum RouteTarget { + User(u64), + Iota(u64), +} + +impl RouteTarget { + pub fn wire_id(self) -> Option { + let (kind, id) = match self { + Self::User(id) => (USER_TARGET_KIND, id), + Self::Iota(id) => (IOTA_TARGET_KIND, id), + }; + (id > 0 && id <= TARGET_ID_MASK).then_some(kind | id) + } + + pub fn from_wire_id(value: u64) -> Option { + let id = value & TARGET_ID_MASK; + if id == 0 || value & !(TARGET_KIND_MASK | TARGET_ID_MASK) != 0 { + return None; + } + + match value & TARGET_KIND_MASK { + USER_TARGET_KIND => Some(Self::User(id)), + IOTA_TARGET_KIND => Some(Self::Iota(id)), + _ => None, + } + } + + pub const fn id(self) -> u64 { + match self { + Self::User(id) | Self::Iota(id) => id, + } + } +} + +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub enum RelaySource { + Client { iota_id: u64 }, + Iota { iota_id: u64 }, +} + +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub enum MessageSecurityClass { + RelayOnly, + AuthenticatedPeerControl, + AuthenticatedLocalRequest, +} + +pub fn message_security_class(frame: &CommunicationValue) -> MessageSecurityClass { + const RELAY_ONLY_TYPES: &[CommunicationType] = &[ + CommunicationType::MessageSend, + CommunicationType::MessageLive, + CommunicationType::MessageState, + CommunicationType::MessageEdit, + CommunicationType::MessageEditLive, + CommunicationType::MessageReactionAdd, + CommunicationType::MessageReactionRemove, + CommunicationType::MessageReactionLive, + CommunicationType::MessageDelete, + CommunicationType::MessageDeleteLive, + CommunicationType::MessageOtherIota, + CommunicationType::SetChatSecret, + CommunicationType::SendChat, + CommunicationType::SettingsSave, + CommunicationType::GlobalSettingsSave, + CommunicationType::AddConversation, + CommunicationType::AddCommunity, + CommunicationType::RemoveCommunity, + ]; + + if RELAY_ONLY_TYPES.iter().any(|kind| frame.is_type(*kind)) { + MessageSecurityClass::RelayOnly + } else if frame.is_type(CommunicationType::GetChatSecret) + || frame.is_type(CommunicationType::MessageGet) + || frame.is_type(CommunicationType::MessagesGet) + { + MessageSecurityClass::AuthenticatedPeerControl + } else { + MessageSecurityClass::AuthenticatedLocalRequest + } +} + +#[derive(Debug, Error)] +pub enum RelayRouteError { + #[error("relay has no next-hop receiver")] + MissingReceiver, + #[error("relay has invalid route target {0}")] + InvalidRouteTarget(u64), + #[error("client relay destination {actual} is not its associated Iota {expected}")] + InvalidClientRoute { expected: u64, actual: u64 }, + #[error("relay destination Iota is not connected to this Omikron")] + DestinationIotaNotLocal, + #[error("relay destination client is offline")] + ClientOffline, + #[error("relay route resolves back to its source Iota")] + RouteLoop, + #[error("relay send failed: {0}")] + Send(String), + #[error(transparent)] + Relay(#[from] RelayError), +} + +pub fn error_response_type(error: &RelayRouteError) -> CommunicationType { + match error { + RelayRouteError::DestinationIotaNotLocal | RelayRouteError::ClientOffline => { + CommunicationType::ErrorNoIota + } + RelayRouteError::Send(_) => CommunicationType::ErrorInternal, + RelayRouteError::MissingReceiver + | RelayRouteError::InvalidRouteTarget(_) + | RelayRouteError::InvalidClientRoute { .. } + | RelayRouteError::RouteLoop + | RelayRouteError::Relay(_) => CommunicationType::ErrorInvalidData, + } +} + +pub async fn route_relay( + state: &Arc, + source: RelaySource, + frame: CommunicationValue, +) -> Result<(), RelayRouteError> { + let (next_hop, frame) = prepare_frame(ensure_relay_frame_id(frame))?; + validate_source_next_hop(source, next_hop)?; + + match source { + RelaySource::Client { iota_id } => { + let rho = state + .rho + .get_by_iota( + i64::try_from(iota_id) + .map_err(|_| RelayRouteError::InvalidRouteTarget(iota_id))?, + ) + .await + .ok_or(RelayRouteError::DestinationIotaNotLocal)?; + let response = rho + .await_relay_to_iota(&frame) + .await + .map_err(RelayRouteError::Send)?; + route_response(response) + } + RelaySource::Iota { iota_id } => { + route_from_iota(&state.rho, &state.omega, iota_id, next_hop, frame).await + } + } +} + +pub async fn route_from_omega( + rho: &RhoManager, + frame: CommunicationValue, +) -> Result<(), RelayRouteError> { + let (next_hop, frame) = prepare_frame(ensure_relay_frame_id(frame))?; + let RouteTarget::Iota(iota_id) = next_hop else { + return Err(RelayRouteError::InvalidRouteTarget(next_hop.id())); + }; + let target = rho + .get_by_iota( + i64::try_from(iota_id).map_err(|_| RelayRouteError::InvalidRouteTarget(iota_id))?, + ) + .await + .ok_or(RelayRouteError::DestinationIotaNotLocal)?; + let response = target + .await_relay_to_iota(&frame) + .await + .map_err(RelayRouteError::Send)?; + route_response(response) +} + +async fn route_from_iota( + rho: &RhoManager, + omega: &OmegaConnection, + source_iota_id: u64, + next_hop: RouteTarget, + frame: CommunicationValue, +) -> Result<(), RelayRouteError> { + match next_hop { + RouteTarget::User(user_id) => { + let target = rho + .get_for_user( + i64::try_from(user_id) + .map_err(|_| RelayRouteError::InvalidRouteTarget(user_id))?, + ) + .await + .ok_or(RelayRouteError::ClientOffline)?; + if !target.has_local_client(user_id).await { + return Err(RelayRouteError::ClientOffline); + } + target.send_relay_to_client(&frame).await.map_err(|error| { + if error == "client offline" { + RelayRouteError::ClientOffline + } else { + RelayRouteError::Send(error) + } + }) + } + RouteTarget::Iota(iota_id) => { + if iota_id == source_iota_id { + return Err(RelayRouteError::RouteLoop); + } + if let Some(target) = rho + .get_by_iota( + i64::try_from(iota_id) + .map_err(|_| RelayRouteError::InvalidRouteTarget(iota_id))?, + ) + .await + { + let response = target + .await_relay_to_iota(&frame) + .await + .map_err(RelayRouteError::Send)?; + return route_response(response); + } + let response = omega + .await_response(&frame, Some(Duration::from_secs(20))) + .await + .map_err(RelayRouteError::Send)?; + route_response(response) + } + } +} + +fn route_response(response: CommunicationValue) -> Result<(), RelayRouteError> { + if response.is_type(CommunicationType::Success) { + Ok(()) + } else { + Err(RelayRouteError::Send(format!( + "next Relay hop rejected the frame with {}", + response.get_type() + ))) + } +} + +pub fn ensure_relay_frame_id(frame: CommunicationValue) -> CommunicationValue { + if frame.id().is_some_and(|id| id != 0) { + return frame; + } + let id = NEXT_RELAY_FRAME_ID.fetch_add(1, Ordering::Relaxed).max(1); + frame.with_id(id) +} + +fn prepare_frame( + frame: CommunicationValue, +) -> Result<(RouteTarget, CommunicationValue), RelayRouteError> { + let next_hop = frame.receiver().ok_or(RelayRouteError::MissingReceiver)?; + let target = + RouteTarget::from_wire_id(next_hop).ok_or(RelayRouteError::InvalidRouteTarget(next_hop))?; + let frame = forward_relay_frame(&frame, next_hop)?; + Ok((target, frame)) +} + +fn validate_source_next_hop( + source: RelaySource, + next_hop: RouteTarget, +) -> Result<(), RelayRouteError> { + match source { + RelaySource::Client { iota_id } => match next_hop { + RouteTarget::Iota(actual) if actual == iota_id => Ok(()), + _ => Err(RelayRouteError::InvalidClientRoute { + expected: iota_id, + actual: next_hop.id(), + }), + }, + RelaySource::Iota { iota_id } => { + if next_hop == RouteTarget::Iota(iota_id) { + Err(RelayRouteError::RouteLoop) + } else { + Ok(()) + } + } + } +} + +#[cfg(test)] +mod tests { + use mtp::codec::{CommunicationType, CommunicationValue, DataValue}; + + use super::{ + RelayRouteError, RelaySource, RouteTarget, prepare_frame, validate_source_next_hop, + }; + + fn wire(target: RouteTarget) -> u64 { + let Some(value) = target.wire_id() else { + panic!("valid route target was rejected"); + }; + value + } + + fn relay(receiver: Option) -> CommunicationValue { + let frame = CommunicationValue::new(CommunicationType::Relay) + .without_sender() + .with_payload(DataValue::Bytes(vec![1, 2, 3])); + receiver.map_or(frame.clone(), |id| frame.with_receiver(id)) + } + + #[test] + fn relay_requires_a_next_hop() { + assert!(matches!( + prepare_frame(relay(None)), + Err(RelayRouteError::MissingReceiver) + )); + } + + #[test] + fn relay_rejects_an_outer_sender() { + let frame = relay(Some(wire(RouteTarget::Iota(7)))).with_sender(9); + assert!(matches!( + prepare_frame(frame), + Err(RelayRouteError::Relay( + mtp::codec::RelayError::OuterSenderPresent + )) + )); + } + + #[test] + fn relay_forwarding_preserves_payload_and_next_hop() { + let frame = relay(Some(wire(RouteTarget::Iota(7)))); + let (next_hop, forwarded) = match prepare_frame(frame.clone()) { + Ok(value) => value, + Err(error) => panic!("valid relay was rejected: {error}"), + }; + assert_eq!(next_hop, RouteTarget::Iota(7)); + assert_eq!(forwarded.receiver(), Some(wire(RouteTarget::Iota(7)))); + assert_eq!(forwarded.sender(), None); + assert_eq!(forwarded.payload(), frame.payload()); + } + + #[test] + fn non_relay_frames_are_rejected() { + let frame = CommunicationValue::new(CommunicationType::Ping) + .with_receiver(wire(RouteTarget::Iota(7))); + assert!(matches!( + prepare_frame(frame), + Err(RelayRouteError::Relay(mtp::codec::RelayError::NotRelay)) + )); + } + + #[test] + fn client_cannot_address_another_iota() { + assert!(matches!( + validate_source_next_hop(RelaySource::Client { iota_id: 7 }, RouteTarget::Iota(8)), + Err(RelayRouteError::InvalidClientRoute { + expected: 7, + actual: 8 + }) + )); + } + + #[test] + fn client_cannot_address_a_user() { + assert!(matches!( + validate_source_next_hop(RelaySource::Client { iota_id: 7 }, RouteTarget::User(8)), + Err(RelayRouteError::InvalidClientRoute { + expected: 7, + actual: 8 + }) + )); + } + + #[test] + fn iota_cannot_route_to_itself() { + assert!(matches!( + validate_source_next_hop(RelaySource::Iota { iota_id: 7 }, RouteTarget::Iota(7)), + Err(RelayRouteError::RouteLoop) + )); + } + + #[tokio::test] + async fn omega_does_not_forward_to_another_omikron() { + let rho = crate::rho::rho_manager::RhoManager::new(); + let result = super::route_from_omega(&rho, relay(Some(wire(RouteTarget::Iota(7))))).await; + assert!(matches!( + result, + Err(RelayRouteError::DestinationIotaNotLocal) + )); + } + + #[test] + fn route_target_namespace_is_explicit() { + assert_eq!( + RouteTarget::from_wire_id(wire(RouteTarget::User(7))), + Some(RouteTarget::User(7)) + ); + assert_eq!( + RouteTarget::from_wire_id(wire(RouteTarget::Iota(7))), + Some(RouteTarget::Iota(7)) + ); + assert_eq!(RouteTarget::from_wire_id(7), None); + } +} diff --git a/src/rho/rho_connection.rs b/src/rho/rho_connection.rs index 9944572..094bb1b 100644 --- a/src/rho/rho_connection.rs +++ b/src/rho/rho_connection.rs @@ -1,8 +1,15 @@ use super::{client_connection::ClientConnection, iota_connection::IotaConnection}; -use crate::{log_err, rho::app_connection::AppConnection}; +use super::relay_router::RouteTarget; +use crate::{ + log_err, + rho::{ + app_connection::AppConnection, + connection::{MtpValueCompat, OptionalDataValueCompat}, + }, +}; use dashmap::DashMap; -use mtp::codec::{CommunicationValue, DataType}; +use mtp::codec::{CommunicationType, CommunicationValue, DataType, DataValue}; use std::sync::Arc; use tokio::sync::RwLock; use uuid::Uuid; @@ -91,7 +98,8 @@ impl RhoConnection { for client in clients { client.clear_rho_connection().await; } - self.client_connections.retain(|(id, _), _| *id != user_id as u64); + self.client_connections + .retain(|(id, _), _| *id != user_id as u64); } #[allow(unused)] @@ -265,11 +273,62 @@ impl RhoConnection { } } - /// Send message to Iota pub async fn message_to_iota(&self, cv: CommunicationValue) { self.iota_connection.send_message(&cv).await; } + pub async fn has_local_client(&self, user_id: u64) -> bool { + self.client_connections + .iter() + .any(|entry| entry.key().0 == user_id) + } + + pub async fn send_relay_to_client(&self, cv: &CommunicationValue) -> Result<(), String> { + let receiver_id = cv + .receiver() + .and_then(RouteTarget::from_wire_id) + .and_then(|target| match target { + RouteTarget::User(user_id) => Some(user_id), + RouteTarget::Iota(_) => None, + }) + .ok_or_else(|| "client offline".to_string())?; + let connections = self + .get_client_connections_for_user(receiver_id as i64) + .await; + if connections.is_empty() { + return Err("client offline".to_string()); + } + + let mut send_error = None; + for connection in connections { + if let Err(error) = connection.send_relay(cv).await { + send_error = Some(error); + } + } + send_error.map_or(Ok(()), Err) + } + + pub async fn send_relay_to_iota(&self, cv: &CommunicationValue) -> Result<(), String> { + self.iota_connection.send_relay(cv).await + } + + pub async fn await_relay_to_iota( + &self, + cv: &CommunicationValue, + ) -> Result { + self.iota_connection + .clone() + .await_response(cv, Some(std::time::Duration::from_secs(20))) + .await + } + + pub async fn forward_relay_ack(&self, user_id: u64, frame_id: u32) { + let acknowledgement = CommunicationValue::new(CommunicationType::Success) + .with_id(frame_id) + .add_typed_default(DataType::UserId, DataValue::UnsignedNumber(user_id.into())); + self.iota_connection.send_message(&acknowledgement).await; + } + /// Check if this RhoConnection contains a specific user ID #[allow(dead_code)] pub async fn contains_user(&self, user_id: &i64) -> bool { diff --git a/src/rho/rho_manager.rs b/src/rho/rho_manager.rs index 5cb7ce3..2eb4632 100644 --- a/src/rho/rho_manager.rs +++ b/src/rho/rho_manager.rs @@ -66,7 +66,10 @@ impl RhoManager { ) .await; - let replacement = user_ids.iter().copied().collect::>(); + let replacement = user_ids + .iter() + .copied() + .collect::>(); for user_id in previous_users { if !replacement.contains(&user_id) { rho.detach_user_clients(user_id).await; diff --git a/src/rho/server.rs b/src/rho/server.rs index ef3c85b..5f158e9 100644 --- a/src/rho/server.rs +++ b/src/rho/server.rs @@ -9,7 +9,7 @@ use crate::{ app_state::AppState, log, log_err, omega::omega_connection::OmegaConnection, - rho::connection::GeneralConnection, + rho::connection::{GeneralConnection, OptionalDataValueCompat}, util::{file_util::load_file_vec, logger::PrintType}, }; use mtp::codec::{CommunicationType, CommunicationValue, DataType, DataValue}; @@ -98,7 +98,7 @@ pub async fn complete_register( }; match response.get_data(DataType::IotaId) { - DataValue::SignedNumber(id) => *id as u64, + Some(DataValue::SignedNumber(id)) => *id as u64, _ => 0, } } diff --git a/src/util/logger.rs b/src/util/logger.rs index 97b17fa..6e688cb 100644 --- a/src/util/logger.rs +++ b/src/util/logger.rs @@ -7,7 +7,9 @@ use std::{ }; use ansi_term::Color; -use mtp::codec::{CommunicationValue, DataTypeId, DataValue, Version}; +use mtp::codec::{CommunicationType, CommunicationValue, DataTypeId, DataValue, Version}; + +use crate::rho::connection::MtpValueCompat; static LOGGER: OnceLock> = OnceLock::new(); @@ -179,11 +181,16 @@ pub fn format_cv(cv: &CommunicationValue) -> String { .unwrap_or_else(|| cv.get_type().to_string()); parts.push(format!("{} (id={})", comm_type, cv.get_id())); - let data = cv.data(); + if cv.is_type(CommunicationType::Relay) { + parts.push("".to_string()); + return parts.join(": "); + } + + let data = cv.data().unwrap_or(&[]); let formated_data = format_data_container( data.iter().map(|(k, v)| (*k, v.clone())).collect(), - Version(1, 0), + Version(3, 0), ); parts.push(format!("{}", formated_data));