From 9e9cc74c2c3d65fcad93cdc4c420d973332b022e Mon Sep 17 00:00:00 2001 From: Rasensprenger Date: Tue, 1 Sep 2026 19:00:35 +0300 Subject: [PATCH 1/2] Lock file maintenance --- Cargo.lock | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/Cargo.lock b/Cargo.lock index f2ea815..7c9d80d 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -2306,9 +2306,9 @@ checksum = "0c790de23124f9ab44544d7ac05d60440adc586479ce501c1d6d7da3cd8c9cf5" [[package]] name = "smallvec" -version = "1.15.2" +version = "1.16.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "8ed6a63f02c8539c91a8685a86f4099661ba3da017932f6ebbea6de3f0fa7c90" +checksum = "b9be42f50aa861c555654aa3a37f52f4b1074bacf4e48fe0ef7fa584e80f1f0f" dependencies = [ "serde", ] From 6d5bac3a095bf3c5a1335c9392bba610ecfacf7f Mon Sep 17 00:00:00 2001 From: Alex Emmet <111742636+Alex-Emmet@users.noreply.github.com> Date: Wed, 2 Sep 2026 22:42:23 +0200 Subject: [PATCH 2/2] [Add] Proper User managment --- migrations/009_user_invitations.sql | 29 +++++++++ mtp-type-maps | 2 +- src/db/user_repo.rs | 99 +++++++++++++++++++++++++++++ src/transport/connection.rs | 1 + src/transport/handlers/account.rs | 41 ++++++++++++ src/transport/handlers/presence.rs | 1 + src/transport/omikron_connection.rs | 4 ++ src/transport/omikron_manager.rs | 30 +++++++++ 8 files changed, 206 insertions(+), 1 deletion(-) create mode 100644 migrations/009_user_invitations.sql diff --git a/migrations/009_user_invitations.sql b/migrations/009_user_invitations.sql new file mode 100644 index 0000000..622fc84 --- /dev/null +++ b/migrations/009_user_invitations.sql @@ -0,0 +1,29 @@ +CREATE TABLE user_invitations ( + invitation_id CHAR(36) NOT NULL, + token_hash BINARY(32) NOT NULL, + iota_id BIGINT NOT NULL, + created_at TIMESTAMP NOT NULL DEFAULT CURRENT_TIMESTAMP, + expires_at TIMESTAMP NULL, + state ENUM('pending', 'redeemed', 'revoked', 'expired') NOT NULL DEFAULT 'pending', + redeemed_user_id BIGINT NULL, + redeemed_at TIMESTAMP NULL, + revoked_at TIMESTAMP NULL, + PRIMARY KEY (invitation_id), + KEY idx_user_invitations_target_state (iota_id, state), + CONSTRAINT fk_user_invitations_iota + FOREIGN KEY (iota_id) REFERENCES iotas(id) ON DELETE CASCADE +); + +CREATE TABLE pending_iota_user_provisioning ( + user_id BIGINT NOT NULL, + iota_id BIGINT NOT NULL, + invitation_id CHAR(36) NOT NULL, + username VARBINARY(255) NOT NULL, + public_key BLOB NOT NULL, + created_at TIMESTAMP NOT NULL DEFAULT CURRENT_TIMESTAMP, + PRIMARY KEY (user_id, iota_id), + CONSTRAINT fk_pending_iota_user_provisioning_iota + FOREIGN KEY (iota_id) REFERENCES iotas(id) ON DELETE CASCADE, + CONSTRAINT fk_pending_iota_user_provisioning_invitation + FOREIGN KEY (invitation_id) REFERENCES user_invitations(invitation_id) ON DELETE CASCADE +); diff --git a/mtp-type-maps b/mtp-type-maps index f4e45aa..909b397 160000 --- a/mtp-type-maps +++ b/mtp-type-maps @@ -1 +1 @@ -Subproject commit f4e45aa3a3ad0e3c3a257f66857b904a1af7901c +Subproject commit 909b3977cb6a233e74a925eb467412fb384844bc diff --git a/src/db/user_repo.rs b/src/db/user_repo.rs index 950be67..31771a6 100644 --- a/src/db/user_repo.rs +++ b/src/db/user_repo.rs @@ -11,6 +11,105 @@ use std::collections::HashMap; pub const MAX_PROTOCOL_ID: i64 = (1_i64 << 48) - 1; const ID_ALLOCATION_ATTEMPTS: usize = 16; +#[derive(Clone, Debug)] +pub struct PendingIotaUserProvisioning { + pub user_id: UserId, + pub iota_id: IotaId, + pub invitation_id: String, + pub username: String, + pub public_key: PublicKeyBundle, +} + +pub async fn create_invitation( + invitation_id: &str, + token_hash: &[u8], + iota_id: IotaId, +) -> Result<()> { + if uuid::Uuid::parse_str(invitation_id).is_err() + || token_hash.len() != 32 + || !valid_protocol_id(iota_id.0) + { + return Err(OmegaError::Validation("invalid invitation".into())); + } + sqlx::query( + "INSERT INTO user_invitations (invitation_id, token_hash, iota_id) VALUES (?, ?, ?)", + ) + .bind(invitation_id) + .bind(token_hash) + .bind(iota_id.0) + .execute(&pool().await?) + .await?; + Ok(()) +} + +pub async fn revoke_invitation(invitation_id: &str, iota_id: IotaId) -> Result { + let result = sqlx::query( + "UPDATE user_invitations SET state = 'revoked', revoked_at = UTC_TIMESTAMP() WHERE invitation_id = ? AND iota_id = ? AND state = 'pending'", + ) + .bind(invitation_id) + .bind(iota_id.0) + .execute(&pool().await?) + .await?; + Ok(result.rows_affected() == 1) +} + +pub async fn enqueue_iota_user_provisioning( + provisioning: &PendingIotaUserProvisioning, +) -> Result<()> { + if !valid_protocol_id(provisioning.user_id.0) + || !valid_protocol_id(provisioning.iota_id.0) + || !valid_username(&provisioning.username) + { + return Err(OmegaError::Validation("invalid user provisioning".into())); + } + sqlx::query( + "INSERT INTO pending_iota_user_provisioning (user_id, iota_id, invitation_id, username, public_key) VALUES (?, ?, ?, ?, ?)", + ) + .bind(provisioning.user_id.0) + .bind(provisioning.iota_id.0) + .bind(&provisioning.invitation_id) + .bind(&provisioning.username) + .bind(provisioning.public_key.try_as_bytes()?) + .execute(&pool().await?) + .await?; + Ok(()) +} + +pub async fn pending_iota_user_provisioning( + iota_id: IotaId, +) -> Result> { + let rows = sqlx::query( + "SELECT user_id, iota_id, invitation_id, username, public_key FROM pending_iota_user_provisioning WHERE iota_id = ? ORDER BY created_at", + ) + .bind(iota_id.0) + .fetch_all(&pool().await?) + .await?; + rows.into_iter() + .map(|row| { + let public_key = PublicKeyBundle::from_bytes(&row.get::, _>("public_key")) + .map_err(|error| OmegaError::Validation(error.to_string()))?; + Ok(PendingIotaUserProvisioning { + user_id: UserId::from(row.get::("user_id")), + iota_id: IotaId::from(row.get::("iota_id")), + invitation_id: row.get("invitation_id"), + username: String::from_utf8(row.get::, _>("username")) + .map_err(|error| OmegaError::Validation(error.to_string()))?, + public_key, + }) + }) + .collect() +} + +pub async fn acknowledge_iota_user_provisioning(user_id: UserId, iota_id: IotaId) -> Result { + let result = + sqlx::query("DELETE FROM pending_iota_user_provisioning WHERE user_id = ? AND iota_id = ?") + .bind(user_id.0) + .bind(iota_id.0) + .execute(&pool().await?) + .await?; + Ok(result.rows_affected() == 1) +} + pub fn generate_protocol_id() -> UserId { loop { let value = rand::random::() & ((1_u64 << 48) - 1); diff --git a/src/transport/connection.rs b/src/transport/connection.rs index 92644dc..3728ebb 100644 --- a/src/transport/connection.rs +++ b/src/transport/connection.rs @@ -79,6 +79,7 @@ pub(crate) fn validate_dispatch_fields(value: &CommunicationValue) -> OmikronRes | mtp::codec::CommunicationType::DeleteUserCredentialBegin | mtp::codec::CommunicationType::DeleteUserCredentialComplete | mtp::codec::CommunicationType::EraseHostedUserDataAck + | mtp::codec::CommunicationType::AcknowledgeIotaUserProvision | mtp::codec::CommunicationType::ReleaseUserFromIota | mtp::codec::CommunicationType::DeleteIota | mtp::codec::CommunicationType::GetNotifications diff --git a/src/transport/handlers/account.rs b/src/transport/handlers/account.rs index 9de94a5..b6d741c 100644 --- a/src/transport/handlers/account.rs +++ b/src/transport/handlers/account.rs @@ -436,3 +436,44 @@ pub async fn erase_hosted_user_data_ack( } } } + +pub async fn acknowledge_iota_user_provisioning( + connection: Arc, + value: CommunicationValue, +) -> OmikronResult<()> { + value.require_id()?; + let Some(user_id) = value + .get_data(DataType::UserId) + .as_signed_number() + .and_then(|id| i64::try_from(id).ok()) + .filter(|id| *id > 0) + else { + return connection + .send_error_response(value.require_id()?, CommunicationType::ErrorInvalidUserId) + .await; + }; + let iota_id = IotaId::from(value.require_sender_i64()?); + match user_repo::acknowledge_iota_user_provisioning(UserId::from(user_id), iota_id).await { + Ok(true) => { + connection + .send( + &CommunicationValue::new(CommunicationType::Success) + .with_id(value.require_id()?), + ) + .await + } + Ok(false) => { + connection + .send_error_response( + value.require_id()?, + CommunicationType::ErrorNotAuthenticated, + ) + .await + } + Err(_) => { + connection + .send_error_response(value.require_id()?, CommunicationType::ErrorInternal) + .await + } + } +} diff --git a/src/transport/handlers/presence.rs b/src/transport/handlers/presence.rs index 048dadf..30903bf 100644 --- a/src/transport/handlers/presence.rs +++ b/src/transport/handlers/presence.rs @@ -337,6 +337,7 @@ pub async fn iota_connected( .add_typed_default(DataType::UserIds, DataValue::Array(user_ids)); connection.clone().send(&response).await?; crate::transport::omikron_manager::deliver_pending_erasures(iota_id).await; + crate::transport::omikron_manager::deliver_pending_user_provisioning(iota_id).await; publish_changed_states(&state, &before, &users).await; connection .send(&CommunicationValue::new(CommunicationType::Success).with_id(value.require_id()?)) diff --git a/src/transport/omikron_connection.rs b/src/transport/omikron_connection.rs index b4be514..d1328d8 100644 --- a/src/transport/omikron_connection.rs +++ b/src/transport/omikron_connection.rs @@ -476,6 +476,10 @@ impl OmikronConnection { Some(CommunicationType::EraseHostedUserDataAck) => { crate::transport::handlers::account::erase_hosted_user_data_ack(self, value).await } + Some(CommunicationType::AcknowledgeIotaUserProvision) => { + crate::transport::handlers::account::acknowledge_iota_user_provisioning(self, value) + .await + } Some(CommunicationType::ReleaseUserFromIota) => { crate::transport::handlers::account::release_from_iota(self, value).await } diff --git a/src/transport/omikron_manager.rs b/src/transport/omikron_manager.rs index 9c337f5..8af8026 100644 --- a/src/transport/omikron_manager.rs +++ b/src/transport/omikron_manager.rs @@ -213,3 +213,33 @@ pub async fn deliver_pending_erasures(iota_id: i64) { let _ = connection.clone().send(&request).await; } } + +/* Provisioning retries until the destination Iota has persisted public + * account metadata and acknowledges it. Credentials never enter this path. */ +pub async fn deliver_pending_user_provisioning(iota_id: i64) { + let Ok(users) = + user_repo::pending_iota_user_provisioning(crate::models::IotaId::from(iota_id)).await + else { + return; + }; + let Some(omikron_id) = get_iota_primary_omikron_connection(iota_id) else { + return; + }; + let Some(connection) = get_connected_omikron(omikron_id) else { + return; + }; + for user in users { + let Ok(public_key) = user.public_key.try_to_base64() else { + continue; + }; + let request = CommunicationValue::new(CommunicationType::ProvisionIotaUser) + .add_typed_default( + DataType::UserId, + DataValue::SignedNumber(user.user_id.0.into()), + ) + .add_typed_default(DataType::Username, DataValue::Str(user.username)) + .add_typed_default(DataType::PublicKey, DataValue::Str(public_key)) + .add_typed_default(DataType::InvitationId, DataValue::Str(user.invitation_id)); + let _ = connection.clone().send(&request).await; + } +}