From b814c3776dde04f3108611ea6f66f9b42d0c8fdb Mon Sep 17 00:00:00 2001 From: Alex Emmet <111742636+Alex-Emmet@users.noreply.github.com> Date: Tue, 18 Aug 2026 22:37:14 +0200 Subject: [PATCH] [WIP] 0.3.0 mtp update --- Cargo.lock | 335 ++++++++++++++---------- Cargo.toml | 3 +- src/db/user_repo.rs | 39 ++- src/server/api.rs | 12 +- src/sql/user_online_tracker.rs | 32 ++- src/state.rs | 40 ++- src/transport/connection.rs | 49 +++- src/transport/handlers/account.rs | 316 ++++++++++++++++++---- src/transport/handlers/links.rs | 8 +- src/transport/handlers/notifications.rs | 4 +- src/transport/handlers/presence.rs | 21 +- src/transport/handlers/register.rs | 4 +- src/transport/handlers/states.rs | 6 +- src/transport/handlers/user_data.rs | 37 ++- src/transport/mod.rs | 1 + src/transport/omikron_connection.rs | 97 ++++++- src/transport/omikron_manager.rs | 35 ++- src/transport/relay_router.rs | 243 +++++++++++++++++ src/util/logger.rs | 5 +- 19 files changed, 1009 insertions(+), 278 deletions(-) create mode 100644 src/transport/relay_router.rs diff --git a/Cargo.lock b/Cargo.lock index 1e03155..cd6c90d 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -44,6 +44,18 @@ dependencies = [ "winapi", ] +[[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" @@ -56,7 +68,7 @@ dependencies = [ "nom", "num-traits", "rusticata-macros", - "thiserror", + "thiserror 2.0.20", "time", ] @@ -85,9 +97,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", @@ -175,6 +187,15 @@ dependencies = [ "serde_core", ] +[[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" @@ -223,9 +244,9 @@ dependencies = [ [[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", @@ -740,9 +761,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 = "flate2" @@ -800,9 +821,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", @@ -815,9 +836,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", @@ -825,15 +846,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", @@ -853,38 +874,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", @@ -936,9 +957,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", @@ -1126,9 +1147,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", @@ -1223,9 +1244,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", @@ -1237,9 +1258,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", @@ -1250,9 +1271,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", @@ -1264,16 +1285,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", @@ -1284,15 +1306,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", @@ -1376,7 +1398,7 @@ dependencies = [ "jni-sys", "log", "simd_cesu8", - "thiserror", + "thiserror 2.0.20", "walkdir", "windows-link", ] @@ -1425,9 +1447,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", @@ -1482,14 +1504,14 @@ checksum = "b6d2cec3eae94f9f509c767b45932f1ada8350c4bdb85af2fcab4a3c14807981" [[package]] name = "libredox" -version = "0.1.19" +version = "0.1.20" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "2026a5056764a10b2bf5d56488cba40da507f5493a6a429340e2004d9ed085fa" +checksum = "28d0a00925a9f930d679b6789b721e3a7f9ed110f41b86d2497caa780c3a070a" dependencies = [ "bitflags", "libc", "plain", - "redox_syscall 0.9.1", + "redox_syscall 0.9.2", ] [[package]] @@ -1504,9 +1526,9 @@ dependencies = [ [[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 = "lock_api" @@ -1642,8 +1664,8 @@ dependencies = [ [[package]] name = "mtp" -version = "0.2.0" -source = "git+https://git.methanium.net/methanium/mtp#b067614a684eb1856bc5db7b3fd82148c036ce6b" +version = "0.3.0" +source = "git+https://git.methanium.net/Methanium/mtp.git#b331b9f6a3943d0331d8fdcb7c696d6c2bc5e8b8" dependencies = [ "mtp-client", "mtp-codec", @@ -1658,8 +1680,8 @@ dependencies = [ [[package]] name = "mtp-client" -version = "0.2.0" -source = "git+https://git.methanium.net/methanium/mtp#b067614a684eb1856bc5db7b3fd82148c036ce6b" +version = "0.3.0" +source = "git+https://git.methanium.net/Methanium/mtp.git#b331b9f6a3943d0331d8fdcb7c696d6c2bc5e8b8" dependencies = [ "mtp-codec", "mtp-common", @@ -1671,8 +1693,8 @@ dependencies = [ [[package]] name = "mtp-codec" -version = "0.2.0" -source = "git+https://git.methanium.net/methanium/mtp#b067614a684eb1856bc5db7b3fd82148c036ce6b" +version = "0.3.0" +source = "git+https://git.methanium.net/Methanium/mtp.git#b331b9f6a3943d0331d8fdcb7c696d6c2bc5e8b8" dependencies = [ "base64 0.23.1", "byteorder", @@ -1680,25 +1702,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#b067614a684eb1856bc5db7b3fd82148c036ce6b" +version = "0.3.0" +source = "git+https://git.methanium.net/Methanium/mtp.git#b331b9f6a3943d0331d8fdcb7c696d6c2bc5e8b8" dependencies = [ "quinn", "rustls", - "thiserror", + "thiserror 2.0.20", "wtransport", ] [[package]] name = "mtp-crypto" -version = "0.2.0" -source = "git+https://git.methanium.net/methanium/mtp#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", @@ -1706,36 +1730,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", + "thiserror 1.0.69", "tokio", "zeroize", ] [[package]] name = "mtp-files" -version = "0.2.0" -source = "git+https://git.methanium.net/methanium/mtp#b067614a684eb1856bc5db7b3fd82148c036ce6b" +version = "0.3.0" +source = "git+https://git.methanium.net/Methanium/mtp.git#b331b9f6a3943d0331d8fdcb7c696d6c2bc5e8b8" dependencies = [ "mtp-crypto", "rand 0.10.2", - "thiserror", + "thiserror 2.0.20", "zeroize", ] [[package]] name = "mtp-host" -version = "0.2.0" -source = "git+https://git.methanium.net/methanium/mtp#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", @@ -1743,13 +1768,14 @@ dependencies = [ [[package]] name = "mtp-transport" -version = "0.2.0" -source = "git+https://git.methanium.net/methanium/mtp#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", @@ -1757,12 +1783,13 @@ dependencies = [ "tokio", "tracing", "wtransport", + "zeroize", ] [[package]] name = "mtp-type-map" -version = "0.2.0" -source = "git+https://git.methanium.net/methanium/mtp#b067614a684eb1856bc5db7b3fd82148c036ce6b" +version = "0.3.0" +source = "git+https://git.methanium.net/Methanium/mtp.git#b331b9f6a3943d0331d8fdcb7c696d6c2bc5e8b8" dependencies = [ "serde", "serde_yaml", @@ -1770,8 +1797,8 @@ dependencies = [ [[package]] name = "mtp-webserver" -version = "0.2.0" -source = "git+https://git.methanium.net/methanium/mtp#b067614a684eb1856bc5db7b3fd82148c036ce6b" +version = "0.3.0" +source = "git+https://git.methanium.net/Methanium/mtp.git#b331b9f6a3943d0331d8fdcb7c696d6c2bc5e8b8" dependencies = [ "async-trait", "bytes", @@ -1788,9 +1815,8 @@ dependencies = [ "mtp-host", "mtp-transport", "quinn", - "rand 0.10.2", "rustls", - "thiserror", + "thiserror 2.0.20", "tokio", "tokio-rustls", "tokio-stream", @@ -1841,9 +1867,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", ] @@ -1903,7 +1929,7 @@ dependencies = [ "sqlx", "strum", "strum_macros", - "thiserror", + "thiserror 2.0.20", "tokio", "uuid", "zip", @@ -1956,6 +1982,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 = "pbkdf2" version = "0.13.0" @@ -2039,9 +2076,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 = "plain" @@ -2062,15 +2099,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", ] @@ -2120,7 +2157,7 @@ dependencies = [ "rustc-hash", "rustls", "socket2", - "thiserror", + "thiserror 2.0.20", "tokio", "tracing", "web-time", @@ -2128,9 +2165,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", @@ -2145,7 +2182,7 @@ dependencies = [ "rustls-pki-types", "rustls-platform-verifier", "slab", - "thiserror", + "thiserror 2.0.20", "tinyvec", "tracing", "web-time", @@ -2238,9 +2275,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", @@ -2262,9 +2299,9 @@ dependencies = [ [[package]] name = "redox_syscall" -version = "0.9.1" +version = "0.9.2" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "07507be7b4a5f9f26eeb41eeaebb1f5a7ff29dfb29739facc21d35bf8b11c21e" +checksum = "f1c93da5bb2c5d4e6c0ef7abeead62c89169a0a4882bfb83ac892f2423aea2fe" dependencies = [ "bitflags", ] @@ -2432,9 +2469,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", @@ -2804,7 +2841,7 @@ dependencies = [ "serde_json", "sha2 0.10.9", "smallvec", - "thiserror", + "thiserror 2.0.20", "tokio", "tokio-stream", "tracing", @@ -2886,7 +2923,7 @@ dependencies = [ "smallvec", "sqlx-core", "stringprep", - "thiserror", + "thiserror 2.0.20", "tracing", "whoami", ] @@ -2923,7 +2960,7 @@ dependencies = [ "smallvec", "sqlx-core", "stringprep", - "thiserror", + "thiserror 2.0.20", "tracing", "whoami", ] @@ -2947,7 +2984,7 @@ dependencies = [ "serde", "serde_urlencoded", "sqlx-core", - "thiserror", + "thiserror 2.0.20", "tracing", "url", ] @@ -3058,18 +3095,38 @@ dependencies = [ [[package]] name = "thiserror" -version = "2.0.19" +version = "1.0.69" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "09a43598840e33d5b0331f38c5e30d13bb11c11210a4b58f0d9b18a5a5eefcd9" +checksum = "b6aaf5339b578ea85b50e080feb250a3e8ae8cfcdff9a461c9ec2904bc923f52" dependencies = [ - "thiserror-impl", + "thiserror-impl 1.0.69", +] + +[[package]] +name = "thiserror" +version = "2.0.20" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "ec86235f5fcc2a73650310756d2ac5b138a5780bbbdfae3eeccec992c435ba4f" +dependencies = [ + "thiserror-impl 2.0.20", ] [[package]] name = "thiserror-impl" -version = "2.0.19" +version = "1.0.69" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "43cbfe0cf76104d42a574802844187e84a305e531ed54455f11fbde0f10541cd" +checksum = "4fee6c4efc90059e10f81e6d42c60a18f76588c3d74cb83a0b242a2b6c7504c1" +dependencies = [ + "proc-macro2", + "quote", + "syn 2.0.119", +] + +[[package]] +name = "thiserror-impl" +version = "2.0.20" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "bc04cd3e1236dd4a98afca4569f2deb3f120e5422a4023be2cb683f8486292af" dependencies = [ "proc-macro2", "quote", @@ -3109,9 +3166,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", @@ -3365,9 +3422,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", @@ -3419,9 +3476,9 @@ checksum = "b8dad83b4f25e74f184f64c43b150b91efe7647395b42289f38e50566d82855b" [[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", @@ -3432,9 +3489,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", @@ -3442,9 +3499,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", @@ -3452,9 +3509,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", @@ -3465,18 +3522,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", @@ -3727,15 +3784,15 @@ checksum = "589f6da84c646204747d1270a2a5661ea66ed1cced2631d546fdfb155959f9ec" [[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", @@ -3746,7 +3803,7 @@ dependencies = [ "rustls-pki-types", "sha2 0.11.0", "socket2", - "thiserror", + "thiserror 2.0.20", "time", "tokio", "tracing", @@ -3757,13 +3814,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", + "thiserror 2.0.20", "url", ] @@ -3794,7 +3851,7 @@ dependencies = [ "oid-registry", "ring", "rusticata-macros", - "thiserror", + "thiserror 2.0.20", "time", ] @@ -3894,9 +3951,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", @@ -3905,9 +3962,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", @@ -3916,13 +3973,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 3b4f6a3..28e732a 100755 --- a/Cargo.toml +++ b/Cargo.toml @@ -4,9 +4,10 @@ version = "0.1.0" edition = "2024" [dependencies] -mtp = { git = "https://git.methanium.net/methanium/mtp", features = [ +mtp = { git = "https://git.methanium.net/Methanium/mtp.git", features = [ "crypto", "files", + "raw", "web-server", ] } diff --git a/src/db/user_repo.rs b/src/db/user_repo.rs index df99a18..79b9034 100644 --- a/src/db/user_repo.rs +++ b/src/db/user_repo.rs @@ -380,27 +380,44 @@ pub async fn delete_user_with_pending_erasure(id: UserId) -> Result = row.get("iota_id"); if let Some(iota_id) = iota_id { - sqlx::query("INSERT IGNORE INTO pending_iota_user_erasure (user_id, iota_id) VALUES (?, ?)") - .bind(id.0) - .bind(iota_id) - .execute(&mut *tx) - .await?; + sqlx::query( + "INSERT IGNORE INTO pending_iota_user_erasure (user_id, iota_id) VALUES (?, ?)", + ) + .bind(id.0) + .bind(iota_id) + .execute(&mut *tx) + .await?; } - sqlx::query("DELETE FROM registration_leases WHERE user_id = ?").bind(id.0).execute(&mut *tx).await?; - sqlx::query("DELETE FROM users WHERE id = ?").bind(id.0).execute(&mut *tx).await?; + sqlx::query("DELETE FROM registration_leases WHERE user_id = ?") + .bind(id.0) + .execute(&mut *tx) + .await?; + sqlx::query("DELETE FROM users WHERE id = ?") + .bind(id.0) + .execute(&mut *tx) + .await?; tx.commit().await?; Ok(iota_id.map(IotaId::from)) } pub async fn pending_erasures_for_iota(iota_id: IotaId) -> Result> { let rows = sqlx::query("SELECT user_id FROM pending_iota_user_erasure WHERE iota_id = ?") - .bind(iota_id.0).fetch_all(&pool().await?).await?; - Ok(rows.into_iter().map(|row| UserId::from(row.get::("user_id"))).collect()) + .bind(iota_id.0) + .fetch_all(&pool().await?) + .await?; + Ok(rows + .into_iter() + .map(|row| UserId::from(row.get::("user_id"))) + .collect()) } pub async fn acknowledge_pending_erasure(user_id: UserId, iota_id: IotaId) -> Result { - let result = sqlx::query("DELETE FROM pending_iota_user_erasure WHERE user_id = ? AND iota_id = ?") - .bind(user_id.0).bind(iota_id.0).execute(&pool().await?).await?; + let result = + sqlx::query("DELETE FROM pending_iota_user_erasure WHERE user_id = ? AND iota_id = ?") + .bind(user_id.0) + .bind(iota_id.0) + .execute(&pool().await?) + .await?; Ok(result.rows_affected() == 1) } diff --git a/src/server/api.rs b/src/server/api.rs index 0319607..23debe4 100644 --- a/src/server/api.rs +++ b/src/server/api.rs @@ -85,11 +85,13 @@ async fn route(path_parts: &[&str]) -> Result<(StatusCode, String)> { let user = get_by_user_id(UserId::from(id)).await?; match user.iota_id { Some(iota_id) => get_iota_primary_omikron_connection(iota_id.0), - None => get_random_omikron() - .await - .map_err(|_| OmegaError::NotFound)? - .get_omikron_id() - .await, + None => { + get_random_omikron() + .await + .map_err(|_| OmegaError::NotFound)? + .get_omikron_id() + .await + } } .ok_or(OmegaError::NotFound)? }; diff --git a/src/sql/user_online_tracker.rs b/src/sql/user_online_tracker.rs index 8017c3d..be3e87c 100644 --- a/src/sql/user_online_tracker.rs +++ b/src/sql/user_online_tracker.rs @@ -1044,20 +1044,24 @@ mod tests { assert_eq!(tracker.subscribers(22).len(), 1); tracker.remove_session(7, 3, 42); assert!(tracker.check_index_consistency().is_ok()); - assert!(tracker - .routes - .read() - .unwrap() - .indices - .sessions_by_user - .is_empty()); - assert!(tracker - .routes - .read() - .unwrap() - .indices - .targets_by_subscriber - .is_empty()); + assert!( + tracker + .routes + .read() + .unwrap() + .indices + .sessions_by_user + .is_empty() + ); + assert!( + tracker + .routes + .read() + .unwrap() + .indices + .targets_by_subscriber + .is_empty() + ); } #[test] diff --git a/src/state.rs b/src/state.rs index 075d268..02c899b 100644 --- a/src/state.rs +++ b/src/state.rs @@ -1,10 +1,13 @@ use crate::sql::user_online_tracker::PresenceTracker; -use std::sync::Arc; use dashmap::DashMap; +use std::sync::Arc; use std::time::{Duration, Instant}; #[derive(Clone, Copy, Debug, PartialEq, Eq, Hash)] -pub enum AccountChallengeOperation { Attach, Delete } +pub enum AccountChallengeOperation { + Attach, + Delete, +} #[derive(Clone, Debug)] pub struct AccountChallenge { @@ -30,15 +33,38 @@ impl Default for OmegaState { } impl OmegaState { - pub fn issue_challenge(&self, operation: AccountChallengeOperation, user_id: i64, requester_iota_id: i64) -> u64 { + pub fn issue_challenge( + &self, + operation: AccountChallengeOperation, + user_id: i64, + requester_iota_id: i64, + ) -> u64 { let nonce = rand::random::(); - self.challenges.insert((operation, user_id, requester_iota_id), AccountChallenge { operation, user_id, requester_iota_id, nonce, created_at: Instant::now() }); + self.challenges.insert( + (operation, user_id, requester_iota_id), + AccountChallenge { + operation, + user_id, + requester_iota_id, + nonce, + created_at: Instant::now(), + }, + ); nonce } - pub fn consume_challenge(&self, operation: AccountChallengeOperation, user_id: i64, requester_iota_id: i64, nonce: u64) -> bool { - self.challenges.remove(&(operation, user_id, requester_iota_id)).is_some_and(|(_, value)| - value.nonce == nonce && value.created_at.elapsed() <= Duration::from_secs(120)) + pub fn consume_challenge( + &self, + operation: AccountChallengeOperation, + user_id: i64, + requester_iota_id: i64, + nonce: u64, + ) -> bool { + self.challenges + .remove(&(operation, user_id, requester_iota_id)) + .is_some_and(|(_, value)| { + value.nonce == nonce && value.created_at.elapsed() <= Duration::from_secs(120) + }) } } diff --git a/src/transport/connection.rs b/src/transport/connection.rs index eff0bb1..6cada3e 100644 --- a/src/transport/connection.rs +++ b/src/transport/connection.rs @@ -1 +1,48 @@ -pub(crate) use super::omikron_connection::OmikronConnection; +pub(crate) use super::omikron_connection::{OmikronConnection, OmikronResult}; + +use mtp::codec::{CommunicationValue, DataValue}; + +pub(crate) trait MtpValueCompat { + fn get_id(&self) -> u32; + fn get_sender(&self) -> u64; + fn get_receiver(&self) -> u64; +} + +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() + } +} + +pub(crate) trait OptionalDataValueCompat { + 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_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()) + } +} diff --git a/src/transport/handlers/account.rs b/src/transport/handlers/account.rs index 360076d..0374f78 100644 --- a/src/transport/handlers/account.rs +++ b/src/transport/handlers/account.rs @@ -1,10 +1,15 @@ -use super::super::omikron_connection::{OmikronConnection, OmikronResult}; +use super::super::connection::{ + MtpValueCompat, OmikronConnection, OmikronResult, OptionalDataValueCompat, +}; use crate::{ db::{iota_repo, user_repo}, models::{IotaId, UserId}, state::AccountChallengeOperation, }; -use mtp::{codec::{CommunicationType, CommunicationValue, DataType, DataValue}, crypto::{verify_ed25519, verify_ml_dsa}}; +use mtp::{ + codec::{CommunicationType, CommunicationValue, DataType, DataValue}, + crypto::{verify_ed25519, verify_ml_dsa}, +}; use std::sync::Arc; async fn delete( @@ -67,14 +72,22 @@ pub async fn release_from_iota( let previous_iota = user.iota_id; match user_repo::change_iota_id(user.id, None).await { Ok(()) => { - if let Some(iota) = previous_iota { crate::transport::omikron_manager::publish_iota_user_snapshot(iota.0).await; } - connection.send(&CommunicationValue::new(CommunicationType::Success).with_id(value.get_id())).await - }, - Err(error) => connection - .send(&CommunicationValue::new(CommunicationType::ErrorInternal) - .with_id(value.get_id()) - .add_typed_default(DataType::ErrorType, DataValue::Str(error.to_string()))) - .await, + if let Some(iota) = previous_iota { + crate::transport::omikron_manager::publish_iota_user_snapshot(iota.0).await; + } + connection + .send(&CommunicationValue::new(CommunicationType::Success).with_id(value.get_id())) + .await + } + Err(error) => { + connection + .send( + &CommunicationValue::new(CommunicationType::ErrorInternal) + .with_id(value.get_id()) + .add_typed_default(DataType::ErrorType, DataValue::Str(error.to_string())), + ) + .await + } } } @@ -87,43 +100,121 @@ fn lifecycle_payload(domain: &[u8], user_id: i64, iota_id: i64, nonce: u64) -> V payload } -pub async fn attach_begin(connection: Arc, value: CommunicationValue) -> OmikronResult<()> { - let Some(user_id) = value.get_data(DataType::UserId).as_signed_number().and_then(|v| i64::try_from(v).ok()).filter(|v| *v > 0) else { - return connection.send_error_response(value.get_id(), CommunicationType::ErrorInvalidUserId).await; +pub async fn attach_begin( + connection: Arc, + value: CommunicationValue, +) -> OmikronResult<()> { + let Some(user_id) = value + .get_data(DataType::UserId) + .as_signed_number() + .and_then(|v| i64::try_from(v).ok()) + .filter(|v| *v > 0) + else { + return connection + .send_error_response(value.get_id(), CommunicationType::ErrorInvalidUserId) + .await; }; - if user_repo::get_by_user_id(UserId::from(user_id)).await.is_err() { - return connection.send_error_response(value.get_id(), CommunicationType::ErrorNotFound).await; + if user_repo::get_by_user_id(UserId::from(user_id)) + .await + .is_err() + { + return connection + .send_error_response(value.get_id(), CommunicationType::ErrorNotFound) + .await; } let requester = value.get_sender() as i64; - let nonce = connection.state().issue_challenge(AccountChallengeOperation::Attach, user_id, requester); - connection.send(&CommunicationValue::new(CommunicationType::AttachUserChallenge).with_id(value.get_id()) - .add_typed_default(DataType::UserId, DataValue::SignedNumber(user_id.into())) - .add_typed_default(DataType::ServerNonce, DataValue::SignedNumber(nonce.into()))).await + let nonce = + connection + .state() + .issue_challenge(AccountChallengeOperation::Attach, user_id, requester); + connection + .send( + &CommunicationValue::new(CommunicationType::AttachUserChallenge) + .with_id(value.get_id()) + .add_typed_default(DataType::UserId, DataValue::SignedNumber(user_id.into())) + .add_typed_default(DataType::ServerNonce, DataValue::SignedNumber(nonce.into())), + ) + .await } -pub async fn attach_complete(connection: Arc, value: CommunicationValue) -> OmikronResult<()> { - let Some(user_id) = value.get_data(DataType::UserId).as_signed_number().and_then(|v| i64::try_from(v).ok()).filter(|v| *v > 0) else { return connection.send_error_response(value.get_id(), CommunicationType::ErrorInvalidUserId).await; }; +pub async fn attach_complete( + connection: Arc, + value: CommunicationValue, +) -> OmikronResult<()> { + let Some(user_id) = value + .get_data(DataType::UserId) + .as_signed_number() + .and_then(|v| i64::try_from(v).ok()) + .filter(|v| *v > 0) + else { + return connection + .send_error_response(value.get_id(), CommunicationType::ErrorInvalidUserId) + .await; + }; let requester = value.get_sender() as i64; - let Some(nonce) = value.get_data(DataType::ServerNonce).as_signed_number().and_then(|v| u64::try_from(v).ok()) else { return connection.send_error_response(value.get_id(), CommunicationType::ErrorInvalidChallenge).await; }; + let Some(nonce) = value + .get_data(DataType::ServerNonce) + .as_signed_number() + .and_then(|v| u64::try_from(v).ok()) + else { + return connection + .send_error_response(value.get_id(), CommunicationType::ErrorInvalidChallenge) + .await; + }; let signature = value.get_data(DataType::Signature).as_bytes(); let pq_signature = value.get_data(DataType::PqSignature).as_bytes(); - let (Some(signature), Some(pq_signature)) = (signature, pq_signature) else { return connection.send_error_response(value.get_id(), CommunicationType::ErrorInvalidChallenge).await; }; - if !connection.state().consume_challenge(AccountChallengeOperation::Attach, user_id, requester, nonce) { return connection.send_error_response(value.get_id(), CommunicationType::ErrorInvalidChallenge).await; } - let Ok(user) = user_repo::get_by_user_id(UserId::from(user_id)).await else { return connection.send_error_response(value.get_id(), CommunicationType::ErrorNotFound).await; }; + let (Some(signature), Some(pq_signature)) = (signature, pq_signature) else { + return connection + .send_error_response(value.get_id(), CommunicationType::ErrorInvalidChallenge) + .await; + }; + if !connection.state().consume_challenge( + AccountChallengeOperation::Attach, + user_id, + requester, + nonce, + ) { + return connection + .send_error_response(value.get_id(), CommunicationType::ErrorInvalidChallenge) + .await; + } + let Ok(user) = user_repo::get_by_user_id(UserId::from(user_id)).await else { + return connection + .send_error_response(value.get_id(), CommunicationType::ErrorNotFound) + .await; + }; let payload = lifecycle_payload(b"tensamin:user-attach:v1\0", user_id, requester, nonce); - if verify_ed25519(&user.public_key.sig_cl_public_key, &payload, &signature).is_err() || verify_ml_dsa(&user.public_key.sig_pq_public_key, &payload, &pq_signature).is_err() { return connection.send_error_response(value.get_id(), CommunicationType::ErrorNotAuthenticated).await; } + if verify_ed25519(&user.public_key.sig_cl_public_key, &payload, &signature).is_err() + || verify_ml_dsa(&user.public_key.sig_pq_public_key, &payload, &pq_signature).is_err() + { + return connection + .send_error_response(value.get_id(), CommunicationType::ErrorNotAuthenticated) + .await; + } let previous_iota = user.iota_id; match user_repo::change_iota_id(user.id, Some(IotaId::from(requester))).await { Ok(()) => { - if let Some(iota) = previous_iota.filter(|id| id.0 != requester) { crate::transport::omikron_manager::publish_iota_user_snapshot(iota.0).await; } + if let Some(iota) = previous_iota.filter(|id| id.0 != requester) { + crate::transport::omikron_manager::publish_iota_user_snapshot(iota.0).await; + } crate::transport::omikron_manager::publish_iota_user_snapshot(requester).await; - connection.send(&CommunicationValue::new(CommunicationType::Success).with_id(value.get_id())).await - }, - Err(_) => connection.send_error_response(value.get_id(), CommunicationType::ErrorInternal).await, + connection + .send(&CommunicationValue::new(CommunicationType::Success).with_id(value.get_id())) + .await + } + Err(_) => { + connection + .send_error_response(value.get_id(), CommunicationType::ErrorInternal) + .await + } } } -async fn complete_delete(connection: Arc, value: CommunicationValue, user_id: UserId) -> OmikronResult<()> { +async fn complete_delete( + connection: Arc, + value: CommunicationValue, + user_id: UserId, +) -> OmikronResult<()> { match user_repo::delete_user_with_pending_erasure(user_id).await { Ok(iota_id) => { let cleanup_pending = iota_id.is_some(); @@ -131,45 +222,160 @@ async fn complete_delete(connection: Arc, value: Communicatio crate::transport::omikron_manager::publish_iota_user_snapshot(iota_id.0).await; crate::transport::omikron_manager::deliver_pending_erasures(iota_id.0).await; } - connection.send(&CommunicationValue::new(CommunicationType::Success).with_id(value.get_id()) - .add_typed_default(DataType::CleanupPending, DataValue::Bool(cleanup_pending))).await + connection + .send( + &CommunicationValue::new(CommunicationType::Success) + .with_id(value.get_id()) + .add_typed_default( + DataType::CleanupPending, + DataValue::Bool(cleanup_pending), + ), + ) + .await + } + Err(crate::error::OmegaError::NotFound) => { + connection + .send_error_response(value.get_id(), CommunicationType::ErrorNotFound) + .await + } + Err(error) => { + connection + .send( + &CommunicationValue::new(CommunicationType::ErrorInternal) + .with_id(value.get_id()) + .add_typed_default(DataType::ErrorType, DataValue::Str(error.to_string())), + ) + .await } - Err(crate::error::OmegaError::NotFound) => connection.send_error_response(value.get_id(), CommunicationType::ErrorNotFound).await, - Err(error) => connection.send(&CommunicationValue::new(CommunicationType::ErrorInternal).with_id(value.get_id()).add_typed_default(DataType::ErrorType, DataValue::Str(error.to_string()))).await, } } -pub async fn delete_credential_begin(connection: Arc, value: CommunicationValue) -> OmikronResult<()> { - let Some(user_id) = value.get_data(DataType::UserId).as_signed_number().and_then(|v| i64::try_from(v).ok()).filter(|v| *v > 0) else { - return connection.send_error_response(value.get_id(), CommunicationType::ErrorInvalidUserId).await; +pub async fn delete_credential_begin( + connection: Arc, + value: CommunicationValue, +) -> OmikronResult<()> { + let Some(user_id) = value + .get_data(DataType::UserId) + .as_signed_number() + .and_then(|v| i64::try_from(v).ok()) + .filter(|v| *v > 0) + else { + return connection + .send_error_response(value.get_id(), CommunicationType::ErrorInvalidUserId) + .await; }; - if user_repo::get_by_user_id(UserId::from(user_id)).await.is_err() { return connection.send_error_response(value.get_id(), CommunicationType::ErrorNotFound).await; } + if user_repo::get_by_user_id(UserId::from(user_id)) + .await + .is_err() + { + return connection + .send_error_response(value.get_id(), CommunicationType::ErrorNotFound) + .await; + } let requester = value.get_sender() as i64; - let nonce = connection.state().issue_challenge(AccountChallengeOperation::Delete, user_id, requester); - connection.send(&CommunicationValue::new(CommunicationType::DeleteUserCredentialChallenge).with_id(value.get_id()) - .add_typed_default(DataType::UserId, DataValue::SignedNumber(user_id.into())) - .add_typed_default(DataType::ServerNonce, DataValue::SignedNumber(nonce.into()))).await + let nonce = + connection + .state() + .issue_challenge(AccountChallengeOperation::Delete, user_id, requester); + connection + .send( + &CommunicationValue::new(CommunicationType::DeleteUserCredentialChallenge) + .with_id(value.get_id()) + .add_typed_default(DataType::UserId, DataValue::SignedNumber(user_id.into())) + .add_typed_default(DataType::ServerNonce, DataValue::SignedNumber(nonce.into())), + ) + .await } -pub async fn delete_credential_complete(connection: Arc, value: CommunicationValue) -> OmikronResult<()> { - let Some(user_id) = value.get_data(DataType::UserId).as_signed_number().and_then(|v| i64::try_from(v).ok()).filter(|v| *v > 0) else { return connection.send_error_response(value.get_id(), CommunicationType::ErrorInvalidUserId).await; }; +pub async fn delete_credential_complete( + connection: Arc, + value: CommunicationValue, +) -> OmikronResult<()> { + let Some(user_id) = value + .get_data(DataType::UserId) + .as_signed_number() + .and_then(|v| i64::try_from(v).ok()) + .filter(|v| *v > 0) + else { + return connection + .send_error_response(value.get_id(), CommunicationType::ErrorInvalidUserId) + .await; + }; let requester = value.get_sender() as i64; - let Some(nonce) = value.get_data(DataType::ServerNonce).as_signed_number().and_then(|v| u64::try_from(v).ok()) else { return connection.send_error_response(value.get_id(), CommunicationType::ErrorInvalidChallenge).await; }; - let (Some(signature), Some(pq_signature)) = (value.get_data(DataType::Signature).as_bytes(), value.get_data(DataType::PqSignature).as_bytes()) else { return connection.send_error_response(value.get_id(), CommunicationType::ErrorInvalidChallenge).await; }; - if !connection.state().consume_challenge(AccountChallengeOperation::Delete, user_id, requester, nonce) { return connection.send_error_response(value.get_id(), CommunicationType::ErrorInvalidChallenge).await; } - let Ok(user) = user_repo::get_by_user_id(UserId::from(user_id)).await else { return connection.send_error_response(value.get_id(), CommunicationType::ErrorNotFound).await; }; + let Some(nonce) = value + .get_data(DataType::ServerNonce) + .as_signed_number() + .and_then(|v| u64::try_from(v).ok()) + else { + return connection + .send_error_response(value.get_id(), CommunicationType::ErrorInvalidChallenge) + .await; + }; + let (Some(signature), Some(pq_signature)) = ( + value.get_data(DataType::Signature).as_bytes(), + value.get_data(DataType::PqSignature).as_bytes(), + ) else { + return connection + .send_error_response(value.get_id(), CommunicationType::ErrorInvalidChallenge) + .await; + }; + if !connection.state().consume_challenge( + AccountChallengeOperation::Delete, + user_id, + requester, + nonce, + ) { + return connection + .send_error_response(value.get_id(), CommunicationType::ErrorInvalidChallenge) + .await; + } + let Ok(user) = user_repo::get_by_user_id(UserId::from(user_id)).await else { + return connection + .send_error_response(value.get_id(), CommunicationType::ErrorNotFound) + .await; + }; let payload = lifecycle_payload(b"tensamin:user-delete:v1\0", user_id, requester, nonce); - if verify_ed25519(&user.public_key.sig_cl_public_key, &payload, &signature).is_err() || verify_ml_dsa(&user.public_key.sig_pq_public_key, &payload, &pq_signature).is_err() { return connection.send_error_response(value.get_id(), CommunicationType::ErrorNotAuthenticated).await; } + if verify_ed25519(&user.public_key.sig_cl_public_key, &payload, &signature).is_err() + || verify_ml_dsa(&user.public_key.sig_pq_public_key, &payload, &pq_signature).is_err() + { + return connection + .send_error_response(value.get_id(), CommunicationType::ErrorNotAuthenticated) + .await; + } complete_delete(connection, value, user.id).await } -pub async fn erase_hosted_user_data_ack(connection: Arc, value: CommunicationValue) -> OmikronResult<()> { - let Some(user_id) = value.get_data(DataType::UserId).as_signed_number().and_then(|v| i64::try_from(v).ok()).filter(|v| *v > 0) else { return connection.send_error_response(value.get_id(), CommunicationType::ErrorInvalidUserId).await; }; +pub async fn erase_hosted_user_data_ack( + connection: Arc, + value: CommunicationValue, +) -> OmikronResult<()> { + let Some(user_id) = value + .get_data(DataType::UserId) + .as_signed_number() + .and_then(|v| i64::try_from(v).ok()) + .filter(|v| *v > 0) + else { + return connection + .send_error_response(value.get_id(), CommunicationType::ErrorInvalidUserId) + .await; + }; let iota_id = IotaId::from(value.get_sender() as i64); match user_repo::acknowledge_pending_erasure(UserId::from(user_id), iota_id).await { - Ok(true) => connection.send(&CommunicationValue::new(CommunicationType::Success).with_id(value.get_id())).await, - Ok(false) => connection.send_error_response(value.get_id(), CommunicationType::ErrorNotAuthenticated).await, - Err(_) => connection.send_error_response(value.get_id(), CommunicationType::ErrorInternal).await, + Ok(true) => { + connection + .send(&CommunicationValue::new(CommunicationType::Success).with_id(value.get_id())) + .await + } + Ok(false) => { + connection + .send_error_response(value.get_id(), CommunicationType::ErrorNotAuthenticated) + .await + } + Err(_) => { + connection + .send_error_response(value.get_id(), CommunicationType::ErrorInternal) + .await + } } } diff --git a/src/transport/handlers/links.rs b/src/transport/handlers/links.rs index a43cb99..ad45c05 100644 --- a/src/transport/handlers/links.rs +++ b/src/transport/handlers/links.rs @@ -1,4 +1,6 @@ -use super::super::omikron_connection::{OmikronConnection, OmikronResult}; +use super::super::connection::{ + MtpValueCompat, OmikronConnection, OmikronResult, OptionalDataValueCompat, +}; use crate::server::short_link::add_short_link; use mtp::codec::{CommunicationType, CommunicationValue, DataType, DataValue}; use std::sync::Arc; @@ -7,8 +9,8 @@ pub async fn shorten( connection: Arc, value: CommunicationValue, ) -> OmikronResult<()> { - let link = value - .get_data(DataType::Link) + let link_data = value.get_data(DataType::Link); + let link = link_data .as_str() .ok_or(crate::error::OmegaError::InvalidResponse)?; let short = add_short_link(link) diff --git a/src/transport/handlers/notifications.rs b/src/transport/handlers/notifications.rs index 5f23474..9405bf0 100644 --- a/src/transport/handlers/notifications.rs +++ b/src/transport/handlers/notifications.rs @@ -1,4 +1,6 @@ -use super::super::omikron_connection::{OmikronConnection, OmikronResult}; +use super::super::connection::{ + MtpValueCompat, OmikronConnection, OmikronResult, OptionalDataValueCompat, +}; use crate::{db::notification_repo, log, models::UserId}; use mtp::codec::{CommunicationType, CommunicationValue, DataType, DataValue}; use mtp::type_map::TypeMap; diff --git a/src/transport/handlers/presence.rs b/src/transport/handlers/presence.rs index de8a275..5c631bf 100644 --- a/src/transport/handlers/presence.rs +++ b/src/transport/handlers/presence.rs @@ -1,4 +1,6 @@ -use super::super::omikron_connection::{OmikronConnection, OmikronResult}; +use super::super::connection::{ + MtpValueCompat, OmikronConnection, OmikronResult, OptionalDataValueCompat, +}; use crate::{ db::user_repo, log_in, models::IotaId, sql::connection_status::UserStatus, state::OmegaState, }; @@ -19,7 +21,7 @@ fn parse_subscription(value: &CommunicationValue) -> Result<(i64, i64, Vec) .and_then(|id| i64::try_from(id).ok()) .filter(|id| *id > 0) .ok_or("session_id")?; - let DataValue::Array(values) = value.get_data(DataType::UserIds) else { + let Some(DataValue::Array(values)) = value.get_data(DataType::UserIds) else { return Err("user_ids"); }; @@ -51,9 +53,10 @@ fn states_for_users(state: &OmegaState, users: &[crate::models::User]) -> HashMa .map(|user| { ( user.id.0, - state - .presence - .resolve_public_state(user.id.0, user.iota_id.map(|id| id.0).unwrap_or_default()), + state.presence.resolve_public_state( + user.id.0, + user.iota_id.map(|id| id.0).unwrap_or_default(), + ), ) }) .collect() @@ -349,7 +352,7 @@ pub async fn set_user_state( .send_error_response(value.get_id(), CommunicationType::ErrorNoUserId) .await; }; - if let Some(requested_user) = value.get_data_opt(DataType::UserId) { + if let Some(requested_user) = value.get_data(DataType::UserId) { let Some(requested_user_id) = requested_user .as_number() .and_then(|id| i64::try_from(id).ok()) @@ -538,12 +541,12 @@ pub async fn sync_status( omikron_id: i64, ) -> OmikronResult<()> { let request_id = value.get_id(); - let DataValue::Array(iota_values) = value.get_data(DataType::IotaIds) else { + let Some(DataValue::Array(iota_values)) = value.get_data(DataType::IotaIds) else { return connection .send_error_response(request_id, CommunicationType::ErrorInvalidData) .await; }; - let DataValue::Array(session_values) = value.get_data(DataType::UserStates) else { + let Some(DataValue::Array(session_values)) = value.get_data(DataType::UserStates) else { return connection .send_error_response(request_id, CommunicationType::ErrorInvalidData) .await; @@ -572,7 +575,7 @@ pub async fn sync_status( } if !connection.peer_capabilities().session_snapshot_v1 { - let DataValue::Array(user_values) = value.get_data(DataType::UserIds) else { + let Some(DataValue::Array(user_values)) = value.get_data(DataType::UserIds) else { return connection .send_error_response(request_id, CommunicationType::ErrorInvalidData) .await; diff --git a/src/transport/handlers/register.rs b/src/transport/handlers/register.rs index 3502269..8799893 100644 --- a/src/transport/handlers/register.rs +++ b/src/transport/handlers/register.rs @@ -1,4 +1,6 @@ -use super::super::omikron_connection::{OmikronConnection, OmikronResult}; +use super::super::connection::{ + MtpValueCompat, OmikronConnection, OmikronResult, OptionalDataValueCompat, +}; use crate::{ db::{iota_repo, user_repo}, models::{IotaId, UserId}, diff --git a/src/transport/handlers/states.rs b/src/transport/handlers/states.rs index d147f27..e1fc12d 100644 --- a/src/transport/handlers/states.rs +++ b/src/transport/handlers/states.rs @@ -1,4 +1,6 @@ -use super::super::omikron_connection::{OmikronConnection, OmikronResult}; +use super::super::connection::{ + MtpValueCompat, OmikronConnection, OmikronResult, OptionalDataValueCompat, +}; use crate::db::user_repo; use mtp::{ codec::{CommunicationType, CommunicationValue, DataType, DataValue}, @@ -29,7 +31,7 @@ pub async fn get( ) -> OmikronResult<()> { let state = connection.state(); let legacy_peer = !connection.peer_capabilities().client_state_push_v1; - let DataValue::Array(ids) = value.get_data(DataType::UserIds) else { + let Some(DataValue::Array(ids)) = value.get_data(DataType::UserIds) else { return send_error( connection, value.get_id(), diff --git a/src/transport/handlers/user_data.rs b/src/transport/handlers/user_data.rs index e331125..1c248fa 100644 --- a/src/transport/handlers/user_data.rs +++ b/src/transport/handlers/user_data.rs @@ -1,4 +1,6 @@ -use super::super::omikron_connection::{OmikronConnection, OmikronResult}; +use super::super::connection::{ + MtpValueCompat, OmikronConnection, OmikronResult, OptionalDataValueCompat, +}; use crate::{ db::{iota_repo, user_repo}, models::{IotaId, UserId}, @@ -91,7 +93,9 @@ pub async fn get_user( } state.presence.resolve_private_state(id) } else { - iota_id.map(|iota_id| state.presence.resolve_public_state(id, iota_id)).unwrap_or(crate::sql::connection_status::UserStatus::user_offline) + iota_id + .map(|iota_id| state.presence.resolve_public_state(id, iota_id)) + .unwrap_or(crate::sql::connection_status::UserStatus::user_offline) }; response = response .add_typed_default( @@ -100,10 +104,13 @@ pub async fn get_user( ) .add_typed_default( DataType::OmikronConnections, - iota_id.map(|iota_id| connections(&connection, iota_id)).unwrap_or_else(|| DataValue::Array(Vec::new())), + iota_id + .map(|iota_id| connections(&connection, iota_id)) + .unwrap_or_else(|| DataValue::Array(Vec::new())), ); if let Some(iota_id) = iota_id { - response = response.add_typed_default(DataType::IotaId, DataValue::SignedNumber(iota_id.into())); + response = + response.add_typed_default(DataType::IotaId, DataValue::SignedNumber(iota_id.into())); } if let Some(route) = route { response = response.add_typed_default( @@ -138,10 +145,14 @@ pub async fn get_iota( } else if let Some(name) = value.get_data(DataType::Username).as_str() { if let Ok(user) = user_repo::get_by_username(name).await { match user.iota_id { - Some(iota_id) => iota_repo::get_iota_by_id(iota_id) - .await - .ok() - .map(|iota| (iota.id.0, iota.public_key, Some(user.id.0), Some(name.to_owned()))), + Some(iota_id) => iota_repo::get_iota_by_id(iota_id).await.ok().map(|iota| { + ( + iota.id.0, + iota.public_key, + Some(user.id.0), + Some(name.to_owned()), + ) + }), None => None, } } else { @@ -245,12 +256,14 @@ pub async fn change_iota( connection: Arc, value: CommunicationValue, ) -> OmikronResult<()> { - let Some(reset) = value.get_data(DataType::ResetToken).as_str() else { + let reset_data = value.get_data(DataType::ResetToken); + let Some(reset) = reset_data.as_str() else { return connection .send_error_response(value.get_id(), CommunicationType::ErrorInvalidData) .await; }; - let Some(new_token) = value.get_data(DataType::NewToken).as_str() else { + let new_token_data = value.get_data(DataType::NewToken); + let Some(new_token) = new_token_data.as_str() else { return connection .send_error_response(value.get_id(), CommunicationType::ErrorInvalidData) .await; @@ -270,7 +283,9 @@ pub async fn change_iota( .await; } let result = - match user_repo::change_iota_id(user_id, Some(IotaId::from(value.get_sender() as i64))).await { + match user_repo::change_iota_id(user_id, Some(IotaId::from(value.get_sender() as i64))) + .await + { Ok(()) => user_repo::change_token(user_id, new_token.to_owned()).await, Err(error) => Err(error), }; diff --git a/src/transport/mod.rs b/src/transport/mod.rs index 52c5fe8..3107217 100644 --- a/src/transport/mod.rs +++ b/src/transport/mod.rs @@ -3,3 +3,4 @@ pub mod connection; pub mod handlers; pub mod omikron_connection; pub mod omikron_manager; +pub mod relay_router; diff --git a/src/transport/omikron_connection.rs b/src/transport/omikron_connection.rs index 7ad0136..82b9ece 100644 --- a/src/transport/omikron_connection.rs +++ b/src/transport/omikron_connection.rs @@ -10,7 +10,7 @@ use dashmap::DashMap; use mtp::{ codec::{CommunicationType, CommunicationValue}, crypto::PublicKeyBundle, - host::{AuthenticationPolicy, HostConfig, Policy, SendMode}, + host::{AuthState, AuthenticationPolicy, HostConfig, Policy, SendMode}, webserver::{MTPWebServer, WebMtpReceiver, WebMtpSender}, }; use std::{ @@ -21,7 +21,10 @@ use std::{ }, time::{Duration, Instant}, }; -use tokio::{sync::Mutex, time::interval}; +use tokio::{ + sync::{Mutex, mpsc}, + time::interval, +}; const CLEANUP_INTERVAL: Duration = Duration::from_secs(30); const MAX_WAITING_AGE: Duration = Duration::from_secs(60); @@ -70,7 +73,11 @@ impl OmikronConnection { id: u64, description: Option<&str>, state: Arc, + authenticated: bool, ) -> Option> { + if !authenticated { + return None; + } let peer_capabilities = PeerCapabilities::from_identification_description(description).ok()?; Some(Arc::new(Self { @@ -87,7 +94,6 @@ impl OmikronConnection { &self.peer_capabilities } - pub async fn handle(self: Arc, receiver: &mut WebMtpReceiver) { log_in!( self.id as i64, @@ -158,7 +164,12 @@ impl OmikronConnection { async fn process_message(self: Arc, value: CommunicationValue) -> OmikronResult<()> { log_cv_in!(PrintType::Omikron, &value); - if let Some((_, task)) = self.waiting_tasks.remove(&value.get_id()) { + if value.is_type(CommunicationType::Relay) { + return self.dispatch(value).await; + } + if let Some(message_id) = value.id() + && let Some((_, task)) = self.waiting_tasks.remove(&message_id) + { let _ = (task.task)(self.clone(), value); return Ok(()); } @@ -168,6 +179,36 @@ impl OmikronConnection { async fn dispatch(self: Arc, value: CommunicationValue) -> OmikronResult<()> { let id = self.id as i64; let state = self.state.clone(); + if value.is_type(CommunicationType::Relay) { + let value = crate::transport::relay_router::ensure_relay_frame_id(value); + let request_id = value.id(); + let result = crate::transport::relay_router::route_from_omikron(id, value).await; + let response = match &result { + Ok(()) => request_id.map(|request_id| { + CommunicationValue::new(CommunicationType::Success).with_id(request_id) + }), + Err(error) => { + log_err!(id, PrintType::Omega, "Relay routing failed: {}", error); + request_id.map(|request_id| { + CommunicationValue::new( + crate::transport::relay_router::error_response_type(error), + ) + .with_id(request_id) + }) + } + }; + if let Some(response) = response + && let Err(send_error) = self.clone().send(&response).await + { + log_err!( + id, + PrintType::Omega, + "Relay routing response failed: {}", + send_error + ); + } + return result.map_err(|error| crate::error::OmegaError::Transport(error.to_string())); + } match value.get_comm_type_enum() { Some(CommunicationType::ShortenLink) => { crate::transport::handlers::links::shorten(self, value).await @@ -179,9 +220,6 @@ impl OmikronConnection { crate::transport::handlers::presence::user_disconnected(state, self, value, id) .await } - Some(CommunicationType::SetUserState) => { - crate::transport::handlers::presence::set_user_state(state, self, value, id).await - } Some(CommunicationType::IotaConnected) => { crate::transport::handlers::presence::iota_connected(state, self, value, id).await } @@ -297,6 +335,47 @@ impl OmikronConnection { } Ok(()) } + + pub(crate) async fn await_response( + self: Arc, + value: &CommunicationValue, + timeout: Duration, + ) -> OmikronResult { + let (tx, mut rx) = mpsc::channel(1); + let message_id = value.id().unwrap_or_default(); + let task_tx = tx.clone(); + self.waiting_tasks.insert( + message_id, + WaitingTask { + task: Box::new(move |_, response| { + let task_tx = task_tx.clone(); + tokio::spawn(async move { + let _ = task_tx.send(response).await; + }); + true + }), + inserted_at: Instant::now(), + }, + ); + + if let Err(error) = self.clone().send(value).await { + self.waiting_tasks.remove(&message_id); + return Err(error); + } + + match tokio::time::timeout(timeout, rx.recv()).await { + Ok(Some(response)) => Ok(response), + Ok(None) => Err(crate::error::OmegaError::Transport( + "Relay response channel closed".into(), + )), + Err(_) => { + self.waiting_tasks.remove(&message_id); + Err(crate::error::OmegaError::Transport( + "Relay response timed out".into(), + )) + } + } + } pub(crate) async fn send_error_response( self: Arc, message_id: u32, @@ -439,16 +518,18 @@ pub async fn start(port: u16, state: Arc) -> Result<(), Box 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, Error)] +pub enum RelayRouteError { + #[error("relay has no destination Iota")] + MissingDestinationIota, + #[error("relay has an outer sender")] + OuterSenderPresent, + #[error("relay has invalid route target {0}")] + InvalidDestinationTarget(u64), + #[error("relay destination Iota is outside Omega's ID range")] + DestinationIotaOutOfRange, + #[error("destination Iota is offline")] + IotaOffline, + #[error("destination Omikron is offline")] + OmikronOffline, + #[error("relay route resolves back to source Omikron")] + RouteLoop, + #[error(transparent)] + Relay(#[from] RelayError), + #[error("sending relay to destination Omikron failed: {0}")] + Send(String), +} + +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) +} + +pub fn error_response_type(error: &RelayRouteError) -> CommunicationType { + match error { + RelayRouteError::IotaOffline | RelayRouteError::OmikronOffline => { + CommunicationType::ErrorNoIota + } + RelayRouteError::Send(_) => CommunicationType::ErrorInternal, + RelayRouteError::MissingDestinationIota + | RelayRouteError::OuterSenderPresent + | RelayRouteError::InvalidDestinationTarget(_) + | RelayRouteError::DestinationIotaOutOfRange + | RelayRouteError::RouteLoop + | RelayRouteError::Relay(_) => CommunicationType::ErrorInvalidData, + } +} + +pub async fn route_from_omikron( + source_omikron_id: i64, + frame: CommunicationValue, +) -> Result<(), RelayRouteError> { + let frame = ensure_relay_frame_id(frame); + if !frame.is_type(CommunicationType::Relay) { + return Err(RelayRouteError::Relay(RelayError::NotRelay)); + } + if frame.sender().is_some() { + return Err(RelayRouteError::OuterSenderPresent); + } + let Some(destination_wire_id) = frame.receiver() else { + log_err!( + source_omikron_id, + PrintType::Omega, + "Relay routing failed: missing destination Iota" + ); + return Err(RelayRouteError::MissingDestinationIota); + }; + let Some(RouteTarget::Iota(destination_iota)) = RouteTarget::from_wire_id(destination_wire_id) + else { + return Err(RelayRouteError::InvalidDestinationTarget( + destination_wire_id, + )); + }; + let destination_iota_i64 = + i64::try_from(destination_iota).map_err(|_| RelayRouteError::DestinationIotaOutOfRange)?; + let frame = forward_relay_frame(&frame, destination_wire_id)?; + + let Some(destination_omikron) = + omikron_manager::get_iota_primary_omikron_connection(destination_iota_i64) + else { + log_err!( + source_omikron_id, + PrintType::Omega, + "Relay destination Iota {} is offline", + destination_iota + ); + return Err(RelayRouteError::IotaOffline); + }; + if destination_omikron == source_omikron_id { + log_err!( + source_omikron_id, + PrintType::Omega, + "Relay route loop for destination Iota {} and Omikron {}", + destination_iota, + destination_omikron + ); + return Err(RelayRouteError::RouteLoop); + } + let Some(connection) = omikron_manager::get_connected_omikron(destination_omikron) else { + log_err!( + source_omikron_id, + PrintType::Omega, + "Relay destination Iota {} resolves to disconnected Omikron {}", + destination_iota, + destination_omikron + ); + return Err(RelayRouteError::OmikronOffline); + }; + let response = connection + .await_response(&frame, Duration::from_secs(20)) + .await + .map_err(|error| { + log_err!( + source_omikron_id, + PrintType::Omega, + "Relay send to destination Iota {} via Omikron {} failed: {}", + destination_iota, + destination_omikron, + error + ); + RelayRouteError::Send(error.to_string()) + })?; + if response.is_type(CommunicationType::Success) { + Ok(()) + } else { + Err(RelayRouteError::Send(format!( + "destination Omikron rejected the Relay with {}", + response.get_type() + ))) + } +} + +#[cfg(test)] +mod tests { + use super::*; + use mtp::codec::DataValue; + + fn wire(target: RouteTarget) -> u64 { + let Some(value) = target.wire_id() else { + panic!("valid route target was rejected"); + }; + value + } + + fn relay_frame() -> CommunicationValue { + CommunicationValue::new(CommunicationType::Relay) + .without_sender() + .with_receiver(wire(RouteTarget::Iota(42))) + .with_payload(DataValue::Bytes(vec![1, 2, 3, 4])) + } + + #[test] + fn forwarding_preserves_relay_payload_and_next_hop() { + let frame = relay_frame().with_receiver(wire(RouteTarget::Iota(7))); + let result = forward_relay_frame(&frame, wire(RouteTarget::Iota(42))); + assert!(result.is_ok()); + let Ok(forwarded) = result else { return }; + assert_eq!(forwarded.receiver(), Some(wire(RouteTarget::Iota(42)))); + assert_eq!(forwarded.sender(), None); + assert_eq!(forwarded.payload(), frame.payload()); + assert_eq!(forwarded.id(), frame.id()); + } + + #[tokio::test] + async fn outer_sender_is_rejected_before_route_lookup() { + let frame = relay_frame().with_sender(9); + let result = route_from_omikron(1, frame).await; + assert!(matches!(result, Err(RelayRouteError::OuterSenderPresent))); + } + + #[tokio::test] + async fn missing_destination_is_rejected_before_route_lookup() { + let frame = relay_frame().without_receiver(); + let result = route_from_omikron(1, frame).await; + assert!(matches!( + result, + Err(RelayRouteError::MissingDestinationIota) + )); + } + + #[tokio::test] + async fn user_route_target_is_rejected_by_opaque_omega_router() { + let frame = relay_frame().with_receiver(wire(RouteTarget::User(42))); + let result = route_from_omikron(1, frame).await; + assert!(matches!( + result, + Err(RelayRouteError::InvalidDestinationTarget(_)) + )); + } + + #[tokio::test] + async fn unavailable_destination_is_reported_as_iota_offline() { + let result = route_from_omikron(1, relay_frame()).await; + assert!(matches!(result, Err(RelayRouteError::IotaOffline))); + } +} diff --git a/src/util/logger.rs b/src/util/logger.rs index 4276cc0..9bd211a 100644 --- a/src/util/logger.rs +++ b/src/util/logger.rs @@ -10,6 +10,7 @@ use std::{ use ansi_term::Color; use mtp::codec::{CommunicationValue, DataTypeId, DataValue, Version}; +use crate::transport::connection::MtpValueCompat; use crate::util::file_util::get_directory; static LOGGER: OnceLock> = OnceLock::new(); @@ -250,11 +251,11 @@ 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(); + 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));