Lock file maintenance

Cargo.toml

@@ -7,28 +7,28 @@ edition = "2024"
 mtp = { git = "https://git.methanium.net/Methanium/mtp.git",  features = [
     "crypto",
     "files",
-    "raw",
     "web-server",
 ] }
 
 ansi_term = "0.12.1"
-base64 = "0.23.1"
+base64 = "0.22.1"
 bytes = "1"
 dashmap = "6.2.1"
 dotenv = "0.15.0"
 http = "1"
 once_cell = "1.21.4"
 rand = "0.10.2"
-rustls = { version = "0.23.43", default-features = false, features = [
+rustls = { version = "0.23.42", default-features = false, features = [
     "std",
     "tls12",
     "aws-lc-rs",
     "prefer-post-quantum",
 ] }
-sqlx = { version = "0.9.0", features = ["mysql", "runtime-tokio", "migrate"] }
+sqlx = { version = "0.8.6", features = ["mysql", "runtime-tokio", "migrate"] }
 tokio = { version = "*", features = ["full"] }
 tokio-util = { version = "0.7.19", features = ["rt"] }
-uuid = { version = "1.26.0", features = ["v4", "v7"] }
-thiserror = "2.0.20"
+uuid = { version = "1.24.0", features = ["v4", "v7"] }
+thiserror = "2.0.19"
 serde = { version = "1.0.229", features = ["derive"] }
 serde_json = "1.0.151"
+zeroize = "1.9"

README.md

@@ -1,7 +1,7 @@
 # Omega
 The Omega is Tensamin's central Server. It maintains the centralized user Registry & manages Omikron useage.
 
-On first start Omega creates an unencrypted private keyring at `./omega.mk` and the matching public bundle at `./omega.mpkb`. Existing identity files are loaded fail-closed, so a malformed keyring or mismatched public bundle stops startup. A missing public bundle is rebuilt from a valid private keyring without generating a new identity.
+Omega requires the non-empty `OMEGA_IDENTITY_SECRET` environment variable at startup. On first start it creates the protected private identity at `./omega.mk` and the matching public bundle at `./omega.mpkb`. Existing identity files are loaded fail-closed, so a malformed keyring or mismatched public bundle stops startup. A missing public bundle is rebuilt from a valid private keyring without generating a new identity.
 
 ## MTP routing contract
 

migrations/001_initial_schema.sql

@@ -15,7 +15,6 @@ CREATE TABLE IF NOT EXISTS users (
     sub_end BIGINT NOT NULL DEFAULT 0,
     public_key BLOB NOT NULL,
     token BLOB NOT NULL,
-    created_at TIMESTAMP(3) NOT NULL DEFAULT CURRENT_TIMESTAMP(3),
 
     UNIQUE KEY uk_users_username (username),
     KEY idx_users_iota_id (iota_id),

migrations/007_iota_snapshot_outbox.sql

@@ -1,4 +0,0 @@
-CREATE TABLE iota_snapshot_outbox (
-    iota_id BIGINT NOT NULL PRIMARY KEY,
-    updated_at DATETIME NOT NULL
-);

migrations/008_user_created_at.sql

@@ -1,2 +0,0 @@
-ALTER TABLE users
-    ADD COLUMN IF NOT EXISTS created_at TIMESTAMP(3) NOT NULL DEFAULT CURRENT_TIMESTAMP(3);

mtp-type-maps

@@ -1 +1 @@
-Subproject commit f4e45aa3a3ad0e3c3a257f66857b904a1af7901c
+Subproject commit 486541b9483356ff49ff3ec7016f87d3ecbeaa0e

src/db/iota_repo.rs

@@ -22,7 +22,7 @@ pub async fn get_iota_by_id(id: IotaId) -> Result<Iota> {
 
 pub async fn create_new_iota(public_key: PublicKeyBundle) -> Result<IotaId> {
     for _ in 0..16 {
-        let id = crate::db::user_repo::generate_protocol_id();
+        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),

src/db/user_repo.rs

@@ -11,13 +11,22 @@ use std::collections::HashMap;
 pub const MAX_PROTOCOL_ID: i64 = (1_i64 << 48) - 1;
 const ID_ALLOCATION_ATTEMPTS: usize = 16;
 
-pub fn generate_protocol_id() -> UserId {
-    loop {
-        let value = rand::random::<u64>() & ((1_u64 << 48) - 1);
-        if value != 0 {
-            return UserId::from(value as i64);
-        }
+pub async fn get_register_id() -> Result<UserId> {
+    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 (1..=MAX_PROTOCOL_ID).contains(&ts) {
+        return Ok(UserId::from(ts));
     }
+
+    // Fall back to random if the timestamp is outside the 48-bit range.
+    let id = (rand::random::<u64>() & ((1_u64 << 48) - 1)) as i64;
+    Ok(UserId::from(id.max(1)))
 }
 
 pub fn valid_protocol_id(id: i64) -> bool {
@@ -33,24 +42,8 @@ pub async fn allocate_registration(iota_id: IotaId, request_id: u32) -> Result<(
         ));
     }
 
-    let database = pool().await?;
-    sqlx::query(
-        "UPDATE registration_leases SET request_id = NULL \
-         WHERE request_id IS NOT NULL AND expires_at < UTC_TIMESTAMP()",
-    )
-    .execute(&database)
-    .await?;
-
     for _ in 0..ID_ALLOCATION_ATTEMPTS {
-        let id = generate_protocol_id();
-        let user_exists = sqlx::query("SELECT 1 FROM users WHERE id = ? LIMIT 1")
-            .bind(id.0)
-            .fetch_optional(&database)
-            .await?
-            .is_some();
-        if user_exists {
-            continue;
-        }
+        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) \
@@ -60,7 +53,7 @@ pub async fn allocate_registration(iota_id: IotaId, request_id: u32) -> Result<(
         .bind(id.0)
         .bind(iota_id.0)
         .bind(request_id)
-        .execute(&database)
+        .execute(&pool().await?)
         .await;
         match result {
             Ok(_) => return Ok((id, token)),
@@ -71,7 +64,7 @@ pub async fn allocate_registration(iota_id: IotaId, request_id: u32) -> Result<(
                 )
                 .bind(iota_id.0)
                 .bind(request_id)
-                .fetch_optional(&database)
+                .fetch_optional(&pool().await?)
                 .await?;
                 if let Some(existing) = existing {
                     let current: i8 = existing.get("current");
@@ -101,9 +94,9 @@ pub(crate) fn is_duplicate_key(error: &sqlx::Error) -> bool {
     })
 }
 
-const USER_BY_USERNAME_QUERY: &str = "SELECT id, iota_id, username, display, status, presence_preference, about, avatar, sub_level, sub_end, public_key, token, CAST(UNIX_TIMESTAMP(created_at) * 1000 AS SIGNED) AS created_at FROM users WHERE username = ?";
-const USER_BY_ID_QUERY: &str = "SELECT id, iota_id, username, display, status, presence_preference, about, avatar, sub_level, sub_end, public_key, token, CAST(UNIX_TIMESTAMP(created_at) * 1000 AS SIGNED) AS created_at FROM users WHERE id = ?";
-const USER_COLUMNS: &str = "SELECT id, iota_id, username, display, status, presence_preference, about, avatar, sub_level, sub_end, public_key, token, CAST(UNIX_TIMESTAMP(created_at) * 1000 AS SIGNED) AS created_at FROM users";
+const USER_BY_USERNAME_QUERY: &str = "SELECT id, iota_id, username, display, status, presence_preference, 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, presence_preference, about, avatar, sub_level, sub_end, public_key, token FROM users WHERE id = ?";
+const USER_COLUMNS: &str = "SELECT id, iota_id, username, display, status, presence_preference, about, avatar, sub_level, sub_end, public_key, token FROM users";
 
 #[derive(FromRow)]
 struct UserRow {
@@ -119,7 +112,6 @@ struct UserRow {
     sub_end: i64,
     public_key: Vec<u8>,
     token: Vec<u8>,
-    created_at: i64,
 }
 
 impl TryFrom<UserRow> for User {
@@ -143,7 +135,6 @@ impl TryFrom<UserRow> for User {
             sub_end: row.sub_end,
             public_key,
             token: decode(row.token)?,
-            created_at: row.created_at,
         })
     }
 }
@@ -181,7 +172,7 @@ fn normalized_ids(ids: &[i64]) -> Vec<i64> {
     ids
 }
 
-fn append_in_clause(query: &mut QueryBuilder<MySql>, ids: &[i64]) {
+fn append_in_clause(query: &mut QueryBuilder<'_, MySql>, ids: &[i64]) {
     query.push("(");
     for (index, id) in ids.iter().enumerate() {
         if index > 0 {
@@ -192,7 +183,7 @@ fn append_in_clause(query: &mut QueryBuilder<MySql>, ids: &[i64]) {
     query.push(")");
 }
 
-async fn fetch_users(mut query: QueryBuilder<MySql>) -> Result<Vec<User>> {
+async fn fetch_users(mut query: QueryBuilder<'_, MySql>) -> Result<Vec<User>> {
     query
         .build_query_as::<UserRow>()
         .fetch_all(&pool().await?)
@@ -345,59 +336,21 @@ pub async fn change_status(id: UserId, value: String) -> Result<()> {
 }
 
 pub async fn change_iota_id(id: UserId, value: Option<IotaId>) -> Result<()> {
-    let mut transaction = pool().await?.begin().await?;
-    let previous = sqlx::query("SELECT iota_id FROM users WHERE id = ? FOR UPDATE")
-        .bind(id.0)
-        .fetch_optional(&mut *transaction)
-        .await?
-        .ok_or(OmegaError::NotFound)?;
-    let previous_iota_id: Option<i64> = previous.get("iota_id");
     sqlx::query("UPDATE users SET iota_id = ? WHERE id = ?")
         .bind(value.map(|id| id.0))
         .bind(id.0)
-        .execute(&mut *transaction)
-        .await?;
-    if let Some(iota_id) = previous_iota_id {
-        enqueue_iota_snapshot(&mut transaction, iota_id).await?;
-    }
-    if let Some(iota_id) = value {
-        enqueue_iota_snapshot(&mut transaction, iota_id.0).await?;
-    }
-    transaction.commit().await?;
-    Ok(())
-}
-
-async fn enqueue_iota_snapshot(
-    transaction: &mut sqlx::Transaction<'_, MySql>,
-    iota_id: i64,
-) -> Result<()> {
-    sqlx::query(
-        "INSERT INTO iota_snapshot_outbox (iota_id, updated_at) VALUES (?, UTC_TIMESTAMP()) \
-         ON DUPLICATE KEY UPDATE updated_at = VALUES(updated_at)",
-    )
-    .bind(iota_id)
-    .execute(&mut **transaction)
-    .await?;
-    Ok(())
-}
-
-pub async fn pending_iota_snapshots() -> Result<Vec<IotaId>> {
-    let rows = sqlx::query("SELECT iota_id FROM iota_snapshot_outbox ORDER BY updated_at")
-        .fetch_all(&pool().await?)
-        .await?;
-    Ok(rows
-        .into_iter()
-        .map(|row| IotaId::from(row.get::<i64, _>("iota_id")))
-        .collect())
-}
-
-pub async fn complete_iota_snapshot(iota_id: IotaId) -> Result<()> {
-    sqlx::query("DELETE FROM iota_snapshot_outbox WHERE iota_id = ?")
-        .bind(iota_id.0)
         .execute(&pool().await?)
         .await?;
     Ok(())
 }
+pub async fn change_token(id: UserId, value: String) -> Result<()> {
+    update(
+        id,
+        "UPDATE users SET token = ? WHERE id = ?",
+        value.into_bytes(),
+    )
+    .await
+}
 /// Delete the central identity while retaining a durable instruction for the
 /// last hosting Iota. The pending row is intentionally independent of users:
 /// it must outlive the account row.
@@ -535,7 +488,6 @@ pub async fn register_complete_user(
         }
     };
     result?;
-    enqueue_iota_snapshot(&mut transaction, iota_id.0).await?;
     sqlx::query("UPDATE registration_leases SET completed_at = UTC_TIMESTAMP() WHERE token = ?")
         .bind(&registration_token)
         .execute(&mut *transaction)
@@ -546,20 +498,19 @@ pub async fn register_complete_user(
 
 fn valid_username(username: &str) -> bool {
     !username.is_empty()
-        && username.len() <= 15
-        && username
-            .bytes()
-            .all(|byte| byte.is_ascii_lowercase() || byte.is_ascii_digit())
+        && username.chars().count() <= 15
+        && !username.chars().any(char::is_control)
+        && !username.contains(['/', '\\'])
 }
 
 #[cfg(test)]
 mod tests {
-    use super::{MAX_PROTOCOL_ID, generate_protocol_id, valid_protocol_id};
+    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(generate_protocol_id().0));
+            assert!(valid_protocol_id(get_register_id().await.unwrap().0));
         }
         assert!(!valid_protocol_id(0));
         assert!(valid_protocol_id(MAX_PROTOCOL_ID));

src/identity.rs

@@ -1,7 +1,7 @@
 use crate::error::{IdentityError, Result};
 use mtp::crypto::{Keyring, PublicKeyBundle};
 use mtp::files::{
-    FileError, load_keyring_raw, load_public_key_bundle, save_keyring_raw, save_public_key_bundle,
+    FileError, load_keyring, load_public_key_bundle, save_keyring, save_public_key_bundle,
 };
 use std::{
     fs,
@@ -22,20 +22,25 @@ pub struct OmegaIdentity {
 static PUBLIC_BUNDLE_TEMP_COUNTER: AtomicU64 = AtomicU64::new(0);
 
 impl OmegaIdentity {
-    pub fn load_or_create() -> Result<Self> {
-        Self::load_or_create_at(Path::new(KEYRING_PATH), Path::new(PUBLIC_KEY_PATH))
+    pub fn load_or_create(passphrase: &[u8]) -> Result<Self> {
+        Self::load_or_create_at(
+            Path::new(KEYRING_PATH),
+            Path::new(PUBLIC_KEY_PATH),
+            passphrase,
+        )
     }
 
     pub(crate) fn load_or_create_at(
         keyring_path: impl AsRef<Path>,
         public_key_path: impl AsRef<Path>,
+        passphrase: &[u8],
     ) -> Result<Self> {
         let keyring_path = keyring_path.as_ref();
         let public_key_path = public_key_path.as_ref();
-        let keyring = match load_keyring_raw(keyring_path) {
+        let keyring = match load_keyring(keyring_path, passphrase) {
             Ok(keyring) => keyring,
             Err(FileError::Io(error)) if error.kind() == std::io::ErrorKind::NotFound => {
-                return Self::create_at(keyring_path, public_key_path);
+                return Self::create_at(keyring_path, public_key_path, passphrase);
             }
             Err(error) => {
                 return Err(IdentityError::Storage {
@@ -67,11 +72,13 @@ impl OmegaIdentity {
         Ok(identity)
     }
 
-    fn create_at(keyring_path: &Path, public_key_path: &Path) -> Result<Self> {
+    fn create_at(keyring_path: &Path, public_key_path: &Path, passphrase: &[u8]) -> Result<Self> {
         let keyring = Keyring::generate();
-        save_keyring_raw(&keyring, keyring_path).map_err(|error| IdentityError::Storage {
-            path: keyring_path.to_path_buf(),
-            source: error,
+        save_keyring(&keyring, keyring_path, passphrase).map_err(|error| {
+            IdentityError::Storage {
+                path: keyring_path.to_path_buf(),
+                source: error,
+            }
         })?;
         Self::persist_public_bundle(&keyring.public_key_bundle(), public_key_path)?;
         Ok(Self {
@@ -181,7 +188,9 @@ mod tests {
         let directory = test_directory();
         let keyring_path = directory.join("omega.mk");
         let public_key_path = directory.join("omega.mpkb");
-        let first = OmegaIdentity::load_or_create_at(&keyring_path, &public_key_path)
+        let passphrase = b"test-passphrase";
+
+        let first = OmegaIdentity::load_or_create_at(&keyring_path, &public_key_path, passphrase)
             .expect("create identity");
         let first_bundle = first.public_key_bundle().try_as_bytes().expect("bundle");
 
@@ -190,8 +199,9 @@ mod tests {
         assert_eq!(&keyring_bytes[..4], b"MTMK");
         assert_eq!(&bundle_bytes[..4], b"MPKB");
 
-        let restarted = OmegaIdentity::load_or_create_at(&keyring_path, &public_key_path)
-            .expect("reload identity");
+        let restarted =
+            OmegaIdentity::load_or_create_at(&keyring_path, &public_key_path, passphrase)
+                .expect("reload identity");
         assert_eq!(
             restarted
                 .public_key_bundle()
@@ -217,7 +227,8 @@ mod tests {
         let public_key_path = directory.join("omega.mpkb");
         fs::write(&keyring_path, b"not-a-keyring").expect("write invalid keyring");
 
-        let result = OmegaIdentity::load_or_create_at(&keyring_path, &public_key_path);
+        let result =
+            OmegaIdentity::load_or_create_at(&keyring_path, &public_key_path, b"test-passphrase");
         assert!(result.is_err());
         assert!(!public_key_path.exists());
 
@@ -229,12 +240,16 @@ mod tests {
         let directory = test_directory();
         let keyring_path = directory.join("omega.mk");
         let public_key_path = directory.join("omega.mpkb");
-        OmegaIdentity::load_or_create_at(&keyring_path, &public_key_path).expect("create identity");
+        let passphrase = b"test-passphrase";
+
+        OmegaIdentity::load_or_create_at(&keyring_path, &public_key_path, passphrase)
+            .expect("create identity");
         let original_keyring = fs::read(&keyring_path).expect("read keyring");
         fs::remove_file(&public_key_path).expect("remove bundle");
 
-        let repaired = OmegaIdentity::load_or_create_at(&keyring_path, &public_key_path)
-            .expect("repair bundle");
+        let repaired =
+            OmegaIdentity::load_or_create_at(&keyring_path, &public_key_path, passphrase)
+                .expect("repair bundle");
 
         assert_eq!(
             fs::read(&keyring_path).expect("read keyring"),
@@ -256,13 +271,16 @@ mod tests {
         let directory = test_directory();
         let keyring_path = directory.join("omega.mk");
         let public_key_path = directory.join("omega.mpkb");
-        OmegaIdentity::load_or_create_at(&keyring_path, &public_key_path).expect("create identity");
+        let passphrase = b"test-passphrase";
+
+        OmegaIdentity::load_or_create_at(&keyring_path, &public_key_path, passphrase)
+            .expect("create identity");
         let other_keyring = Keyring::generate();
         save_public_key_bundle(&other_keyring.public_key_bundle(), &public_key_path)
             .expect("save mismatched bundle");
 
         assert!(matches!(
-            OmegaIdentity::load_or_create_at(&keyring_path, &public_key_path),
+            OmegaIdentity::load_or_create_at(&keyring_path, &public_key_path, passphrase),
             Err(crate::OmegaError::Identity(
                 IdentityError::PublicBundleMismatch { .. }
             ))
@@ -272,25 +290,24 @@ mod tests {
     }
 
     #[test]
-    fn existing_raw_keyring_is_reloaded_without_a_passphrase() {
+    fn wrong_passphrase_does_not_replace_existing_keyring() {
         let directory = test_directory();
         let keyring_path = directory.join("omega.mk");
         let public_key_path = directory.join("omega.mpkb");
+        let passphrase = b"test-passphrase";
 
-        let first = OmegaIdentity::load_or_create_at(&keyring_path, &public_key_path)
+        OmegaIdentity::load_or_create_at(&keyring_path, &public_key_path, passphrase)
             .expect("create identity");
         let original_keyring = fs::read(&keyring_path).expect("read keyring");
 
-        let reloaded = OmegaIdentity::load_or_create_at(&keyring_path, &public_key_path)
-            .expect("reload identity");
+        assert!(
+            OmegaIdentity::load_or_create_at(&keyring_path, &public_key_path, b"wrong-passphrase")
+                .is_err()
+        );
         assert_eq!(
             fs::read(&keyring_path).expect("read keyring"),
             original_keyring
         );
-        assert_eq!(
-            reloaded.public_key_bundle().try_as_bytes().expect("bundle"),
-            first.public_key_bundle().try_as_bytes().expect("bundle")
-        );
 
         fs::remove_dir_all(directory).expect("remove test directory");
     }

src/main.rs

@@ -15,7 +15,6 @@ pub use error::{OmegaError, Result};
 use crate::db::initialize;
 use crate::state::OmegaState;
 use crate::transport::omikron_connection;
-use crate::transport::omikron_manager;
 use crate::util::file_util::get_directory;
 use crate::util::logger::PrintType;
 use crate::util::logger::startup;
@@ -26,6 +25,7 @@ use std::env;
 use std::path::Path;
 use std::time::Duration;
 use tokio::time::interval;
+use zeroize::Zeroizing;
 
 #[tokio::main]
 async fn main() {
@@ -38,6 +38,18 @@ async fn main() {
     log_in!("Incoming messages");
     log_out!("Outgoing messages");
 
+    let identity_secret = match env::var("OMEGA_IDENTITY_SECRET") {
+        Ok(secret) if !secret.is_empty() => secret,
+        Ok(_) => {
+            log!("[FATAL] OMEGA_IDENTITY_SECRET must not be empty");
+            return;
+        }
+        Err(error) => {
+            log!("[FATAL] Unable to load OMEGA_IDENTITY_SECRET: {}", error);
+            return;
+        }
+    };
+    let identity_secret = Zeroizing::new(identity_secret);
     let config = match OmegaConfig::from_env() {
         Ok(config) => config,
         Err(error) => {
@@ -49,13 +61,14 @@ async fn main() {
         log!("[FATAL] Omega rate-limit configuration was initialized more than once");
         return;
     }
-    let identity = match identity::OmegaIdentity::load_or_create() {
+    let identity = match identity::OmegaIdentity::load_or_create(identity_secret.as_bytes()) {
         Ok(identity) => identity,
         Err(error) => {
             log!("[FATAL] Omega identity initialization failed: {}", error);
             return;
         }
     };
+    drop(identity_secret);
     let state = OmegaState::new(identity, config);
 
     log!("Started");
@@ -84,7 +97,6 @@ async fn main() {
             }
         }
     });
-    let snapshot_outbox_worker = omikron_manager::spawn_iota_snapshot_outbox_worker();
     let port: u16 = env::var("PORT")
         .ok()
         .and_then(|s| s.parse().ok())
@@ -102,5 +114,4 @@ async fn main() {
     }
     rate_limit_cleanup.abort();
     short_link_cleanup.abort();
-    snapshot_outbox_worker.abort();
 }

src/models/user.rs

@@ -17,5 +17,4 @@ pub struct User {
     pub public_key: PublicKeyBundle,
     #[serde(skip_serializing)]
     pub token: String,
-    pub created_at: i64,
 }

src/server/api.rs

@@ -13,7 +13,6 @@ use crate::models::UserId;
 use crate::server::{
     middleware,
     validation::{parse_positive_id, validate_non_empty},
-    with_cors,
 };
 use crate::transport::omikron_manager::{
     get_all_connections, get_connected_omikron, get_iota_primary_omikron_connection,
@@ -194,34 +193,39 @@ pub async fn handle(
     let method = request.method;
     let path = request.uri.path().to_string();
     if method != Method::OPTIONS && !middleware::allow(request.remote_addr.ip(), &path) {
-        return with_cors(response.status(StatusCode::TOO_MANY_REQUESTS).body(json(
-            &StatusResponse {
+        return response
+            .status(StatusCode::TOO_MANY_REQUESTS)
+            .header("access-control-allow-origin", crate::config::cors_origin())
+            .body(json(&StatusResponse {
                 status: "error_rate_limited",
-            },
-        )));
+            }));
     }
     if method == Method::OPTIONS {
-        return with_cors(response.status(StatusCode::NO_CONTENT));
+        return response
+            .status(StatusCode::OK)
+            .header("access-control-allow-origin", crate::config::cors_origin())
+            .header("access-control-allow-methods", "GET, POST, OPTIONS")
+            .header("access-control-allow-headers", "*");
     }
     let path_parts: Vec<&str> = path.split('/').filter(|part| !part.is_empty()).collect();
     if let ["api", "download", "iota_frontend"] = path_parts.as_slice() {
         let file_path = format!("{}/downloads/iota_frontend.zip", get_directory());
         return match std::fs::read(file_path) {
-            Ok(bytes) => with_cors(
-                response
-                    .status(StatusCode::OK)
-                    .header("content-type", "application/zip")
-                    .header(
-                        "content-disposition",
-                        "attachment; filename=\"iota_frontend.zip\"",
-                    )
-                    .body(Bytes::from(bytes)),
-            ),
-            Err(_) => with_cors(response.status(StatusCode::NOT_FOUND).body(json(
-                &StatusResponse {
+            Ok(bytes) => response
+                .status(StatusCode::OK)
+                .header("access-control-allow-origin", crate::config::cors_origin())
+                .header("content-type", "application/zip")
+                .header(
+                    "content-disposition",
+                    "attachment; filename=\"iota_frontend.zip\"",
+                )
+                .body(Bytes::from(bytes)),
+            Err(_) => response
+                .status(StatusCode::NOT_FOUND)
+                .header("access-control-allow-origin", crate::config::cors_origin())
+                .body(json(&StatusResponse {
                     status: "error_not_found",
-                },
-            ))),
+                })),
         };
     }
     if let ["direct", short @ ..] = path_parts.as_slice() {
@@ -229,16 +233,19 @@ pub async fn handle(
         let location = crate::server::short_link::get_short_link(&short)
             .await
             .unwrap_or_else(|_| "https://tensamin.net".to_string());
-        return with_cors(
-            response
-                .status(StatusCode::TEMPORARY_REDIRECT)
-                .header("location", &location),
-        );
+        return response
+            .status(StatusCode::TEMPORARY_REDIRECT)
+            .header("location", &location);
     }
     let (status, body) = route(&path_parts, &identity)
         .await
         .unwrap_or_else(|error| (error.status_code(), error_body(&error)));
-    with_cors(response.status(status).body(body))
+    response
+        .status(status)
+        .header("access-control-allow-origin", crate::config::cors_origin())
+        .header("access-control-allow-headers", "*")
+        .header("access-control-allow-methods", "GET, POST, OPTIONS")
+        .body(body)
 }
 
 pub async fn handle_pattern(

src/server/index.rs

@@ -1,8 +1,6 @@
 use http::StatusCode;
 use mtp::webserver::HttpResponse;
 
-use crate::server::with_cors;
-
 pub fn index_handler(response: HttpResponse) -> HttpResponse {
     let documentation = r#"
 Omega API Server
@@ -35,10 +33,8 @@ live in-memory state; it is empty after an Omega restart until Omikrons sync.
 All other routes will return this documentation.
 "#;
 
-    with_cors(
-        response
-            .status(StatusCode::OK)
-            .header("content-type", "text/plain; charset=utf-8")
-            .body(documentation),
-    )
+    response
+        .status(StatusCode::OK)
+        .header("content-type", "text/plain; charset=utf-8")
+        .body(documentation)
 }

src/server/mod.rs

@@ -4,29 +4,3 @@ pub mod middleware;
 pub mod short_link;
 pub mod validation;
 pub mod web;
-
-use mtp::webserver::HttpResponse;
-
-pub(crate) fn with_cors(response: HttpResponse) -> HttpResponse {
-    response
-        .header("access-control-allow-origin", crate::config::cors_origin())
-        .header("access-control-allow-methods", "GET, POST, OPTIONS")
-        .header("access-control-allow-headers", "*")
-}
-
-#[cfg(test)]
-mod tests {
-    use super::with_cors;
-    use mtp::webserver::HttpResponse;
-
-    #[test]
-    fn web_responses_allow_all_origins() {
-        let response = with_cors(HttpResponse::default());
-        assert_eq!(response.headers["access-control-allow-origin"], "*");
-        assert_eq!(
-            response.headers["access-control-allow-methods"],
-            "GET, POST, OPTIONS"
-        );
-        assert_eq!(response.headers["access-control-allow-headers"], "*");
-    }
-}

src/server/web.rs

@@ -1,6 +1,5 @@
 use crate::identity::OmegaIdentity;
 use crate::server::{api, index::index_handler};
-use http::{Method, StatusCode};
 use mtp::webserver::{HttpRequest, HttpResponse, RouteParams, WebServerConfig};
 use std::{future::Future, pin::Pin, sync::Arc};
 
@@ -44,11 +43,5 @@ pub fn build_web_config(
         )?
         .route_pattern("/api/get/user/{id}", api_pattern_handler(identity.clone()))?
         .route_pattern("/direct/{short}", api_pattern_handler(identity))?
-        .fallback(|request, response| async move {
-            if request.method == Method::OPTIONS {
-                crate::server::with_cors(response.status(StatusCode::NO_CONTENT))
-            } else {
-                index_handler(response)
-            }
-        })
+        .fallback(|_request, response| async move { index_handler(response) })
 }

src/transport/connection.rs

@@ -73,6 +73,7 @@ pub(crate) fn validate_dispatch_fields(value: &CommunicationValue) -> OmikronRes
         message_type,
         mtp::codec::CommunicationType::GetUserData
             | mtp::codec::CommunicationType::ChangeUserData
+            | mtp::codec::CommunicationType::ChangeIotaData
             | mtp::codec::CommunicationType::DeleteUser
             | mtp::codec::CommunicationType::AttachUserBegin
             | mtp::codec::CommunicationType::AttachUserComplete
@@ -146,6 +147,7 @@ mod tests {
         for message_type in [
             CommunicationType::GetUserData,
             CommunicationType::ChangeUserData,
+            CommunicationType::ChangeIotaData,
             CommunicationType::DeleteUser,
             CommunicationType::AttachUserBegin,
             CommunicationType::AttachUserComplete,

src/transport/handlers/account.rs

@@ -84,7 +84,7 @@ pub async fn release_from_iota(
     match user_repo::change_iota_id(user.id, None).await {
         Ok(()) => {
             if let Some(iota) = previous_iota {
-                let _ = crate::transport::omikron_manager::publish_iota_user_snapshot(iota.0).await;
+                crate::transport::omikron_manager::publish_iota_user_snapshot(iota.0).await;
             }
             connection
                 .send(
@@ -225,9 +225,9 @@ pub async fn attach_complete(
     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) {
-                let _ = crate::transport::omikron_manager::publish_iota_user_snapshot(iota.0).await;
+                crate::transport::omikron_manager::publish_iota_user_snapshot(iota.0).await;
             }
-            let _ = crate::transport::omikron_manager::publish_iota_user_snapshot(requester).await;
+            crate::transport::omikron_manager::publish_iota_user_snapshot(requester).await;
             connection
                 .send(
                     &CommunicationValue::new(CommunicationType::Success)
@@ -252,8 +252,7 @@ async fn complete_delete(
         Ok(iota_id) => {
             let cleanup_pending = iota_id.is_some();
             if let Some(iota_id) = iota_id {
-                let _ =
-                    crate::transport::omikron_manager::publish_iota_user_snapshot(iota_id.0).await;
+                crate::transport::omikron_manager::publish_iota_user_snapshot(iota_id.0).await;
                 crate::transport::omikron_manager::deliver_pending_erasures(iota_id.0).await;
             }
             connection

src/transport/handlers/presence.rs

@@ -562,15 +562,6 @@ pub async fn sync_status(
         &affected_iota_ids.iter().copied().collect::<Vec<_>>(),
     )
     .await?;
-    if sessions.iter().any(|(user_id, _, iota_id)| {
-        !users
-            .iter()
-            .any(|user| user.id.0 == *user_id && user.iota_id == Some(IotaId::from(*iota_id)))
-    }) {
-        return connection
-            .send_error_response(request_id, CommunicationType::ErrorInvalidData)
-            .await;
-    }
     let user_ids = users.iter().map(|user| user.id.0).collect::<Vec<_>>();
     apply_preferences(
         &state,

src/transport/handlers/register.rs

@@ -135,17 +135,6 @@ pub async fn complete_user(
     .await
     {
         Ok(()) => {
-            if let Err(error) =
-                crate::transport::omikron_manager::publish_iota_user_snapshot(iota_id).await
-            {
-                crate::log_in!(
-                    crate::util::logger::PrintType::General,
-                    "Could not publish Iota snapshot after registering user {} on Iota {}: {}",
-                    user_id,
-                    iota_id,
-                    error
-                );
-            }
             connection
                 .send(
                     &CommunicationValue::new(CommunicationType::Success)

src/transport/handlers/user_data.rs

@@ -64,10 +64,6 @@ pub async fn get_user(
             DataValue::Str(user.public_key.try_to_base64()?),
         )
         .add_typed_default(DataType::UserId, DataValue::SignedNumber(id.into()))
-        .add_typed_default(
-            DataType::CreatedAt,
-            DataValue::SignedNumber(user.created_at.into()),
-        )
         .add_typed_default(DataType::Display, DataValue::Str(display))
         .add_typed_default(
             DataType::SubLevel,
@@ -268,3 +264,49 @@ pub async fn change_user(
 ) -> OmikronResult<()> {
     update_user(connection, value).await
 }
+
+pub async fn change_iota(
+    connection: Arc<OmikronConnection>,
+    value: CommunicationValue,
+) -> OmikronResult<()> {
+    let request_id = value.require_id()?;
+    let reset_data = value.get_data(DataType::ResetToken);
+    let Some(reset) = reset_data.as_str() else {
+        return connection
+            .send_error_response(value.require_id()?, CommunicationType::ErrorInvalidData)
+            .await;
+    };
+    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.require_id()?, CommunicationType::ErrorInvalidData)
+            .await;
+    };
+    let user_id = UserId::from(value.require_sender_i64()?);
+    let user = match user_repo::get_by_user_id(user_id).await {
+        Ok(user) => user,
+        Err(_) => {
+            return connection
+                .send_error_response(request_id, CommunicationType::ErrorNotFound)
+                .await;
+        }
+    };
+    if user.token != reset {
+        return connection
+            .send_error_response(request_id, CommunicationType::ErrorInvalidChallenge)
+            .await;
+    }
+    let result =
+        match user_repo::change_iota_id(user_id, Some(IotaId::from(value.require_sender_i64()?)))
+            .await
+        {
+            Ok(()) => user_repo::change_token(user_id, new_token.to_owned()).await,
+            Err(error) => Err(error),
+        };
+    let response = match result {
+        Ok(()) => CommunicationValue::new(CommunicationType::Success),
+        Err(error) => CommunicationValue::new(CommunicationType::ErrorInternal)
+            .add_typed_default(DataType::ErrorType, DataValue::Str(error.to_string())),
+    };
+    connection.send(&response.with_id(request_id)).await
+}

src/transport/omikron_connection.rs

@@ -394,7 +394,9 @@ impl OmikronConnection {
             let request_id = value.id();
             let result = crate::transport::relay_router::route_from_omikron(id, value).await;
             let response = match &result {
-                Ok(response) => request_id.map(|request_id| response.clone().with_id(request_id)),
+                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| {
@@ -415,9 +417,7 @@ impl OmikronConnection {
                     send_error
                 );
             }
-            return result
-                .map(|_| ())
-                .map_err(|error| crate::error::OmegaError::Transport(error.to_string()));
+            return result.map_err(|error| crate::error::OmegaError::Transport(error.to_string()));
         }
         match value.get_comm_type_enum() {
             Some(CommunicationType::ShortenLink) => {
@@ -449,6 +449,9 @@ impl OmikronConnection {
             Some(CommunicationType::ChangeUserData) => {
                 crate::transport::handlers::user_data::change_user(self, value).await
             }
+            Some(CommunicationType::ChangeIotaData) => {
+                crate::transport::handlers::user_data::change_iota(self, value).await
+            }
             Some(CommunicationType::GetRegister) => {
                 crate::transport::handlers::register::get_register(self, value).await
             }
@@ -761,9 +764,8 @@ pub async fn start(port: u16, state: Arc<OmegaState>) -> Result<(), Box<dyn std:
 
 #[cfg(test)]
 mod tests {
-    use super::{DispatchClass, OmikronConnection, parse_bind_address};
+    use super::{DispatchClass, OmikronConnection};
     use mtp::codec::{CommunicationType, CommunicationValue};
-    use std::net::{IpAddr, Ipv4Addr};
 
     #[test]
     fn relay_dispatch_is_not_on_the_ordered_state_lane() {
@@ -774,14 +776,6 @@ mod tests {
         );
     }
 
-    #[test]
-    fn parses_configured_bind_address() {
-        assert_eq!(
-            parse_bind_address(Some("10.200.2.0")),
-            Ok(IpAddr::V4(Ipv4Addr::new(10, 200, 2, 0)))
-        );
-    }
-
     #[test]
     fn presence_lifecycle_dispatch_is_ordered() {
         let value = CommunicationValue::new(CommunicationType::UserConnected);
@@ -812,12 +806,4 @@ mod tests {
             &CommunicationValue::new(CommunicationType::GetUserData).with_id(1)
         ));
     }
-
-    #[test]
-    fn defaults_bind_address_to_all_interfaces() {
-        assert_eq!(
-            parse_bind_address(None),
-            Ok(IpAddr::V4(Ipv4Addr::UNSPECIFIED))
-        );
-    }
 }

src/transport/omikron_manager.rs

@@ -7,9 +7,6 @@ use mtp::codec::{CommunicationType, CommunicationValue, DataType, DataValue};
 use once_cell::sync::Lazy;
 use rand::prelude::IteratorRandom;
 use std::sync::Arc;
-use std::time::Duration;
-use tokio::task::JoinHandle;
-use tokio::time::interval;
 
 pub static OMIKRON_CONNECTIONS: Lazy<DashMap<i64, Arc<OmikronConnection>>> =
     Lazy::new(DashMap::new);
@@ -26,7 +23,6 @@ pub async fn add_omikron(conn: Arc<OmikronConnection>) {
     if let Some(old) = OMIKRON_CONNECTIONS.insert(id, conn.clone()) {
         old.close().await;
     }
-    let _ = flush_iota_snapshot_outbox().await;
 }
 
 pub async fn remove_omikron(omikron_id: i64, connection: &Arc<OmikronConnection>) -> bool {
@@ -67,18 +63,12 @@ pub async fn get_all_connections()
                 .await
                 .map_err(|_| ())?;
             for user in users {
-                if let Some(iota_id) = user.iota_id {
-                    for omikron_id in state
-                        .presence
-                        .iota_connections(iota_id.0)
-                        .unwrap_or_default()
+                for route in state.presence.routes_for_user(user.id.0) {
+                    if let Some(iotas) = result.get_mut(&route.omikron_id)
+                        && let Some(iota_id) = user.iota_id
+                        && let Some(users) = iotas.get_mut(&iota_id.0)
                     {
-                        if let Some(users) = result
-                            .get_mut(&omikron_id)
-                            .and_then(|iotas| iotas.get_mut(&iota_id.0))
-                        {
-                            users.push(user.id.0);
-                        }
+                        users.push(user.id.0);
                     }
                 }
             }
@@ -125,18 +115,19 @@ pub async fn send_to_user(user_id: i64, cv: &CommunicationValue) {
     }
 }
 
-/*
- * Publish each Iota membership snapshot to every live relay route. Each
- * Omikron keeps a local authorization index, so sending only a primary route
- * leaves the remaining relays stale after registration or migration.
- */
-pub async fn publish_iota_user_snapshot(iota_id: i64) -> OmikronResult<()> {
-    let state = get_state().ok_or(crate::error::OmegaError::NotConnected)?;
-    let omikron_ids = state
-        .presence
-        .iota_connections(iota_id)
-        .ok_or(crate::error::OmegaError::NotConnected)?;
-    let users = user_repo::get_users_by_iota_id(crate::models::IotaId::from(iota_id)).await?;
+/// Publish the authoritative membership list after an attach, migration, or
+/// release. Omikron replaces its full local index from this snapshot.
+pub async fn publish_iota_user_snapshot(iota_id: i64) {
+    let Some(omikron_id) = get_iota_primary_omikron_connection(iota_id) else {
+        return;
+    };
+    let Some(connection) = get_connected_omikron(omikron_id) else {
+        return;
+    };
+    let Ok(users) = user_repo::get_users_by_iota_id(crate::models::IotaId::from(iota_id)).await
+    else {
+        return;
+    };
     let user_ids = users
         .into_iter()
         .map(|user| DataValue::SignedNumber(user.id.0.into()))
@@ -144,54 +135,7 @@ pub async fn publish_iota_user_snapshot(iota_id: i64) -> OmikronResult<()> {
     let snapshot = CommunicationValue::new(CommunicationType::IotaUserData)
         .add_typed_default(DataType::IotaId, DataValue::SignedNumber(iota_id.into()))
         .add_typed_default(DataType::UserIds, DataValue::Array(user_ids));
-    for omikron_id in omikron_ids {
-        let connection =
-            get_connected_omikron(omikron_id).ok_or(crate::error::OmegaError::NotConnected)?;
-        connection.send(&snapshot).await?;
-    }
-    Ok(())
-}
-
-pub async fn flush_iota_snapshot_outbox() -> OmikronResult<()> {
-    for iota_id in user_repo::pending_iota_snapshots().await? {
-        match publish_iota_user_snapshot(iota_id.0).await {
-            Ok(()) => {
-                if let Err(error) = user_repo::complete_iota_snapshot(iota_id).await {
-                    crate::log_in!(
-                        crate::util::logger::PrintType::General,
-                        "Could not complete Iota snapshot outbox entry for {}: {}",
-                        iota_id.0,
-                        error
-                    );
-                }
-            }
-            Err(error) => {
-                crate::log_in!(
-                    crate::util::logger::PrintType::General,
-                    "Could not publish Iota snapshot for {}: {}",
-                    iota_id.0,
-                    error
-                );
-            }
-        }
-    }
-    Ok(())
-}
-
-pub fn spawn_iota_snapshot_outbox_worker() -> JoinHandle<()> {
-    tokio::spawn(async {
-        let mut retry = interval(Duration::from_secs(30));
-        loop {
-            retry.tick().await;
-            if let Err(error) = flush_iota_snapshot_outbox().await {
-                crate::log_in!(
-                    crate::util::logger::PrintType::General,
-                    "Could not load Iota snapshot outbox: {}",
-                    error
-                );
-            }
-        }
-    })
+    let _ = connection.send(&snapshot).await;
 }
 
 pub async fn deliver_pending_erasures(iota_id: i64) {

src/transport/relay_router.rs

@@ -82,7 +82,7 @@ pub fn error_response_type(error: &RelayRouteError) -> CommunicationType {
 pub async fn route_from_omikron(
     source_omikron_id: i64,
     frame: CommunicationValue,
-) -> Result<CommunicationValue, RelayRouteError> {
+) -> Result<(), RelayRouteError> {
     let frame = ensure_relay_frame_id(frame);
     if !frame.is_type(CommunicationType::Relay) {
         return Err(RelayRouteError::Relay(RelayError::NotRelay));
@@ -154,7 +154,7 @@ pub async fn route_from_omikron(
             RelayRouteError::Send(error.to_string())
         })?;
     if response.is_type(CommunicationType::Success) {
-        Ok(response)
+        Ok(())
     } else {
         Err(RelayRouteError::Send(format!(
             "destination Omikron rejected the Relay with {}",