mirror of
https://github.com/logos-messaging/libchat.git
synced 2026-07-09 09:29:39 +00:00
Remove test group type
This commit is contained in:
Jazz Turner-Baggs
parent
90c975634a
commit
05a64f10af
@ -1,5 +1,4 @@
|
||||
mod direct_v1;
|
||||
mod group_cent_info;
|
||||
pub mod group_v1;
|
||||
mod group_v2;
|
||||
pub mod mls_extensions;
|
||||
@ -11,7 +10,6 @@ use crate::proto::EncryptedPayload;
|
||||
use crate::service_context::{ExternalServices, ServiceContext};
|
||||
use crate::types::ConvoMetadata;
|
||||
pub use direct_v1::DirectV1Convo;
|
||||
pub use group_cent_info::GroupCentInfoConvo;
|
||||
pub use group_v1::GroupV1Convo;
|
||||
pub use group_v2::GroupV2Convo;
|
||||
pub use privatev1::PrivateV1Convo;
|
||||
|
||||
@ -1,371 +0,0 @@
|
||||
/// GroupV1 is a conversationType which provides effecient handling of multiple participants
|
||||
/// Properties:
|
||||
/// - Harvest Now Decrypt Later (HNDL) protection provided by XWING
|
||||
/// - Multiple
|
||||
use blake2::{Blake2b, Digest, digest::consts::U6};
|
||||
use chat_proto::logoschat::encryption::{EncryptedPayload, Plaintext, encrypted_payload};
|
||||
use chat_proto::logoschat::reliability::ReliablePayload;
|
||||
use openmls::prelude::tls_codec::Deserialize;
|
||||
use openmls::prelude::*;
|
||||
use prost::Message as _;
|
||||
use shared_traits::IdentIdRef;
|
||||
use std::collections::VecDeque;
|
||||
use tracing::debug;
|
||||
|
||||
use crate::account_directory::{AccountDirectory, resolve_device_ids};
|
||||
use crate::conversation::ConversationIdRef;
|
||||
use crate::inbox_v2::MlsProvider;
|
||||
use crate::service_context::{ExternalServices, ServiceContext};
|
||||
use crate::types::ConvoMetadata;
|
||||
use crate::utils::{blake2b_hex, hash_size};
|
||||
use crate::{
|
||||
DeliveryService, IdentityProvider,
|
||||
conversation::{ChatError, Convo, GroupConvo, Identified},
|
||||
outcomes::{Content, ConvoOutcome},
|
||||
service_traits::KeyPackageProvider,
|
||||
types::AddressedEncryptedPayload,
|
||||
};
|
||||
|
||||
use super::mls_extensions::{
|
||||
ConvoMetaInfo, GROUP_METADATA_EXTENSION_TYPE, capabilities_with_group_metadata,
|
||||
};
|
||||
|
||||
const OUTBOUND_HASH_CACHE_SIZE: usize = 25;
|
||||
|
||||
pub struct GroupCentInfoConvo {
|
||||
mls_group: MlsGroup,
|
||||
convo_id: String,
|
||||
// Cache outbound message Id's to filter out re-entrant messages
|
||||
outbound_msgs: VecDeque<String>,
|
||||
}
|
||||
|
||||
impl std::fmt::Debug for GroupCentInfoConvo {
|
||||
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
|
||||
f.debug_struct("GroupCentInfoConvo")
|
||||
.field("convo_id", &self.convo_id)
|
||||
.field("mls_epoch", &self.mls_group.epoch())
|
||||
.finish_non_exhaustive()
|
||||
}
|
||||
}
|
||||
|
||||
impl GroupCentInfoConvo {
|
||||
// Create a new conversation with the creator as the only participant.
|
||||
pub fn new<S: ExternalServices>(
|
||||
cx: &mut ServiceContext<S>,
|
||||
name: &str,
|
||||
desc: &str,
|
||||
) -> Result<Self, ChatError> {
|
||||
let config = Self::mls_create_config(cx, name, desc);
|
||||
let mls_group = MlsGroup::new(
|
||||
&cx.mls_provider,
|
||||
&cx.mls_identity,
|
||||
&config,
|
||||
cx.mls_identity.get_credential(),
|
||||
)
|
||||
.unwrap();
|
||||
let convo_id = hex::encode(mls_group.group_id().as_slice());
|
||||
Self::subscribe(&mut cx.ds, &convo_id)?;
|
||||
|
||||
Ok(Self {
|
||||
mls_group,
|
||||
convo_id,
|
||||
outbound_msgs: VecDeque::new(),
|
||||
})
|
||||
}
|
||||
|
||||
// Constructs a new conversation upon receiving a MlsWelcome message.
|
||||
pub fn new_from_welcome<S: ExternalServices>(
|
||||
cx: &mut ServiceContext<S>,
|
||||
welcome: Welcome,
|
||||
) -> Result<Self, ChatError> {
|
||||
let mls_group =
|
||||
StagedWelcome::build_from_welcome(&cx.mls_provider, &Self::mls_join_config(), welcome)
|
||||
.unwrap()
|
||||
.build()
|
||||
.unwrap()
|
||||
.into_group(&cx.mls_provider)
|
||||
.unwrap();
|
||||
|
||||
let convo_id = hex::encode(mls_group.group_id().as_slice());
|
||||
Self::subscribe(&mut cx.ds, &convo_id)?;
|
||||
|
||||
Ok(Self {
|
||||
mls_group,
|
||||
convo_id,
|
||||
outbound_msgs: VecDeque::new(),
|
||||
})
|
||||
}
|
||||
|
||||
// Configure the delivery service to listen for the required delivery addresses.
|
||||
fn subscribe(ds: &mut impl DeliveryService, convo_id: &str) -> Result<(), ChatError> {
|
||||
ds.subscribe(&Self::delivery_address_from_id(convo_id))
|
||||
.map_err(ChatError::generic)?;
|
||||
|
||||
Ok(())
|
||||
}
|
||||
|
||||
fn mls_create_config<S: ExternalServices>(
|
||||
cx: &mut ServiceContext<S>,
|
||||
name: &str,
|
||||
desc: &str,
|
||||
) -> MlsGroupCreateConfig {
|
||||
let meta = ConvoMetaInfo::new(name, cx.mls_identity.id(), desc);
|
||||
|
||||
let extensions = Extensions::from_vec(vec![Extension::Unknown(
|
||||
GROUP_METADATA_EXTENSION_TYPE,
|
||||
UnknownExtension(meta.to_extension_bytes()),
|
||||
)])
|
||||
.expect("failed to create extensions");
|
||||
|
||||
MlsGroupCreateConfig::builder()
|
||||
.ciphersuite(cx.mls_provider.crypto().supported_ciphersuites()[0])
|
||||
.capabilities(capabilities_with_group_metadata())
|
||||
.use_ratchet_tree_extension(true) // Embed the ratchet tree in the Welcome so joiners can build the group
|
||||
.with_group_context_extensions(extensions)
|
||||
.build()
|
||||
}
|
||||
|
||||
fn mls_join_config() -> MlsGroupJoinConfig {
|
||||
MlsGroupJoinConfig::builder().build()
|
||||
}
|
||||
|
||||
fn delivery_address_from_id(convo_id: &str) -> String {
|
||||
let hash = Blake2b::<U6>::new()
|
||||
.chain_update("delivery_addr|")
|
||||
.chain_update(convo_id)
|
||||
.finalize();
|
||||
hex::encode(hash)
|
||||
}
|
||||
|
||||
fn delivery_address(&self) -> String {
|
||||
Self::delivery_address_from_id(&self.convo_id)
|
||||
}
|
||||
|
||||
/// Resolve an account to a KeyPackage for *every* device it authorizes.
|
||||
///
|
||||
/// First resolves the account to its device ids through the account
|
||||
/// directory ([`resolve_device_ids`]), then fetches each device's
|
||||
/// KeyPackage. When the account never published a bundle, resolution falls
|
||||
/// back to a single device id equal to the account id — the pre-directory
|
||||
/// behaviour — so single-device accounts are unaffected.
|
||||
fn key_packages_for_account(
|
||||
&self,
|
||||
ident: IdentIdRef,
|
||||
provider: &impl MlsProvider,
|
||||
registry: &(impl KeyPackageProvider + AccountDirectory),
|
||||
) -> Result<Vec<KeyPackage>, ChatError> {
|
||||
let device_ids =
|
||||
resolve_device_ids(registry, ident).map_err(|e| ChatError::Generic(e.to_string()))?;
|
||||
|
||||
let mut keypackages = Vec::with_capacity(device_ids.len());
|
||||
for device_id in &device_ids {
|
||||
let retrieved = registry
|
||||
.retrieve(device_id)
|
||||
.map_err(|e| ChatError::Generic(e.to_string()))?;
|
||||
let Some(keypkg_bytes) = retrieved else {
|
||||
return Err(ChatError::Protocol(format!(
|
||||
"no keypackage for device {device_id} of account {ident}"
|
||||
)));
|
||||
};
|
||||
|
||||
let key_package_in = KeyPackageIn::tls_deserialize(&mut keypkg_bytes.as_slice())?;
|
||||
let keypkg = key_package_in.validate(provider.crypto(), ProtocolVersion::Mls10)?; //TODO: P3 - Hardcoded Protocol Version
|
||||
keypackages.push(keypkg);
|
||||
}
|
||||
Ok(keypackages)
|
||||
}
|
||||
|
||||
fn send_message<S: ExternalServices>(
|
||||
&mut self,
|
||||
content: &[u8],
|
||||
cx: &mut ServiceContext<S>,
|
||||
) -> Result<(), ChatError> {
|
||||
let sender_id = cx.mls_identity.id().as_str();
|
||||
let reliable = cx.causal.on_send(&self.convo_id, sender_id, content);
|
||||
let wire = reliable.encode_to_vec();
|
||||
|
||||
let mls_message_out = self
|
||||
.mls_group
|
||||
.create_message(&cx.mls_provider, &cx.mls_identity, &wire)
|
||||
.unwrap();
|
||||
|
||||
let msg_bytes = mls_message_out.to_bytes().unwrap();
|
||||
self.send_payload(cx, msg_bytes)
|
||||
}
|
||||
|
||||
// Publish outboubound payloads to the DeliveryService
|
||||
fn send_payload<S: ExternalServices>(
|
||||
&mut self,
|
||||
cx: &mut ServiceContext<S>,
|
||||
msg_bytes: Vec<u8>,
|
||||
) -> Result<(), ChatError> {
|
||||
// Hash and Cache to detect inbound messages
|
||||
let msg_hash = blake2b_hex::<hash_size::MessageId>(&[&msg_bytes]);
|
||||
self.outbound_msgs.push_back(msg_hash);
|
||||
if self.outbound_msgs.len() > OUTBOUND_HASH_CACHE_SIZE {
|
||||
let _ = self.outbound_msgs.remove(0);
|
||||
}
|
||||
|
||||
// Wrap in Payload frames
|
||||
let aep = AddressedEncryptedPayload {
|
||||
delivery_address: self.delivery_address(),
|
||||
data: EncryptedPayload {
|
||||
encryption: Some(encrypted_payload::Encryption::Plaintext(Plaintext {
|
||||
payload: msg_bytes.into(),
|
||||
})),
|
||||
},
|
||||
};
|
||||
let env = aep.into_envelope(self.convo_id.clone());
|
||||
|
||||
// Send via DS
|
||||
cx.ds
|
||||
.publish(env)
|
||||
.map_err(|e| ChatError::Delivery(e.to_string()))
|
||||
}
|
||||
}
|
||||
|
||||
impl Identified for GroupCentInfoConvo {
|
||||
fn id(&self) -> ConversationIdRef<'_> {
|
||||
&self.convo_id
|
||||
}
|
||||
}
|
||||
|
||||
impl<S: ExternalServices> Convo<S> for GroupCentInfoConvo {
|
||||
fn send_content(
|
||||
&mut self,
|
||||
cx: &mut ServiceContext<S>,
|
||||
content: &[u8],
|
||||
) -> Result<(), ChatError> {
|
||||
self.send_message(content, cx)
|
||||
}
|
||||
|
||||
fn handle_frame(
|
||||
&mut self,
|
||||
cx: &mut ServiceContext<S>,
|
||||
encoded_payload: EncryptedPayload,
|
||||
) -> Result<ConvoOutcome, ChatError> {
|
||||
let bytes = match encoded_payload.encryption {
|
||||
Some(encrypted_payload::Encryption::Plaintext(pt)) => pt.payload,
|
||||
_ => {
|
||||
return Err(ChatError::ProtocolExpectation(
|
||||
"None",
|
||||
"Some(Encryption::Plaintext)".into(),
|
||||
));
|
||||
}
|
||||
};
|
||||
|
||||
// Bail early if we sent this message
|
||||
let msg_hash = blake2b_hex::<hash_size::MessageId>(&[bytes.as_ref()]);
|
||||
if self.outbound_msgs.contains(&msg_hash) {
|
||||
debug!("Dropping message, sent from self");
|
||||
return Ok(ConvoOutcome::empty(self.convo_id.to_string()));
|
||||
}
|
||||
|
||||
let mls_message: MlsMessageIn =
|
||||
MlsMessageIn::tls_deserialize_exact_bytes(&bytes).map_err(ChatError::generic)?;
|
||||
|
||||
let protocol_message: ProtocolMessage = mls_message
|
||||
.try_into_protocol_message()
|
||||
.map_err(ChatError::generic)?;
|
||||
|
||||
if protocol_message.epoch() < self.mls_group.epoch() {
|
||||
// TODO: (P1) Add logging for messages arriving from past epoch.
|
||||
return Ok(ConvoOutcome::empty(self.id().to_string()));
|
||||
}
|
||||
|
||||
let processed = self
|
||||
.mls_group
|
||||
.process_message(&cx.mls_provider, protocol_message)
|
||||
.map_err(ChatError::generic)?;
|
||||
|
||||
let cred_bytes = processed.credential().serialized_content().to_vec();
|
||||
|
||||
let content = match processed.into_content() {
|
||||
ProcessedMessageContent::ApplicationMessage(msg) => {
|
||||
let reliable = ReliablePayload::decode(msg.into_bytes().as_slice())?;
|
||||
cx.causal.on_receive(&self.convo_id, &reliable);
|
||||
Some(Content {
|
||||
bytes: reliable.content.to_vec(),
|
||||
encoded_credential: cred_bytes,
|
||||
})
|
||||
}
|
||||
ProcessedMessageContent::StagedCommitMessage(commit) => {
|
||||
self.mls_group
|
||||
.merge_staged_commit(&cx.mls_provider, *commit)
|
||||
.map_err(ChatError::generic)?;
|
||||
None
|
||||
}
|
||||
_ => {
|
||||
// TODO: (P2) Log unknown message type
|
||||
None
|
||||
}
|
||||
};
|
||||
Ok(ConvoOutcome {
|
||||
convo_id: self.id().to_string(),
|
||||
content,
|
||||
})
|
||||
}
|
||||
|
||||
fn wakeup(&mut self, _: &mut ServiceContext<S>) -> Result<(), ChatError> {
|
||||
Ok(())
|
||||
}
|
||||
}
|
||||
|
||||
impl<S: ExternalServices> GroupConvo<S> for GroupCentInfoConvo {
|
||||
// add_members returns:
|
||||
// commit — the Commit message Alice broadcasts to all members
|
||||
// welcome — the Welcome message sent privately to each new joiner
|
||||
// _group_info — used for external joins; ignore for now
|
||||
fn add_member(
|
||||
&mut self,
|
||||
cx: &mut ServiceContext<S>,
|
||||
members: &[IdentIdRef],
|
||||
) -> Result<(), ChatError> {
|
||||
if members.len() > 50 {
|
||||
// This is a temporary limit that originates from the the De-MLS epoch time.
|
||||
return Err(ChatError::Protocol(
|
||||
"Cannot add more than 50 Members at a time".into(),
|
||||
));
|
||||
}
|
||||
|
||||
// Resolve each account to a KeyPackage per authorized device and flatten
|
||||
// them into one list — every device of every invitee becomes an MLS
|
||||
// leaf, so all of a user's installations join the group.
|
||||
let mut keypkgs = Vec::with_capacity(members.len());
|
||||
for ident in members {
|
||||
keypkgs.extend(self.key_packages_for_account(ident, &cx.mls_provider, &cx.registry)?);
|
||||
}
|
||||
|
||||
let (commit, welcome, _group_info) = self
|
||||
.mls_group
|
||||
.add_members(
|
||||
&cx.mls_provider,
|
||||
&cx.mls_identity,
|
||||
keypkgs.iter().as_slice(),
|
||||
)
|
||||
.unwrap();
|
||||
|
||||
self.mls_group
|
||||
.merge_pending_commit(&cx.mls_provider)
|
||||
.unwrap();
|
||||
|
||||
for ident in members {
|
||||
crate::inbox_v2::invite_user_group_cent_info(&mut cx.ds, ident, &welcome)?;
|
||||
}
|
||||
|
||||
self.send_payload(cx, commit.to_bytes()?)
|
||||
}
|
||||
|
||||
fn metadata(&self) -> ConvoMetadata {
|
||||
let res = self.mls_group.extensions().iter().find_map(|ext| {
|
||||
if let Extension::Unknown(ext_type, UnknownExtension(bytes)) = ext {
|
||||
if *ext_type == GROUP_METADATA_EXTENSION_TYPE {
|
||||
return ConvoMetaInfo::from_extension_bytes(bytes).ok();
|
||||
};
|
||||
}
|
||||
None
|
||||
});
|
||||
|
||||
res.map(Into::into).unwrap_or_else(ConvoMetadata::empty)
|
||||
}
|
||||
}
|
||||
@ -1,5 +1,4 @@
|
||||
use crate::causal_history::{CausalHistoryStore, MissingMessage};
|
||||
use crate::conversation::GroupCentInfoConvo;
|
||||
use crate::conversation::{
|
||||
ConversationIdRef, DirectV1Convo, GroupV1Convo, GroupV2Convo, Identified, PrivateV1Convo,
|
||||
};
|
||||
@ -270,24 +269,6 @@ impl<'a, S: ExternalServices + 'static> Core<S> {
|
||||
Ok(convo_id)
|
||||
}
|
||||
|
||||
pub fn create_group_cent_info(
|
||||
&mut self,
|
||||
participants: &[IdentIdRef],
|
||||
name: &str,
|
||||
desc: &str,
|
||||
) -> Result<ConversationId, ChatError> {
|
||||
// TODO: (P1) Ensure errors are handled properly. This is a high chance for
|
||||
// desynchronized state: MlsGroup persistence, conversation persistence, and
|
||||
// invite delivery all happen separately.
|
||||
let mut convo = GroupCentInfoConvo::new(&mut self.services, name, desc)?;
|
||||
convo.add_member(&mut self.services, participants)?;
|
||||
let convo_id = convo.id().to_string();
|
||||
|
||||
self.register_convo(ConvoTypeOwned::Group(Box::new(convo)))?;
|
||||
|
||||
Ok(convo_id)
|
||||
}
|
||||
|
||||
/// Add members to an existing group conversation.
|
||||
pub fn group_add_member(
|
||||
&mut self,
|
||||
|
||||
@ -16,7 +16,6 @@ pub(crate) use mls_provider::MlsEphemeralPqProvider;
|
||||
use crate::ChatError;
|
||||
use crate::DeliveryService;
|
||||
use crate::RegistrationService;
|
||||
use crate::conversation::GroupCentInfoConvo;
|
||||
use crate::conversation::GroupConvo;
|
||||
use crate::conversation::GroupV1Convo;
|
||||
use crate::conversation::GroupV2Convo;
|
||||
@ -142,20 +141,6 @@ impl InboxV2 {
|
||||
let convo = GroupV2Convo::new_from_welcome(service_ctx, &mw)?;
|
||||
Ok(Some(Box::new(convo)))
|
||||
}
|
||||
InviteType::GroupV3(welcome_bytes) => {
|
||||
let (msg_in, _remaining) =
|
||||
MlsMessageIn::tls_deserialize_bytes(welcome_bytes.as_slice())?;
|
||||
|
||||
let MlsMessageBodyIn::Welcome(welcome) = msg_in.extract() else {
|
||||
return Err(ChatError::ProtocolExpectation(
|
||||
"something else",
|
||||
"Welcome".into(),
|
||||
));
|
||||
};
|
||||
|
||||
let convo = GroupCentInfoConvo::new_from_welcome(service_ctx, welcome)?;
|
||||
Ok(Some(Box::new(convo)))
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
@ -232,8 +217,6 @@ pub enum InviteType {
|
||||
GroupV1(GroupV1HeavyInvite),
|
||||
#[prost(bytes, tag = "2")]
|
||||
GroupV2(Vec<u8>),
|
||||
#[prost(bytes, tag = "3")]
|
||||
GroupV3(Vec<u8>),
|
||||
}
|
||||
|
||||
#[derive(Clone, PartialEq, Message)]
|
||||
|
||||
@ -1,66 +0,0 @@
|
||||
use integration_tests_core::TestHarness;
|
||||
use std::time::Duration;
|
||||
|
||||
#[test]
|
||||
fn dev() {
|
||||
let _ = tracing_subscriber::fmt()
|
||||
.with_max_level(tracing::Level::DEBUG)
|
||||
.with_test_writer()
|
||||
.try_init();
|
||||
|
||||
let mut harness = TestHarness::<3>::new(|_, _| {});
|
||||
|
||||
let raya_id = harness.raya().ident_id().clone();
|
||||
let pax_id = harness.pax().ident_id().clone();
|
||||
|
||||
const M_R1: &[u8; 12] = b"Hi From Raya";
|
||||
const M_P1: &[u8; 13] = b"Hey it's Pax!";
|
||||
|
||||
// Step: Saro Create Convo with Raya
|
||||
|
||||
let convo_id = harness
|
||||
.saro()
|
||||
.create_group_cent_info(
|
||||
&[&raya_id],
|
||||
"Saro<>Raya",
|
||||
"this is a DM Between Raya and Saro",
|
||||
)
|
||||
.expect("Saro invite Raya ");
|
||||
harness.process_until(|h| h.raya().list_conversations().unwrap().len() == 1);
|
||||
|
||||
// Step: Raya Send Content
|
||||
|
||||
println!("{:?}", harness.raya().convo_metadata(&convo_id));
|
||||
|
||||
harness
|
||||
.raya()
|
||||
.send_content(&convo_id, M_R1)
|
||||
.expect("Raya send Msg");
|
||||
|
||||
harness.process_until(|h| h.saro().received_messages().len() == 1);
|
||||
|
||||
// Step: Saro add Pax
|
||||
|
||||
harness
|
||||
.saro()
|
||||
.group_add_member(&convo_id, &[&pax_id])
|
||||
.expect("Saro invite pax");
|
||||
harness.process_until(|h| h.pax().list_conversations().unwrap().len() == 1);
|
||||
|
||||
// Step: Pax send Content
|
||||
|
||||
harness
|
||||
.pax()
|
||||
.send_content(&convo_id, M_P1)
|
||||
.expect("Pax send");
|
||||
harness.process(Duration::from_millis(500));
|
||||
|
||||
assert!(harness.saro().check(&convo_id, M_R1));
|
||||
assert!(harness.saro().check(&convo_id, M_P1));
|
||||
|
||||
assert!(!harness.raya().check(&convo_id, M_R1));
|
||||
assert!(harness.raya().check(&convo_id, M_P1));
|
||||
|
||||
assert!(!harness.pax().check(&convo_id, M_R1));
|
||||
assert!(!harness.pax().check(&convo_id, M_P1));
|
||||
}
|
||||
Loading…
x
Reference in New Issue
Block a user