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

Merge origin/main

Browse source
This commit is contained in:
Alex Emmet 2026-07-06 18:54:20 +02:00
commit 1e0d2b9691
5 changed files with 667 additions and 593 deletions
Download patch file Download diff file Expand all files Collapse all files

2
client/Cargo.toml
View file

@ -4,7 +4,7 @@ version = "0.1.0"
edition = "2024" edition = "2024"
[dependencies] [dependencies]
mtp = { git = "https://git.methanium.net/Methanium/mtp.git" } mtp = { git = "https://git.methanium.net/Methanium/mtp.git", features = ["client"] }
iota-logger = { path = "../iota-logger" } iota-logger = { path = "../iota-logger" }
iota-util = { path = "../iota-util" } iota-util = { path = "../iota-util" }
iota-storage = { path = "../iota-storage" } iota-storage = { path = "../iota-storage" }

349
client/src/client_connection.rs
View file

@ -6,16 +6,14 @@ use iota_storage::util::chat_files::{MessageState, change_message_state};
use iota_storage::util::chats_util::{get_user, mod_user}; use iota_storage::util::chats_util::{get_user, mod_user};
use iota_storage::util::communities_util::CommunitiesUtil; use iota_storage::util::communities_util::CommunitiesUtil;
use iota_storage::util::config_util::CONFIG; use iota_storage::util::config_util::CONFIG;
use iota_storage::util::e2ee_storage::{self, ChatSecretQuery, StoredChatSecret};
use iota_storage::util::{chat_files, chats_util}; use iota_storage::util::{chat_files, chats_util};
use iota_storage::util::e2ee_storage::{
self, EncryptedDeviceSecretQuery, EncryptedMessageQuery,
StoredEncryptedDeviceSecret, StoredEncryptedMessage,
};
use iota_util::crypto_helper::keyring_from_base64; use iota_util::crypto_helper::keyring_from_base64;
use iota_util::crypto_util::{self}; use iota_util::crypto_util::{self};
use iota_util::file_util::{get_children, load_file, save_file}; use iota_util::file_util::{get_children, load_file, save_file};
use mtp::client::{Receiver, Sender}; use mtp::client::{Receiver, Sender};
use mtp::codec::{CommunicationType, CommunicationValue, DataType, DataValue}; use mtp::codec::{CommunicationType, CommunicationValue, DataType, DataValue};
use mtp::type_map::TypeMap;
use std::sync::Arc; use std::sync::Arc;
use std::time::{Duration, SystemTime, UNIX_EPOCH}; use std::time::{Duration, SystemTime, UNIX_EPOCH};
@ -34,8 +32,6 @@ fn typed_container(items: Vec<(DataType, DataValue)>) -> DataValue {
) )
} }
fn data_string(cv: &CommunicationValue, dt: DataType) -> Option<String> { fn data_string(cv: &CommunicationValue, dt: DataType) -> Option<String> {
cv.get_data(dt) cv.get_data(dt)
.as_str() .as_str()
@ -48,12 +44,102 @@ fn data_i64(cv: &CommunicationValue, dt: DataType) -> Option<i64> {
cv.get_data(dt) cv.get_data(dt)
.as_number() .as_number()
.and_then(|n| i64::try_from(n).ok()) .and_then(|n| i64::try_from(n).ok())
.or_else(|| cv.get_data(dt).as_signed_number().and_then(|n| i64::try_from(n).ok())) .or_else(|| {
cv.get_data(dt)
.as_signed_number()
.and_then(|n| i64::try_from(n).ok())
})
.or_else(|| cv.get_data(dt).as_str().and_then(|s| s.parse::<i64>().ok())) .or_else(|| cv.get_data(dt).as_str().and_then(|s| s.parse::<i64>().ok()))
} }
fn data_bytes(cv: &CommunicationValue, dt: DataType) -> Option<Vec<u8>> { #[derive(Debug, Clone)]
cv.get_bytes(dt).map(|bytes| bytes.to_vec()) struct ChatSecretRecipient {
user_id: String,
encrypted_secret: Vec<u8>,
kem_ciphertext: Vec<u8>,
}
fn recipient_from_value(value: &DataValue) -> Option<ChatSecretRecipient> {
let tm = TypeMap::latest();
let user_id = value
.get_field(DataType::UserId.to_id(&tm))?
.as_str()
.map(|s| s.to_string())
.or_else(|| {
value
.get_field(DataType::UserId.to_id(&tm))?
.as_number()
.map(|n| n.to_string())
})?;
let encrypted_secret = value
.get_field(DataType::EncryptedSecret.to_id(&tm))?
.as_bytes()?;
let kem_ciphertext = value
.get_field(DataType::KemCiphertext.to_id(&tm))?
.as_bytes()?;
Some(ChatSecretRecipient {
user_id,
encrypted_secret,
kem_ciphertext,
})
}
fn chat_secret_recipients(cv: &CommunicationValue) -> Option<Vec<ChatSecretRecipient>> {
let recipients = cv.get_data(DataType::Recipients).as_array()?;
let parsed = recipients
.iter()
.map(recipient_from_value)
.collect::<Option<Vec<_>>>()?;
if parsed.is_empty() {
None
} else {
Some(parsed)
}
}
fn set_chat_secret_cv_for_recipient(
source: &CommunicationValue,
recipient: &ChatSecretRecipient,
) -> CommunicationValue {
let recipient_value = typed_container(vec![
(DataType::UserId, DataValue::Str(recipient.user_id.clone())),
(
DataType::EncryptedSecret,
DataValue::Bytes(recipient.encrypted_secret.clone()),
),
(
DataType::KemCiphertext,
DataValue::Bytes(recipient.kem_ciphertext.clone()),
),
]);
CommunicationValue::new(CommunicationType::SetChatSecret)
.with_id(source.get_id())
.with_sender(source.get_sender())
.with_receiver(recipient.user_id.parse::<u64>().unwrap_or(0))
.add_typed_default(DataType::ChatId, source.get_data(DataType::ChatId).clone())
.add_typed_default(
DataType::SecretId,
source.get_data(DataType::SecretId).clone(),
)
.add_typed_default(
DataType::VersionNumber,
source.get_data(DataType::VersionNumber).clone(),
)
.add_typed_default(
DataType::WrappingScheme,
source.get_data(DataType::WrappingScheme).clone(),
)
.add_typed_default(
DataType::CreatedAt,
source.get_data(DataType::CreatedAt).clone(),
)
.add_typed_default(
DataType::Recipients,
DataValue::Array(vec![recipient_value]),
)
} }
fn now_millis_i64() -> i64 { fn now_millis_i64() -> i64 {
@ -182,187 +268,141 @@ impl ClientConnection {
return; return;
} }
if cv.is_type(CommunicationType::SetEncryptedDeviceSecret) { if cv.is_type(CommunicationType::SetChatSecret) {
let sender_id = cv.get_sender().to_string(); let sender_id = cv.get_sender().to_string();
if data_string(&cv, DataType::UserId).as_deref() != Some(sender_id.as_str()) { let recipients = match chat_secret_recipients(&cv) {
self.send_message(&error_response(&cv, CommunicationType::ErrorInvalidData)).await; Some(recipients) => recipients,
return; None => {
} self.send_message(&error_response(&cv, CommunicationType::ErrorInvalidData))
.await;
return;
}
};
let now = now_millis_i64(); let now = now_millis_i64();
let record = data_string(&cv, DataType::UserId) let chat_id = data_string(&cv, DataType::ChatId);
.zip(data_string(&cv, DataType::DeviceId)) let secret_id = data_string(&cv, DataType::SecretId);
.zip(data_string(&cv, DataType::SecretId)) let version = data_i64(&cv, DataType::VersionNumber);
.zip(data_i64(&cv, DataType::VersionNumber)) let wrapping_scheme = data_string(&cv, DataType::WrappingScheme);
.zip(data_bytes(&cv, DataType::EncryptedSecret)) let created_at = data_i64(&cv, DataType::CreatedAt).unwrap_or(now);
.zip(data_string(&cv, DataType::WrappingScheme))
.map(|(((((user_id, device_id), secret_id), version), encrypted_secret), wrapping_scheme)| {
StoredEncryptedDeviceSecret {
user_id,
device_id,
secret_id,
version,
encrypted_secret,
wrapping_public_key_id: data_string(&cv, DataType::WrappingPublicKeyId),
wrapping_scheme,
created_at: data_i64(&cv, DataType::CreatedAt).unwrap_or(now),
updated_at: now,
}
});
match record.map(e2ee_storage::put_encrypted_device_secret) { let Some((((chat_id, secret_id), version), wrapping_scheme)) =
Some(Ok(())) => { chat_id.zip(secret_id).zip(version).zip(wrapping_scheme)
self.send_message(&error_response(&cv, CommunicationType::Success)).await; else {
} self.send_message(&error_response(&cv, CommunicationType::ErrorInvalidData))
_ => { .await;
self.send_message(&error_response(&cv, CommunicationType::ErrorInvalidData)).await; return;
};
for recipient in recipients.iter().filter(|item| item.user_id == sender_id) {
if e2ee_storage::put_chat_secret(StoredChatSecret {
user_id: recipient.user_id.clone(),
chat_id: chat_id.clone(),
secret_id: secret_id.clone(),
version,
encrypted_secret: recipient.encrypted_secret.clone(),
kem_ciphertext: recipient.kem_ciphertext.clone(),
wrapping_scheme: wrapping_scheme.clone(),
created_at,
updated_at: now,
})
.is_err()
{
self.send_message(&error_response(&cv, CommunicationType::ErrorInvalidData))
.await;
return;
} }
} }
for recipient in recipients.iter().filter(|item| item.user_id != sender_id) {
self.send_message(&set_chat_secret_cv_for_recipient(&cv, recipient))
.await;
}
self.send_message(&error_response(&cv, CommunicationType::Success))
.await;
return; return;
} }
if cv.is_type(CommunicationType::GetEncryptedDeviceSecret) { if cv.is_type(CommunicationType::GetChatSecret) {
let Some(user_id) = data_string(&cv, DataType::UserId) else { let Some(user_id) = data_string(&cv, DataType::UserId) else {
self.send_message(&error_response(&cv, CommunicationType::ErrorInvalidData)).await; self.send_message(&error_response(&cv, CommunicationType::ErrorInvalidData))
.await;
return; return;
}; };
let sender_id = cv.get_sender().to_string(); let sender_id = cv.get_sender().to_string();
if user_id != sender_id { if user_id != sender_id {
self.send_message(&error_response(&cv, CommunicationType::ErrorNotFound)).await; self.send_message(&error_response(&cv, CommunicationType::ErrorNotFound))
.await;
return; return;
} }
let Some(chat_id) = data_string(&cv, DataType::ChatId) else {
self.send_message(&error_response(&cv, CommunicationType::ErrorInvalidData))
.await;
return;
};
match e2ee_storage::get_encrypted_device_secret(EncryptedDeviceSecretQuery { match e2ee_storage::get_chat_secret(ChatSecretQuery {
user_id, user_id,
device_id: data_string(&cv, DataType::DeviceId), chat_id,
secret_id: data_string(&cv, DataType::SecretId), secret_id: data_string(&cv, DataType::SecretId),
}) { }) {
Ok(Some(record)) => { Ok(Some(record)) => {
let mut response = CommunicationValue::new(CommunicationType::EncryptedDeviceSecretResponse) let response = CommunicationValue::new(CommunicationType::ChatSecretResponse)
.with_id(cv.get_id()) .with_id(cv.get_id())
.with_receiver(cv.get_sender()) .with_receiver(cv.get_sender())
.add_typed_default(DataType::UserId, DataValue::Str(record.user_id)) .add_typed_default(DataType::UserId, DataValue::Str(record.user_id))
.add_typed_default(DataType::DeviceId, DataValue::Str(record.device_id)) .add_typed_default(DataType::ChatId, DataValue::Str(record.chat_id))
.add_typed_default(DataType::SecretId, DataValue::Str(record.secret_id)) .add_typed_default(DataType::SecretId, DataValue::Str(record.secret_id))
.add_typed_default(DataType::VersionNumber, DataValue::SignedNumber(record.version as i128)) .add_typed_default(
.add_typed_default(DataType::EncryptedSecret, DataValue::Bytes(record.encrypted_secret)) DataType::VersionNumber,
.add_typed_default(DataType::WrappingScheme, DataValue::Str(record.wrapping_scheme)) DataValue::SignedNumber(record.version as i128),
.add_typed_default(DataType::CreatedAt, DataValue::SignedNumber(record.created_at as i128)) )
.add_typed_default(DataType::UpdatedAt, DataValue::SignedNumber(record.updated_at as i128)); .add_typed_default(
if let Some(value) = record.wrapping_public_key_id { DataType::EncryptedSecret,
response = response.add_typed_default(DataType::WrappingPublicKeyId, DataValue::Str(value)); DataValue::Bytes(record.encrypted_secret),
} )
.add_typed_default(
DataType::KemCiphertext,
DataValue::Bytes(record.kem_ciphertext),
)
.add_typed_default(
DataType::WrappingScheme,
DataValue::Str(record.wrapping_scheme),
)
.add_typed_default(
DataType::CreatedAt,
DataValue::SignedNumber(record.created_at as i128),
)
.add_typed_default(
DataType::UpdatedAt,
DataValue::SignedNumber(record.updated_at as i128),
);
self.send_message(&response).await; self.send_message(&response).await;
} }
Ok(None) => { Ok(None) => {
self.send_message(&error_response(&cv, CommunicationType::ErrorNotFound)).await; self.send_message(&error_response(&cv, CommunicationType::ErrorNotSet))
.await;
} }
Err(_) => { Err(_) => {
self.send_message(&error_response(&cv, CommunicationType::ErrorInvalidData)).await; self.send_message(&error_response(&cv, CommunicationType::ErrorInvalidData))
.await;
} }
} }
return; return;
} }
if cv.is_type(CommunicationType::EncryptedMessage) { if cv.is_type(CommunicationType::ChatSecretForward) {
let sender_id = cv.get_sender().to_string(); let sender_id = cv.get_sender().to_string();
if data_string(&cv, DataType::SenderUserId).as_deref() != Some(sender_id.as_str()) { let recipient_id = data_string(&cv, DataType::RecipientUserId).unwrap_or_default();
self.send_message(&error_response(&cv, CommunicationType::ErrorInvalidData)).await; if data_string(&cv, DataType::SenderUserId).as_deref() != Some(sender_id.as_str())
|| recipient_id.is_empty()
{
self.send_message(&error_response(&cv, CommunicationType::ErrorInvalidData))
.await;
return; return;
} }
let record = data_string(&cv, DataType::MessageId) self.send_message(&cv.with_receiver(recipient_id.parse::<u64>().unwrap_or(0)))
.zip(data_string(&cv, DataType::ConversationId)) .await;
.zip(data_string(&cv, DataType::SenderClientId))
.zip(data_string(&cv, DataType::RecipientClientId))
.zip(data_bytes(&cv, DataType::EncryptedPayload))
.map(|((((message_id, conversation_id), sender_client_id), recipient_client_id), encrypted_payload)| {
StoredEncryptedMessage {
message_id,
conversation_id,
sender_client_id,
recipient_client_id,
sender_user_id: data_string(&cv, DataType::SenderUserId),
recipient_user_id: data_string(&cv, DataType::RecipientUserId),
created_at: data_i64(&cv, DataType::CreatedAt).unwrap_or_else(now_millis_i64),
encryption_version: data_i64(&cv, DataType::EncryptionVersion).unwrap_or(1),
encrypted_payload,
}
});
if let Some(record) = record {
let message_id = record.message_id.clone();
let conversation_id = record.conversation_id.clone();
let recipient_client_id = record.recipient_client_id.clone();
match e2ee_storage::put_encrypted_message(record) {
Ok(()) => {
self.send_message(&CommunicationValue::new(CommunicationType::EncryptedMessageAck)
.with_id(cv.get_id())
.with_receiver(cv.get_sender())
.add_typed_default(DataType::MessageId, DataValue::Str(message_id))
.add_typed_default(DataType::ConversationId, DataValue::Str(conversation_id))
.add_typed_default(DataType::RecipientClientId, DataValue::Str(recipient_client_id))
.add_typed_default(DataType::GetTime, DataValue::SignedNumber(now_millis_i64() as i128))
).await;
}
Err(_) => {
self.send_message(&error_response(&cv, CommunicationType::ErrorInvalidData)).await;
}
}
} else {
self.send_message(&error_response(&cv, CommunicationType::ErrorInvalidData)).await;
}
return;
}
if cv.is_type(CommunicationType::EncryptedMessagesGet) {
let requester_user_id = cv.get_sender().to_string();
let limit = data_i64(&cv, DataType::Limit).map(|v| v as i64);
let since = data_i64(&cv, DataType::Since);
let conversation_id = data_string(&cv, DataType::ConversationId);
let peer_client_id = data_string(&cv, DataType::PeerClientId);
match e2ee_storage::get_encrypted_messages(EncryptedMessageQuery {
sender_user_id: data_string(&cv, DataType::SenderUserId),
recipient_client_id: None,
recipient_user_id: Some(requester_user_id.clone()),
conversation_id,
limit,
offset: since.map(|v| v.max(0)),
}) {
Ok(records) => {
let messages = records
.into_iter()
.filter(|record| {
peer_client_id
.as_ref()
.map(|peer| &record.sender_client_id == peer || &record.recipient_client_id == peer)
.unwrap_or(true)
})
.map(|record| {
typed_container(vec![
(DataType::MessageId, DataValue::Str(record.message_id)),
(DataType::ConversationId, DataValue::Str(record.conversation_id)),
(DataType::SenderClientId, DataValue::Str(record.sender_client_id)),
(DataType::RecipientClientId, DataValue::Str(record.recipient_client_id)),
(DataType::SenderUserId, DataValue::Str(record.sender_user_id.unwrap_or_default())),
(DataType::RecipientUserId, DataValue::Str(record.recipient_user_id.unwrap_or_default())),
(DataType::CreatedAt, DataValue::SignedNumber(record.created_at as i128)),
(DataType::EncryptionVersion, DataValue::SignedNumber(record.encryption_version as i128)),
(DataType::EncryptedPayload, DataValue::Bytes(record.encrypted_payload)),
])
})
.collect::<Vec<_>>();
self.send_message(&CommunicationValue::new(CommunicationType::EncryptedMessagesResponse)
.with_id(cv.get_id())
.with_receiver(cv.get_sender())
.add_typed_default(DataType::Messages, DataValue::Array(messages))
.add_typed_default(DataType::HasMore, DataValue::Bool(false))
).await;
}
Err(_) => {
self.send_message(&error_response(&cv, CommunicationType::ErrorInvalidData)).await;
}
}
return; return;
} }
@ -551,7 +591,8 @@ impl ClientConnection {
// Incoming storsed message: store for the recipient, attempt local delivery, notify sender. // Incoming storsed message: store for the recipient, attempt local delivery, notify sender.
if cv.is_type(CommunicationType::MessageSend) { if cv.is_type(CommunicationType::MessageSend) {
self.send_message(&error_response(&cv, CommunicationType::ErrorInvalidData)).await; self.send_message(&error_response(&cv, CommunicationType::ErrorInvalidData))
.await;
return; return;
} }

450
iota-storage/src/util/e2ee_storage.rs
View file

@ -5,46 +5,36 @@ use std::sync::{Arc, LazyLock, Mutex};
pub type StorageError = String; pub type StorageError = String;
#[derive(Debug, Clone, PartialEq, Eq)] #[derive(Debug, Clone, PartialEq, Eq)]
pub struct StoredEncryptedMessage { pub struct StoredChatSecret {
pub message_id: String,
pub conversation_id: String,
pub sender_client_id: String,
pub recipient_client_id: String,
pub sender_user_id: Option<String>,
pub recipient_user_id: Option<String>,
pub created_at: i64,
pub encryption_version: i64,
pub encrypted_payload: Vec<u8>,
}
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct StoredEncryptedDeviceSecret {
pub user_id: String, pub user_id: String,
pub device_id: String, pub chat_id: String,
pub secret_id: String, pub secret_id: String,
pub version: i64, pub version: i64,
pub encrypted_secret: Vec<u8>, pub encrypted_secret: Vec<u8>,
pub wrapping_public_key_id: Option<String>, pub kem_ciphertext: Vec<u8>,
pub wrapping_scheme: String, pub wrapping_scheme: String,
pub created_at: i64, pub created_at: i64,
pub updated_at: i64, pub updated_at: i64,
} }
#[derive(Debug, Clone, Default)] #[derive(Debug, Clone, Default)]
pub struct EncryptedMessageQuery { pub struct ChatSecretQuery {
pub sender_user_id: Option<String>, pub user_id: String,
pub recipient_client_id: Option<String>, pub chat_id: String,
pub recipient_user_id: Option<String>, pub secret_id: Option<String>,
pub conversation_id: Option<String>,
pub limit: Option<i64>,
pub offset: Option<i64>,
} }
#[derive(Debug, Clone, Default)] #[derive(Debug, Clone, PartialEq, Eq)]
pub struct EncryptedDeviceSecretQuery { pub struct PendingChatSecretForward {
pub user_id: String, pub recipient_user_id: String,
pub device_id: Option<String>, pub chat_id: String,
pub secret_id: Option<String>, pub sender_user_id: String,
pub secret_id: String,
pub version: i64,
pub encrypted_secret: Vec<u8>,
pub kem_ciphertext: Vec<u8>,
pub wrapping_scheme: String,
pub created_at: i64,
} }
static E2EE_DB: LazyLock<Arc<Mutex<rusqlite::Connection>>> = LazyLock::new(|| { static E2EE_DB: LazyLock<Arc<Mutex<rusqlite::Connection>>> = LazyLock::new(|| {
@ -54,163 +44,68 @@ static E2EE_DB: LazyLock<Arc<Mutex<rusqlite::Connection>>> = LazyLock::new(|| {
PRAGMA journal_mode = WAL; PRAGMA journal_mode = WAL;
PRAGMA synchronous = NORMAL; PRAGMA synchronous = NORMAL;
CREATE TABLE IF NOT EXISTS encrypted_messages ( DROP TABLE IF EXISTS encrypted_messages;
message_id TEXT NOT NULL PRIMARY KEY, DROP TABLE IF EXISTS encrypted_device_secrets;
conversation_id TEXT NOT NULL,
sender_client_id TEXT NOT NULL,
recipient_client_id TEXT NOT NULL,
sender_user_id TEXT,
recipient_user_id TEXT,
created_at INTEGER NOT NULL,
encryption_version INTEGER NOT NULL,
encrypted_payload BLOB NOT NULL,
acked_at INTEGER
);
CREATE INDEX IF NOT EXISTS idx_encrypted_messages_recipient CREATE TABLE IF NOT EXISTS chat_secrets (
ON encrypted_messages (recipient_client_id, created_at ASC);
CREATE INDEX IF NOT EXISTS idx_encrypted_messages_recipient_user
ON encrypted_messages (recipient_user_id, created_at ASC);
CREATE INDEX IF NOT EXISTS idx_encrypted_messages_conversation
ON encrypted_messages (conversation_id, created_at ASC);
CREATE TABLE IF NOT EXISTS encrypted_device_secrets (
user_id TEXT NOT NULL, user_id TEXT NOT NULL,
device_id TEXT NOT NULL, chat_id TEXT NOT NULL,
secret_id TEXT NOT NULL, secret_id TEXT NOT NULL,
version INTEGER NOT NULL, version INTEGER NOT NULL,
encrypted_secret BLOB NOT NULL, encrypted_secret BLOB NOT NULL,
wrapping_public_key_id TEXT, kem_ciphertext BLOB NOT NULL,
wrapping_scheme TEXT NOT NULL, wrapping_scheme TEXT NOT NULL,
created_at INTEGER NOT NULL, created_at INTEGER NOT NULL,
updated_at INTEGER NOT NULL, updated_at INTEGER NOT NULL,
PRIMARY KEY (user_id, device_id, secret_id) PRIMARY KEY (user_id, chat_id, secret_id)
); );
CREATE INDEX IF NOT EXISTS idx_encrypted_device_secrets_owner CREATE INDEX IF NOT EXISTS idx_chat_secrets_owner
ON encrypted_device_secrets (user_id, device_id, secret_id); ON chat_secrets (user_id, chat_id, secret_id);
CREATE TABLE IF NOT EXISTS pending_chat_secret_forwards (
recipient_user_id TEXT NOT NULL,
chat_id TEXT NOT NULL,
sender_user_id TEXT NOT NULL,
secret_id TEXT NOT NULL,
version INTEGER NOT NULL,
encrypted_secret BLOB NOT NULL,
kem_ciphertext BLOB NOT NULL,
wrapping_scheme TEXT NOT NULL,
created_at INTEGER NOT NULL,
PRIMARY KEY (recipient_user_id, chat_id, secret_id)
);
CREATE INDEX IF NOT EXISTS idx_pending_chat_secret_forwards_recipient
ON pending_chat_secret_forwards (recipient_user_id, created_at);
"#, "#,
) )
.expect("Failed to create or initialize E2EE DB") .expect("Failed to create or initialize E2EE DB")
}); });
pub fn put_encrypted_message(record: StoredEncryptedMessage) -> Result<(), StorageError> { pub fn put_chat_secret(record: StoredChatSecret) -> Result<(), StorageError> {
db::with_conn(&E2EE_DB, |conn| { db::with_conn(&E2EE_DB, |conn| {
conn.execute( conn.execute(
r#" r#"
INSERT OR REPLACE INTO encrypted_messages ( INSERT INTO chat_secrets (
message_id, conversation_id, sender_client_id, recipient_client_id, user_id, chat_id, secret_id, version, encrypted_secret,
sender_user_id, recipient_user_id, created_at, encryption_version, kem_ciphertext, wrapping_scheme, created_at, updated_at
encrypted_payload
) VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8, ?9) ) VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8, ?9)
"#, ON CONFLICT(user_id, chat_id, secret_id) DO UPDATE SET
params![
record.message_id,
record.conversation_id,
record.sender_client_id,
record.recipient_client_id,
record.sender_user_id,
record.recipient_user_id,
record.created_at,
record.encryption_version,
record.encrypted_payload,
],
)?;
Ok(())
})
}
pub fn get_encrypted_messages(
query: EncryptedMessageQuery,
) -> Result<Vec<StoredEncryptedMessage>, StorageError> {
let limit = query.limit.unwrap_or(100).clamp(1, 500);
let offset = query.offset.unwrap_or(0).max(0);
db::with_conn(&E2EE_DB, |conn| {
let mut stmt = conn.prepare(
r#"
SELECT message_id, conversation_id, sender_client_id, recipient_client_id,
sender_user_id, recipient_user_id, created_at, encryption_version,
encrypted_payload
FROM encrypted_messages
WHERE (?2 IS NULL OR recipient_client_id = ?2)
AND (
(?1 IS NULL AND (?3 IS NULL OR sender_user_id = ?3 OR recipient_user_id = ?3))
OR (?1 IS NOT NULL AND ?3 IS NOT NULL AND (
(sender_user_id = ?1 AND recipient_user_id = ?3)
OR (sender_user_id = ?3 AND recipient_user_id = ?1)
))
OR (?1 IS NOT NULL AND ?3 IS NULL AND (sender_user_id = ?1 OR recipient_user_id = ?1))
)
AND (?4 IS NULL OR conversation_id = ?4)
ORDER BY created_at ASC
LIMIT ?5 OFFSET ?6
"#,
)?;
let rows = stmt.query_map(
params![
query.sender_user_id,
query.recipient_client_id,
query.recipient_user_id,
query.conversation_id,
limit,
offset,
],
encrypted_message_from_row,
)?;
let mut out = Vec::new();
for row in rows {
out.push(row?);
}
Ok(out)
})
}
pub fn ack_encrypted_message(
message_id: &str,
recipient_client_id: &str,
) -> Result<(), StorageError> {
let now = now_millis();
db::with_conn(&E2EE_DB, |conn| {
conn.execute(
r#"
UPDATE encrypted_messages
SET acked_at = ?1
WHERE message_id = ?2 AND recipient_client_id = ?3
"#,
params![now, message_id, recipient_client_id],
)?;
Ok(())
})
}
pub fn put_encrypted_device_secret(
record: StoredEncryptedDeviceSecret,
) -> Result<(), StorageError> {
db::with_conn(&E2EE_DB, |conn| {
conn.execute(
r#"
INSERT INTO encrypted_device_secrets (
user_id, device_id, secret_id, version, encrypted_secret,
wrapping_public_key_id, wrapping_scheme, created_at, updated_at
) VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8, ?9)
ON CONFLICT(user_id, device_id, secret_id) DO UPDATE SET
version = excluded.version, version = excluded.version,
encrypted_secret = excluded.encrypted_secret, encrypted_secret = excluded.encrypted_secret,
wrapping_public_key_id = excluded.wrapping_public_key_id, kem_ciphertext = excluded.kem_ciphertext,
wrapping_scheme = excluded.wrapping_scheme, wrapping_scheme = excluded.wrapping_scheme,
created_at = excluded.created_at, created_at = excluded.created_at,
updated_at = excluded.updated_at updated_at = excluded.updated_at
"#, "#,
params![ params![
record.user_id, record.user_id,
record.device_id, record.chat_id,
record.secret_id, record.secret_id,
record.version, record.version,
record.encrypted_secret, record.encrypted_secret,
record.wrapping_public_key_id, record.kem_ciphertext,
record.wrapping_scheme, record.wrapping_scheme,
record.created_at, record.created_at,
record.updated_at, record.updated_at,
@ -220,68 +115,131 @@ pub fn put_encrypted_device_secret(
}) })
} }
pub fn get_encrypted_device_secret( pub fn put_pending_chat_secret_forward(
query: EncryptedDeviceSecretQuery, record: PendingChatSecretForward,
) -> Result<Option<StoredEncryptedDeviceSecret>, StorageError> { ) -> Result<(), StorageError> {
if query.user_id.is_empty() { db::with_conn(&E2EE_DB, |conn| {
conn.execute(
r#"
INSERT INTO pending_chat_secret_forwards (
recipient_user_id, chat_id, sender_user_id, secret_id, version,
encrypted_secret, kem_ciphertext, wrapping_scheme, created_at
) VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8, ?9)
ON CONFLICT(recipient_user_id, chat_id, secret_id) DO UPDATE SET
sender_user_id = excluded.sender_user_id,
version = excluded.version,
encrypted_secret = excluded.encrypted_secret,
kem_ciphertext = excluded.kem_ciphertext,
wrapping_scheme = excluded.wrapping_scheme,
created_at = excluded.created_at
"#,
params![
record.recipient_user_id,
record.chat_id,
record.sender_user_id,
record.secret_id,
record.version,
record.encrypted_secret,
record.kem_ciphertext,
record.wrapping_scheme,
record.created_at,
],
)?;
Ok(())
})
}
pub fn get_pending_chat_secret_forwards(
limit: i64,
) -> Result<Vec<PendingChatSecretForward>, StorageError> {
db::with_conn(&E2EE_DB, |conn| {
let mut stmt = conn.prepare(
r#"
SELECT recipient_user_id, chat_id, sender_user_id, secret_id, version,
encrypted_secret, kem_ciphertext, wrapping_scheme, created_at
FROM pending_chat_secret_forwards
ORDER BY created_at ASC
LIMIT ?1
"#,
)?;
let rows = stmt.query_map(params![limit.clamp(1, 500)], pending_forward_from_row)?;
let mut out = Vec::new();
for row in rows {
out.push(row?);
}
Ok(out)
})
}
pub fn delete_pending_chat_secret_forward(
recipient_user_id: &str,
chat_id: &str,
secret_id: &str,
) -> Result<(), StorageError> {
db::with_conn(&E2EE_DB, |conn| {
conn.execute(
r#"
DELETE FROM pending_chat_secret_forwards
WHERE recipient_user_id = ?1 AND chat_id = ?2 AND secret_id = ?3
"#,
params![recipient_user_id, chat_id, secret_id],
)?;
Ok(())
})
}
pub fn get_chat_secret(query: ChatSecretQuery) -> Result<Option<StoredChatSecret>, StorageError> {
if query.user_id.is_empty() || query.chat_id.is_empty() {
return Ok(None); return Ok(None);
} }
db::with_conn(&E2EE_DB, |conn| { db::with_conn(&E2EE_DB, |conn| {
conn.query_row( conn.query_row(
r#" r#"
SELECT user_id, device_id, secret_id, version, encrypted_secret, SELECT user_id, chat_id, secret_id, version, encrypted_secret,
wrapping_public_key_id, wrapping_scheme, created_at, updated_at kem_ciphertext, wrapping_scheme, created_at, updated_at
FROM encrypted_device_secrets FROM chat_secrets
WHERE user_id = ?1 WHERE user_id = ?1
AND (?2 IS NULL OR device_id = ?2) AND chat_id = ?2
AND (?3 IS NULL OR secret_id = ?3) AND (?3 IS NULL OR secret_id = ?3)
ORDER BY updated_at DESC ORDER BY updated_at DESC
LIMIT 1 LIMIT 1
"#, "#,
params![query.user_id, query.device_id, query.secret_id], params![query.user_id, query.chat_id, query.secret_id],
encrypted_device_secret_from_row, chat_secret_from_row,
) )
.optional() .optional()
}) })
} }
fn encrypted_message_from_row(row: &rusqlite::Row<'_>) -> rusqlite::Result<StoredEncryptedMessage> { fn chat_secret_from_row(row: &rusqlite::Row<'_>) -> rusqlite::Result<StoredChatSecret> {
Ok(StoredEncryptedMessage { Ok(StoredChatSecret {
message_id: row.get(0)?,
conversation_id: row.get(1)?,
sender_client_id: row.get(2)?,
recipient_client_id: row.get(3)?,
sender_user_id: row.get(4)?,
recipient_user_id: row.get(5)?,
created_at: row.get(6)?,
encryption_version: row.get(7)?,
encrypted_payload: row.get(8)?,
})
}
fn encrypted_device_secret_from_row(
row: &rusqlite::Row<'_>,
) -> rusqlite::Result<StoredEncryptedDeviceSecret> {
Ok(StoredEncryptedDeviceSecret {
user_id: row.get(0)?, user_id: row.get(0)?,
device_id: row.get(1)?, chat_id: row.get(1)?,
secret_id: row.get(2)?, secret_id: row.get(2)?,
version: row.get(3)?, version: row.get(3)?,
encrypted_secret: row.get(4)?, encrypted_secret: row.get(4)?,
wrapping_public_key_id: row.get(5)?, kem_ciphertext: row.get(5)?,
wrapping_scheme: row.get(6)?, wrapping_scheme: row.get(6)?,
created_at: row.get(7)?, created_at: row.get(7)?,
updated_at: row.get(8)?, updated_at: row.get(8)?,
}) })
} }
fn now_millis() -> i64 { fn pending_forward_from_row(
use std::time::{SystemTime, UNIX_EPOCH}; row: &rusqlite::Row<'_>,
SystemTime::now() ) -> rusqlite::Result<PendingChatSecretForward> {
.duration_since(UNIX_EPOCH) Ok(PendingChatSecretForward {
.unwrap_or_default() recipient_user_id: row.get(0)?,
.as_millis() as i64 chat_id: row.get(1)?,
sender_user_id: row.get(2)?,
secret_id: row.get(3)?,
version: row.get(4)?,
encrypted_secret: row.get(5)?,
kem_ciphertext: row.get(6)?,
wrapping_scheme: row.get(7)?,
created_at: row.get(8)?,
})
} }
#[cfg(test)] #[cfg(test)]
@ -289,120 +247,44 @@ mod tests {
use super::*; use super::*;
fn suffix(name: &str) -> String { fn suffix(name: &str) -> String {
format!("{name}-{}", now_millis()) let unix = std::time::SystemTime::now()
.duration_since(std::time::UNIX_EPOCH)
.unwrap_or_default()
.as_nanos();
format!("{name}-{unix}")
} }
#[test] #[test]
fn stores_encrypted_message_as_opaque_ciphertext() { fn stores_and_retrieves_chat_secret_blob_for_owner_chat() {
let message_id = suffix("msg");
put_encrypted_message(StoredEncryptedMessage {
message_id: message_id.clone(),
conversation_id: suffix("conversation"),
sender_client_id: "sender".to_string(),
recipient_client_id: "recipient".to_string(),
sender_user_id: Some("sender-user".to_string()),
recipient_user_id: Some(suffix("recipient-user")),
created_at: now_millis(),
encryption_version: 1,
encrypted_payload: vec![9, 8, 7],
})
.unwrap();
let messages = get_encrypted_messages(EncryptedMessageQuery {
recipient_client_id: Some("recipient".to_string()),
limit: Some(10),
..Default::default()
})
.unwrap();
assert!(messages.iter().any(|m| m.message_id == message_id && m.encrypted_payload == vec![9, 8, 7]));
}
#[test]
fn encrypted_history_query_returns_both_directions_for_peer() {
let user = suffix("user");
let peer = suffix("peer");
let incoming_id = suffix("incoming");
let outgoing_id = suffix("outgoing");
let other_id = suffix("other");
put_encrypted_message(StoredEncryptedMessage {
message_id: incoming_id.clone(),
conversation_id: suffix("conversation"),
sender_client_id: "peer-client".to_string(),
recipient_client_id: "user-client".to_string(),
sender_user_id: Some(peer.clone()),
recipient_user_id: Some(user.clone()),
created_at: now_millis(),
encryption_version: 1,
encrypted_payload: vec![1],
})
.unwrap();
put_encrypted_message(StoredEncryptedMessage {
message_id: outgoing_id.clone(),
conversation_id: suffix("conversation"),
sender_client_id: "user-client".to_string(),
recipient_client_id: "peer-client".to_string(),
sender_user_id: Some(user.clone()),
recipient_user_id: Some(peer.clone()),
created_at: now_millis() + 1,
encryption_version: 1,
encrypted_payload: vec![2],
})
.unwrap();
put_encrypted_message(StoredEncryptedMessage {
message_id: other_id.clone(),
conversation_id: suffix("conversation"),
sender_client_id: "other-client".to_string(),
recipient_client_id: "user-client".to_string(),
sender_user_id: Some(suffix("other")),
recipient_user_id: Some(user.clone()),
created_at: now_millis() + 2,
encryption_version: 1,
encrypted_payload: vec![3],
})
.unwrap();
let messages = get_encrypted_messages(EncryptedMessageQuery {
sender_user_id: Some(peer),
recipient_user_id: Some(user),
limit: Some(10),
..Default::default()
})
.unwrap();
assert!(messages.iter().any(|m| m.message_id == incoming_id));
assert!(messages.iter().any(|m| m.message_id == outgoing_id));
assert!(!messages.iter().any(|m| m.message_id == other_id));
}
#[test]
fn stores_and_retrieves_encrypted_device_secret_blob_for_owner_only() {
let user_id = suffix("user"); let user_id = suffix("user");
put_encrypted_device_secret(StoredEncryptedDeviceSecret { let chat_id = suffix("chat");
put_chat_secret(StoredChatSecret {
user_id: user_id.clone(), user_id: user_id.clone(),
device_id: "device".to_string(), chat_id: chat_id.clone(),
secret_id: "main".to_string(), secret_id: "main".to_string(),
version: 1, version: 1,
encrypted_secret: vec![42, 43], encrypted_secret: vec![42, 43],
wrapping_public_key_id: None, kem_ciphertext: vec![9, 8, 7],
wrapping_scheme: "mtp-chacha20poly1305-hkdf-sha256-v1".to_string(), wrapping_scheme: "mtp-kem-chacha20poly1305-hkdf-sha256-v1".to_string(),
created_at: 1, created_at: 1,
updated_at: 2, updated_at: 2,
}) })
.unwrap(); .unwrap();
let found = get_encrypted_device_secret(EncryptedDeviceSecretQuery { let found = get_chat_secret(ChatSecretQuery {
user_id: user_id.clone(), user_id: user_id.clone(),
device_id: Some("device".to_string()), chat_id: chat_id.clone(),
secret_id: Some("main".to_string()), secret_id: Some("main".to_string()),
}) })
.unwrap() .unwrap()
.unwrap(); .unwrap();
assert_eq!(found.encrypted_secret, vec![42, 43]); assert_eq!(found.encrypted_secret, vec![42, 43]);
assert_eq!(found.kem_ciphertext, vec![9, 8, 7]);
let denied = get_encrypted_device_secret(EncryptedDeviceSecretQuery { let denied = get_chat_secret(ChatSecretQuery {
user_id: suffix("other-user"), user_id: suffix("other-user"),
device_id: Some("device".to_string()), chat_id,
secret_id: Some("main".to_string()), secret_id: Some("main".to_string()),
}) })
.unwrap(); .unwrap();

469
omikron-connector/src/omikron_connection.rs
View file

@ -5,10 +5,9 @@ use iota_storage::users::contact::Contact;
use iota_storage::util::chat_files::{self, MessageState, change_message_state}; use iota_storage::util::chat_files::{self, MessageState, change_message_state};
use iota_storage::util::chats_util::{self, get_user, mod_user}; use iota_storage::util::chats_util::{self, get_user, mod_user};
use iota_storage::util::communities_util::CommunitiesUtil; use iota_storage::util::communities_util::CommunitiesUtil;
use iota_storage::util::config_util::{modify_config, CONFIG}; use iota_storage::util::config_util::{CONFIG, modify_config};
use iota_storage::util::e2ee_storage::{ use iota_storage::util::e2ee_storage::{
self, EncryptedDeviceSecretQuery, EncryptedMessageQuery, self, ChatSecretQuery, PendingChatSecretForward, StoredChatSecret,
StoredEncryptedDeviceSecret, StoredEncryptedMessage,
}; };
use iota_util::crypto_helper::{self, keyring_from_base64}; use iota_util::crypto_helper::{self, keyring_from_base64};
use iota_util::crypto_util::{self}; use iota_util::crypto_util::{self};
@ -16,6 +15,7 @@ use iota_util::file_util::{get_children, has_file, load_file, save_file};
use mtp::client::{Client, ClientConfig, Policy, Receiver, SendMode, Sender}; use mtp::client::{Client, ClientConfig, Policy, Receiver, SendMode, Sender};
use mtp::codec::{CommunicationType, CommunicationValue, DataType, DataValue}; use mtp::codec::{CommunicationType, CommunicationValue, DataType, DataValue};
use mtp::crypto::{Keyring, PublicKeyBundle}; use mtp::crypto::{Keyring, PublicKeyBundle};
use mtp::type_map::TypeMap;
use std::collections::HashMap; use std::collections::HashMap;
use std::env; use std::env;
use std::sync::{Arc, LazyLock}; use std::sync::{Arc, LazyLock};
@ -50,12 +50,102 @@ fn data_i64(cv: &CommunicationValue, dt: DataType) -> Option<i64> {
cv.get_data(dt) cv.get_data(dt)
.as_number() .as_number()
.and_then(|n| i64::try_from(n).ok()) .and_then(|n| i64::try_from(n).ok())
.or_else(|| cv.get_data(dt).as_signed_number().and_then(|n| i64::try_from(n).ok())) .or_else(|| {
cv.get_data(dt)
.as_signed_number()
.and_then(|n| i64::try_from(n).ok())
})
.or_else(|| cv.get_data(dt).as_str().and_then(|s| s.parse::<i64>().ok())) .or_else(|| cv.get_data(dt).as_str().and_then(|s| s.parse::<i64>().ok()))
} }
fn data_bytes(cv: &CommunicationValue, dt: DataType) -> Option<Vec<u8>> { #[derive(Debug, Clone)]
cv.get_bytes(dt).map(|bytes| bytes.to_vec()) struct ChatSecretRecipient {
user_id: String,
encrypted_secret: Vec<u8>,
kem_ciphertext: Vec<u8>,
}
fn recipient_from_value(value: &DataValue) -> Option<ChatSecretRecipient> {
let tm = TypeMap::latest();
let user_id = value
.get_field(DataType::UserId.to_id(&tm))?
.as_str()
.map(|s| s.to_string())
.or_else(|| {
value
.get_field(DataType::UserId.to_id(&tm))?
.as_number()
.map(|n| n.to_string())
})?;
let encrypted_secret = value
.get_field(DataType::EncryptedSecret.to_id(&tm))?
.as_bytes()?;
let kem_ciphertext = value
.get_field(DataType::KemCiphertext.to_id(&tm))?
.as_bytes()?;
Some(ChatSecretRecipient {
user_id,
encrypted_secret,
kem_ciphertext,
})
}
fn chat_secret_recipients(cv: &CommunicationValue) -> Option<Vec<ChatSecretRecipient>> {
let recipients = cv.get_data(DataType::Recipients).as_array()?;
let parsed = recipients
.iter()
.map(recipient_from_value)
.collect::<Option<Vec<_>>>()?;
if parsed.is_empty() {
None
} else {
Some(parsed)
}
}
fn set_chat_secret_cv_for_recipient(
source: &CommunicationValue,
recipient: &ChatSecretRecipient,
) -> CommunicationValue {
let recipient_value = typed_container(vec![
(DataType::UserId, DataValue::Str(recipient.user_id.clone())),
(
DataType::EncryptedSecret,
DataValue::Bytes(recipient.encrypted_secret.clone()),
),
(
DataType::KemCiphertext,
DataValue::Bytes(recipient.kem_ciphertext.clone()),
),
]);
CommunicationValue::new(CommunicationType::SetChatSecret)
.with_id(source.get_id())
.with_sender(source.get_sender())
.with_receiver(recipient.user_id.parse::<u64>().unwrap_or(0))
.add_typed_default(DataType::ChatId, source.get_data(DataType::ChatId).clone())
.add_typed_default(
DataType::SecretId,
source.get_data(DataType::SecretId).clone(),
)
.add_typed_default(
DataType::VersionNumber,
source.get_data(DataType::VersionNumber).clone(),
)
.add_typed_default(
DataType::WrappingScheme,
source.get_data(DataType::WrappingScheme).clone(),
)
.add_typed_default(
DataType::CreatedAt,
source.get_data(DataType::CreatedAt).clone(),
)
.add_typed_default(
DataType::Recipients,
DataValue::Array(vec![recipient_value]),
)
} }
fn now_millis_i64() -> i64 { fn now_millis_i64() -> i64 {
@ -65,6 +155,63 @@ fn now_millis_i64() -> i64 {
.as_millis() as i64 .as_millis() as i64
} }
fn pending_chat_secret_forward_from_cv(
cv: &CommunicationValue,
) -> Option<PendingChatSecretForward> {
let recipient = chat_secret_recipients(cv)?.into_iter().next()?;
Some(PendingChatSecretForward {
recipient_user_id: recipient.user_id,
chat_id: data_string(cv, DataType::ChatId)?,
sender_user_id: cv.get_sender().to_string(),
secret_id: data_string(cv, DataType::SecretId)?,
version: data_i64(cv, DataType::VersionNumber)?,
encrypted_secret: recipient.encrypted_secret,
kem_ciphertext: recipient.kem_ciphertext,
wrapping_scheme: data_string(cv, DataType::WrappingScheme)?,
created_at: data_i64(cv, DataType::CreatedAt).unwrap_or_else(now_millis_i64),
})
}
fn chat_secret_forward_cv(record: &PendingChatSecretForward) -> CommunicationValue {
let recipient = typed_container(vec![
(
DataType::UserId,
DataValue::Str(record.recipient_user_id.clone()),
),
(
DataType::EncryptedSecret,
DataValue::Bytes(record.encrypted_secret.clone()),
),
(
DataType::KemCiphertext,
DataValue::Bytes(record.kem_ciphertext.clone()),
),
]);
CommunicationValue::new(CommunicationType::SetChatSecret)
.with_sender(record.sender_user_id.parse::<u64>().unwrap_or(0))
.with_receiver(record.recipient_user_id.parse::<u64>().unwrap_or(0))
.add_typed_default(DataType::ChatId, DataValue::Str(record.chat_id.clone()))
.add_typed_default(
DataType::SenderUserId,
DataValue::Str(record.sender_user_id.clone()),
)
.add_typed_default(DataType::SecretId, DataValue::Str(record.secret_id.clone()))
.add_typed_default(
DataType::VersionNumber,
DataValue::SignedNumber(record.version as i128),
)
.add_typed_default(
DataType::WrappingScheme,
DataValue::Str(record.wrapping_scheme.clone()),
)
.add_typed_default(
DataType::CreatedAt,
DataValue::SignedNumber(record.created_at as i128),
)
.add_typed_default(DataType::Recipients, DataValue::Array(vec![recipient]))
}
fn error_response(request: &CommunicationValue, ty: CommunicationType) -> CommunicationValue { fn error_response(request: &CommunicationValue, ty: CommunicationType) -> CommunicationValue {
CommunicationValue::new(ty) CommunicationValue::new(ty)
.with_id(request.get_id()) .with_id(request.get_id())
@ -563,10 +710,50 @@ impl OmikronConnection {
break; break;
} }
self.flush_pending_chat_secret_forwards().await;
self.send_ping().await; self.send_ping().await;
} }
} }
async fn forward_chat_secret(&self, cv: &CommunicationValue) -> bool {
self.await_response(cv, Some(Duration::from_secs(10)))
.await
.is_ok()
}
async fn store_pending_chat_secret_forward(&self, cv: &CommunicationValue) {
let Some(record) = pending_chat_secret_forward_from_cv(cv) else {
return;
};
let _ = e2ee_storage::put_pending_chat_secret_forward(record);
}
async fn flush_pending_chat_secret_forwards(&self) {
let Ok(records) = e2ee_storage::get_pending_chat_secret_forwards(100) else {
return;
};
for record in records {
let Ok(recipient) = record.recipient_user_id.parse::<u64>() else {
let _ = e2ee_storage::delete_pending_chat_secret_forward(
&record.recipient_user_id,
&record.chat_id,
&record.secret_id,
);
continue;
};
let message = chat_secret_forward_cv(&record).with_receiver(recipient);
if self.forward_chat_secret(&message).await {
let _ = e2ee_storage::delete_pending_chat_secret_forward(
&record.recipient_user_id,
&record.chat_id,
&record.secret_id,
);
}
}
}
// ------------------------------------------------------------------------- // -------------------------------------------------------------------------
// Message Handling (Preserved from original) // Message Handling (Preserved from original)
// ------------------------------------------------------------------------- // -------------------------------------------------------------------------
@ -590,187 +777,161 @@ impl OmikronConnection {
return; return;
} }
if cv.is_type(CommunicationType::SetEncryptedDeviceSecret) { if cv.is_type(CommunicationType::SetChatSecret) {
let sender_id = cv.get_sender().to_string(); let sender_id = cv.get_sender().to_string();
if data_string(&cv, DataType::UserId).as_deref() != Some(sender_id.as_str()) { let recipients = match chat_secret_recipients(&cv) {
self.send_message(&error_response(&cv, CommunicationType::ErrorInvalidData)).await; Some(recipients) => recipients,
return; None => {
} self.send_message(&error_response(&cv, CommunicationType::ErrorInvalidData))
.await;
return;
}
};
let now = now_millis_i64(); let now = now_millis_i64();
let record = data_string(&cv, DataType::UserId) let chat_id = data_string(&cv, DataType::ChatId);
.zip(data_string(&cv, DataType::DeviceId)) let secret_id = data_string(&cv, DataType::SecretId);
.zip(data_string(&cv, DataType::SecretId)) let version = data_i64(&cv, DataType::VersionNumber);
.zip(data_i64(&cv, DataType::VersionNumber)) let wrapping_scheme = data_string(&cv, DataType::WrappingScheme);
.zip(data_bytes(&cv, DataType::EncryptedSecret)) let created_at = data_i64(&cv, DataType::CreatedAt).unwrap_or(now);
.zip(data_string(&cv, DataType::WrappingScheme))
.map(|(((((user_id, device_id), secret_id), version), encrypted_secret), wrapping_scheme)| {
StoredEncryptedDeviceSecret {
user_id,
device_id,
secret_id,
version,
encrypted_secret,
wrapping_public_key_id: data_string(&cv, DataType::WrappingPublicKeyId),
wrapping_scheme,
created_at: data_i64(&cv, DataType::CreatedAt).unwrap_or(now),
updated_at: now,
}
});
match record.map(e2ee_storage::put_encrypted_device_secret) { let Some((((chat_id, secret_id), version), wrapping_scheme)) =
Some(Ok(())) => self.send_message(&error_response(&cv, CommunicationType::Success)).await, chat_id.zip(secret_id).zip(version).zip(wrapping_scheme)
_ => self.send_message(&error_response(&cv, CommunicationType::ErrorInvalidData)).await, else {
self.send_message(&error_response(&cv, CommunicationType::ErrorInvalidData))
.await;
return;
};
for recipient in &recipients {
let recipient_id = recipient.user_id.parse::<i64>().unwrap_or(0);
let is_local = iota_storage::users::user_manager::get_user(recipient_id).is_some();
if is_local {
if e2ee_storage::put_chat_secret(StoredChatSecret {
user_id: recipient.user_id.clone(),
chat_id: chat_id.clone(),
secret_id: secret_id.clone(),
version,
encrypted_secret: recipient.encrypted_secret.clone(),
kem_ciphertext: recipient.kem_ciphertext.clone(),
wrapping_scheme: wrapping_scheme.clone(),
created_at,
updated_at: now,
})
.is_err()
{
self.send_message(&error_response(
&cv,
CommunicationType::ErrorInvalidData,
))
.await;
return;
}
continue;
}
if recipient.user_id != sender_id {
let forward = set_chat_secret_cv_for_recipient(&cv, recipient);
if !self.forward_chat_secret(&forward).await {
self.store_pending_chat_secret_forward(&forward).await;
}
}
} }
self.send_message(&error_response(&cv, CommunicationType::Success))
.await;
return; return;
} }
if cv.is_type(CommunicationType::GetEncryptedDeviceSecret) { if cv.is_type(CommunicationType::GetChatSecret) {
let Some(user_id) = data_string(&cv, DataType::UserId) else { let Some(user_id) = data_string(&cv, DataType::UserId) else {
self.send_message(&error_response(&cv, CommunicationType::ErrorInvalidData)).await; self.send_message(&error_response(&cv, CommunicationType::ErrorInvalidData))
.await;
return; return;
}; };
if user_id != cv.get_sender().to_string() { if user_id != cv.get_sender().to_string() {
self.send_message(&error_response(&cv, CommunicationType::ErrorNotFound)).await; self.send_message(&error_response(&cv, CommunicationType::ErrorNotFound))
.await;
return; return;
} }
let Some(chat_id) = data_string(&cv, DataType::ChatId) else {
self.send_message(&error_response(&cv, CommunicationType::ErrorInvalidData))
.await;
return;
};
match e2ee_storage::get_encrypted_device_secret(EncryptedDeviceSecretQuery { match e2ee_storage::get_chat_secret(ChatSecretQuery {
user_id, user_id,
device_id: data_string(&cv, DataType::DeviceId), chat_id,
secret_id: data_string(&cv, DataType::SecretId), secret_id: data_string(&cv, DataType::SecretId),
}) { }) {
Ok(Some(record)) => { Ok(Some(record)) => {
let mut response = CommunicationValue::new(CommunicationType::EncryptedDeviceSecretResponse) let response = CommunicationValue::new(CommunicationType::ChatSecretResponse)
.with_id(cv.get_id()) .with_id(cv.get_id())
.with_receiver(cv.get_sender()) .with_receiver(cv.get_sender())
.add_typed_default(DataType::UserId, DataValue::Str(record.user_id)) .add_typed_default(DataType::UserId, DataValue::Str(record.user_id))
.add_typed_default(DataType::DeviceId, DataValue::Str(record.device_id)) .add_typed_default(DataType::ChatId, DataValue::Str(record.chat_id))
.add_typed_default(DataType::SecretId, DataValue::Str(record.secret_id)) .add_typed_default(DataType::SecretId, DataValue::Str(record.secret_id))
.add_typed_default(DataType::VersionNumber, DataValue::SignedNumber(record.version as i128)) .add_typed_default(
.add_typed_default(DataType::EncryptedSecret, DataValue::Bytes(record.encrypted_secret)) DataType::VersionNumber,
.add_typed_default(DataType::WrappingScheme, DataValue::Str(record.wrapping_scheme)) DataValue::SignedNumber(record.version as i128),
.add_typed_default(DataType::CreatedAt, DataValue::SignedNumber(record.created_at as i128)) )
.add_typed_default(DataType::UpdatedAt, DataValue::SignedNumber(record.updated_at as i128)); .add_typed_default(
if let Some(value) = record.wrapping_public_key_id { DataType::EncryptedSecret,
response = response.add_typed_default(DataType::WrappingPublicKeyId, DataValue::Str(value)); DataValue::Bytes(record.encrypted_secret),
} )
.add_typed_default(
DataType::KemCiphertext,
DataValue::Bytes(record.kem_ciphertext),
)
.add_typed_default(
DataType::WrappingScheme,
DataValue::Str(record.wrapping_scheme),
)
.add_typed_default(
DataType::CreatedAt,
DataValue::SignedNumber(record.created_at as i128),
)
.add_typed_default(
DataType::UpdatedAt,
DataValue::SignedNumber(record.updated_at as i128),
);
self.send_message(&response).await; self.send_message(&response).await;
} }
Ok(None) => self.send_message(&error_response(&cv, CommunicationType::ErrorNotFound)).await, Ok(None) => {
Err(_) => self.send_message(&error_response(&cv, CommunicationType::ErrorInvalidData)).await, self.send_message(&error_response(&cv, CommunicationType::ErrorNotSet))
.await
}
Err(_) => {
self.send_message(&error_response(&cv, CommunicationType::ErrorInvalidData))
.await
}
} }
return; return;
} }
if cv.is_type(CommunicationType::EncryptedMessage) { if cv.is_type(CommunicationType::ChatSecretForward) {
let sender_id = cv.get_sender().to_string(); let sender_id = cv.get_sender().to_string();
if data_string(&cv, DataType::SenderUserId).as_deref() != Some(sender_id.as_str()) { let recipient_user_id = data_string(&cv, DataType::RecipientUserId).unwrap_or_default();
self.send_message(&error_response(&cv, CommunicationType::ErrorInvalidData)).await; if data_string(&cv, DataType::SenderUserId).as_deref() != Some(sender_id.as_str())
|| recipient_user_id.is_empty()
|| pending_chat_secret_forward_from_cv(&cv).is_none()
{
self.send_message(&error_response(&cv, CommunicationType::ErrorInvalidData))
.await;
return; return;
} }
let record = data_string(&cv, DataType::MessageId)
.zip(data_string(&cv, DataType::ConversationId))
.zip(data_string(&cv, DataType::SenderClientId))
.zip(data_string(&cv, DataType::RecipientClientId))
.zip(data_bytes(&cv, DataType::EncryptedPayload))
.map(|((((message_id, conversation_id), sender_client_id), recipient_client_id), encrypted_payload)| {
StoredEncryptedMessage {
message_id,
conversation_id,
sender_client_id,
recipient_client_id,
sender_user_id: data_string(&cv, DataType::SenderUserId),
recipient_user_id: data_string(&cv, DataType::RecipientUserId),
created_at: data_i64(&cv, DataType::CreatedAt).unwrap_or_else(now_millis_i64),
encryption_version: data_i64(&cv, DataType::EncryptionVersion).unwrap_or(1),
encrypted_payload,
}
});
if let Some(record) = record { let forward = cv
let message_id = record.message_id.clone(); .clone()
let conversation_id = record.conversation_id.clone(); .with_receiver(recipient_user_id.parse::<u64>().unwrap_or(0));
let recipient_client_id = record.recipient_client_id.clone(); if self.forward_chat_secret(&forward).await {
let sender_client_id = record.sender_client_id.clone(); self.send_message(&error_response(&cv, CommunicationType::Success))
let sender_user_id = record.sender_user_id.clone().unwrap_or_default(); .await;
let recipient_user_id = record.recipient_user_id.clone().unwrap_or_default();
match e2ee_storage::put_encrypted_message(record) {
Ok(()) => {
self.send_message(&CommunicationValue::new(CommunicationType::EncryptedMessageAck)
.with_id(cv.get_id())
.with_receiver(cv.get_sender())
.add_typed_default(DataType::MessageId, DataValue::Str(message_id))
.add_typed_default(DataType::ConversationId, DataValue::Str(conversation_id))
.add_typed_default(DataType::RecipientClientId, DataValue::Str(recipient_client_id.clone()))
.add_typed_default(DataType::GetTime, DataValue::SignedNumber(now_millis_i64() as i128))
).await;
if !recipient_user_id.is_empty()
&& recipient_user_id != sender_user_id
&& recipient_client_id != sender_client_id
{
if cv.get_receiver().to_string() != recipient_user_id {
self.send_message(&cv.clone().with_receiver(recipient_user_id.parse::<u64>().unwrap_or(0))).await;
} else {
self.send_message(&cv).await;
}
}
}
Err(_) => self.send_message(&error_response(&cv, CommunicationType::ErrorInvalidData)).await,
}
} else { } else {
self.send_message(&error_response(&cv, CommunicationType::ErrorInvalidData)).await; self.store_pending_chat_secret_forward(&cv).await;
} self.send_message(&error_response(&cv, CommunicationType::Success))
return; .await;
}
if cv.is_type(CommunicationType::EncryptedMessagesGet) {
let requester_user_id = cv.get_sender().to_string();
let limit = data_i64(&cv, DataType::Limit).map(|v| v as i64);
let since = data_i64(&cv, DataType::Since);
let conversation_id = data_string(&cv, DataType::ConversationId);
let peer_client_id = data_string(&cv, DataType::PeerClientId);
match e2ee_storage::get_encrypted_messages(EncryptedMessageQuery {
sender_user_id: data_string(&cv, DataType::SenderUserId),
recipient_client_id: None,
recipient_user_id: Some(requester_user_id.clone()),
conversation_id,
limit,
offset: since.map(|v| v.max(0)),
}) {
Ok(records) => {
let messages = records
.into_iter()
.filter(|record| {
peer_client_id
.as_ref()
.map(|peer| &record.sender_client_id == peer || &record.recipient_client_id == peer)
.unwrap_or(true)
})
.map(|record| {
typed_container(vec![
(DataType::MessageId, DataValue::Str(record.message_id)),
(DataType::ConversationId, DataValue::Str(record.conversation_id)),
(DataType::SenderClientId, DataValue::Str(record.sender_client_id)),
(DataType::RecipientClientId, DataValue::Str(record.recipient_client_id)),
(DataType::SenderUserId, DataValue::Str(record.sender_user_id.unwrap_or_default())),
(DataType::RecipientUserId, DataValue::Str(record.recipient_user_id.unwrap_or_default())),
(DataType::CreatedAt, DataValue::SignedNumber(record.created_at as i128)),
(DataType::EncryptionVersion, DataValue::SignedNumber(record.encryption_version as i128)),
(DataType::EncryptedPayload, DataValue::Bytes(record.encrypted_payload)),
])
})
.collect::<Vec<_>>();
self.send_message(&CommunicationValue::new(CommunicationType::EncryptedMessagesResponse)
.with_id(cv.get_id())
.with_receiver(cv.get_sender())
.add_typed_default(DataType::Messages, DataValue::Array(messages))
.add_typed_default(DataType::HasMore, DataValue::Bool(false))
).await;
}
Err(_) => self.send_message(&error_response(&cv, CommunicationType::ErrorInvalidData)).await,
} }
return; return;
} }

26
type-maps.yaml
View file

@ -99,13 +99,11 @@ type_maps:
AppChallengeResponse: 133 AppChallengeResponse: 133
AppIdentificationResponse: 134 AppIdentificationResponse: 134
LoadTxtRecord: 135 LoadTxtRecord: 135
SetEncryptedDeviceSecret: 139 ErrorNotSet: 136
GetEncryptedDeviceSecret: 140 SetChatSecret: 139
EncryptedDeviceSecretResponse: 141 GetChatSecret: 140
EncryptedMessage: 142 ChatSecretResponse: 141
EncryptedMessageAck: 143 ChatSecretForward: 142
EncryptedMessagesGet: 144
EncryptedMessagesResponse: 145
DataTypes: DataTypes:
ErrorType: 32 ErrorType: 32
ErrorProtocol: 33 ErrorProtocol: 33
@ -213,18 +211,10 @@ type_maps:
SecretId: 142 SecretId: 142
VersionNumber: 143 VersionNumber: 143
EncryptedSecret: 144 EncryptedSecret: 144
WrappingPublicKeyId: 145
WrappingScheme: 146 WrappingScheme: 146
UpdatedAt: 147 UpdatedAt: 147
MessageId: 148 ChatId: 148
ConversationId: 149 KemCiphertext: 149
SenderClientId: 150
RecipientClientId: 151
SenderUserId: 152 SenderUserId: 152
RecipientUserId: 153 RecipientUserId: 153
EncryptionVersion: 154 Recipients: 154
HasMore: 155
NextCursor: 156
Since: 157
Limit: 158
PeerClientId: 159