tensamin/omega
Watch
6
0
Fork
You've already forked omega
 0
Code Issues Pull requests 1 Projects Releases Packages Wiki Activity Actions

[WIP] 0.3.0 mtp update

Browse source
This commit is contained in:
Alex Emmet 2026-08-18 22:37:14 +02:00
commit 7709fe599a
No known key found for this signature in database
19 changed files with 1009 additions and 278 deletions
Download patch file Download diff file Expand all files Collapse all files

39
src/db/user_repo.rs
View file

@ -380,27 +380,44 @@ pub async fn delete_user_with_pending_erasure(id: UserId) -> Result<Option<IotaI
.ok_or(OmegaError::NotFound)?;
let iota_id: Option<i64> = row.get("iota_id");
if let Some(iota_id) = iota_id {
sqlx::query("INSERT IGNORE INTO pending_iota_user_erasure (user_id, iota_id) VALUES (?, ?)")
.bind(id.0)
.bind(iota_id)
.execute(&mut *tx)
.await?;
sqlx::query(
"INSERT IGNORE INTO pending_iota_user_erasure (user_id, iota_id) VALUES (?, ?)",
)
.bind(id.0)
.bind(iota_id)
.execute(&mut *tx)
.await?;
}
sqlx::query("DELETE FROM registration_leases WHERE user_id = ?").bind(id.0).execute(&mut *tx).await?;
sqlx::query("DELETE FROM users WHERE id = ?").bind(id.0).execute(&mut *tx).await?;
sqlx::query("DELETE FROM registration_leases WHERE user_id = ?")
.bind(id.0)
.execute(&mut *tx)
.await?;
sqlx::query("DELETE FROM users WHERE id = ?")
.bind(id.0)
.execute(&mut *tx)
.await?;
tx.commit().await?;
Ok(iota_id.map(IotaId::from))
}
pub async fn pending_erasures_for_iota(iota_id: IotaId) -> Result<Vec<UserId>> {
let rows = sqlx::query("SELECT user_id FROM pending_iota_user_erasure WHERE iota_id = ?")
.bind(iota_id.0).fetch_all(&pool().await?).await?;
Ok(rows.into_iter().map(|row| UserId::from(row.get::<i64, _>("user_id"))).collect())
.bind(iota_id.0)
.fetch_all(&pool().await?)
.await?;
Ok(rows
.into_iter()
.map(|row| UserId::from(row.get::<i64, _>("user_id")))
.collect())
}
pub async fn acknowledge_pending_erasure(user_id: UserId, iota_id: IotaId) -> Result<bool> {
let result = sqlx::query("DELETE FROM pending_iota_user_erasure WHERE user_id = ? AND iota_id = ?")
.bind(user_id.0).bind(iota_id.0).execute(&pool().await?).await?;
let result =
sqlx::query("DELETE FROM pending_iota_user_erasure WHERE user_id = ? AND iota_id = ?")
.bind(user_id.0)
.bind(iota_id.0)
.execute(&pool().await?)
.await?;
Ok(result.rows_affected() == 1)
}

12
src/server/api.rs
View file

@ -85,11 +85,13 @@ async fn route(path_parts: &[&str]) -> Result<(StatusCode, String)> {
let user = get_by_user_id(UserId::from(id)).await?;
match user.iota_id {
Some(iota_id) => get_iota_primary_omikron_connection(iota_id.0),
None => get_random_omikron()
.await
.map_err(|_| OmegaError::NotFound)?
.get_omikron_id()
.await,
None => {
get_random_omikron()
.await
.map_err(|_| OmegaError::NotFound)?
.get_omikron_id()
.await
}
}
.ok_or(OmegaError::NotFound)?
};

32
src/sql/user_online_tracker.rs
View file

@ -1044,20 +1044,24 @@ mod tests {
assert_eq!(tracker.subscribers(22).len(), 1);
tracker.remove_session(7, 3, 42);
assert!(tracker.check_index_consistency().is_ok());
assert!(tracker
.routes
.read()
.unwrap()
.indices
.sessions_by_user
.is_empty());
assert!(tracker
.routes
.read()
.unwrap()
.indices
.targets_by_subscriber
.is_empty());
assert!(
tracker
.routes
.read()
.unwrap()
.indices
.sessions_by_user
.is_empty()
);
assert!(
tracker
.routes
.read()
.unwrap()
.indices
.targets_by_subscriber
.is_empty()
);
}
#[test]

40
src/state.rs
View file

@ -1,10 +1,13 @@
use crate::sql::user_online_tracker::PresenceTracker;
use std::sync::Arc;
use dashmap::DashMap;
use std::sync::Arc;
use std::time::{Duration, Instant};
#[derive(Clone, Copy, Debug, PartialEq, Eq, Hash)]
pub enum AccountChallengeOperation { Attach, Delete }
pub enum AccountChallengeOperation {
Attach,
Delete,
}
#[derive(Clone, Debug)]
pub struct AccountChallenge {
@ -30,15 +33,38 @@ impl Default for OmegaState {
}
impl OmegaState {
pub fn issue_challenge(&self, operation: AccountChallengeOperation, user_id: i64, requester_iota_id: i64) -> u64 {
pub fn issue_challenge(
&self,
operation: AccountChallengeOperation,
user_id: i64,
requester_iota_id: i64,
) -> u64 {
let nonce = rand::random::<u64>();
self.challenges.insert((operation, user_id, requester_iota_id), AccountChallenge { operation, user_id, requester_iota_id, nonce, created_at: Instant::now() });
self.challenges.insert(
(operation, user_id, requester_iota_id),
AccountChallenge {
operation,
user_id,
requester_iota_id,
nonce,
created_at: Instant::now(),
},
);
nonce
}
pub fn consume_challenge(&self, operation: AccountChallengeOperation, user_id: i64, requester_iota_id: i64, nonce: u64) -> bool {
self.challenges.remove(&(operation, user_id, requester_iota_id)).is_some_and(|(_, value)|
value.nonce == nonce && value.created_at.elapsed() <= Duration::from_secs(120))
pub fn consume_challenge(
&self,
operation: AccountChallengeOperation,
user_id: i64,
requester_iota_id: i64,
nonce: u64,
) -> bool {
self.challenges
.remove(&(operation, user_id, requester_iota_id))
.is_some_and(|(_, value)| {
value.nonce == nonce && value.created_at.elapsed() <= Duration::from_secs(120)
})
}
}

49
src/transport/connection.rs
View file

@ -1 +1,48 @@
pub(crate) use super::omikron_connection::OmikronConnection;
pub(crate) use super::omikron_connection::{OmikronConnection, OmikronResult};
use mtp::codec::{CommunicationValue, DataValue};
pub(crate) trait MtpValueCompat {
fn get_id(&self) -> u32;
fn get_sender(&self) -> u64;
fn get_receiver(&self) -> u64;
}
impl MtpValueCompat for CommunicationValue {
fn get_id(&self) -> u32 {
self.id().unwrap_or_default()
}
fn get_sender(&self) -> u64 {
self.sender().unwrap_or_default()
}
fn get_receiver(&self) -> u64 {
self.receiver().unwrap_or_default()
}
}
pub(crate) trait OptionalDataValueCompat {
fn as_number(&self) -> Option<i128>;
fn as_signed_number(&self) -> Option<i128>;
fn as_str(&self) -> Option<&str>;
fn as_bytes(&self) -> Option<Vec<u8>>;
}
impl OptionalDataValueCompat for Option<&DataValue> {
fn as_number(&self) -> Option<i128> {
self.and_then(|value| value.as_number())
}
fn as_signed_number(&self) -> Option<i128> {
self.and_then(|value| value.as_signed_number())
}
fn as_str(&self) -> Option<&str> {
self.and_then(|value| value.as_str())
}
fn as_bytes(&self) -> Option<Vec<u8>> {
self.and_then(|value| value.as_bytes())
}
}

316
src/transport/handlers/account.rs
View file

@ -1,10 +1,15 @@
use super::super::omikron_connection::{OmikronConnection, OmikronResult};
use super::super::connection::{
MtpValueCompat, OmikronConnection, OmikronResult, OptionalDataValueCompat,
};
use crate::{
db::{iota_repo, user_repo},
models::{IotaId, UserId},
state::AccountChallengeOperation,
};
use mtp::{codec::{CommunicationType, CommunicationValue, DataType, DataValue}, crypto::{verify_ed25519, verify_ml_dsa}};
use mtp::{
codec::{CommunicationType, CommunicationValue, DataType, DataValue},
crypto::{verify_ed25519, verify_ml_dsa},
};
use std::sync::Arc;
async fn delete(
@ -67,14 +72,22 @@ pub async fn release_from_iota(
let previous_iota = user.iota_id;
match user_repo::change_iota_id(user.id, None).await {
Ok(()) => {
if let Some(iota) = previous_iota { crate::transport::omikron_manager::publish_iota_user_snapshot(iota.0).await; }
connection.send(&CommunicationValue::new(CommunicationType::Success).with_id(value.get_id())).await
},
Err(error) => connection
.send(&CommunicationValue::new(CommunicationType::ErrorInternal)
.with_id(value.get_id())
.add_typed_default(DataType::ErrorType, DataValue::Str(error.to_string())))
.await,
if let Some(iota) = previous_iota {
crate::transport::omikron_manager::publish_iota_user_snapshot(iota.0).await;
}
connection
.send(&CommunicationValue::new(CommunicationType::Success).with_id(value.get_id()))
.await
}
Err(error) => {
connection
.send(
&CommunicationValue::new(CommunicationType::ErrorInternal)
.with_id(value.get_id())
.add_typed_default(DataType::ErrorType, DataValue::Str(error.to_string())),
)
.await
}
}
}
@ -87,43 +100,121 @@ fn lifecycle_payload(domain: &[u8], user_id: i64, iota_id: i64, nonce: u64) -> V
payload
}
pub async fn attach_begin(connection: Arc<OmikronConnection>, value: CommunicationValue) -> OmikronResult<()> {
let Some(user_id) = value.get_data(DataType::UserId).as_signed_number().and_then(|v| i64::try_from(v).ok()).filter(|v| *v > 0) else {
return connection.send_error_response(value.get_id(), CommunicationType::ErrorInvalidUserId).await;
pub async fn attach_begin(
connection: Arc<OmikronConnection>,
value: CommunicationValue,
) -> OmikronResult<()> {
let Some(user_id) = value
.get_data(DataType::UserId)
.as_signed_number()
.and_then(|v| i64::try_from(v).ok())
.filter(|v| *v > 0)
else {
return connection
.send_error_response(value.get_id(), CommunicationType::ErrorInvalidUserId)
.await;
};
if user_repo::get_by_user_id(UserId::from(user_id)).await.is_err() {
return connection.send_error_response(value.get_id(), CommunicationType::ErrorNotFound).await;
if user_repo::get_by_user_id(UserId::from(user_id))
.await
.is_err()
{
return connection
.send_error_response(value.get_id(), CommunicationType::ErrorNotFound)
.await;
}
let requester = value.get_sender() as i64;
let nonce = connection.state().issue_challenge(AccountChallengeOperation::Attach, user_id, requester);
connection.send(&CommunicationValue::new(CommunicationType::AttachUserChallenge).with_id(value.get_id())
.add_typed_default(DataType::UserId, DataValue::SignedNumber(user_id.into()))
.add_typed_default(DataType::ServerNonce, DataValue::SignedNumber(nonce.into()))).await
let nonce =
connection
.state()
.issue_challenge(AccountChallengeOperation::Attach, user_id, requester);
connection
.send(
&CommunicationValue::new(CommunicationType::AttachUserChallenge)
.with_id(value.get_id())
.add_typed_default(DataType::UserId, DataValue::SignedNumber(user_id.into()))
.add_typed_default(DataType::ServerNonce, DataValue::SignedNumber(nonce.into())),
)
.await
}
pub async fn attach_complete(connection: Arc<OmikronConnection>, value: CommunicationValue) -> OmikronResult<()> {
let Some(user_id) = value.get_data(DataType::UserId).as_signed_number().and_then(|v| i64::try_from(v).ok()).filter(|v| *v > 0) else { return connection.send_error_response(value.get_id(), CommunicationType::ErrorInvalidUserId).await; };
pub async fn attach_complete(
connection: Arc<OmikronConnection>,
value: CommunicationValue,
) -> OmikronResult<()> {
let Some(user_id) = value
.get_data(DataType::UserId)
.as_signed_number()
.and_then(|v| i64::try_from(v).ok())
.filter(|v| *v > 0)
else {
return connection
.send_error_response(value.get_id(), CommunicationType::ErrorInvalidUserId)
.await;
};
let requester = value.get_sender() as i64;
let Some(nonce) = value.get_data(DataType::ServerNonce).as_signed_number().and_then(|v| u64::try_from(v).ok()) else { return connection.send_error_response(value.get_id(), CommunicationType::ErrorInvalidChallenge).await; };
let Some(nonce) = value
.get_data(DataType::ServerNonce)
.as_signed_number()
.and_then(|v| u64::try_from(v).ok())
else {
return connection
.send_error_response(value.get_id(), CommunicationType::ErrorInvalidChallenge)
.await;
};
let signature = value.get_data(DataType::Signature).as_bytes();
let pq_signature = value.get_data(DataType::PqSignature).as_bytes();
let (Some(signature), Some(pq_signature)) = (signature, pq_signature) else { return connection.send_error_response(value.get_id(), CommunicationType::ErrorInvalidChallenge).await; };
if !connection.state().consume_challenge(AccountChallengeOperation::Attach, user_id, requester, nonce) { return connection.send_error_response(value.get_id(), CommunicationType::ErrorInvalidChallenge).await; }
let Ok(user) = user_repo::get_by_user_id(UserId::from(user_id)).await else { return connection.send_error_response(value.get_id(), CommunicationType::ErrorNotFound).await; };
let (Some(signature), Some(pq_signature)) = (signature, pq_signature) else {
return connection
.send_error_response(value.get_id(), CommunicationType::ErrorInvalidChallenge)
.await;
};
if !connection.state().consume_challenge(
AccountChallengeOperation::Attach,
user_id,
requester,
nonce,
) {
return connection
.send_error_response(value.get_id(), CommunicationType::ErrorInvalidChallenge)
.await;
}
let Ok(user) = user_repo::get_by_user_id(UserId::from(user_id)).await else {
return connection
.send_error_response(value.get_id(), CommunicationType::ErrorNotFound)
.await;
};
let payload = lifecycle_payload(b"tensamin:user-attach:v1\0", user_id, requester, nonce);
if verify_ed25519(&user.public_key.sig_cl_public_key, &payload, &signature).is_err() || verify_ml_dsa(&user.public_key.sig_pq_public_key, &payload, &pq_signature).is_err() { return connection.send_error_response(value.get_id(), CommunicationType::ErrorNotAuthenticated).await; }
if verify_ed25519(&user.public_key.sig_cl_public_key, &payload, &signature).is_err()
|| verify_ml_dsa(&user.public_key.sig_pq_public_key, &payload, &pq_signature).is_err()
{
return connection
.send_error_response(value.get_id(), CommunicationType::ErrorNotAuthenticated)
.await;
}
let previous_iota = user.iota_id;
match user_repo::change_iota_id(user.id, Some(IotaId::from(requester))).await {
Ok(()) => {
if let Some(iota) = previous_iota.filter(|id| id.0 != requester) { crate::transport::omikron_manager::publish_iota_user_snapshot(iota.0).await; }
if let Some(iota) = previous_iota.filter(|id| id.0 != requester) {
crate::transport::omikron_manager::publish_iota_user_snapshot(iota.0).await;
}
crate::transport::omikron_manager::publish_iota_user_snapshot(requester).await;
connection.send(&CommunicationValue::new(CommunicationType::Success).with_id(value.get_id())).await
},
Err(_) => connection.send_error_response(value.get_id(), CommunicationType::ErrorInternal).await,
connection
.send(&CommunicationValue::new(CommunicationType::Success).with_id(value.get_id()))
.await
}
Err(_) => {
connection
.send_error_response(value.get_id(), CommunicationType::ErrorInternal)
.await
}
}
}
async fn complete_delete(connection: Arc<OmikronConnection>, value: CommunicationValue, user_id: UserId) -> OmikronResult<()> {
async fn complete_delete(
connection: Arc<OmikronConnection>,
value: CommunicationValue,
user_id: UserId,
) -> OmikronResult<()> {
match user_repo::delete_user_with_pending_erasure(user_id).await {
Ok(iota_id) => {
let cleanup_pending = iota_id.is_some();
@ -131,45 +222,160 @@ async fn complete_delete(connection: Arc<OmikronConnection>, value: Communicatio
crate::transport::omikron_manager::publish_iota_user_snapshot(iota_id.0).await;
crate::transport::omikron_manager::deliver_pending_erasures(iota_id.0).await;
}
connection.send(&CommunicationValue::new(CommunicationType::Success).with_id(value.get_id())
.add_typed_default(DataType::CleanupPending, DataValue::Bool(cleanup_pending))).await
connection
.send(
&CommunicationValue::new(CommunicationType::Success)
.with_id(value.get_id())
.add_typed_default(
DataType::CleanupPending,
DataValue::Bool(cleanup_pending),
),
)
.await
}
Err(crate::error::OmegaError::NotFound) => {
connection
.send_error_response(value.get_id(), CommunicationType::ErrorNotFound)
.await
}
Err(error) => {
connection
.send(
&CommunicationValue::new(CommunicationType::ErrorInternal)
.with_id(value.get_id())
.add_typed_default(DataType::ErrorType, DataValue::Str(error.to_string())),
)
.await
}
Err(crate::error::OmegaError::NotFound) => connection.send_error_response(value.get_id(), CommunicationType::ErrorNotFound).await,
Err(error) => connection.send(&CommunicationValue::new(CommunicationType::ErrorInternal).with_id(value.get_id()).add_typed_default(DataType::ErrorType, DataValue::Str(error.to_string()))).await,
}
}
pub async fn delete_credential_begin(connection: Arc<OmikronConnection>, value: CommunicationValue) -> OmikronResult<()> {
let Some(user_id) = value.get_data(DataType::UserId).as_signed_number().and_then(|v| i64::try_from(v).ok()).filter(|v| *v > 0) else {
return connection.send_error_response(value.get_id(), CommunicationType::ErrorInvalidUserId).await;
pub async fn delete_credential_begin(
connection: Arc<OmikronConnection>,
value: CommunicationValue,
) -> OmikronResult<()> {
let Some(user_id) = value
.get_data(DataType::UserId)
.as_signed_number()
.and_then(|v| i64::try_from(v).ok())
.filter(|v| *v > 0)
else {
return connection
.send_error_response(value.get_id(), CommunicationType::ErrorInvalidUserId)
.await;
};
if user_repo::get_by_user_id(UserId::from(user_id)).await.is_err() { return connection.send_error_response(value.get_id(), CommunicationType::ErrorNotFound).await; }
if user_repo::get_by_user_id(UserId::from(user_id))
.await
.is_err()
{
return connection
.send_error_response(value.get_id(), CommunicationType::ErrorNotFound)
.await;
}
let requester = value.get_sender() as i64;
let nonce = connection.state().issue_challenge(AccountChallengeOperation::Delete, user_id, requester);
connection.send(&CommunicationValue::new(CommunicationType::DeleteUserCredentialChallenge).with_id(value.get_id())
.add_typed_default(DataType::UserId, DataValue::SignedNumber(user_id.into()))
.add_typed_default(DataType::ServerNonce, DataValue::SignedNumber(nonce.into()))).await
let nonce =
connection
.state()
.issue_challenge(AccountChallengeOperation::Delete, user_id, requester);
connection
.send(
&CommunicationValue::new(CommunicationType::DeleteUserCredentialChallenge)
.with_id(value.get_id())
.add_typed_default(DataType::UserId, DataValue::SignedNumber(user_id.into()))
.add_typed_default(DataType::ServerNonce, DataValue::SignedNumber(nonce.into())),
)
.await
}
pub async fn delete_credential_complete(connection: Arc<OmikronConnection>, value: CommunicationValue) -> OmikronResult<()> {
let Some(user_id) = value.get_data(DataType::UserId).as_signed_number().and_then(|v| i64::try_from(v).ok()).filter(|v| *v > 0) else { return connection.send_error_response(value.get_id(), CommunicationType::ErrorInvalidUserId).await; };
pub async fn delete_credential_complete(
connection: Arc<OmikronConnection>,
value: CommunicationValue,
) -> OmikronResult<()> {
let Some(user_id) = value
.get_data(DataType::UserId)
.as_signed_number()
.and_then(|v| i64::try_from(v).ok())
.filter(|v| *v > 0)
else {
return connection
.send_error_response(value.get_id(), CommunicationType::ErrorInvalidUserId)
.await;
};
let requester = value.get_sender() as i64;
let Some(nonce) = value.get_data(DataType::ServerNonce).as_signed_number().and_then(|v| u64::try_from(v).ok()) else { return connection.send_error_response(value.get_id(), CommunicationType::ErrorInvalidChallenge).await; };
let (Some(signature), Some(pq_signature)) = (value.get_data(DataType::Signature).as_bytes(), value.get_data(DataType::PqSignature).as_bytes()) else { return connection.send_error_response(value.get_id(), CommunicationType::ErrorInvalidChallenge).await; };
if !connection.state().consume_challenge(AccountChallengeOperation::Delete, user_id, requester, nonce) { return connection.send_error_response(value.get_id(), CommunicationType::ErrorInvalidChallenge).await; }
let Ok(user) = user_repo::get_by_user_id(UserId::from(user_id)).await else { return connection.send_error_response(value.get_id(), CommunicationType::ErrorNotFound).await; };
let Some(nonce) = value
.get_data(DataType::ServerNonce)
.as_signed_number()
.and_then(|v| u64::try_from(v).ok())
else {
return connection
.send_error_response(value.get_id(), CommunicationType::ErrorInvalidChallenge)
.await;
};
let (Some(signature), Some(pq_signature)) = (
value.get_data(DataType::Signature).as_bytes(),
value.get_data(DataType::PqSignature).as_bytes(),
) else {
return connection
.send_error_response(value.get_id(), CommunicationType::ErrorInvalidChallenge)
.await;
};
if !connection.state().consume_challenge(
AccountChallengeOperation::Delete,
user_id,
requester,
nonce,
) {
return connection
.send_error_response(value.get_id(), CommunicationType::ErrorInvalidChallenge)
.await;
}
let Ok(user) = user_repo::get_by_user_id(UserId::from(user_id)).await else {
return connection
.send_error_response(value.get_id(), CommunicationType::ErrorNotFound)
.await;
};
let payload = lifecycle_payload(b"tensamin:user-delete:v1\0", user_id, requester, nonce);
if verify_ed25519(&user.public_key.sig_cl_public_key, &payload, &signature).is_err() || verify_ml_dsa(&user.public_key.sig_pq_public_key, &payload, &pq_signature).is_err() { return connection.send_error_response(value.get_id(), CommunicationType::ErrorNotAuthenticated).await; }
if verify_ed25519(&user.public_key.sig_cl_public_key, &payload, &signature).is_err()
|| verify_ml_dsa(&user.public_key.sig_pq_public_key, &payload, &pq_signature).is_err()
{
return connection
.send_error_response(value.get_id(), CommunicationType::ErrorNotAuthenticated)
.await;
}
complete_delete(connection, value, user.id).await
}
pub async fn erase_hosted_user_data_ack(connection: Arc<OmikronConnection>, value: CommunicationValue) -> OmikronResult<()> {
let Some(user_id) = value.get_data(DataType::UserId).as_signed_number().and_then(|v| i64::try_from(v).ok()).filter(|v| *v > 0) else { return connection.send_error_response(value.get_id(), CommunicationType::ErrorInvalidUserId).await; };
pub async fn erase_hosted_user_data_ack(
connection: Arc<OmikronConnection>,
value: CommunicationValue,
) -> OmikronResult<()> {
let Some(user_id) = value
.get_data(DataType::UserId)
.as_signed_number()
.and_then(|v| i64::try_from(v).ok())
.filter(|v| *v > 0)
else {
return connection
.send_error_response(value.get_id(), CommunicationType::ErrorInvalidUserId)
.await;
};
let iota_id = IotaId::from(value.get_sender() as i64);
match user_repo::acknowledge_pending_erasure(UserId::from(user_id), iota_id).await {
Ok(true) => connection.send(&CommunicationValue::new(CommunicationType::Success).with_id(value.get_id())).await,
Ok(false) => connection.send_error_response(value.get_id(), CommunicationType::ErrorNotAuthenticated).await,
Err(_) => connection.send_error_response(value.get_id(), CommunicationType::ErrorInternal).await,
Ok(true) => {
connection
.send(&CommunicationValue::new(CommunicationType::Success).with_id(value.get_id()))
.await
}
Ok(false) => {
connection
.send_error_response(value.get_id(), CommunicationType::ErrorNotAuthenticated)
.await
}
Err(_) => {
connection
.send_error_response(value.get_id(), CommunicationType::ErrorInternal)
.await
}
}
}

8
src/transport/handlers/links.rs
View file

@ -1,4 +1,6 @@
use super::super::omikron_connection::{OmikronConnection, OmikronResult};
use super::super::connection::{
MtpValueCompat, OmikronConnection, OmikronResult, OptionalDataValueCompat,
};
use crate::server::short_link::add_short_link;
use mtp::codec::{CommunicationType, CommunicationValue, DataType, DataValue};
use std::sync::Arc;
@ -7,8 +9,8 @@ pub async fn shorten(
connection: Arc<OmikronConnection>,
value: CommunicationValue,
) -> OmikronResult<()> {
let link = value
.get_data(DataType::Link)
let link_data = value.get_data(DataType::Link);
let link = link_data
.as_str()
.ok_or(crate::error::OmegaError::InvalidResponse)?;
let short = add_short_link(link)

4
src/transport/handlers/notifications.rs
View file

@ -1,4 +1,6 @@
use super::super::omikron_connection::{OmikronConnection, OmikronResult};
use super::super::connection::{
MtpValueCompat, OmikronConnection, OmikronResult, OptionalDataValueCompat,
};
use crate::{db::notification_repo, log, models::UserId};
use mtp::codec::{CommunicationType, CommunicationValue, DataType, DataValue};
use mtp::type_map::TypeMap;

21
src/transport/handlers/presence.rs
View file

@ -1,4 +1,6 @@
use super::super::omikron_connection::{OmikronConnection, OmikronResult};
use super::super::connection::{
MtpValueCompat, OmikronConnection, OmikronResult, OptionalDataValueCompat,
};
use crate::{
db::user_repo, log_in, models::IotaId, sql::connection_status::UserStatus, state::OmegaState,
};
@ -19,7 +21,7 @@ fn parse_subscription(value: &CommunicationValue) -> Result<(i64, i64, Vec<i64>)
.and_then(|id| i64::try_from(id).ok())
.filter(|id| *id > 0)
.ok_or("session_id")?;
let DataValue::Array(values) = value.get_data(DataType::UserIds) else {
let Some(DataValue::Array(values)) = value.get_data(DataType::UserIds) else {
return Err("user_ids");
};
@ -51,9 +53,10 @@ fn states_for_users(state: &OmegaState, users: &[crate::models::User]) -> HashMa
.map(|user| {
(
user.id.0,
state
.presence
.resolve_public_state(user.id.0, user.iota_id.map(|id| id.0).unwrap_or_default()),
state.presence.resolve_public_state(
user.id.0,
user.iota_id.map(|id| id.0).unwrap_or_default(),
),
)
})
.collect()
@ -349,7 +352,7 @@ pub async fn set_user_state(
.send_error_response(value.get_id(), CommunicationType::ErrorNoUserId)
.await;
};
if let Some(requested_user) = value.get_data_opt(DataType::UserId) {
if let Some(requested_user) = value.get_data(DataType::UserId) {
let Some(requested_user_id) = requested_user
.as_number()
.and_then(|id| i64::try_from(id).ok())
@ -538,12 +541,12 @@ pub async fn sync_status(
omikron_id: i64,
) -> OmikronResult<()> {
let request_id = value.get_id();
let DataValue::Array(iota_values) = value.get_data(DataType::IotaIds) else {
let Some(DataValue::Array(iota_values)) = value.get_data(DataType::IotaIds) else {
return connection
.send_error_response(request_id, CommunicationType::ErrorInvalidData)
.await;
};
let DataValue::Array(session_values) = value.get_data(DataType::UserStates) else {
let Some(DataValue::Array(session_values)) = value.get_data(DataType::UserStates) else {
return connection
.send_error_response(request_id, CommunicationType::ErrorInvalidData)
.await;
@ -572,7 +575,7 @@ pub async fn sync_status(
}
if !connection.peer_capabilities().session_snapshot_v1 {
let DataValue::Array(user_values) = value.get_data(DataType::UserIds) else {
let Some(DataValue::Array(user_values)) = value.get_data(DataType::UserIds) else {
return connection
.send_error_response(request_id, CommunicationType::ErrorInvalidData)
.await;

4
src/transport/handlers/register.rs
View file

@ -1,4 +1,6 @@
use super::super::omikron_connection::{OmikronConnection, OmikronResult};
use super::super::connection::{
MtpValueCompat, OmikronConnection, OmikronResult, OptionalDataValueCompat,
};
use crate::{
db::{iota_repo, user_repo},
models::{IotaId, UserId},

6
src/transport/handlers/states.rs
View file

@ -1,4 +1,6 @@
use super::super::omikron_connection::{OmikronConnection, OmikronResult};
use super::super::connection::{
MtpValueCompat, OmikronConnection, OmikronResult, OptionalDataValueCompat,
};
use crate::db::user_repo;
use mtp::{
codec::{CommunicationType, CommunicationValue, DataType, DataValue},
@ -29,7 +31,7 @@ pub async fn get(
) -> OmikronResult<()> {
let state = connection.state();
let legacy_peer = !connection.peer_capabilities().client_state_push_v1;
let DataValue::Array(ids) = value.get_data(DataType::UserIds) else {
let Some(DataValue::Array(ids)) = value.get_data(DataType::UserIds) else {
return send_error(
connection,
value.get_id(),

37
src/transport/handlers/user_data.rs
View file

@ -1,4 +1,6 @@
use super::super::omikron_connection::{OmikronConnection, OmikronResult};
use super::super::connection::{
MtpValueCompat, OmikronConnection, OmikronResult, OptionalDataValueCompat,
};
use crate::{
db::{iota_repo, user_repo},
models::{IotaId, UserId},
@ -91,7 +93,9 @@ pub async fn get_user(
}
state.presence.resolve_private_state(id)
} else {
iota_id.map(|iota_id| state.presence.resolve_public_state(id, iota_id)).unwrap_or(crate::sql::connection_status::UserStatus::user_offline)
iota_id
.map(|iota_id| state.presence.resolve_public_state(id, iota_id))
.unwrap_or(crate::sql::connection_status::UserStatus::user_offline)
};
response = response
.add_typed_default(
@ -100,10 +104,13 @@ pub async fn get_user(
)
.add_typed_default(
DataType::OmikronConnections,
iota_id.map(|iota_id| connections(&connection, iota_id)).unwrap_or_else(|| DataValue::Array(Vec::new())),
iota_id
.map(|iota_id| connections(&connection, iota_id))
.unwrap_or_else(|| DataValue::Array(Vec::new())),
);
if let Some(iota_id) = iota_id {
response = response.add_typed_default(DataType::IotaId, DataValue::SignedNumber(iota_id.into()));
response =
response.add_typed_default(DataType::IotaId, DataValue::SignedNumber(iota_id.into()));
}
if let Some(route) = route {
response = response.add_typed_default(
@ -138,10 +145,14 @@ pub async fn get_iota(
} else if let Some(name) = value.get_data(DataType::Username).as_str() {
if let Ok(user) = user_repo::get_by_username(name).await {
match user.iota_id {
Some(iota_id) => iota_repo::get_iota_by_id(iota_id)
.await
.ok()
.map(|iota| (iota.id.0, iota.public_key, Some(user.id.0), Some(name.to_owned()))),
Some(iota_id) => iota_repo::get_iota_by_id(iota_id).await.ok().map(|iota| {
(
iota.id.0,
iota.public_key,
Some(user.id.0),
Some(name.to_owned()),
)
}),
None => None,
}
} else {
@ -245,12 +256,14 @@ pub async fn change_iota(
connection: Arc<OmikronConnection>,
value: CommunicationValue,
) -> OmikronResult<()> {
let Some(reset) = value.get_data(DataType::ResetToken).as_str() else {
let reset_data = value.get_data(DataType::ResetToken);
let Some(reset) = reset_data.as_str() else {
return connection
.send_error_response(value.get_id(), CommunicationType::ErrorInvalidData)
.await;
};
let Some(new_token) = value.get_data(DataType::NewToken).as_str() else {
let new_token_data = value.get_data(DataType::NewToken);
let Some(new_token) = new_token_data.as_str() else {
return connection
.send_error_response(value.get_id(), CommunicationType::ErrorInvalidData)
.await;
@ -270,7 +283,9 @@ pub async fn change_iota(
.await;
}
let result =
match user_repo::change_iota_id(user_id, Some(IotaId::from(value.get_sender() as i64))).await {
match user_repo::change_iota_id(user_id, Some(IotaId::from(value.get_sender() as i64)))
.await
{
Ok(()) => user_repo::change_token(user_id, new_token.to_owned()).await,
Err(error) => Err(error),
};

1
src/transport/mod.rs
View file

@ -3,3 +3,4 @@ pub mod connection;
pub mod handlers;
pub mod omikron_connection;
pub mod omikron_manager;
pub mod relay_router;

97
src/transport/omikron_connection.rs
View file

@ -10,7 +10,7 @@ use dashmap::DashMap;
use mtp::{
codec::{CommunicationType, CommunicationValue},
crypto::PublicKeyBundle,
host::{AuthenticationPolicy, HostConfig, Policy, SendMode},
host::{AuthState, AuthenticationPolicy, HostConfig, Policy, SendMode},
webserver::{MTPWebServer, WebMtpReceiver, WebMtpSender},
};
use std::{
@ -21,7 +21,10 @@ use std::{
},
time::{Duration, Instant},
};
use tokio::{sync::Mutex, time::interval};
use tokio::{
sync::{Mutex, mpsc},
time::interval,
};
const CLEANUP_INTERVAL: Duration = Duration::from_secs(30);
const MAX_WAITING_AGE: Duration = Duration::from_secs(60);
@ -70,7 +73,11 @@ impl OmikronConnection {
id: u64,
description: Option<&str>,
state: Arc<OmegaState>,
authenticated: bool,
) -> Option<Arc<Self>> {
if !authenticated {
return None;
}
let peer_capabilities =
PeerCapabilities::from_identification_description(description).ok()?;
Some(Arc::new(Self {
@ -87,7 +94,6 @@ impl OmikronConnection {
&self.peer_capabilities
}
pub async fn handle(self: Arc<Self>, receiver: &mut WebMtpReceiver) {
log_in!(
self.id as i64,
@ -158,7 +164,12 @@ impl OmikronConnection {
async fn process_message(self: Arc<Self>, value: CommunicationValue) -> OmikronResult<()> {
log_cv_in!(PrintType::Omikron, &value);
if let Some((_, task)) = self.waiting_tasks.remove(&value.get_id()) {
if value.is_type(CommunicationType::Relay) {
return self.dispatch(value).await;
}
if let Some(message_id) = value.id()
&& let Some((_, task)) = self.waiting_tasks.remove(&message_id)
{
let _ = (task.task)(self.clone(), value);
return Ok(());
}
@ -168,6 +179,36 @@ impl OmikronConnection {
async fn dispatch(self: Arc<Self>, value: CommunicationValue) -> OmikronResult<()> {
let id = self.id as i64;
let state = self.state.clone();
if value.is_type(CommunicationType::Relay) {
let value = crate::transport::relay_router::ensure_relay_frame_id(value);
let request_id = value.id();
let result = crate::transport::relay_router::route_from_omikron(id, value).await;
let response = match &result {
Ok(()) => request_id.map(|request_id| {
CommunicationValue::new(CommunicationType::Success).with_id(request_id)
}),
Err(error) => {
log_err!(id, PrintType::Omega, "Relay routing failed: {}", error);
request_id.map(|request_id| {
CommunicationValue::new(
crate::transport::relay_router::error_response_type(error),
)
.with_id(request_id)
})
}
};
if let Some(response) = response
&& let Err(send_error) = self.clone().send(&response).await
{
log_err!(
id,
PrintType::Omega,
"Relay routing response failed: {}",
send_error
);
}
return result.map_err(|error| crate::error::OmegaError::Transport(error.to_string()));
}
match value.get_comm_type_enum() {
Some(CommunicationType::ShortenLink) => {
crate::transport::handlers::links::shorten(self, value).await
@ -179,9 +220,6 @@ impl OmikronConnection {
crate::transport::handlers::presence::user_disconnected(state, self, value, id)
.await
}
Some(CommunicationType::SetUserState) => {
crate::transport::handlers::presence::set_user_state(state, self, value, id).await
}
Some(CommunicationType::IotaConnected) => {
crate::transport::handlers::presence::iota_connected(state, self, value, id).await
}
@ -297,6 +335,47 @@ impl OmikronConnection {
}
Ok(())
}
pub(crate) async fn await_response(
self: Arc<Self>,
value: &CommunicationValue,
timeout: Duration,
) -> OmikronResult<CommunicationValue> {
let (tx, mut rx) = mpsc::channel(1);
let message_id = value.id().unwrap_or_default();
let task_tx = tx.clone();
self.waiting_tasks.insert(
message_id,
WaitingTask {
task: Box::new(move |_, response| {
let task_tx = task_tx.clone();
tokio::spawn(async move {
let _ = task_tx.send(response).await;
});
true
}),
inserted_at: Instant::now(),
},
);
if let Err(error) = self.clone().send(value).await {
self.waiting_tasks.remove(&message_id);
return Err(error);
}
match tokio::time::timeout(timeout, rx.recv()).await {
Ok(Some(response)) => Ok(response),
Ok(None) => Err(crate::error::OmegaError::Transport(
"Relay response channel closed".into(),
)),
Err(_) => {
self.waiting_tasks.remove(&message_id);
Err(crate::error::OmegaError::Transport(
"Relay response timed out".into(),
))
}
}
}
pub(crate) async fn send_error_response(
self: Arc<Self>,
message_id: u32,
@ -439,16 +518,18 @@ pub async fn start(port: u16, state: Arc<OmegaState>) -> Result<(), Box<dyn std:
);
continue;
}
let authenticated = matches!(&conn.auth_state, AuthState::Authenticated);
let Some(connection) = OmikronConnection::new(
conn.sender,
conn.client_id,
conn.description.as_deref(),
state.clone(),
authenticated,
) else {
log_err!(
0,
PrintType::Omega,
"Rejected Omikron connection with invalid capabilities"
"Rejected Omikron connection without authenticated transport state or valid capabilities"
);
continue;
};

35
src/transport/omikron_manager.rs
View file

@ -66,7 +66,8 @@ pub async fn get_all_connections()
for route in state.presence.routes_for_user(user.id.0) {
if let Some(iotas) = result.get_mut(&route.omikron_id) {
if let Some(iota_id) = user.iota_id
&& let Some(users) = iotas.get_mut(&iota_id.0) {
&& let Some(users) = iotas.get_mut(&iota_id.0)
{
users.push(user.id.0);
}
}
@ -118,10 +119,20 @@ pub async fn send_to_user(user_id: i64, cv: &CommunicationValue) {
/// 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())).collect();
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()))
.collect();
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));
@ -129,9 +140,17 @@ pub async fn publish_iota_user_snapshot(iota_id: i64) {
}
pub async fn deliver_pending_erasures(iota_id: i64) {
let Ok(users) = user_repo::pending_erasures_for_iota(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; };
let Ok(users) =
user_repo::pending_erasures_for_iota(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_id in users {
let request = CommunicationValue::new(CommunicationType::EraseHostedUserData)
.add_typed_default(DataType::UserId, DataValue::SignedNumber(user_id.0.into()))

243
src/transport/relay_router.rs Normal file
View file

@ -0,0 +1,243 @@
use super::omikron_manager;
use crate::{log_err, util::logger::PrintType};
use mtp::codec::{CommunicationType, CommunicationValue, RelayError, forward_relay_frame};
use std::{
convert::TryFrom,
sync::atomic::{AtomicU32, Ordering},
time::Duration,
};
use thiserror::Error;
const TARGET_KIND_MASK: u64 = 0xC000_0000_0000_0000;
const TARGET_ID_MASK: u64 = (1_u64 << 48) - 1;
const USER_TARGET_KIND: u64 = 0x4000_0000_0000_0000;
const IOTA_TARGET_KIND: u64 = 0x8000_0000_0000_0000;
static NEXT_RELAY_FRAME_ID: AtomicU32 = AtomicU32::new(1);
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum RouteTarget {
User(u64),
Iota(u64),
}
impl RouteTarget {
pub fn wire_id(self) -> Option<u64> {
let (kind, id) = match self {
Self::User(id) => (USER_TARGET_KIND, id),
Self::Iota(id) => (IOTA_TARGET_KIND, id),
};
(id > 0 && id <= TARGET_ID_MASK).then_some(kind | id)
}
pub fn from_wire_id(value: u64) -> Option<Self> {
let id = value & TARGET_ID_MASK;
if id == 0 || value & !(TARGET_KIND_MASK | TARGET_ID_MASK) != 0 {
return None;
}
match value & TARGET_KIND_MASK {
USER_TARGET_KIND => Some(Self::User(id)),
IOTA_TARGET_KIND => Some(Self::Iota(id)),
_ => None,
}
}
pub const fn id(self) -> u64 {
match self {
Self::User(id) | Self::Iota(id) => id,
}
}
}
#[derive(Debug, Error)]
pub enum RelayRouteError {
#[error("relay has no destination Iota")]
MissingDestinationIota,
#[error("relay has an outer sender")]
OuterSenderPresent,
#[error("relay has invalid route target {0}")]
InvalidDestinationTarget(u64),
#[error("relay destination Iota is outside Omega's ID range")]
DestinationIotaOutOfRange,
#[error("destination Iota is offline")]
IotaOffline,
#[error("destination Omikron is offline")]
OmikronOffline,
#[error("relay route resolves back to source Omikron")]
RouteLoop,
#[error(transparent)]
Relay(#[from] RelayError),
#[error("sending relay to destination Omikron failed: {0}")]
Send(String),
}
pub fn ensure_relay_frame_id(frame: CommunicationValue) -> CommunicationValue {
if frame.id().is_some_and(|id| id != 0) {
return frame;
}
let id = NEXT_RELAY_FRAME_ID.fetch_add(1, Ordering::Relaxed).max(1);
frame.with_id(id)
}
pub fn error_response_type(error: &RelayRouteError) -> CommunicationType {
match error {
RelayRouteError::IotaOffline | RelayRouteError::OmikronOffline => {
CommunicationType::ErrorNoIota
}
RelayRouteError::Send(_) => CommunicationType::ErrorInternal,
RelayRouteError::MissingDestinationIota
| RelayRouteError::OuterSenderPresent
| RelayRouteError::InvalidDestinationTarget(_)
| RelayRouteError::DestinationIotaOutOfRange
| RelayRouteError::RouteLoop
| RelayRouteError::Relay(_) => CommunicationType::ErrorInvalidData,
}
}
pub async fn route_from_omikron(
source_omikron_id: i64,
frame: CommunicationValue,
) -> Result<(), RelayRouteError> {
let frame = ensure_relay_frame_id(frame);
if !frame.is_type(CommunicationType::Relay) {
return Err(RelayRouteError::Relay(RelayError::NotRelay));
}
if frame.sender().is_some() {
return Err(RelayRouteError::OuterSenderPresent);
}
let Some(destination_wire_id) = frame.receiver() else {
log_err!(
source_omikron_id,
PrintType::Omega,
"Relay routing failed: missing destination Iota"
);
return Err(RelayRouteError::MissingDestinationIota);
};
let Some(RouteTarget::Iota(destination_iota)) = RouteTarget::from_wire_id(destination_wire_id)
else {
return Err(RelayRouteError::InvalidDestinationTarget(
destination_wire_id,
));
};
let destination_iota_i64 =
i64::try_from(destination_iota).map_err(|_| RelayRouteError::DestinationIotaOutOfRange)?;
let frame = forward_relay_frame(&frame, destination_wire_id)?;
let Some(destination_omikron) =
omikron_manager::get_iota_primary_omikron_connection(destination_iota_i64)
else {
log_err!(
source_omikron_id,
PrintType::Omega,
"Relay destination Iota {} is offline",
destination_iota
);
return Err(RelayRouteError::IotaOffline);
};
if destination_omikron == source_omikron_id {
log_err!(
source_omikron_id,
PrintType::Omega,
"Relay route loop for destination Iota {} and Omikron {}",
destination_iota,
destination_omikron
);
return Err(RelayRouteError::RouteLoop);
}
let Some(connection) = omikron_manager::get_connected_omikron(destination_omikron) else {
log_err!(
source_omikron_id,
PrintType::Omega,
"Relay destination Iota {} resolves to disconnected Omikron {}",
destination_iota,
destination_omikron
);
return Err(RelayRouteError::OmikronOffline);
};
let response = connection
.await_response(&frame, Duration::from_secs(20))
.await
.map_err(|error| {
log_err!(
source_omikron_id,
PrintType::Omega,
"Relay send to destination Iota {} via Omikron {} failed: {}",
destination_iota,
destination_omikron,
error
);
RelayRouteError::Send(error.to_string())
})?;
if response.is_type(CommunicationType::Success) {
Ok(())
} else {
Err(RelayRouteError::Send(format!(
"destination Omikron rejected the Relay with {}",
response.get_type()
)))
}
}
#[cfg(test)]
mod tests {
use super::*;
use mtp::codec::DataValue;
fn wire(target: RouteTarget) -> u64 {
let Some(value) = target.wire_id() else {
panic!("valid route target was rejected");
};
value
}
fn relay_frame() -> CommunicationValue {
CommunicationValue::new(CommunicationType::Relay)
.without_sender()
.with_receiver(wire(RouteTarget::Iota(42)))
.with_payload(DataValue::Bytes(vec![1, 2, 3, 4]))
}
#[test]
fn forwarding_preserves_relay_payload_and_next_hop() {
let frame = relay_frame().with_receiver(wire(RouteTarget::Iota(7)));
let result = forward_relay_frame(&frame, wire(RouteTarget::Iota(42)));
assert!(result.is_ok());
let Ok(forwarded) = result else { return };
assert_eq!(forwarded.receiver(), Some(wire(RouteTarget::Iota(42))));
assert_eq!(forwarded.sender(), None);
assert_eq!(forwarded.payload(), frame.payload());
assert_eq!(forwarded.id(), frame.id());
}
#[tokio::test]
async fn outer_sender_is_rejected_before_route_lookup() {
let frame = relay_frame().with_sender(9);
let result = route_from_omikron(1, frame).await;
assert!(matches!(result, Err(RelayRouteError::OuterSenderPresent)));
}
#[tokio::test]
async fn missing_destination_is_rejected_before_route_lookup() {
let frame = relay_frame().without_receiver();
let result = route_from_omikron(1, frame).await;
assert!(matches!(
result,
Err(RelayRouteError::MissingDestinationIota)
));
}
#[tokio::test]
async fn user_route_target_is_rejected_by_opaque_omega_router() {
let frame = relay_frame().with_receiver(wire(RouteTarget::User(42)));
let result = route_from_omikron(1, frame).await;
assert!(matches!(
result,
Err(RelayRouteError::InvalidDestinationTarget(_))
));
}
#[tokio::test]
async fn unavailable_destination_is_reported_as_iota_offline() {
let result = route_from_omikron(1, relay_frame()).await;
assert!(matches!(result, Err(RelayRouteError::IotaOffline)));
}
}

5
src/util/logger.rs
View file

@ -10,6 +10,7 @@ use std::{
use ansi_term::Color;
use mtp::codec::{CommunicationValue, DataTypeId, DataValue, Version};
use crate::transport::connection::MtpValueCompat;
use crate::util::file_util::get_directory;
static LOGGER: OnceLock<mpsc::Sender<LogMessage>> = OnceLock::new();
@ -250,11 +251,11 @@ pub fn format_cv(cv: &CommunicationValue) -> String {
.unwrap_or_else(|| cv.get_type().to_string());
parts.push(format!("{} (id={})", comm_type, cv.get_id()));
let data = cv.data();
let data = cv.data().unwrap_or(&[]);
let formated_data = format_data_container(
data.iter().map(|(k, v)| (*k, v.clone())).collect(),
Version(1, 0),
Version(3, 0),
);
parts.push(format!("{}", formated_data));