use std::collections::HashSet; use std::sync::Arc; use std::thread::{self, JoinHandle}; use components::{ThreadedWakeupService, WakeupEvent}; use crossbeam_channel::{Receiver, Sender, select}; use crypto::Ed25519VerifyingKey; use libchat::{ ConversationId, ConvoOutcome, Core, DeliveryService, GroupV2Config, IdentId, IdentIdRef, InboxOutcome, Introduction, PayloadOutcome, RegistrationService, }; use logos_account::{AccountDirectory, resolve_device_ids}; use parking_lot::Mutex; use storage::ChatStore; use crate::delegate::{DelegateCredential, DelegateIdentity, DelegateSigner}; use crate::errors::ClientError; use crate::event::{Event, MessageSender}; type ClientCore = Core<(DelegateIdentity, T, R, ThreadedWakeupService, S)>; type AccountAddressRef<'a> = &'a str; type LocalSignerId = IdentId; /// A member of a group conversation's roster. /// /// Shares [`MessageSender`]'s field semantics: `account` is set only when the /// member's credential claimed an account *and* the directory confirmed this /// device belongs to it. Unlike a message sender, an unconfirmable claim does /// not hide the member: it is cryptographically in the group, so it is listed /// by `local_identity` (its device) with `account: None`. #[derive(Debug, Clone, PartialEq, Eq)] pub struct GroupMember { pub account: Option, pub local_identity: IdentId, } /// The transport as the client sees it: a [`DeliveryService`] for outbound /// publishing plus the inbound payload stream the worker drains. One object owns /// both directions of the boundary. pub trait Transport: DeliveryService + Send + 'static { /// Hand over the inbound payload stream. Called once, at client construction, /// before the [`Core`] takes ownership of the service. fn inbound(&mut self) -> Receiver>; } /// High-level chat client. /// /// Owns the synchronous [`Core`] behind an `Arc>` and a background /// worker that consumes inbound payloads off the transport's channel, drives /// the core, and forwards observations as [`Event`]s. Construction returns the /// handle together with the `Receiver` the application drains on its own /// schedule. /// /// Outbound calls (`send_message`, `create_conversation`, …) run on the /// caller's thread: they briefly lock the core, invoke it, and return — no /// message-passing round-trip. The `Arc`/`Mutex`/threads live entirely here; /// the core never mentions threads. pub struct ChatClient where T: Transport + Send + 'static, R: RegistrationService + AccountDirectory + Clone + Send + 'static, S: ChatStore + Send + 'static, { /// `parking_lot::Mutex` for its eventual fairness: an inbound burst can't /// starve caller operations of the lock. core: Arc>>, /// The account → device directory. On testnet the registration service /// doubles as the directory (one deployed registry serves both roles), so /// the client keeps its own clone of `R`; the core sees key packages only. directory: R, /// Dropped on `Drop` to wake the worker's `select!` and shut it down. shutdown: Option>, worker: Option>, address: String, } // -- GenericChatClient impl ChatClient where T: Transport + Send + 'static, R: RegistrationService + AccountDirectory + Clone + Send + 'static, S: ChatStore + Send + 'static, { pub fn new( ident: DelegateSigner, account: String, mut transport: T, reg: R, storage: S, group_v2: Option, ) -> Result<(Self, Receiver), ClientError> { let inbound = transport.inbound(); let (wakeup_tx, wakeup_rx) = crossbeam_channel::unbounded(); let wakeup_service = ThreadedWakeupService::new(wakeup_tx); let directory = reg.clone(); let ident = DelegateIdentity::new(ident, &account); let mut core = Core::new_with_name(ident, transport, reg, wakeup_service, storage)?; if let Some(config) = group_v2 { core.set_group_v2_config(config); } Ok(Self::spawn(core, directory, account, inbound, wakeup_rx)) } fn spawn( core: ClientCore, directory: R, address: String, inbound: Receiver>, wakeup_events: Receiver, ) -> (Self, Receiver) { let core = Arc::new(Mutex::new(core)); let (event_tx, event_rx) = crossbeam_channel::unbounded(); let (shutdown_tx, shutdown_rx) = crossbeam_channel::bounded::<()>(0); let worker = thread::spawn({ let core = Arc::clone(&core); let directory = directory.clone(); move || { worker_loop( core, directory, inbound, wakeup_events, shutdown_rx, event_tx, ) } }); ( Self { core, directory, shutdown: Some(shutdown_tx), worker: Some(worker), address, }, event_rx, ) } /// The account address peers use to reach this client. pub fn addr(&self) -> &str { &self.address } /// Returns the installation name (identity label) of this client. pub fn installation_name(&self) -> String { self.core.lock().installation_name().to_string() } /// Produce a serialised introduction bundle for sharing out-of-band. pub fn create_intro_bundle(&mut self) -> Result, ClientError> { self.core.lock().create_intro_bundle().map_err(Into::into) } // Creates a conversation between two Accounts. pub fn create_direct_conversation( &mut self, account: AccountAddressRef, ) -> Result { let signers = self.signers_from_account(account)?; let signer_refs: Vec = signers.iter().collect(); self.core .lock() .create_direct_convo(&signer_refs) .map_err(Into::into) } /// Create a GroupV2 conversation with the given accounts' devices. Each /// account resolves to the signer ids its directory bundle endorses; the /// group invite goes to every one of them. An empty slice creates a group /// with only this client, to grow via [`Self::add_group_members`]. pub fn create_group_conversation( &mut self, accounts: &[AccountAddressRef], ) -> Result { let signers = self.signers_from_accounts(accounts)?; let signer_refs: Vec = signers.iter().collect(); self.core .lock() .create_group_convo(&signer_refs) .map_err(Into::into) } /// Add accounts' devices to an existing group conversation. The add is /// staged as an MLS proposal and merged by the group's next commit (driven /// asynchronously by the wakeup loop); each joiner's welcome is sent when /// that commit lands, not when this call returns. pub fn add_group_members( &mut self, convo_id: &str, accounts: &[AccountAddressRef], ) -> Result<(), ClientError> { let signers = self.signers_from_accounts(accounts)?; let signer_refs: Vec = signers.iter().collect(); self.core .lock() .group_add_member(convo_id, &signer_refs) .map_err(Into::into) } /// The group's roster, one [`GroupMember`] per account (self included). An /// account's several devices collapse to a single entry surfacing that /// account; a member whose account claim the directory can't confirm stays /// on the roster individually, keyed by its device. Costs one directory /// lookup per member that claims an account, the same per-member cost a /// received message's sender check pays. pub fn group_members(&mut self, convo_id: &str) -> Result, ClientError> { let credentials = self.core.lock().group_members(convo_id)?; let members = credentials .iter() .filter_map(|credential| roster_member(&self.directory, credential)); Ok(dedup_members(members)) } /// Parse intro bundle bytes and initiate a private conversation. Outbound /// envelopes are published by the core. Returns this side's conversation ID. /// /// This function will be deprecated in the future. Use `create_direct_conversation` pub fn create_conversation( &mut self, intro_bundle: &[u8], initial_content: &[u8], ) -> Result { let intro = Introduction::try_from(intro_bundle)?; self.core .lock() .create_private_convo_v1(&intro, initial_content) .map_err(Into::into) } /// List all conversation IDs known to this client. pub fn list_conversations(&self) -> Result, ClientError> { self.core.lock().list_conversations().map_err(Into::into) } /// Encrypt and send `content` to an existing conversation. The core /// publishes the outbound envelope. pub fn send_message(&mut self, convo_id: &str, content: &[u8]) -> Result<(), ClientError> { self.core .lock() .send_content(convo_id, content) .map_err(Into::into) } /// Resolve an account address to the signer (device) ids its published /// directory bundle endorses. A reachable account has published at least /// one signer; anything else is an error. fn signers_from_account( &self, account: AccountAddressRef, ) -> Result, ClientError> { let account = IdentId::new(account.to_string()); let device_ids = resolve_device_ids(&self.directory, &account) .map_err(|e| ClientError::AccountResolution(e.to_string()))?; Ok(device_ids.into_iter().map(IdentId::new).collect()) } /// Resolve each account to its signer ids and flatten them, failing on the /// first unresolvable account. fn signers_from_accounts( &self, accounts: &[AccountAddressRef], ) -> Result, ClientError> { let mut signers = Vec::new(); for account in accounts { signers.extend(self.signers_from_account(account)?); } Ok(signers) } } impl Drop for ChatClient where T: Transport + Send + 'static, R: RegistrationService + AccountDirectory + Clone + Send + 'static, S: ChatStore + Send + 'static, { fn drop(&mut self) { // Dropping the sender disconnects the worker's shutdown channel, waking // its `select!` so it can exit; then we join it. self.shutdown.take(); if let Some(handle) = self.worker.take() { let _ = handle.join(); } } } /// Background loop: block until an inbound payload or shutdown arrives, drive /// the core on each payload, and forward events. No polling — `select!` parks /// the thread until one of the channels is ready. fn worker_loop( core: Arc>>, directory: R, inbound: Receiver>, wakeup_events: Receiver, shutdown: Receiver<()>, event_tx: Sender, ) where T: DeliveryService + Send + 'static, R: RegistrationService + AccountDirectory + Send + 'static, { loop { select! { recv(inbound) -> msg => { let Ok(bytes) = msg else { return; // transport's sender dropped }; let events = { let mut core = core.lock(); match core.handle_payload(&bytes) { Ok(outcome) => events_from_inbound(outcome, &directory), Err(e) => { tracing::warn!("inbound handle_payload failed: {e:?}"); vec![Event::InboundError { message: e.to_string(), }] } } }; for event in events { if event_tx.send(event).is_err() { return; // application dropped the receiver } } } recv(wakeup_events) -> msg => { let Ok(WakeupEvent { convo_id }) = msg else { return; // wakeup service's sender dropped }; if let Err(e) = core.lock().wakeup(&convo_id) { tracing::warn!("wakeup failed: {e:?}"); } } recv(shutdown) -> _ => return, } } } /// Walk a [`PayloadOutcome`] in causal order and emit one `Event` per /// observation. For an `Inbox` outcome, [`Event::ConversationStarted`] /// precedes the message event. The convo id is wrapped into `Arc` once /// per outcome and shared across the events it produces. fn events_from_inbound(result: PayloadOutcome, directory: &impl AccountDirectory) -> Vec { match result { PayloadOutcome::Empty => Vec::new(), PayloadOutcome::Convo(co) => convo_events(co, directory), PayloadOutcome::Inbox(io) => inbox_events(io, directory), } } /// Interpret a hex account address as an Ed25519 account verifying key. fn account_key_from_hex(addr: &str) -> Option { let bytes: [u8; 32] = hex::decode(addr).ok()?.try_into().ok()?; Ed25519VerifyingKey::from_bytes(&bytes).ok() } /// Why a message's sender could not be accepted, so the message is dropped. #[derive(Debug, PartialEq, Eq)] enum SenderError { /// No credential at all, so no sender can be attributed. Every delivered /// message must carry an explicit sender. Missing, /// Credential bytes were not valid hex. NotHex, /// Credential bytes did not decode to a delegate credential. Malformed, /// The claimed account address is not an Ed25519 verifying key. AccountNotAKey, /// The account → device mapping is wrong or could not be confirmed: the /// device is not in the account's published set, the account published none, /// or the directory lookup failed. Unverified, } /// The resolution of a credential's account claim against the directory. enum AccountClaim { /// The credential claimed no account. None, /// Confirmed: the directory lists this device under the claimed account. Verified(IdentId), /// An account was claimed but could not be confirmed (see [`SenderError`]). Unverified(SenderError), } /// Parse a wire credential into the device it names and the resolution of any /// account claim, checked against the account → device directory. `Err` only /// when no device can be attributed at all (missing or unparseable credential). /// /// The account-claim policy is left to the caller: a message drops on an /// unconfirmable claim, a roster entry keeps the device and forgoes the account. fn parse_credential( directory: &impl AccountDirectory, encoded: &[u8], ) -> Result<(IdentId, AccountClaim), SenderError> { // No credential at all: there is no device to attribute. if encoded.is_empty() { return Err(SenderError::Missing); } let Ok(data) = hex::decode(encoded) else { tracing::warn!("credential is not valid hex"); return Err(SenderError::NotHex); }; let Ok(cred) = DelegateCredential::try_from(data) else { tracing::warn!("malformed credential"); return Err(SenderError::Malformed); }; let device = IdentId::new(hex::encode(cred.delegate_id().as_ref())); // An unassociated delegate asserts no account → device mapping. let Some(account_addr) = cred.account_addr() else { return Ok((device, AccountClaim::None)); }; let Some(account_key) = account_key_from_hex(account_addr) else { tracing::warn!(account_addr, "account address is not a verifying key"); return Ok(( device, AccountClaim::Unverified(SenderError::AccountNotAKey), )); }; let claim = match directory.fetch(&account_key) { Ok(Some(set)) if set.devices.iter().any(|d| d.as_str() == device.as_str()) => { AccountClaim::Verified(IdentId::new(account_addr.to_string())) } _ => { tracing::warn!(account_addr, device = %device.as_str(), "account → device mapping is wrong or unconfirmable"); AccountClaim::Unverified(SenderError::Unverified) } }; Ok((device, claim)) } /// Decode and verify a message's sender from its credential, checked against the /// account → device directory (our account store). /// /// `Ok(sender)` — deliver with the sender; its `account` is set only when the /// directory confirmed the device, so it is always verified. `Err` — drop the /// message (including when no credential is present, since every delivered /// message must carry an explicit sender). fn decode_sender( directory: &impl AccountDirectory, encoded: &[u8], ) -> Result { let (device, claim) = parse_credential(directory, encoded)?; match claim { AccountClaim::None => Ok(MessageSender { account: None, local_identity: device, }), AccountClaim::Verified(account) => Ok(MessageSender { account: Some(account), local_identity: device, }), // An unconfirmable account claim drops the message: every delivered // message must carry a verified sender. AccountClaim::Unverified(err) => Err(err), } } /// Map a group member's credential (as reported by MLS, in the same hex-encoded /// form a message carries as its sender) to a roster entry, tolerating an /// unconfirmable account claim by listing the device without an account. `None` /// only when the credential cannot be parsed, which does not happen for a real /// MLS leaf. fn roster_member(directory: &impl AccountDirectory, encoded: &[u8]) -> Option { let (device, claim) = parse_credential(directory, encoded).ok()?; let account = match claim { AccountClaim::Verified(account) => Some(account), AccountClaim::None | AccountClaim::Unverified(_) => None, }; Some(GroupMember { account, local_identity: device, }) } /// The key that decides whether two roster entries are the same member: a /// verified account, so an account's several devices count once; or, for a /// member with no confirmed account, its device — unique per MLS leaf, so it /// never merges with another. fn member_key(member: &GroupMember) -> &str { member .account .as_ref() .unwrap_or(&member.local_identity) .as_str() } /// Collapse a roster to one entry per account (keeping the first-seen device as /// the account's representative) while leaving account-less members individual, /// order preserved. fn dedup_members(members: impl IntoIterator) -> Vec { let mut seen = HashSet::new(); members .into_iter() .filter(|member| seen.insert(member_key(member).to_owned())) .collect() } fn convo_events(outcome: ConvoOutcome, directory: &impl AccountDirectory) -> Vec { let ConvoOutcome { convo_id, content } = outcome; content .and_then(|c| { let sender = decode_sender(directory, &c.encoded_credential).ok()?; Some(Event::MessageReceived { convo_id: Arc::from(convo_id), content: c.bytes, sender, }) }) .into_iter() .collect() } fn inbox_events(outcome: InboxOutcome, directory: &impl AccountDirectory) -> Vec { let InboxOutcome { new_conversation, initial, } = outcome; let id: Arc = Arc::from(new_conversation.convo_id); let mut events = Vec::with_capacity(2); events.push(Event::ConversationStarted { convo_id: Arc::clone(&id), class: new_conversation.class, }); if let Some(c) = initial.and_then(|co| co.content) && let Ok(sender) = decode_sender(directory, &c.encoded_credential) { events.push(Event::MessageReceived { convo_id: Arc::clone(&id), content: c.bytes, sender, }); } events } #[cfg(test)] mod sender_check_tests { use std::collections::HashMap; use crypto::{Ed25519SigningKey, Ed25519VerifyingKey}; use libchat::IdentId; use logos_account::{DeviceSet, SignedDeviceBundle}; use super::{ GroupMember, MessageSender, SenderError, decode_sender, dedup_members, member_key, roster_member, }; use crate::delegate::DelegateCredential; /// In-test account → device directory. Holds device id sets keyed by the hex /// account key, and can be made to fail to simulate a directory outage. #[derive(Debug, Default)] struct FakeDir { bundles: HashMap>, fail: bool, } impl FakeDir { /// Publish `devices` (verifying keys) as `account`'s device set. fn with_devices(account: &Ed25519VerifyingKey, devices: &[&Ed25519VerifyingKey]) -> Self { let mut bundles = HashMap::new(); bundles.insert( hex::encode(account.as_ref()), devices.iter().map(|d| hex::encode(d.as_ref())).collect(), ); Self { bundles, fail: false, } } } impl logos_account::AccountDirectory for FakeDir { type Error = &'static str; fn publish(&mut self, _: &SignedDeviceBundle) -> Result<(), Self::Error> { Ok(()) } fn fetch(&self, account: &Ed25519VerifyingKey) -> Result, Self::Error> { if self.fail { return Err("directory unavailable"); } Ok(self .bundles .get(&hex::encode(account.as_ref())) .map(|devices| DeviceSet { lamport: 1, devices: devices.clone(), })) } } fn key() -> Ed25519VerifyingKey { Ed25519SigningKey::generate().verifying_key() } /// Encode a credential exactly as it travels on the wire: the hex of the /// serialized TLV, matching the MLS leaf credential's content bytes. fn encoded(cred: DelegateCredential) -> Vec { hex::encode(cred.serialize()).into_bytes() } fn local_id(k: &Ed25519VerifyingKey) -> IdentId { IdentId::new(hex::encode(k.as_ref())) } /// The account published a device set that includes the sending device — the /// claim checks out, so the message is delivered with a verified account. #[test] fn verified_sender_surfaces_account_and_device() { let account = key(); let device = key(); let dir = FakeDir::with_devices(&account, &[&device]); let cred = DelegateCredential::associated(&device, &hex::encode(account.as_ref())); assert_eq!( decode_sender(&dir, &encoded(cred)), Ok(MessageSender { account: Some(local_id(&account)), local_identity: local_id(&device), }) ); } /// The account published a device set that does NOT include the sending /// device — a spoofed account claim, so the message is dropped. #[test] fn contradicted_claim_is_dropped() { let account = key(); let endorsed = key(); let spoofer = key(); let dir = FakeDir::with_devices(&account, &[&endorsed]); let cred = DelegateCredential::associated(&spoofer, &hex::encode(account.as_ref())); assert_eq!( decode_sender(&dir, &encoded(cred)), Err(SenderError::Unverified) ); } /// A delegate that claims no account surfaces its device but no account. #[test] fn unassociated_sender_surfaces_device_only() { let dir = FakeDir::default(); let device = key(); let cred = DelegateCredential::unassociated(&device); assert_eq!( decode_sender(&dir, &encoded(cred)), Ok(MessageSender { account: None, local_identity: local_id(&device), }) ); } /// The claimed account has never published a device set — the mapping is /// missing, so the message is dropped. #[test] fn unpublished_account_is_dropped() { let account = key(); let device = key(); let dir = FakeDir::default(); // nothing published let cred = DelegateCredential::associated(&device, &hex::encode(account.as_ref())); assert_eq!( decode_sender(&dir, &encoded(cred)), Err(SenderError::Unverified) ); } /// A directory outage leaves the mapping unconfirmed, so the message is /// dropped rather than delivered on an unverified claim. #[test] fn directory_error_is_dropped() { let account = key(); let device = key(); let dir = FakeDir { fail: true, ..Default::default() }; let cred = DelegateCredential::associated(&device, &hex::encode(account.as_ref())); assert_eq!( decode_sender(&dir, &encoded(cred)), Err(SenderError::Unverified) ); } /// No credential at all (e.g. the PrivateV1 placeholder) leaves no sender to /// attribute, so the message is dropped. #[test] fn empty_credential_is_dropped() { let dir = FakeDir::default(); assert_eq!(decode_sender(&dir, b""), Err(SenderError::Missing)); } /// Bytes that aren't a well-formed credential leave the sender's mapping /// undeterminable, so the message is dropped. #[test] fn malformed_credential_is_dropped() { let dir = FakeDir::default(); assert_eq!(decode_sender(&dir, b"not hex"), Err(SenderError::NotHex)); assert_eq!( decode_sender(&dir, hex::encode([0u8; 4]).as_bytes()), Err(SenderError::Malformed) ); } /// An account address that isn't a verifying key can't be looked up, so the /// claim is unconfirmable and the message is dropped. #[test] fn non_key_account_address_is_dropped() { let dir = FakeDir::default(); let cred = DelegateCredential::associated(&key(), "user@example.com"); assert_eq!( decode_sender(&dir, &encoded(cred)), Err(SenderError::AccountNotAKey) ); } /// A verified account claim surfaces the member's account and device — the /// same happy path as a message sender. #[test] fn roster_verified_member_surfaces_account() { let account = key(); let device = key(); let dir = FakeDir::with_devices(&account, &[&device]); let cred = DelegateCredential::associated(&device, &hex::encode(account.as_ref())); assert_eq!( roster_member(&dir, &encoded(cred)), Some(GroupMember { account: Some(local_id(&account)), local_identity: local_id(&device), }) ); } /// Unlike a message sender, a spoofed account claim does not hide the /// member: the device is cryptographically in the group, so it is listed /// with no account rather than dropped. #[test] fn roster_contradicted_claim_lists_device_without_account() { let account = key(); let endorsed = key(); let spoofer = key(); let dir = FakeDir::with_devices(&account, &[&endorsed]); let cred = DelegateCredential::associated(&spoofer, &hex::encode(account.as_ref())); assert_eq!( roster_member(&dir, &encoded(cred)), Some(GroupMember { account: None, local_identity: local_id(&spoofer), }) ); } /// A member whose credential claims no account is listed by device only. #[test] fn roster_unassociated_member_lists_device_without_account() { let dir = FakeDir::default(); let device = key(); let cred = DelegateCredential::unassociated(&device); assert_eq!( roster_member(&dir, &encoded(cred)), Some(GroupMember { account: None, local_identity: local_id(&device), }) ); } /// A directory outage leaves the account unconfirmed, but the member stays /// on the roster by device (a message would drop here). #[test] fn roster_directory_outage_lists_device_without_account() { let account = key(); let device = key(); let dir = FakeDir { fail: true, ..Default::default() }; let cred = DelegateCredential::associated(&device, &hex::encode(account.as_ref())); assert_eq!( roster_member(&dir, &encoded(cred)), Some(GroupMember { account: None, local_identity: local_id(&device), }) ); } /// A non-key account address can't be confirmed, so the member is listed by /// device without an account. #[test] fn roster_non_key_account_lists_device_without_account() { let dir = FakeDir::default(); let device = key(); let cred = DelegateCredential::associated(&device, "user@example.com"); assert_eq!( roster_member(&dir, &encoded(cred)), Some(GroupMember { account: None, local_identity: local_id(&device), }) ); } /// The roster collapses an account's several devices into one entry (keeping /// the first device seen) while leaving account-less members individual, /// order preserved. #[test] fn dedup_collapses_account_devices_and_keeps_unknowns() { let with_account = |account: &str, device: &str| GroupMember { account: Some(IdentId::new(account.to_string())), local_identity: IdentId::new(device.to_string()), }; let device_only = |device: &str| GroupMember { account: None, local_identity: IdentId::new(device.to_string()), }; let roster = dedup_members(vec![ with_account("alice", "alice-dev-1"), with_account("alice", "alice-dev-2"), device_only("orphan-x"), with_account("bob", "bob-dev-1"), device_only("orphan-y"), ]); let keys: Vec<&str> = roster.iter().map(member_key).collect(); assert_eq!(keys, ["alice", "orphan-x", "bob", "orphan-y"]); // Alice's collapsed entry keeps her first-seen device. assert_eq!(roster[0].local_identity.as_str(), "alice-dev-1"); } }