From 4f7260419a13d1e9256c96fe293778d8504ba526 Mon Sep 17 00:00:00 2001 From: Alex Date: Mon, 27 Jul 2026 20:36:23 +0200 Subject: [PATCH] [Fix] Stability --- .cargo/config.toml | 2 +- .gitmodules | 3 + Cargo.lock | 85 +++--- .../{001.sql => 001_initial_schema.sql} | 16 +- migrations/002_registration_leases.sql | 13 + .../003_registration_lease_request_id.sql | 8 + mtp-type-maps | 1 + src/config.rs | 4 +- src/db/iota_repo.rs | 26 +- src/db/user_repo.rs | 182 +++++++++++- src/server/api.rs | 37 ++- src/server/index.rs | 29 +- src/server/server.rs | 7 +- src/sql/user_online_tracker.rs | 56 ++++ src/transport/handlers/presence.rs | 59 +++- src/transport/handlers/register.rs | 119 ++++---- src/transport/omikron_connection.rs | 94 +++--- src/transport/omikron_manager.rs | 10 +- src/util/logger.rs | 7 +- type-maps.yaml | 269 ------------------ 20 files changed, 563 insertions(+), 464 deletions(-) create mode 100644 .gitmodules rename migrations/{001.sql => 001_initial_schema.sql} (91%) create mode 100644 migrations/002_registration_leases.sql create mode 100644 migrations/003_registration_lease_request_id.sql create mode 160000 mtp-type-maps delete mode 100644 type-maps.yaml diff --git a/.cargo/config.toml b/.cargo/config.toml index d363b83..46adaad 100644 --- a/.cargo/config.toml +++ b/.cargo/config.toml @@ -1,2 +1,2 @@ [env] -MTP_TYPE_MAPS = { value = "type-maps.yaml", relative = true } +MTP_TYPE_MAPS = { value = "mtp-type-maps/type-maps.yaml", relative = true } diff --git a/.gitmodules b/.gitmodules new file mode 100644 index 0000000..1e9ee12 --- /dev/null +++ b/.gitmodules @@ -0,0 +1,3 @@ +[submodule "mtp-type-maps"] + path = mtp-type-maps + url = https://git.methanium.net/tensamin/mtp-type-maps diff --git a/Cargo.lock b/Cargo.lock index a20fbae..8c1774a 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -20,9 +20,9 @@ dependencies = [ [[package]] name = "aes" -version = "0.9.1" +version = "0.9.2" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "f1fc76eaeac4c9164506c466d4ffdd8ec9d0c5bf57ee97177c4d8eceb3a0e138" +checksum = "f8eb277bec05f56a0e0591f155a484cbd0f4f07ff2905051a48c72f004f7ed58" dependencies = [ "cipher 0.5.2", "cpubits", @@ -91,7 +91,7 @@ checksum = "ae36dc4177970ef04fde5178d3e2429882def40e57a451f919c098f72baa6cec" dependencies = [ "proc-macro2", "quote", - "syn 3.0.2", + "syn 3.0.3", ] [[package]] @@ -217,9 +217,9 @@ dependencies = [ [[package]] name = "cc" -version = "1.3.0" +version = "1.4.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "c89588d05638b5b4594a3348a2d6c20277e43a7f5c5202b05cc56888475a47b8" +checksum = "5add81bb678e6cb321aff7fa0dc7689ad82b112dbc032cea19f91d6b8e3582b9" dependencies = [ "find-msvc-tools", "jobserver", @@ -320,15 +320,6 @@ dependencies = [ "memchr", ] -[[package]] -name = "concurrent-queue" -version = "2.5.0" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "4ca0197aee26d1ae37445ee532fefce43251d24cc7c166799f4d46817f1d3973" -dependencies = [ - "crossbeam-utils", -] - [[package]] name = "const-oid" version = "0.9.6" @@ -641,9 +632,9 @@ dependencies = [ [[package]] name = "either" -version = "1.16.0" +version = "1.17.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "91622ff5e7162018101f2fea40d6ebf4a78bbe5a49736a2020649edf9693679e" +checksum = "9e5e8f6c15a24b9a3ee5efec809ccd006d3b30e8b3bb63c39af737c7f87daa1d" dependencies = [ "serde", ] @@ -686,11 +677,10 @@ dependencies = [ [[package]] name = "event-listener" -version = "5.4.1" +version = "5.4.2" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "e13b66accf52311f30a0db42147dadea9850cb48cd070028831ae5f5d4b856ab" +checksum = "5a23add41df1562121a9393cb065eab5146a1242410f23a644851e90cfd669d2" dependencies = [ - "concurrent-queue", "parking", "pin-project-lite", ] @@ -1124,6 +1114,12 @@ version = "1.10.1" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "6dbf3de79e51f3d586ab4cb9d5c3e2c14aa28ed23d180cf89b4df0454a69cc87" +[[package]] +name = "httpdate" +version = "1.0.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "df3b46402a9d5adb4c86a0cf463f42e19994e3ee891101b1841f30a545cb49a9" + [[package]] name = "hybrid-array" version = "0.4.13" @@ -1148,6 +1144,7 @@ dependencies = [ "http", "http-body", "httparse", + "httpdate", "itoa", "pin-project-lite", "smallvec", @@ -1444,9 +1441,9 @@ checksum = "34b357333733e8260735ba5894eb928c02ecc69c78715f01a8019e7fa7f2db4c" [[package]] name = "libc" -version = "0.2.186" +version = "0.2.189" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "68ab91017fe16c622486840e4c83c9a37afeff978bd239b5293d61ece587de66" +checksum = "3eaf3ede3fee6db1a4c2ee091bf8a8b4dccdc6d17f656fb07896ee72867612f2" [[package]] name = "libm" @@ -1617,7 +1614,7 @@ dependencies = [ [[package]] name = "mtp" version = "0.2.0" -source = "git+https://git.methanium.net/methanium/mtp#04760fd88d2bc3adf548a9ec532fa228227f0a49" +source = "git+https://git.methanium.net/methanium/mtp#88ae866b91857c9b1650ea0ee2af2fdb70b92f10" dependencies = [ "mtp-client", "mtp-codec", @@ -1633,7 +1630,7 @@ dependencies = [ [[package]] name = "mtp-client" version = "0.2.0" -source = "git+https://git.methanium.net/methanium/mtp#04760fd88d2bc3adf548a9ec532fa228227f0a49" +source = "git+https://git.methanium.net/methanium/mtp#88ae866b91857c9b1650ea0ee2af2fdb70b92f10" dependencies = [ "mtp-codec", "mtp-common", @@ -1646,7 +1643,7 @@ dependencies = [ [[package]] name = "mtp-codec" version = "0.2.0" -source = "git+https://git.methanium.net/methanium/mtp#04760fd88d2bc3adf548a9ec532fa228227f0a49" +source = "git+https://git.methanium.net/methanium/mtp#88ae866b91857c9b1650ea0ee2af2fdb70b92f10" dependencies = [ "base64", "byteorder", @@ -1659,7 +1656,7 @@ dependencies = [ [[package]] name = "mtp-common" version = "0.2.0" -source = "git+https://git.methanium.net/methanium/mtp#04760fd88d2bc3adf548a9ec532fa228227f0a49" +source = "git+https://git.methanium.net/methanium/mtp#88ae866b91857c9b1650ea0ee2af2fdb70b92f10" dependencies = [ "quinn", "rustls", @@ -1670,7 +1667,7 @@ dependencies = [ [[package]] name = "mtp-crypto" version = "0.2.0" -source = "git+https://git.methanium.net/methanium/mtp#04760fd88d2bc3adf548a9ec532fa228227f0a49" +source = "git+https://git.methanium.net/methanium/mtp#88ae866b91857c9b1650ea0ee2af2fdb70b92f10" dependencies = [ "base64", "chacha20poly1305", @@ -1692,7 +1689,7 @@ dependencies = [ [[package]] name = "mtp-files" version = "0.2.0" -source = "git+https://git.methanium.net/methanium/mtp#04760fd88d2bc3adf548a9ec532fa228227f0a49" +source = "git+https://git.methanium.net/methanium/mtp#88ae866b91857c9b1650ea0ee2af2fdb70b92f10" dependencies = [ "mtp-crypto", "rand 0.10.2", @@ -1703,7 +1700,7 @@ dependencies = [ [[package]] name = "mtp-host" version = "0.2.0" -source = "git+https://git.methanium.net/methanium/mtp#04760fd88d2bc3adf548a9ec532fa228227f0a49" +source = "git+https://git.methanium.net/methanium/mtp#88ae866b91857c9b1650ea0ee2af2fdb70b92f10" dependencies = [ "mtp-codec", "mtp-common", @@ -1718,7 +1715,7 @@ dependencies = [ [[package]] name = "mtp-transport" version = "0.2.0" -source = "git+https://git.methanium.net/methanium/mtp#04760fd88d2bc3adf548a9ec532fa228227f0a49" +source = "git+https://git.methanium.net/methanium/mtp#88ae866b91857c9b1650ea0ee2af2fdb70b92f10" dependencies = [ "async-trait", "mtp-codec", @@ -1736,7 +1733,7 @@ dependencies = [ [[package]] name = "mtp-type-map" version = "0.2.0" -source = "git+https://git.methanium.net/methanium/mtp#04760fd88d2bc3adf548a9ec532fa228227f0a49" +source = "git+https://git.methanium.net/methanium/mtp#88ae866b91857c9b1650ea0ee2af2fdb70b92f10" dependencies = [ "serde", "serde_yaml", @@ -1745,7 +1742,7 @@ dependencies = [ [[package]] name = "mtp-webserver" version = "0.2.0" -source = "git+https://git.methanium.net/methanium/mtp#04760fd88d2bc3adf548a9ec532fa228227f0a49" +source = "git+https://git.methanium.net/methanium/mtp#88ae866b91857c9b1650ea0ee2af2fdb70b92f10" dependencies = [ "async-trait", "bytes", @@ -1753,6 +1750,9 @@ dependencies = [ "h3-quinn", "h3-webtransport", "http", + "http-body-util", + "hyper", + "hyper-util", "mtp-codec", "mtp-common", "mtp-crypto", @@ -1763,6 +1763,8 @@ dependencies = [ "rustls", "thiserror 2.0.19", "tokio", + "tokio-rustls", + "tokio-stream", "tracing", ] @@ -2355,9 +2357,9 @@ dependencies = [ [[package]] name = "rustls-pki-types" -version = "1.15.0" +version = "1.15.1" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "764899a24af3980067ee14bc143654f297b22eaebfe3c7b6b211920a5a59b046" +checksum = "2f4925028c7eb5d1fcdaf196971378ed9d2c1c4efc7dc5d011256f76c99c0a96" dependencies = [ "web-time", "zeroize", @@ -2494,7 +2496,7 @@ checksum = "e7a5d71263a5a7d47b41f6b3f06ba276f10cc18b0931f1799f710578e2309348" dependencies = [ "proc-macro2", "quote", - "syn 3.0.2", + "syn 3.0.3", ] [[package]] @@ -2966,9 +2968,9 @@ dependencies = [ [[package]] name = "syn" -version = "3.0.2" +version = "3.0.3" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "a207d6d6a2b7fc470b80443726053f18a2481b7e1eee970597051596567987a3" +checksum = "53e9bae58849f64dfa4f5d5ae372c8341f7305f82a3868709269343628b659a3" dependencies = [ "proc-macro2", "quote", @@ -3053,7 +3055,7 @@ checksum = "43cbfe0cf76104d42a574802844187e84a305e531ed54455f11fbde0f10541cd" dependencies = [ "proc-macro2", "quote", - "syn 3.0.2", + "syn 3.0.3", ] [[package]] @@ -3152,9 +3154,9 @@ dependencies = [ [[package]] name = "tokio-stream" -version = "0.1.18" +version = "0.1.19" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "32da49809aab5c3bc678af03902d4ccddea2a87d028d86392a4b1560c6906c70" +checksum = "a3d06f0b082ba57c26b79407372e57cf2a1e28124f78e9479fe80322cf53420b" dependencies = [ "futures-core", "pin-project-lite", @@ -3163,13 +3165,14 @@ dependencies = [ [[package]] name = "tokio-util" -version = "0.7.18" +version = "0.7.19" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "9ae9cec805b01e8fc3fd2fe289f89149a9b66dd16786abd8b19cfa7b48cb0098" +checksum = "494815d09bf52b5548659851081238f0ca39ff638363907596da739561c62c52" dependencies = [ "bytes", "futures-core", "futures-sink", + "libc", "pin-project-lite", "tokio", ] diff --git a/migrations/001.sql b/migrations/001_initial_schema.sql similarity index 91% rename from migrations/001.sql rename to migrations/001_initial_schema.sql index dc3a096..e2b8af5 100644 --- a/migrations/001.sql +++ b/migrations/001_initial_schema.sql @@ -1,3 +1,8 @@ +CREATE TABLE IF NOT EXISTS iotas ( + id BIGINT NOT NULL PRIMARY KEY, + public_key BLOB NOT NULL +); + CREATE TABLE IF NOT EXISTS users ( id BIGINT NOT NULL PRIMARY KEY, iota_id BIGINT NOT NULL, @@ -10,14 +15,15 @@ CREATE TABLE IF NOT EXISTS users ( sub_end BIGINT NOT NULL DEFAULT 0, public_key BLOB NOT NULL, token BLOB NOT NULL, + UNIQUE KEY uk_users_username (username), - UNIQUE KEY uk_users_iota_id (iota_id) + KEY idx_users_iota_id (iota_id), + + CONSTRAINT fk_users_iota + FOREIGN KEY (iota_id) + REFERENCES iotas (id) ); -CREATE TABLE IF NOT EXISTS iotas ( - id BIGINT NOT NULL PRIMARY KEY, - public_key BLOB NOT NULL -); CREATE TABLE IF NOT EXISTS omikrons ( id BIGINT NOT NULL PRIMARY KEY, diff --git a/migrations/002_registration_leases.sql b/migrations/002_registration_leases.sql new file mode 100644 index 0000000..a9148b1 --- /dev/null +++ b/migrations/002_registration_leases.sql @@ -0,0 +1,13 @@ +-- Registration IDs are allocated before the user row is created. Keep the +-- allocation durable and bound to the Iota that requested it so a completed +-- registration can safely be retried after any response is lost. +CREATE TABLE registration_leases ( + token CHAR(36) CHARACTER SET ascii COLLATE ascii_bin NOT NULL PRIMARY KEY, + user_id BIGINT NOT NULL UNIQUE, + iota_id BIGINT NOT NULL, + expires_at DATETIME NOT NULL, + completed_at DATETIME NULL, + CONSTRAINT fk_registration_leases_iota + FOREIGN KEY (iota_id) REFERENCES iotas (id), + INDEX idx_registration_leases_expiry (expires_at) +); diff --git a/migrations/003_registration_lease_request_id.sql b/migrations/003_registration_lease_request_id.sql new file mode 100644 index 0000000..6784177 --- /dev/null +++ b/migrations/003_registration_lease_request_id.sql @@ -0,0 +1,8 @@ +-- A relay retry uses the original MTP request ID. Make allocation idempotent +-- for that authenticated Iota/request pair, so a lost response returns the +-- same user ID and lease rather than allocating another one. +ALTER TABLE registration_leases + -- Existing leases predate retry correlation; leave their request ID NULL + -- rather than assigning a shared sentinel that could violate uniqueness. + ADD COLUMN request_id INT UNSIGNED NULL, + ADD CONSTRAINT uk_registration_leases_iota_request UNIQUE (iota_id, request_id); diff --git a/mtp-type-maps b/mtp-type-maps new file mode 160000 index 0000000..594646a --- /dev/null +++ b/mtp-type-maps @@ -0,0 +1 @@ +Subproject commit 594646ac39d986f0787aa614a99d580035a67318 diff --git a/src/config.rs b/src/config.rs index 5ba9dc2..66f5382 100644 --- a/src/config.rs +++ b/src/config.rs @@ -9,8 +9,8 @@ pub struct RateLimitConfig { pub transport_connections_per_ip: usize, } -pub fn cors_origin() -> String { - env::var("CORS_ORIGIN").unwrap_or_else(|_| "https://tensamin.net".to_string()) +pub const fn cors_origin() -> &'static str { + "*" } impl Default for RateLimitConfig { diff --git a/src/db/iota_repo.rs b/src/db/iota_repo.rs index a5952f1..3eedce8 100644 --- a/src/db/iota_repo.rs +++ b/src/db/iota_repo.rs @@ -21,13 +21,31 @@ pub async fn get_iota_by_id(id: IotaId) -> Result { } pub async fn create_new_iota(public_key: PublicKeyBundle) -> Result { - let id = crate::db::user_repo::get_register_id().await?; - let iota_id = IotaId::from(id.0); - register_complete_iota(iota_id, public_key).await?; - Ok(iota_id) + for _ in 0..16 { + let id = crate::db::user_repo::get_register_id().await?; + let iota_id = IotaId::from(id.0); + match register_complete_iota(iota_id, public_key.clone()).await { + Ok(()) => return Ok(iota_id), + Err(OmegaError::Database(error)) => { + if crate::db::user_repo::is_duplicate_key(&error) { + continue; + } + return Err(OmegaError::Database(error)); + } + Err(error) => return Err(error), + } + } + Err(OmegaError::Validation( + "could not allocate a unique Iota ID".into(), + )) } pub async fn register_complete_iota(id: IotaId, public_key: PublicKeyBundle) -> Result<()> { + if !crate::db::user_repo::valid_protocol_id(id.0) { + return Err(OmegaError::Validation( + "Iota ID is outside the 48-bit protocol range".into(), + )); + } sqlx::query("INSERT INTO iotas (id, public_key) VALUES (?, ?)") .bind(id.0) .bind(public_key.as_bytes()) diff --git a/src/db/user_repo.rs b/src/db/user_repo.rs index ca71dd3..8554566 100644 --- a/src/db/user_repo.rs +++ b/src/db/user_repo.rs @@ -4,17 +4,94 @@ use crate::{ models::{IotaId, User, UserId}, }; use mtp::crypto::PublicKeyBundle; -use sqlx::FromRow; +use sqlx::{FromRow, Row}; + +pub const MAX_PROTOCOL_ID: i64 = (1_i64 << 48) - 1; +const ID_ALLOCATION_ATTEMPTS: usize = 16; pub async fn get_register_id() -> Result { - let bytes = *uuid::Uuid::now_v7().as_bytes(); - let id = - i64::from_be_bytes(bytes[8..].try_into().map_err(|_| { - OmegaError::Validation("generated ID has an invalid length".to_string()) - })?) & i64::MAX; + use std::time::{SystemTime, UNIX_EPOCH}; + + let timestamp = SystemTime::now() + .duration_since(UNIX_EPOCH) + .map(|d| d.as_secs()) + .unwrap_or(0); + + let ts = timestamp as i64; + if ts >= 1 && ts <= MAX_PROTOCOL_ID { + return Ok(UserId::from(ts)); + } + + // Fall back to random if the timestamp is outside the 48-bit range. + let id = (rand::random::() & ((1_u64 << 48) - 1)) as i64; Ok(UserId::from(id.max(1))) } +pub fn valid_protocol_id(id: i64) -> bool { + (1..=MAX_PROTOCOL_ID).contains(&id) +} + +/// Allocate an ID that is durable, short-lived, and bound to the connected +/// Iota. `token` is presented again when completing the registration. +pub async fn allocate_registration(iota_id: IotaId, request_id: u32) -> Result<(UserId, String)> { + if !valid_protocol_id(iota_id.0) { + return Err(OmegaError::Validation( + "Iota ID is outside the 48-bit protocol range".into(), + )); + } + + for _ in 0..ID_ALLOCATION_ATTEMPTS { + let id = get_register_id().await?; + let token = uuid::Uuid::new_v4().to_string(); + let result = sqlx::query( + "INSERT INTO registration_leases (token, user_id, iota_id, request_id, expires_at) \ + VALUES (?, ?, ?, ?, DATE_ADD(UTC_TIMESTAMP(), INTERVAL 10 MINUTE))", + ) + .bind(&token) + .bind(id.0) + .bind(iota_id.0) + .bind(request_id) + .execute(&pool().await?) + .await; + match result { + Ok(_) => return Ok((id, token)), + Err(error) if is_duplicate_key(&error) => { + let existing = sqlx::query( + "SELECT user_id, token, expires_at >= UTC_TIMESTAMP() AS current \ + FROM registration_leases WHERE iota_id = ? AND request_id = ?", + ) + .bind(iota_id.0) + .bind(request_id) + .fetch_optional(&pool().await?) + .await?; + if let Some(existing) = existing { + let current: i8 = existing.get("current"); + if current != 0 { + return Ok(( + UserId::from(existing.get::("user_id")), + existing.get::("token"), + )); + } + } + continue; + } + Err(error) => return Err(error.into()), + } + } + Err(OmegaError::Validation( + "could not allocate a unique registration ID".into(), + )) +} + +pub(crate) fn is_duplicate_key(error: &sqlx::Error) -> bool { + error.as_database_error().is_some_and(|database| { + // MySQL's generic database-error API exposes SQLSTATE (23000) as + // `code()`. The driver-specific duplicate-key number is retained in + // the diagnostic message. + database.code().as_deref() == Some("23000") && database.message().contains("1062") + }) +} + const USER_BY_USERNAME_QUERY: &str = "SELECT id, iota_id, username, display, status, about, avatar, sub_level, sub_end, public_key, token FROM users WHERE username = ?"; const USER_BY_ID_QUERY: &str = "SELECT id, iota_id, username, display, status, about, avatar, sub_level, sub_end, public_key, token FROM users WHERE id = ?"; const USERS_BY_IOTA_ID_QUERY: &str = "SELECT id, iota_id, username, display, status, about, avatar, sub_level, sub_end, public_key, token FROM users WHERE iota_id = ?"; @@ -179,16 +256,99 @@ pub async fn register_complete_user( public_key: PublicKeyBundle, iota_id: IotaId, token: String, + registration_token: String, ) -> Result<()> { - sqlx::query( + if !valid_protocol_id(id.0) || !valid_protocol_id(iota_id.0) { + return Err(OmegaError::Validation( + "user or Iota ID is outside the 48-bit protocol range".into(), + )); + } + if !valid_username(&username) { + return Err(OmegaError::Validation("invalid username".into())); + } + + let mut transaction = pool().await?.begin().await?; + let lease = sqlx::query( + "SELECT iota_id, completed_at IS NOT NULL AS completed, \ + expires_at >= UTC_TIMESTAMP() AS current FROM registration_leases \ + WHERE token = ? AND user_id = ? FOR UPDATE", + ) + .bind(®istration_token) + .bind(id.0) + .fetch_optional(&mut *transaction) + .await? + .ok_or_else(|| OmegaError::Validation("unknown registration lease".into()))?; + let lease_iota_id: i64 = lease.get("iota_id"); + let completed: i8 = lease.get("completed"); + let current: i8 = lease.get("current"); + if lease_iota_id != iota_id.0 || (completed == 0 && current == 0) { + return Err(OmegaError::Validation( + "expired or mismatched registration lease".into(), + )); + } + + let insert_result = sqlx::query( "INSERT INTO users (id, username, public_key, iota_id, token) VALUES (?, ?, ?, ?, ?)", ) .bind(id.0) - .bind(username.into_bytes()) + .bind(username.as_bytes()) .bind(public_key.as_bytes()) .bind(iota_id.0) - .bind(token.into_bytes()) - .execute(&pool().await?) - .await?; + .bind(token.as_bytes()) + .execute(&mut *transaction) + .await; + + let result: Result<()> = match insert_result { + Ok(_) => Ok(()), + Err(insert_error) => { + let existing = sqlx::query_as::<_, UserRow>(USER_BY_ID_QUERY) + .bind(id.0) + .fetch_optional(&mut *transaction) + .await?; + match existing + .map(User::try_from) + .transpose() + .map_err(OmegaError::from)? + { + Some(existing) + if existing.iota_id == iota_id + && existing.username == username + && existing.public_key.as_bytes() == public_key.as_bytes() + && existing.token == token => + { + Ok(()) + } + _ => Err(insert_error.into()), + } + } + }; + result?; + sqlx::query("UPDATE registration_leases SET completed_at = UTC_TIMESTAMP() WHERE token = ?") + .bind(®istration_token) + .execute(&mut *transaction) + .await?; + transaction.commit().await?; Ok(()) } + +fn valid_username(username: &str) -> bool { + !username.is_empty() + && username.chars().count() <= 15 + && !username.chars().any(char::is_control) + && !username.contains(['/', '\\']) +} + +#[cfg(test)] +mod tests { + use super::{MAX_PROTOCOL_ID, get_register_id, valid_protocol_id}; + + #[tokio::test] + async fn generated_registration_ids_fit_the_mtp_wire_range() { + for _ in 0..128 { + assert!(valid_protocol_id(get_register_id().await.unwrap().0)); + } + assert!(!valid_protocol_id(0)); + assert!(valid_protocol_id(MAX_PROTOCOL_ID)); + assert!(!valid_protocol_id(MAX_PROTOCOL_ID + 1)); + } +} diff --git a/src/server/api.rs b/src/server/api.rs index 6dc01aa..ed5bb80 100644 --- a/src/server/api.rs +++ b/src/server/api.rs @@ -15,12 +15,12 @@ use crate::server::{ validation::{parse_positive_id, validate_non_empty}, }; use crate::sql::user_online_tracker::{get_all_connections, get_iota_primary_omikron_connection}; -use crate::transport::omikron_manager::get_random_omikron; +use crate::transport::omikron_manager::{get_connected_omikron, get_random_omikron}; use crate::util::file_util::get_directory; use base64::Engine as _; use bytes::Bytes; use http::{Method, StatusCode}; -use mtp::webserver::{Http3Request, Http3Response, RouteParams}; +use mtp::webserver::{HttpRequest, HttpResponse, RouteParams}; use std::collections::BTreeMap; fn error_body(error: &OmegaError) -> String { @@ -75,20 +75,19 @@ async fn route(path_parts: &[&str]) -> Result<(StatusCode, String)> { } ["api", "get", "omikron", id] => { let id = parse_positive_id(id)?; - let omikron = match get_omikron_by_id(id.into()).await { - Ok(value) => value, - Err(_) => { - let fallback_id = - if let Some(fallback_id) = get_iota_primary_omikron_connection(id) { - fallback_id - } else { - let user = get_by_user_id(UserId::from(id)).await?; - get_iota_primary_omikron_connection(user.iota_id.0) - .ok_or(OmegaError::NotFound)? - }; - get_omikron_by_id(fallback_id.into()).await? - } + let omikron_id = if get_connected_omikron(id).is_some() { + id + } else if let Some(omikron_id) = get_iota_primary_omikron_connection(id) { + omikron_id + } else { + let user = get_by_user_id(UserId::from(id)).await?; + get_iota_primary_omikron_connection(user.iota_id.0).ok_or(OmegaError::NotFound)? }; + + // Database rows describe registered Omikrons. The public discovery + // API must expose only routes backed by a currently live transport. + get_connected_omikron(omikron_id).ok_or(OmegaError::NotFound)?; + let omikron = get_omikron_by_id(omikron_id.into()).await?; Ok(( StatusCode::OK, json(&OmikronResponse { @@ -178,7 +177,7 @@ async fn route(path_parts: &[&str]) -> Result<(StatusCode, String)> { } } -pub async fn handle(request: Http3Request, response: Http3Response) -> Http3Response { +pub async fn handle(request: HttpRequest, response: HttpResponse) -> HttpResponse { let method = request.method; let path = request.uri.path().to_string(); if method != Method::OPTIONS && !middleware::allow(request.remote_addr.ip(), &path) { @@ -238,9 +237,9 @@ pub async fn handle(request: Http3Request, response: Http3Response) -> Http3Resp } pub async fn handle_pattern( - request: Http3Request, - response: Http3Response, + request: HttpRequest, + response: HttpResponse, _params: RouteParams, -) -> Http3Response { +) -> HttpResponse { handle(request, response).await } diff --git a/src/server/index.rs b/src/server/index.rs index b121209..f128c30 100644 --- a/src/server/index.rs +++ b/src/server/index.rs @@ -1,13 +1,34 @@ use http::StatusCode; -use mtp::webserver::Http3Response; +use mtp::webserver::HttpResponse; -pub fn index_handler(response: Http3Response) -> Http3Response { +pub fn index_handler(response: HttpResponse) -> HttpResponse { let documentation = r#" Omega API Server Available Routes: -- /api/* : API endpoints for the Omega server. -- /direct/* : Resolution for shortened links. +- /api/get/omikron + Returns a randomly selected connected Omikron. +- /api/get/omikron/{id} + Returns an Omikron by ID. If {id} is a connected Iota ID or user ID, + returns that account's primary connected Omikron instead. +- /api/get/connections + Returns the current live connection map as + { "status": "success", "connections": { omikron_id: { iota_id: [user_id] } } }. +- /api/get/iota/{id} + Returns the Iota's public identity data. +- /api/get/user/{id} + Returns user profile and public identity data. +- /api/get/id/{username} + Resolves a username to its user and Iota IDs. +- /api/get/public_key + Returns Omega's public key. +- /api/download/iota_frontend + Downloads the Iota frontend archive. +- /direct/{short_key} + Resolves a shortened link with a temporary redirect. + +All IDs must be positive decimal integers. The connections endpoint reports +live in-memory state; it is empty after an Omega restart until Omikrons sync. All other routes will return this documentation. "#; diff --git a/src/server/server.rs b/src/server/server.rs index f88e267..9ac4f06 100644 --- a/src/server/server.rs +++ b/src/server/server.rs @@ -3,10 +3,6 @@ use mtp::webserver::WebServerConfig; pub fn build_web_config() -> Result { WebServerConfig::new() - .route( - "/", - |_request, response| async move { index_handler(response) }, - )? .route("/api/download/iota_frontend", api::handle)? .route("/api/get/omikron", api::handle)? .route("/api/get/connections", api::handle)? @@ -15,5 +11,6 @@ pub fn build_web_config() -> Result>> = Lazy::new(DashMa // UserID -> UserStatus static USER_STATUS_MAP: Lazy> = Lazy::new(DashMap::new); +// The legacy account map is kept for Iota-only/offline compatibility. Client +// transports are tracked independently: two devices must never overwrite each +// other's route merely because they authenticate as the same account. +static USER_SESSION_STATUS_MAP: Lazy> = Lazy::new(DashMap::new); + pub fn track_iota_connection(iota_id: i64, omikron_id: i64, primary: bool) { let mut entry = IOTA_OMIKRON_CONNECTIONS .entry(iota_id) @@ -121,13 +126,63 @@ pub fn track_user_status(user_id: i64, status: UserStatus, omikron_id: i64) { ); } +pub fn track_user_session_status( + user_id: i64, + session_id: i64, + status: UserStatus, + omikron_id: i64, +) { + USER_SESSION_STATUS_MAP.insert( + (user_id, session_id), + UserConnection { + connection_type: status, + omikron_id, + }, + ); +} + pub fn untrack_user_status(user_id: i64, omikron_id: i64) { USER_STATUS_MAP.remove_if(&user_id, |_, connection| { connection.omikron_id == omikron_id }); } +pub fn untrack_user_session_status(user_id: i64, session_id: i64, omikron_id: i64) { + USER_SESSION_STATUS_MAP.remove_if(&(user_id, session_id), |_, connection| { + connection.omikron_id == omikron_id + }); +} + +pub fn update_user_session_status(user_id: i64, status: UserStatus) { + for mut entry in USER_SESSION_STATUS_MAP.iter_mut() { + if entry.key().0 == user_id { + entry.connection_type = status.clone(); + } + } + // Preserve the account preference for legacy routes as well. + if let Some(mut entry) = USER_STATUS_MAP.get_mut(&user_id) { + entry.connection_type = status; + } +} + pub fn get_user_status(user_id: i64) -> Option { + // A connected visible session is preferred. Invisible sessions remain + // routable but are intentionally presented as offline when they are the + // only active routes. + let sessions: Vec = USER_SESSION_STATUS_MAP + .iter() + .filter(|entry| entry.key().0 == user_id) + .map(|entry| entry.value().clone()) + .collect(); + if let Some(status) = sessions + .iter() + .find(|status| status.connection_type != UserStatus::user_invisible) + { + return Some(status.clone()); + } + if let Some(status) = sessions.first() { + return Some(status.clone()); + } USER_STATUS_MAP.get(&user_id).map(|v| v.clone()) } @@ -164,6 +219,7 @@ pub async fn untrack_omikron(omikron_id: i64) { } USER_STATUS_MAP.retain(|_, status| status.omikron_id != omikron_id); + USER_SESSION_STATUS_MAP.retain(|_, status| status.omikron_id != omikron_id); for iota_id in offline_iotas { if let Ok(users) = user_repo::get_users_by_iota_id(IotaId::from(iota_id)).await { diff --git a/src/transport/handlers/presence.rs b/src/transport/handlers/presence.rs index 1dd10d0..adaa1a2 100644 --- a/src/transport/handlers/presence.rs +++ b/src/transport/handlers/presence.rs @@ -21,7 +21,18 @@ pub async fn user_connected( .and_then(UserStatus::from_str) .unwrap_or(UserStatus::user_online); if let Ok(user_id) = i64::try_from(user_id) { - user_online_tracker::track_user_status(user_id, status, omikron_id); + if let Some(session_id) = value + .get_data(DataType::SessionId) + .as_number() + .and_then(|id| i64::try_from(id).ok()) + .filter(|id| *id > 0) + { + user_online_tracker::track_user_session_status( + user_id, session_id, status, omikron_id, + ); + } else { + user_online_tracker::track_user_status(user_id, status, omikron_id); + } } } Ok(()) @@ -34,11 +45,55 @@ pub async fn user_disconnected( ) -> OmikronResult<()> { log_in!(crate::util::logger::PrintType::Omega, "User disconnected"); if let Some(user_id) = value.get_data(DataType::UserId).as_number() { - user_online_tracker::untrack_user_status(user_id as i64, omikron_id); + if let Some(session_id) = value + .get_data(DataType::SessionId) + .as_number() + .and_then(|id| i64::try_from(id).ok()) + .filter(|id| *id > 0) + { + user_online_tracker::untrack_user_session_status( + user_id as i64, + session_id, + omikron_id, + ); + } else { + user_online_tracker::untrack_user_status(user_id as i64, omikron_id); + } } Ok(()) } +pub async fn client_changed( + _: Arc, + value: CommunicationValue, + _: i64, +) -> OmikronResult<()> { + let Some(user_id) = value + .get_data(DataType::UserId) + .as_number() + .and_then(|id| i64::try_from(id).ok()) + else { + return Ok(()); + }; + let Some(status) = value + .get_data(DataType::UserState) + .as_str() + .and_then(UserStatus::from_str) + else { + return Ok(()); + }; + // Connectivity is derived from routes. Clients may choose only public + // presence preferences, never server/offline states. + if matches!( + status, + UserStatus::user_offline | UserStatus::iota_offline | UserStatus::iota_online + ) { + return Ok(()); + } + user_online_tracker::update_user_session_status(user_id, status); + Ok(()) +} + pub async fn iota_connected( connection: Arc, value: CommunicationValue, diff --git a/src/transport/handlers/register.rs b/src/transport/handlers/register.rs index 23bf3e7..3502269 100644 --- a/src/transport/handlers/register.rs +++ b/src/transport/handlers/register.rs @@ -13,13 +13,25 @@ pub async fn get_register( connection: Arc, value: CommunicationValue, ) -> OmikronResult<()> { - let register_id = user_repo::get_register_id().await?; + let iota_id = value + .get_data(DataType::IotaId) + .as_number() + .and_then(|id| i64::try_from(id).ok()) + .filter(|id| user_repo::valid_protocol_id(*id)); + let Some(iota_id) = iota_id else { + return connection + .send_error_response(value.get_id(), CommunicationType::ErrorInvalidData) + .await; + }; + let (register_id, registration_token) = + user_repo::allocate_registration(IotaId::from(iota_id), value.get_id()).await?; let response = CommunicationValue::new(CommunicationType::GetRegister) .with_id(value.get_id()) .add_typed_default( DataType::UserId, DataValue::SignedNumber(register_id.0.into()), - ); + ) + .add_typed_default(DataType::RegisterId, DataValue::Str(registration_token)); connection.send(&response).await } @@ -27,10 +39,6 @@ pub async fn complete_iota( connection: Arc, value: CommunicationValue, ) -> OmikronResult<()> { - let iota_id = value - .get_data(DataType::IotaId) - .as_number() - .map(|id| id as i64); let public_key = value .get_data(DataType::PublicKey) .as_str() @@ -40,57 +48,25 @@ pub async fn complete_iota( .send_error_response(value.get_id(), CommunicationType::ErrorInvalidData) .await; }; - match iota_id { - Some(iota_id) => { - match iota_repo::register_complete_iota(IotaId::from(iota_id), public_key).await { - Ok(()) => { - 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 - } - } + match iota_repo::create_new_iota(public_key).await { + Ok(id) => { + connection + .send( + &CommunicationValue::new(CommunicationType::CompleteRegisterIota) + .with_id(value.get_id()) + .add_typed_default(DataType::IotaId, DataValue::SignedNumber(id.0.into())), + ) + .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 } - None => match iota_repo::create_new_iota(public_key).await { - Ok(id) => { - connection - .send( - &CommunicationValue::new(CommunicationType::CompleteRegisterIota) - .with_id(value.get_id()) - .add_typed_default( - DataType::IotaId, - DataValue::SignedNumber(id.0.into()), - ), - ) - .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 - } - }, } } @@ -101,7 +77,8 @@ pub async fn complete_user( let user_id = value .get_data(DataType::UserId) .as_number() - .map(|id| id as i64); + .and_then(|id| i64::try_from(id).ok()) + .filter(|id| user_repo::valid_protocol_id(*id)); let username = value .get_data(DataType::Username) .as_str() @@ -114,22 +91,44 @@ pub async fn complete_user( .get_data(DataType::ResetToken) .as_str() .map(str::to_owned); - let Some((user_id, username, public_key, reset_token)) = user_id + let registration_token = value + .get_data(DataType::RegisterId) + .as_str() + .filter(|token| uuid::Uuid::parse_str(token).is_ok()) + .map(str::to_owned); + let Some((user_id, username, public_key, reset_token, registration_token)) = user_id .zip(username) .zip(public_key) .zip(reset_token) - .map(|(((id, name), key), token)| (id, name, key, token)) + .zip(registration_token) + .map(|((((id, name), key), token), registration_token)| { + (id, name, key, token, registration_token) + }) else { return connection .send_error_response(value.get_id(), CommunicationType::ErrorInvalidData) .await; }; + // Omikron supplies the authenticated Iota ID in the payload. The lease + // check below binds completion to that Iota rather than trusting sender. + let iota_id = value + .get_data(DataType::IotaId) + .as_number() + .and_then(|id| i64::try_from(id).ok()) + .filter(|id| user_repo::valid_protocol_id(*id)); + let Some(iota_id) = iota_id else { + return connection + .send_error_response(value.get_id(), CommunicationType::ErrorInvalidData) + .await; + }; + match user_repo::register_complete_user( UserId::from(user_id), username, public_key, - IotaId::from(value.get_sender() as i64), + IotaId::from(iota_id), reset_token, + registration_token, ) .await { diff --git a/src/transport/omikron_connection.rs b/src/transport/omikron_connection.rs index 7095f9d..d48120c 100644 --- a/src/transport/omikron_connection.rs +++ b/src/transport/omikron_connection.rs @@ -91,10 +91,29 @@ impl OmikronConnection { .retain(|_, task| task.inserted_at.elapsed() < MAX_WAITING_AGE); } })); - while let Ok(value) = receiver.receive().await { - if let Err(error) = self.clone().process_message(value).await { - log_err!(0, PrintType::Omega, "Error processing message: {}", error); - if matches!(error, crate::error::OmegaError::NotConnected) { + loop { + match receiver.receive().await { + Ok(value) => { + if let Err(error) = self.clone().process_message(value).await { + log_err!( + self.id as i64, + PrintType::Omega, + "Error processing Omikron message: {}", + error + ); + if matches!(error, crate::error::OmegaError::NotConnected) { + break; + } + } + } + Err(error) => { + log_err!( + self.id as i64, + PrintType::Omega, + "Omikron receive loop ended: {}; transport close reason: {:?}", + error, + receiver.close_reason() + ); break; } } @@ -133,6 +152,9 @@ impl OmikronConnection { Some(CommunicationType::UserDisconnected) => { crate::transport::handlers::presence::user_disconnected(self, value, id).await } + Some(CommunicationType::ClientChanged) => { + crate::transport::handlers::presence::client_changed(self, value, id).await + } Some(CommunicationType::IotaConnected) => { crate::transport::handlers::presence::iota_connected(self, value, id).await } @@ -259,40 +281,38 @@ pub async fn complete_register(_: PublicKeyBundle, _: Option) -> u64 { pub async fn start(port: u16) -> Result<(), Box> { let cert_pem = load_file_vec("certs", "cert.pem")?; let key_pem = load_file_vec("certs", "key.pem")?; - let web_config = server::server::build_web_config()?; - let host_config = HostConfig::new( - IpAddr::from(Ipv4Addr::new(0, 0, 0, 0)), - port, - cert_pem, - key_pem, - ) - .with_policy(Policy { - send_mode: SendMode::SingleStreamPerMessage, - max_message_size: 1_000_000_000, - handshake_max_message_size: 1_000_000, - close_frame_len: u32::MAX, - application_close_code: 0, - open_stream_timeout: Duration::from_millis(2_000), - write_timeout: Duration::from_millis(2_000), - accept_stream_timeout: Duration::from_millis(10_000), - read_timeout: Duration::from_millis(30_000), - keep_alive_interval: Some(Duration::from_secs(6)), - max_idle_timeout: Some(Duration::from_secs(30)), - force_close_delay: Duration::from_millis(300), - receiver_queue_capacity: 1000, - max_concurrent_stream_tasks: 10, - persistent_stream_max_retries: 5, - persistent_stream_retry_backoff: Duration::from_secs(5), - max_frames_per_stream: None, - }) - .with_authentication( - load_keyring(), - Box::new(|id, description| Box::pin(get_by_omikron_id(id, description))), - Box::new(|key, description| Box::pin(complete_register(key, description))), - ) - .with_authentication_policy(AuthenticationPolicy::ForceAuthentication); + let web_config = server::server::build_web_config()? + .serve_tcp_https(true) + .max_tcp_connections(256); + let ip = IpAddr::from(Ipv4Addr::new(0, 0, 0, 0)); + let host_config = HostConfig::new(ip, port, cert_pem, key_pem) + .with_policy(Policy { + send_mode: SendMode::SingleStreamPerMessage, + max_message_size: 1_000_000_000, + handshake_max_message_size: 1_000_000, + close_frame_len: u32::MAX, + application_close_code: 0, + open_stream_timeout: Duration::from_millis(5_000), + write_timeout: Duration::from_millis(5_000), + accept_stream_timeout: Duration::from_millis(10_000), + read_timeout: Duration::from_millis(30_000), + keep_alive_interval: Some(Duration::from_secs(6)), + max_idle_timeout: Some(Duration::from_secs(30)), + force_close_delay: Duration::from_millis(300), + receiver_queue_capacity: 1000, + max_concurrent_stream_tasks: 64, + persistent_stream_max_retries: 5, + persistent_stream_retry_backoff: Duration::from_secs(5), + max_frames_per_stream: None, + }) + .with_authentication( + load_keyring(), + Box::new(|id, description| Box::pin(get_by_omikron_id(id, description))), + Box::new(|key, description| Box::pin(complete_register(key, description))), + ) + .with_authentication_policy(AuthenticationPolicy::ForceAuthentication); let mut server = MTPWebServer::new(host_config, web_config).await?; - log!("OmegaServer listening on port {}", port); + log!("OmegaServer listening on {}:{}", ip.to_string(), port); loop { let mut conn = match server.accept().await { Ok(Some(conn)) => conn, diff --git a/src/transport/omikron_manager.rs b/src/transport/omikron_manager.rs index 2de4ea3..0b2f66c 100644 --- a/src/transport/omikron_manager.rs +++ b/src/transport/omikron_manager.rs @@ -26,12 +26,18 @@ pub async fn remove_omikron(omikron_id: i64) { OMIKRON_CONNECTIONS.remove(&omikron_id); } +pub fn get_connected_omikron(omikron_id: i64) -> Option> { + OMIKRON_CONNECTIONS + .get(&omikron_id) + .map(|connection| connection.clone()) +} + pub async fn get_random_omikron() -> Result, ()> { let keys: Vec<_> = OMIKRON_CONNECTIONS.iter().map(|e| *e.key()).collect(); if let Some(key) = keys.into_iter().choose(&mut rand::rng()) { - if let Some(entry) = OMIKRON_CONNECTIONS.get(&key) { - return Ok(entry.clone()); + if let Some(connection) = get_connected_omikron(key) { + return Ok(connection); } } diff --git a/src/util/logger.rs b/src/util/logger.rs index 39eb933..4276cc0 100644 --- a/src/util/logger.rs +++ b/src/util/logger.rs @@ -244,8 +244,11 @@ pub fn format_cv(cv: &CommunicationValue) -> String { parts.push(format!("> {}", receiver)); } - let comm_type = cv.get_type().to_string(); - parts.push(format!("{}", comm_type)); + let comm_type = cv + .get_comm_type_enum() + .map(|kind| kind.to_string()) + .unwrap_or_else(|| cv.get_type().to_string()); + parts.push(format!("{} (id={})", comm_type, cv.get_id())); let data = cv.data(); diff --git a/type-maps.yaml b/type-maps.yaml deleted file mode 100644 index ce3d5e6..0000000 --- a/type-maps.yaml +++ /dev/null @@ -1,269 +0,0 @@ -protocol_version: "1.0" - -# Note that markers 0 to 31 are reserved for default use, manually working with them is not recommended - -# Fixed CommunicationType markers are: -# Error: 0 -# ErrorParsing: 1 -# ErrorBadVersion: 2 -# Disconnect: 3 -# Redirect: 4 -# Shutdown: 5 -# BadRequest: 6 -# Unauthorized: 7 -# Forbidden: 8 -# NotFound: 9 -# TooManyRequests: 10 -# InternalServerError: 11 -# BadGateway: 12 -# ServiceUnavailable: 13 -# GatewayTimeout: 14 -# Identification: 15 -# IdentificationResponse: 16 -# Register: 17 -# RegisterResponse: 18 -# Ping: 19 -# Pong: 20 - -# Fixed Data Type markers are: -# Error: 0 -# ErrorParsing: 1 -# ErrorMessage: 2 -# Version: 3 -# Description: 4 -# Timestamp: 5 -# Id: 6 -# ClientNonce: 7 -# ServerNonce: 8 -# PublicKeys: 9 -# Signature: 10 -# Connected: 11 - -type_maps: - "1.0": - CommunicationTypes: - ErrorProtocol: 33 - ErrorAnonymous: 34 - ErrorInternal: 35 - ErrorInvalidData: 36 - ErrorInvalidUserId: 37 - ErrorInvalidOmikronId: 38 - ErrorNotFound: 39 - ErrorNotAuthenticated: 40 - ErrorNoIota: 41 - ErrorInvalidChallenge: 42 - ErrorInvalidSecret: 43 - ErrorInvalidPrivateKey: 44 - ErrorInvalidPublicKey: 45 - ErrorNoUserId: 46 - ErrorNoCallId: 47 - ErrorInvalidCallId: 48 - Success: 49 - ShortenLink: 50 - SettingsSave: 51 - SettingsLoad: 52 - SettingsList: 53 - GlobalSettingsSave: 54 - GlobalSettingsLoad: 55 - Message: 56 - MessageState: 57 - MessageSend: 58 - MessageLive: 59 - MessageOtherIota: 60 - MessageChunk: 61 - MessageGet: 143 - MessagesGet: 62 - PushNotification: 63 - ReadNotification: 64 - GetNotifications: 65 - TauriIdentification: 66 - ChangeConfirm: 67 - ConfirmReceive: 68 - ConfirmRead: 69 - GetChats: 70 - GetStates: 71 - AddCommunity: 72 - RemoveCommunity: 73 - GetCommunities: 74 - RegisterIota: 81 - RegisterIotaSuccess: 82 - AddConversation: 85 - SendChat: 86 - ClientChanged: 87 - ClientConnected: 88 - ClientDisconnected: 89 - ClientClosed: 90 - PublicKey: 91 - PrivateKey: 92 - WebrtcSdp: 93 - WebrtcIce: 94 - StartStream: 95 - EndStream: 96 - WatchStream: 97 - CallToken: 98 - CallInvite: 99 - CallDisconnectUser: 100 - CallTimeoutUser: 101 - CallSetAnonymousJoining: 102 - CallData: 103 - EndCall: 104 - Function: 105 - Update: 106 - CreateUser: 107 - RhoUpdate: 108 - UserConnected: 109 - UserDisconnected: 110 - IotaConnected: 111 - IotaDisconnected: 112 - SyncClientIotaStatus: 113 - GetUserData: 114 - GetIotaData: 115 - IotaUserData: 116 - ChangeUserData: 117 - ChangeIotaData: 118 - GetRegister: 119 - CompleteRegisterUser: 120 - CompleteRegisterIota: 121 - DeleteUser: 122 - DeleteIota: 123 - StartRegister: 124 - CompleteRegister: 125 - GetApp: 126 - CreateApp: 127 - DeleteApp: 128 - SaveAppData: 129 - LoadAppData: 130 - AppIdentification: 131 - AppChallenge: 132 - AppChallengeResponse: 133 - AppIdentificationResponse: 134 - LoadTxtRecord: 135 - ErrorNotSet: 136 - SetChatSecret: 139 - GetChatSecret: 140 - ChatSecretResponse: 141 - ChatSecretForward: 142 - MessageEditLive: 144 - MessageEdit: 145 - MessageReactionAdd: 146 - MessageReactionRemove: 147 - MessageReactionLive: 148 - MessageDeleteLive: 150 - DataTypes: - ErrorType: 32 - ErrorProtocol: 33 - AcceptedIds: 34 - Uuid: 35 - RegisterId: 36 - Link: 37 - Settings: 38 - SettingsName: 39 - ChatPartnerId: 40 - ChatPartnerName: 41 - IotaId: 42 - UserId: 43 - UserIds: 44 - IotaIds: 45 - UserState: 46 - UserStates: 47 - UserPings: 48 - CallState: 49 - ScreenShare: 50 - PrivateKeyHash: 51 - # Accepted: 52 now part of default MTP - AcceptedProfiles: 53 - DeniedProfiles: 54 - Content: 55 - Messages: 56 - Notifications: 57 - SendTime: 58 - GetTime: 59 - GetVariant: 60 - SharedSecretOwn: 61 - SharedSecretOther: 62 - SharedSecretSign: 63 - SharedSecret: 64 - CallId: 65 - CallToken: 66 - CallSecret: 67 - Untill: 68 - Enabled: 69 - StartDate: 70 - EndDate: 71 - ReceiverId: 72 - SenderId: 73 - Signed: 75 - Message: 76 - MessageState: 77 - LastPing: 78 - PingIota: 79 - PingClients: 80 - Matches: 81 - Omikron: 82 - Offset: 83 - Amount: 84 - Position: 85 - Name: 86 - Path: 87 - Codec: 88 - Function: 89 - Payload: 90 - Result: 91 - Interactables: 92 - WantToWatch: 93 - Watcher: 94 - CreatedAt: 95 - Username: 96 - Display: 97 - Avatar: 98 - About: 99 - Status: 100 - PublicKey: 101 - SubLevel: 102 - SubEnd: 103 - CommunityAddress: 104 - CommunityTitle: 106 - Communities: 107 - RhoConnections: 108 - User: 109 - OnlineStatus: 110 - OmikronId: 111 - OmikronConnections: 112 - ResetToken: 113 - NewToken: 114 - CallInvited: 115 - CallMembers: 116 - Calls: 117 - Timeout: 118 - HasAdmin: 119 - LastMessageAt: 120 - Height: 121 - SentBySelf: 122 - SessionId: 123 - Contacts: 124 - LastMessage: 125 - AppIdentifier: 127 - AppPrivateKey: 128 - AppPublicKey: 129 - AppSession: 130 - AppData: 131 - TauriToken: 132 - Challenge: 133 - EncryptedPayload: 134 - SecurePayload: 135 - DeviceId: 136 - ClientId: 137 - SecretId: 142 - VersionNumber: 143 - EncryptedSecret: 144 - WrappingScheme: 146 - UpdatedAt: 147 - ChatId: 148 - KemCiphertext: 149 - SenderUserId: 152 - RecipientUserId: 153 - Recipients: 154 - Edited: 155 - Reactions: 156 - Reaction: 157 - ReplyId: 158