diff --git a/Cargo.lock b/Cargo.lock index 280e23c..7e7467b 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -1330,7 +1330,7 @@ dependencies = [ [[package]] name = "chat-proto" version = "0.1.0" -source = "git+https://github.com/logos-messaging/chat_proto?rev=37ec98a151f6d50aab2905802ac0a896477e62ea#37ec98a151f6d50aab2905802ac0a896477e62ea" +source = "git+https://github.com/logos-messaging/chat_proto?rev=948f4ad6dedb01b44e279204d8eb30a1eda5a330#948f4ad6dedb01b44e279204d8eb30a1eda5a330" dependencies = [ "prost", ] @@ -1463,14 +1463,15 @@ name = "components" version = "0.1.0" dependencies = [ "base64", + "chat-proto", "crossbeam-channel", "crypto", "hex", "libchat", "logos-account", + "prost", "reqwest 0.12.28", "serde", - "serde_json", "storage", "thiserror", "tracing", @@ -3629,8 +3630,6 @@ version = "0.1.0" dependencies = [ "base64", "crossbeam-channel", - "libchat", - "logos-generic-chat", "serde", "serde_json", "thiserror", diff --git a/Cargo.toml b/Cargo.toml index 1efb279..47d84ea 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -44,6 +44,7 @@ logos-delivery = { path = "extensions/logos-delivery-rust"} logos-generic-chat = { path = "crates/generic-chat" } shared-traits = { path = "core/shared-traits" } storage = { path = "core/storage" } +chat-proto = { git = "https://github.com/logos-messaging/chat_proto", rev = "948f4ad6dedb01b44e279204d8eb30a1eda5a330" } # External Workspace dependency declarations (sorted) blake2 = "0.10" diff --git a/bin/chat-cli/src/main.rs b/bin/chat-cli/src/main.rs index f4910bc..f3f9de1 100644 --- a/bin/chat-cli/src/main.rs +++ b/bin/chat-cli/src/main.rs @@ -10,7 +10,7 @@ use clap::{Parser, ValueEnum}; use crossbeam_channel::Receiver; use logos_chat::{ AccountDirectory, ChatClient, ChatStore, Event, LogosAuthVerifier, LogosConfig, P2pConfig, - RegistrationService, Transport, + RegistrationService, RegistryPublishMode, Transport, }; use app::ChatApp; @@ -65,6 +65,28 @@ struct Cli { /// Example: `--registry-url http://127.0.0.1:18080`. #[arg(long)] registry_url: Option, + + /// How keypackage and account bundles are submitted to the store: over its + /// HTTP POST API, or published on the delivery network for the store to + /// pick up by subscription. Queries always use the HTTP API. + #[arg(long, value_enum, default_value_t = RegistryPublishKind::Http)] + registry_publish: RegistryPublishKind, +} + +#[derive(Copy, Clone, Debug, ValueEnum)] +#[value(rename_all = "kebab-case")] +enum RegistryPublishKind { + Http, + Delivery, +} + +impl From for RegistryPublishMode { + fn from(kind: RegistryPublishKind) -> Self { + match kind { + RegistryPublishKind::Http => RegistryPublishMode::Http, + RegistryPublishKind::Delivery => RegistryPublishMode::Delivery, + } + } } fn main() -> Result<()> { @@ -95,6 +117,7 @@ fn main() -> Result<()> { if let Some(registry_url) = cli.registry_url.as_deref() { config.set_registry_url(registry_url); } + config.set_registry_publish_mode(cli.registry_publish.into()); config.set_p2p_config(p2p_config); let (client, events) = logos_chat::open(config) .map_err(|e| anyhow::anyhow!("{e:?}")) @@ -115,6 +138,7 @@ fn main() -> Result<()> { if let Some(registry_url) = cli.registry_url.as_deref() { config.set_registry_url(registry_url); } + config.set_registry_publish_mode(cli.registry_publish.into()); let (client, events) = logos_chat::open_with_transport(config, transport) .map_err(|e| anyhow::anyhow!("{e:?}")) .context("failed to open chat client")?; diff --git a/bin/chat-cli/src/transport/file.rs b/bin/chat-cli/src/transport/file.rs index 75b1d31..eb17215 100644 --- a/bin/chat-cli/src/transport/file.rs +++ b/bin/chat-cli/src/transport/file.rs @@ -14,7 +14,7 @@ pub enum FileTransportError { Io(#[from] io::Error), } -#[derive(Debug)] +#[derive(Clone, Debug)] pub struct FileTransport { transport_dir: PathBuf, inbound_rx: Option>>, diff --git a/core/conversations/Cargo.toml b/core/conversations/Cargo.toml index 0b3950d..e44b1e4 100644 --- a/core/conversations/Cargo.toml +++ b/core/conversations/Cargo.toml @@ -17,7 +17,7 @@ storage = { workspace = true } # External dependencies (sorted) alloy = "2.0" base64 = "0.22" -chat-proto = { git = "https://github.com/logos-messaging/chat_proto", rev = "37ec98a151f6d50aab2905802ac0a896477e62ea" } +chat-proto = { workspace = true } de-mls = { git = "https://github.com/vacp2p/de-mls", rev = "2c7a8669c1492c749c02efd2c5ac45e93e4926a3"} # Expose Mls Extensions (#131) double-ratchets = { path = "../double-ratchets" } hashgraph-like-consensus = "0.6.0" diff --git a/core/conversations/src/conversation.rs b/core/conversations/src/conversation.rs index 69a1137..f4589e7 100644 --- a/core/conversations/src/conversation.rs +++ b/core/conversations/src/conversation.rs @@ -37,6 +37,10 @@ pub(crate) trait Convo: Identified + Send { /// Advances any time-driven protocol work (de-mls consensus deadlines) and /// reports what it observed, mirroring [`Self::handle_frame`]. fn wakeup(&mut self, service_ctx: &mut ServiceContext) -> Result; + + /// Each current member's MLS leaf-credential content (hex-encoded), self + /// included. + fn members(&self) -> Result, ChatError>; } /// Group-only operations. @@ -47,9 +51,11 @@ pub(crate) trait GroupConvo: Convo + std::fmt::Debug + S members: &[IdentIdRef], ) -> Result<(), ChatError>; - /// Each current member's MLS leaf-credential content (hex-encoded), self - /// included. - fn members(&self) -> Result, ChatError>; + /// Each member this conversation invited and the group has not committed + /// yet, in the same encoding as [`Self::members`]. Covers only invites + /// [`Self::add_member`] made here, and is empty for a conversation kind + /// whose add takes effect within that call. + fn pending_members(&self) -> Result, ChatError>; // All GroupConvos MUST return ConvoMetadata // the return type is Option<_> to support legacy ConvoTypes which // are being phased out. diff --git a/core/conversations/src/conversation/direct_v1.rs b/core/conversations/src/conversation/direct_v1.rs index 8a6cb01..e92c0ec 100644 --- a/core/conversations/src/conversation/direct_v1.rs +++ b/core/conversations/src/conversation/direct_v1.rs @@ -2,7 +2,7 @@ use chat_proto::logoschat::encryption::EncryptedPayload; use shared_traits::IdentIdRef; use crate::{ - ChatError, ExternalServices, + ChatError, ExternalServices, UnverifiedSender, conversation::{ConversationIdRef, Convo, GroupConvo, GroupV1Convo, Identified}, service_context::ServiceContext, }; @@ -61,4 +61,8 @@ where ) -> Result { self.inner_group.wakeup(service_ctx) } + + fn members(&self) -> Result, ChatError> { + Convo::::members(&self.inner_group) + } } diff --git a/core/conversations/src/conversation/group_v1.rs b/core/conversations/src/conversation/group_v1.rs index a87e5f9..9a21155 100644 --- a/core/conversations/src/conversation/group_v1.rs +++ b/core/conversations/src/conversation/group_v1.rs @@ -300,6 +300,17 @@ impl Convo for GroupV1Convo { fn wakeup(&mut self, _: &mut ServiceContext) -> Result { Ok(ConvoOutcome::empty(self.id().to_string())) } + + fn members(&self) -> Result, ChatError> { + Ok(self + .mls_group + .members() + .map(|m| UnverifiedSender { + signer_id: m.signature_key.into(), + cred: m.credential.serialized_content().to_vec(), + }) + .collect()) + } } impl GroupConvo for GroupV1Convo { @@ -349,12 +360,10 @@ impl GroupConvo for GroupV1Convo { self.send_payload(cx, commit.to_bytes()?) } - fn members(&self) -> Result, ChatError> { - Ok(self - .mls_group - .members() - .map(|m| UnverifiedSender::from(m)) - .collect()) + /// Always empty: `add_member` merges its own commit, so an added member is + /// on the roster by the time the call returns. + fn pending_members(&self) -> Result, ChatError> { + Ok(Vec::new()) } fn metadata(&self) -> Option { diff --git a/core/conversations/src/conversation/group_v2.rs b/core/conversations/src/conversation/group_v2.rs index d940c91..3a9f16e 100644 --- a/core/conversations/src/conversation/group_v2.rs +++ b/core/conversations/src/conversation/group_v2.rs @@ -291,6 +291,25 @@ where let events = self.after_op(ctx)?; // publish what poll produced + re-arm alarm Ok(self.outcome_from_events(&events)) } + + fn members(&self) -> Result, ChatError> { + // Guarantee the local member is listed so callers see the full roster. + let mut members = self.conversation.members()?; + let self_id = self.conversation.member_id_bytes().to_vec(); + if !members.contains(&self_id) { + members.push(self_id); + } + + let members = members + .into_iter() + .map(|cred| UnverifiedSender { + // TODO: (!) Replace is actual SenderId. + signer_id: SignerId::from(b"".as_slice()), + cred, + }) + .collect(); + Ok(members) + } } impl GroupConvo for GroupV2Convo @@ -376,23 +395,8 @@ where result.and(flushed) } - fn members(&self) -> Result, ChatError> { - // Guarantee the local member is listed so callers see the full roster. - let mut members = self.conversation.members()?; - let self_id = self.conversation.member_id_bytes().to_vec(); - if !members.contains(&self_id) { - members.push(self_id); - } - - let members = members - .into_iter() - .map(|cred| UnverifiedSender { - // TODO: (!) Replace is actual SenderId. - signer_id: SignerId::from(b"".as_slice()), - cred, - }) - .collect(); - Ok(members) + fn pending_members(&self) -> Result, ChatError> { + Ok(vec![]) } fn metadata(&self) -> Option { diff --git a/core/conversations/src/core.rs b/core/conversations/src/core.rs index 182a429..a436bba 100644 --- a/core/conversations/src/core.rs +++ b/core/conversations/src/core.rs @@ -271,46 +271,48 @@ impl<'a, S: ExternalServices + 'static> Core { convo_id: &str, members: &[IdentIdRef], ) -> Result<(), ChatError> { - if self.cached_convos.contains_key(convo_id) { - let convo = self - .cached_convos - .get_mut(convo_id) - .ok_or_else(|| ChatError::NoConvo(convo_id.to_string()))?; + let convo = self + .cached_convos + .get_mut(convo_id) + .ok_or_else(|| ChatError::NoConvo(convo_id.to_string()))?; - match convo { - ConvoTypeOwned::Group(group_convo) => { - group_convo.add_member(&mut self.services, members) - } - ConvoTypeOwned::Direct(convo) => Err(ChatError::UnsupportedFunction( - convo.id().into(), - "Add Member".into(), - )), + match convo { + ConvoTypeOwned::Group(group_convo) => { + group_convo.add_member(&mut self.services, members) } - } else { - let mut convo = self.load_group_convo(convo_id)?; - convo.add_member(&mut self.services, members) + ConvoTypeOwned::Direct(convo) => Err(ChatError::UnsupportedFunction( + convo.id().into(), + "Add Member".into(), + )), } } - /// Each member's MLS leaf-credential content (hex-encoded); errors if - /// `convo_id` names a direct (non-group) conversation. + /// Each member's MLS leaf-credential content (hex-encoded), for a direct + /// conversation as for a group. pub fn group_members(&mut self, convo_id: &str) -> Result, ChatError> { - if self.cached_convos.contains_key(convo_id) { - let convo = self - .cached_convos - .get(convo_id) - .ok_or_else(|| ChatError::NoConvo(convo_id.to_string()))?; + let convo = self + .cached_convos + .get(convo_id) + .ok_or_else(|| ChatError::NoConvo(convo_id.to_string()))?; - match convo { - ConvoTypeOwned::Group(group_convo) => group_convo.members(), - ConvoTypeOwned::Direct(convo) => Err(ChatError::UnsupportedFunction( - convo.id().into(), - "List Members".into(), - )), - } - } else { - let convo = self.load_group_convo(convo_id)?; - convo.members() + convo.members() + } + + /// Each member invited here and still awaiting the group's commit, in the + /// same encoding as [`Self::group_members`]. A direct conversation has no + /// pending members and reports none. + pub fn group_pending_members( + &mut self, + convo_id: &str, + ) -> Result, ChatError> { + let convo = self + .cached_convos + .get(convo_id) + .ok_or_else(|| ChatError::NoConvo(convo_id.to_string()))?; + + match convo { + ConvoTypeOwned::Group(group_convo) => group_convo.pending_members(), + ConvoTypeOwned::Direct(_) => Ok(Vec::new()), } } @@ -454,17 +456,6 @@ impl<'a, S: ExternalServices + 'static> Core { }) } - /// Rebuilds a group conversation; errors if `convo_id` names a non-group. - fn load_group_convo(&mut self, convo_id: &str) -> Result>, ChatError> { - let record = self.load_conversation_meta(convo_id)?; - match record.kind { - ConversationKind::GroupV1 => Ok(Box::new(self.load_mls_convo(&record.local_convo_id)?)), - ConversationKind::Unknown(_) => { - Err(ChatError::UnsupportedConvoType(record.kind.as_str().into())) - } - } - } - /// Rebuilds a group conversation from storage so an operation can run against it. fn load_mls_convo(&mut self, convo_id: &str) -> Result { let group_id_bytes = hex::decode(convo_id).map_err(ChatError::generic)?; @@ -553,4 +544,11 @@ impl Convo for ConvoTypeOwned { ConvoTypeOwned::Direct(convo) => convo.wakeup(service_ctx), } } + + fn members(&self) -> Result, ChatError> { + match self { + ConvoTypeOwned::Group(group_convo) => group_convo.members(), + ConvoTypeOwned::Direct(convo) => convo.members(), + } + } } diff --git a/crates/generic-chat/src/client.rs b/crates/generic-chat/src/client.rs index 0007e8c..0e14c1b 100644 --- a/crates/generic-chat/src/client.rs +++ b/crates/generic-chat/src/client.rs @@ -42,12 +42,19 @@ type LocalSignerId = IdentId; /// 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`. +/// not hide the member: a committed member is cryptographically in the group, +/// so it is listed by `local_identity` (its device) with `account: None`. +/// +/// `pending` marks a member whose add the group has not committed yet, so it +/// cannot read the conversation. Only invites this client sent are reported; +/// an add another member proposed is invisible until it commits. The flag +/// clears when the commit admitting the member lands, and an invite the group +/// never commits stays pending for the life of the conversation. #[derive(Debug, Clone, PartialEq, Eq)] pub struct GroupMember { pub account: Option, pub local_identity: IdentId, + pub pending: bool, } pub struct MemberWithAuthResult { @@ -588,6 +595,7 @@ fn roster_member(directory: &impl AccountDirectory, encoded: &[u8]) -> Option (TestClient, Receiver, String) { + create_test_client_with(message_bus, reg, fast_group_v2_config()) +} + +/// [`create_test_client`] with explicit GroupV2 timers, for a test that needs +/// to observe the group between two protocol steps. +fn create_test_client_with( message_bus: MessageBus, mut reg: EphemeralRegistry, + config: GroupV2Config, ) -> (TestClient, Receiver, String) { let account = TestLogosAccount::new(); let delegate = DelegateSigner::random(); @@ -53,7 +63,7 @@ fn create_test_client( .ident(delegate) .transport(InProcessDelivery::new(message_bus)) .registration(reg) - .group_v2_config(fast_group_v2_config()) + .group_v2_config(config) .build() .expect("client create"); let addr = client.addr().to_string(); @@ -93,10 +103,11 @@ fn wait_for_group_started(events: &Receiver, label: &str) -> String { }) } -/// Poll a client's roster for `convo_id` until its verified accounts equal -/// `expected` (order-independent), or panic after a timeout. The roster settles -/// asynchronously as each member applies the add commit, so it is polled rather -/// than snapshotted. +/// Poll a client's roster for `convo_id` until its verified *committed* +/// accounts equal `expected` (order-independent), or panic after a timeout. The +/// roster settles asynchronously as each member applies the add commit, so it is +/// polled rather than snapshotted; members still awaiting that commit are +/// skipped so an invite alone never reads as convergence. fn wait_for_members(client: &mut TestClient, convo_id: &str, expected: &[&str]) { use std::collections::BTreeSet; let want: BTreeSet = expected.iter().map(|s| s.to_string()).collect(); @@ -270,6 +281,148 @@ fn group_creator_is_in_own_roster() { assert_eq!(accounts, vec![Some(saro_addr.clone())]); } +/// An invited member joins the roster immediately, flagged pending: the add is +/// staged as a proposal, so the invitee is not a member until the group commits. +#[test] +fn invited_member_is_pending_until_the_group_commits() { + let bus = MessageBus::default(); + let reg = EphemeralRegistry::new(); + + // A commit window far longer than the assertions below, so the add provably + // cannot merge while they run. + let deferred_commit = GroupV2Config { + commit_inactivity_duration: Duration::from_secs(30), + ..fast_group_v2_config() + }; + let (mut saro, _saro_events, saro_addr) = + create_test_client_with(bus.clone(), reg.clone(), deferred_commit); + let (_raya, _raya_events, raya_addr) = create_test_client(bus.clone(), reg.clone()); + + let convo_id = saro + .create_group_conversation(&[], unnamed_group()) + .expect("empty group"); + saro.add_group_members(&convo_id, &[&raya_addr]) + .expect("saro invites raya"); + + let roster = saro.group_members(&convo_id).expect("group_members"); + let accounts = |pending: bool| -> Vec<&str> { + roster + .iter() + .filter(|m| m.pending == pending) + .filter_map(|m| m.account.as_ref().map(|a| a.as_str())) + .collect() + }; + assert_eq!(accounts(false), vec![saro_addr.as_str()]); + assert_eq!(accounts(true), vec![raya_addr.as_str()]); +} + +/// The pending flag is transient: once the group commits the add, the invitee +/// is an ordinary roster member and nothing is left pending. The joiner, which +/// invited nobody, never reports a pending member at all: the flag is local to +/// the client that sent the invite. +#[test] +fn pending_clears_once_the_add_commits() { + let bus = MessageBus::default(); + let reg = EphemeralRegistry::new(); + + let (mut saro, _saro_events, saro_addr) = create_test_client(bus.clone(), reg.clone()); + let (mut raya, raya_events, raya_addr) = create_test_client(bus.clone(), reg.clone()); + + let convo_id = saro + .create_group_conversation(&[], unnamed_group()) + .expect("empty group"); + saro.add_group_members(&convo_id, &[&raya_addr]) + .expect("saro invites raya"); + + let raya_convo_id = wait_for_group_started(&raya_events, "raya ConversationStarted"); + wait_for_members(&mut saro, &convo_id, &[&saro_addr, &raya_addr]); + + let roster = saro.group_members(&convo_id).expect("group_members"); + assert!( + roster.iter().all(|m| !m.pending), + "committed roster still reports a pending member: {roster:?}" + ); + + let joiner_roster = raya + .group_members(&raya_convo_id) + .expect("joiner group_members"); + assert!( + joiner_roster.iter().all(|m| !m.pending), + "joiner reports a pending member it never invited: {joiner_roster:?}" + ); +} + +/// An invited member joins the roster immediately, flagged pending: the add is +/// staged as a proposal, so the invitee is not a member until the group commits. +#[test] +fn invited_member_is_pending_until_the_group_commits() { + let bus = MessageBus::default(); + let reg = EphemeralRegistry::new(); + + // A commit window far longer than the assertions below, so the add provably + // cannot merge while they run. + let deferred_commit = GroupV2Config { + commit_inactivity_duration: Duration::from_secs(30), + ..fast_group_v2_config() + }; + let (mut saro, _saro_events, saro_addr) = + create_test_client_with(bus.clone(), reg.clone(), deferred_commit); + let (_raya, _raya_events, raya_addr) = create_test_client(bus.clone(), reg.clone()); + + let convo_id = saro + .create_group_conversation(&[], unnamed_group()) + .expect("empty group"); + saro.add_group_members(&convo_id, &[&raya_addr]) + .expect("saro invites raya"); + + let roster = saro.group_members(&convo_id).expect("group_members"); + let accounts = |pending: bool| -> Vec<&str> { + roster + .iter() + .filter(|m| m.pending == pending) + .filter_map(|m| m.account.as_ref().map(|a| a.as_str())) + .collect() + }; + assert_eq!(accounts(false), vec![saro_addr.as_str()]); + assert_eq!(accounts(true), vec![raya_addr.as_str()]); +} + +/// The pending flag is transient: once the group commits the add, the invitee +/// is an ordinary roster member and nothing is left pending. The joiner, which +/// invited nobody, never reports a pending member at all: the flag is local to +/// the client that sent the invite. +#[test] +fn pending_clears_once_the_add_commits() { + let bus = MessageBus::default(); + let reg = EphemeralRegistry::new(); + + let (mut saro, _saro_events, saro_addr) = create_test_client(bus.clone(), reg.clone()); + let (mut raya, raya_events, raya_addr) = create_test_client(bus.clone(), reg.clone()); + + let convo_id = saro + .create_group_conversation(&[], unnamed_group()) + .expect("empty group"); + saro.add_group_members(&convo_id, &[&raya_addr]) + .expect("saro invites raya"); + + let raya_convo_id = wait_for_group_started(&raya_events, "raya ConversationStarted"); + wait_for_members(&mut saro, &convo_id, &[&saro_addr, &raya_addr]); + + let roster = saro.group_members(&convo_id).expect("group_members"); + assert!( + roster.iter().all(|m| !m.pending), + "committed roster still reports a pending member: {roster:?}" + ); + + let joiner_roster = raya + .group_members(&raya_convo_id) + .expect("joiner group_members"); + assert!( + joiner_roster.iter().all(|m| !m.pending), + "joiner reports a pending member it never invited: {joiner_roster:?}" + ); +} + /// A batch add is validated before any member is proposed: a member whose /// account is endorsed in the directory but whose device registered no key /// package fails the whole call, the resolvable member in the same batch is diff --git a/crates/generic-chat/tests/saro_and_raya.rs b/crates/generic-chat/tests/saro_and_raya.rs index 25f252c..16be174 100644 --- a/crates/generic-chat/tests/saro_and_raya.rs +++ b/crates/generic-chat/tests/saro_and_raya.rs @@ -307,6 +307,43 @@ fn group_metadata_on_direct_conversation_errors() { .expect_err("direct conversation has no group metadata"); } +/// A direct conversation reports its participants like any conversation. Add +/// Member errors on the creator's handle, though the joiner holds the same +/// conversation as a plain group, so the rejection is not conversation-wide. +#[test] +fn direct_conversation_lists_its_participants() { + let bus = MessageBus::default(); + let reg = EphemeralRegistry::new(); + + let (mut saro, _saro_events) = + create_test_client(bus.clone(), reg.clone()).expect("client create"); + let (raya, _raya_events) = create_test_client(bus.clone(), reg.clone()).expect("client create"); + + let saro_addr = saro.addr().to_string(); + let raya_addr = raya.addr().to_string(); + let convo_id = saro + .create_direct_conversation(&raya_addr) + .expect("convo create"); + + let roster = saro.group_members(&convo_id).expect("group_members"); + let mut accounts: Vec> = roster + .iter() + .map(|m| m.account.as_ref().map(|a| a.as_str())) + .collect(); + accounts.sort(); + let mut expected = vec![Some(saro_addr.as_str()), Some(raya_addr.as_str())]; + expected.sort(); + assert_eq!(accounts, expected); + + let err = saro + .add_group_members(&convo_id, &[&raya_addr]) + .expect_err("add member is unsupported on a direct conversation"); + assert!(matches!( + err, + logos_generic_chat::ClientError::Chat(libchat::ChatError::UnsupportedFunction(..)) + )); +} + #[derive(Debug)] struct FailingDelivery { inbound_tx: Sender>, diff --git a/crates/logos-chat/src/logos.rs b/crates/logos-chat/src/logos.rs index cc9078b..7a5dc14 100644 --- a/crates/logos-chat/src/logos.rs +++ b/crates/logos-chat/src/logos.rs @@ -2,7 +2,8 @@ //! //! [`open`] commits to the Logos service stack so independently built clients //! share the same production services instead of each re-deriving them: a -//! delegate identity, the HTTP keypackage + account registry, and encrypted +//! delegate identity, the keypackage + account registry (queried over HTTP, +//! with submissions over HTTP or the delivery network), and encrypted //! on-disk storage. The stack is generic over the transport — any //! [`Transport`] can be injected via [`open_with_transport`] — and the //! concrete [`LogosChatClient`] commits to the embedded logos-delivery node, @@ -14,7 +15,7 @@ //! constructors off the alias; they are crate-level functions ([`open`], //! [`open_with_transport`]) taking the all-inclusive [`LogosConfig`] instead. -use components::HttpRegistry; +use components::{ContactRegistry, RegistryPublishMode}; use crossbeam_channel::Receiver; use embedded_logos_delivery::{EmbeddedLogosDelivery, P2pConfig}; use libchat::{ChatStorage, StorageConfig}; @@ -41,6 +42,7 @@ pub struct LogosConfig { db_path: String, db_key: String, registry_url: String, + registry_publish_mode: RegistryPublishMode, p2p_config: P2pConfig, group_v2_config: Option, } @@ -54,6 +56,7 @@ impl LogosConfig { db_path: db_path.into(), db_key: db_key.into(), registry_url: REGISTRY_ENDPOINT.to_string(), + registry_publish_mode: RegistryPublishMode::default(), p2p_config: P2pConfig::default(), group_v2_config: None, } @@ -65,6 +68,13 @@ impl LogosConfig { self.registry_url = registry_url.into(); } + /// Choose how keypackage and account bundles are submitted to the store: + /// HTTP POST (the default) or published over the delivery transport for the + /// store to pick up by subscription. Reads always use the HTTP query API. + pub fn set_registry_publish_mode(&mut self, mode: RegistryPublishMode) { + self.registry_publish_mode = mode; + } + /// Override the embedded node's p2p settings (defaults to /// [`P2pConfig::default`]). Only [`open`] starts an embedded node, so /// [`open_with_transport`] ignores this. @@ -107,19 +117,33 @@ pub fn open(config: LogosConfig) -> Result<(LogosChatClient, Receiver), C /// Open a client on the Logos stack per `config` with the injected transport, /// persisting to the encrypted database. +/// +/// The registry publishes per `config`'s +/// [`registry publish mode`](LogosConfig::set_registry_publish_mode): over +/// HTTP (the default), or over a clone of `transport` — sharing the client's +/// own delivery stack, which is why the transport must be `Clone`. #[allow(clippy::type_complexity)] -pub fn open_with_transport( +pub fn open_with_transport( config: LogosConfig, transport: T, -) -> Result<(ChatClient, Receiver), ClientError> -{ +) -> Result< + ( + ChatClient, ChatStorage>, + Receiver, + ), + ClientError, +> { // A fresh account endorsing a fresh delegate each open: the account // key is dropped after publishing the bundle, so devices cannot be // added later. A caller-supplied, custody-holding account replaces // this once the platform provides one. let account = TestLogosAccount::new(); let delegate = DelegateSigner::random(); - let mut registry = HttpRegistry::new(config.registry_url); + let mut registry = ContactRegistry::new( + transport.clone(), + config.registry_url, + config.registry_publish_mode, + ); account .add_delegate_signer(&mut registry, delegate.public_key()) .map_err(|e| ClientError::BundlePublish(e.to_string()))?; @@ -139,11 +163,16 @@ pub fn open_with_transport( } /// The Logos client: a [`ChatClient`] wired to the Logos service stack — a -/// [`DelegateSigner`] identity acting for a fresh dev account, the HTTP -/// keypackage + account registry ([`HttpRegistry`], which is both the -/// keypackage store and the account → device directory), and encrypted -/// [`ChatStorage`] — running an embedded logos-delivery node as its -/// transport. Open one with [`open`], or swap the transport via +/// [`DelegateSigner`] identity acting for a fresh dev account, the keypackage + +/// account registry ([`ContactRegistry`], which is both the keypackage store +/// and the account → device directory; it queries over HTTP and submits over +/// HTTP or the delivery network per [`LogosConfig::set_registry_publish_mode`]), +/// and encrypted [`ChatStorage`] — running an embedded logos-delivery node as +/// its transport. Open one with [`open`], or swap the transport via /// [`open_with_transport`]. -pub type LogosChatClient = - ChatClient; +pub type LogosChatClient = ChatClient< + LogosAuthVerifier, + EmbeddedLogosDelivery, + ContactRegistry, + ChatStorage, +>; diff --git a/extensions/components/Cargo.toml b/extensions/components/Cargo.toml index 7e1a79f..c4d1dd5 100644 --- a/extensions/components/Cargo.toml +++ b/extensions/components/Cargo.toml @@ -12,10 +12,11 @@ storage = { workspace = true } # External dependencies (sorted) base64 = "0.22" +chat-proto = { workspace = true } crossbeam-channel = { workspace = true } hex = "0.4.3" +prost = "0.14.1" reqwest = { version = "0.12", default-features = false, features = ["blocking", "json", "rustls-tls"] } serde = { version = "1.0", features = ["derive"] } -serde_json = "1.0" thiserror = "2" tracing = "0.1" diff --git a/extensions/components/src/contact_registry.rs b/extensions/components/src/contact_registry.rs index 6d63e33..a9e9d92 100644 --- a/extensions/components/src/contact_registry.rs +++ b/extensions/components/src/contact_registry.rs @@ -1,2 +1,2 @@ pub mod ephemeral; -pub mod http; +pub mod store; diff --git a/extensions/components/src/contact_registry/http.rs b/extensions/components/src/contact_registry/http.rs deleted file mode 100644 index 0cf5987..0000000 --- a/extensions/components/src/contact_registry/http.rs +++ /dev/null @@ -1,431 +0,0 @@ -use std::fmt::Debug; -use std::time::{Duration, SystemTime, UNIX_EPOCH}; - -use base64::Engine; -use base64::engine::general_purpose::STANDARD as BASE64; -use crypto::{Ed25519Signature, Ed25519VerifyingKey}; -use libchat::{IdentityProvider, RegistrationService}; -use logos_account::{AccountDirectory, BundleError, DeviceSet, SignedDeviceBundle, verify_bundle}; -use serde::{Deserialize, Serialize}; - -/// HTTP client for the testnet KeyPackage Registry service. -/// -/// Throwaway transport for issue #110 — replaced by λLEZ in v0.3. -/// -/// The wire carries `device_id` (the hex device verifying key), an opaque -/// `payload` blob, and its `signature`. The signed bytes and the transmitted -/// `payload` bytes are identical, so every verifier checks the signature over -/// exactly what it received — no field-by-field reconstruction to keep in sync. -/// The `payload` is opaque to the server: it verifies `signature` over `payload` -/// with `device_id`'s key (proof-of-possession — only the holder of that key can -/// publish under `device_id`) without decoding the payload. -#[derive(Clone)] -pub struct HttpRegistry { - base_url: String, - http: reqwest::blocking::Client, -} - -#[derive(Debug, thiserror::Error)] -pub enum HttpRegistryError { - #[error("http: {0}")] - Http(#[from] reqwest::Error), - #[error("server returned status {0}: {1}")] - Server(u16, String), - #[error("decode: {0}")] - Decode(String), - #[error("clock before unix epoch")] - Clock, - #[error("signature verification failed")] - SignatureInvalid, - #[error("bundle: {0}")] - Bundle(#[from] BundleError), -} - -#[derive(Debug, Serialize)] -struct SubmitRequest { - /// hex of the 32-byte device verifying key — the verification + storage key. - device_id: String, - /// base64 of the canonical signed payload (see [`encode_payload`]). - payload: String, - /// base64 of the 64-byte Ed25519 signature over `payload`. - signature: String, -} - -#[derive(Debug, Deserialize)] -struct FetchResponse { - payload: String, - signature: String, -} - -#[derive(Debug, Serialize)] -struct SubmitAccountRequest { - /// hex of the 32-byte account verifying key — verification + storage key. - account_pub: String, - /// base64 of the canonical signed device-list payload. - payload: String, - /// base64 of the 64-byte account signature over `payload`. - signature: String, -} - -#[derive(Debug, Deserialize)] -struct FetchAccountResponse { - payload: String, - signature: String, - #[allow(dead_code)] // server's prune clock; freshness is taken from the bundle's lamport - updated_at: i64, -} - -impl HttpRegistry { - pub fn new(base_url: impl Into) -> Self { - Self::with_timeout(base_url, Duration::from_secs(10)) - } - - pub fn with_timeout(base_url: impl Into, timeout: Duration) -> Self { - let http = reqwest::blocking::Client::builder() - .timeout(timeout) - .build() - .expect("reqwest client builder is infallible with these options"); - Self { - base_url: base_url.into().trim_end_matches('/').to_string(), - http, - } - } -} - -impl Debug for HttpRegistry { - fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { - f.debug_struct("HttpRegistry") - .field("base_url", &self.base_url) - .finish() - } -} - -impl RegistrationService for HttpRegistry { - type Error = HttpRegistryError; - - fn register( - &mut self, - identity: &dyn IdentityProvider, - key_bundle: Vec, - ) -> Result<(), HttpRegistryError> { - let device_id = hex::encode(identity.public_key().as_ref()); - let timestamp_ms = SystemTime::now() - .duration_since(UNIX_EPOCH) - .map_err(|_| HttpRegistryError::Clock)? - .as_millis() as u64; - - // Sign exactly the bytes that go on the wire. - let payload = encode_payload(timestamp_ms, &key_bundle); - let signature = identity.sign(&payload); - - let req = SubmitRequest { - device_id, - payload: BASE64.encode(&payload), - signature: BASE64.encode(signature.as_ref()), - }; - - let url = format!("{}/v0/keypackage", self.base_url); - let resp = send_retrying(|| self.http.post(&url).json(&req))?; - if !resp.status().is_success() { - let status = resp.status().as_u16(); - let body = resp.text().unwrap_or_default(); - return Err(HttpRegistryError::Server(status, body)); - } - Ok(()) - } - - fn retrieve(&self, device_id: &str) -> Result>, HttpRegistryError> { - let url = format!("{}/v0/keypackage/{}", self.base_url, device_id); - let resp = send_retrying(|| self.http.get(&url))?; - if resp.status().as_u16() == 404 { - return Ok(None); - } - if !resp.status().is_success() { - let status = resp.status().as_u16(); - let body = resp.text().unwrap_or_default(); - return Err(HttpRegistryError::Server(status, body)); - } - let body: FetchResponse = resp.json()?; - - let payload = BASE64 - .decode(&body.payload) - .map_err(|e| HttpRegistryError::Decode(e.to_string()))?; - let signature_arr: [u8; 64] = BASE64 - .decode(&body.signature) - .map_err(|e| HttpRegistryError::Decode(e.to_string()))? - .as_slice() - .try_into() - .map_err(|_| HttpRegistryError::Decode("signature not 64 bytes".into()))?; - - // Verify over the received payload bytes, using the key we asked for - // (`device_id`). A bundle the requested device didn't sign won't verify. - let device_pubkey: [u8; 32] = hex::decode(device_id) - .map_err(|e| HttpRegistryError::Decode(e.to_string()))? - .as_slice() - .try_into() - .map_err(|_| HttpRegistryError::Decode("device_id not a 32-byte key".into()))?; - let verifying_key = Ed25519VerifyingKey::from_bytes(&device_pubkey) - .map_err(|_| HttpRegistryError::Decode("device_id not a valid ed25519 vk".into()))?; - verifying_key - .verify(&payload, &Ed25519Signature::from(signature_arr)) - .map_err(|_| HttpRegistryError::SignatureInvalid)?; - - let (_timestamp_ms, key_package) = decode_payload(&payload) - .ok_or_else(|| HttpRegistryError::Decode("short payload".into()))?; - - Ok(Some(key_package.to_vec())) - } -} - -impl AccountDirectory for HttpRegistry { - type Error = HttpRegistryError; - - fn publish(&mut self, bundle: &SignedDeviceBundle) -> Result<(), Self::Error> { - let req = SubmitAccountRequest { - account_pub: hex::encode(bundle.account_pub.as_ref()), - payload: BASE64.encode(&bundle.payload), - signature: BASE64.encode(bundle.signature.as_ref()), - }; - - let url = format!("{}/v0/account", self.base_url); - let resp = send_retrying(|| self.http.post(&url).json(&req))?; - if !resp.status().is_success() { - let status = resp.status().as_u16(); - let body = resp.text().unwrap_or_default(); - return Err(HttpRegistryError::Server(status, body)); - } - Ok(()) - } - - fn fetch(&self, account: &Ed25519VerifyingKey) -> Result, Self::Error> { - let url = format!( - "{}/v0/account/{}", - self.base_url, - hex::encode(account.as_ref()) - ); - let resp = send_retrying(|| self.http.get(&url))?; - if resp.status().as_u16() == 404 { - return Ok(None); - } - if !resp.status().is_success() { - let status = resp.status().as_u16(); - let body = resp.text().unwrap_or_default(); - return Err(HttpRegistryError::Server(status, body)); - } - let body: FetchAccountResponse = resp.json()?; - - let payload = BASE64 - .decode(&body.payload) - .map_err(|e| HttpRegistryError::Decode(e.to_string()))?; - let signature_arr: [u8; 64] = BASE64 - .decode(&body.signature) - .map_err(|e| HttpRegistryError::Decode(e.to_string()))? - .as_slice() - .try_into() - .map_err(|_| HttpRegistryError::Decode("signature not 64 bytes".into()))?; - - // The directory service is untrusted: verify the account signature over - // the exact received bytes, and that the bundle is bound to the account - // we asked for, before handing back any device keys. - let bundle = SignedDeviceBundle { - account_pub: account.clone(), - payload, - signature: Ed25519Signature::from(signature_arr), - }; - let device_set = verify_bundle(account, &bundle)?; - Ok(Some(device_set)) - } -} - -/// Canonical binary payload — the bytes that are both signed and transmitted -/// verbatim. Opaque to the server; decoded only by consumers: -/// -/// ```text -/// timestamp_ms : u64 little-endian (8 bytes) -/// key_package : remaining bytes (variable, last → no length prefix needed) -/// ``` -/// -/// The fixed-width field first with the one variable field last makes every -/// byte string parse exactly one way — no delimiter, no ambiguity, even though -/// `key_package` is arbitrary bytes. The device verifying key is carried -/// alongside as `device_id`, not embedded here. -fn encode_payload(timestamp_ms: u64, key_package: &[u8]) -> Vec { - let mut out = Vec::with_capacity(8 + key_package.len()); - out.extend_from_slice(×tamp_ms.to_le_bytes()); - out.extend_from_slice(key_package); - out -} - -/// Inverse of [`encode_payload`]. Returns `None` if the payload is shorter than -/// the fixed header (`8`). -fn decode_payload(payload: &[u8]) -> Option<(u64, &[u8])> { - if payload.len() < 8 { - return None; - } - let timestamp_ms = u64::from_le_bytes(payload[..8].try_into().ok()?); - Some((timestamp_ms, &payload[8..])) -} - -/// Retry budget for the registry's transient, load-induced 5xx/429 responses. -/// The service is reliable request-by-request but sheds concurrent bursts, so a -/// few backed-off retries let a request land once the burst clears. On that path -/// each retry returns fast, so the added cost is the ~3s worst-case backoff sum, -/// well inside chat_module's ~20s init IPC budget. A fully unreachable registry -/// instead costs up to MAX_RETRIES times the reqwest timeout, which no retry -/// budget can rescue. -const MAX_RETRIES: u32 = 4; -const RETRY_BASE_MS: u64 = 200; -const RETRY_MAX_BACKOFF_MS: u64 = 2000; - -/// Send a request built by `build`, retrying transient failures — network errors -/// and 5xx/429 responses — with exponential backoff and full jitter. The -/// registry is reliable request-by-request but sheds concurrent bursts with a -/// 5xx, so a backed-off retry lands once the burst clears; a 4xx (and any other -/// final response) is returned to the caller unchanged. `build` is re-invoked per -/// attempt because sending consumes the builder. -fn send_retrying( - build: impl Fn() -> reqwest::blocking::RequestBuilder, -) -> Result { - let mut attempt = 0; - loop { - let outcome = build().send(); - let transient = match &outcome { - Err(_) => true, // network error / timeout: worth another try - Ok(resp) => is_transient_status(resp.status()), - }; - if !transient || attempt >= MAX_RETRIES { - return Ok(outcome?); - } - std::thread::sleep(backoff_with_jitter(attempt)); - attempt += 1; - } -} - -/// Whether a response status is worth retrying: 5xx (the registry sheds -/// concurrent load with these) or 429 (explicit backpressure). A 4xx is the -/// caller's fault and won't change on retry. -fn is_transient_status(status: reqwest::StatusCode) -> bool { - status.is_server_error() || status == reqwest::StatusCode::TOO_MANY_REQUESTS -} - -/// Full-jitter exponential backoff: a random delay in -/// `[0, min(RETRY_MAX_BACKOFF_MS, RETRY_BASE_MS * 2^attempt)]`. The jitter -/// decorrelates concurrent publishers so their retries don't collide into the -/// same burst that failed them. -fn backoff_with_jitter(attempt: u32) -> Duration { - let exp = RETRY_BASE_MS.saturating_mul(1u64 << attempt.min(16)); - Duration::from_millis(jitter_below(exp.min(RETRY_MAX_BACKOFF_MS))) -} - -/// A value in `[0, max]`, seeded from the wall clock's sub-second nanos — enough -/// entropy to spread retries across processes without pulling in an RNG crate. -fn jitter_below(max: u64) -> u64 { - if max == 0 { - return 0; - } - let nanos = SystemTime::now() - .duration_since(UNIX_EPOCH) - .map(|d| d.subsec_nanos() as u64) - .unwrap_or(0); - nanos % (max + 1) -} - -#[cfg(test)] -mod tests { - use super::*; - use crypto::Ed25519SigningKey; - - /// `encode_payload` / `decode_payload` round-trip, including a key_package - /// containing bytes that a delimiter scheme would choke on (`:`, `|`, NUL). - #[test] - fn payload_roundtrips_with_arbitrary_bytes() { - let ts = 1_700_000_000_000u64; - let key_package = b"mls:bytes|with\x00delimiters".to_vec(); - - let payload = encode_payload(ts, &key_package); - let (got_ts, got_kp) = decode_payload(&payload).unwrap(); - assert_eq!(got_ts, ts); - assert_eq!(got_kp, key_package.as_slice()); - } - - #[test] - fn decode_rejects_short_payload() { - assert!(decode_payload(&[0u8; 7]).is_none()); - } - - /// Tampering with any byte of the payload breaks verification. - #[test] - fn signature_binds_payload() { - let signing = Ed25519SigningKey::generate(); - let verifying = signing.verifying_key(); - - let payload = encode_payload(1_700_000_000_000, b"original-keypackage"); - let signature = signing.sign(&payload); - - let tampered = encode_payload(1_700_000_000_000, b"tampered-keypackage"); - verifying - .verify(&tampered, &signature) - .expect_err("signature must not verify against a different payload"); - } - - /// End-to-end of the wire crypto: verify over the received payload bytes - /// using the key recovered from device_id, exactly as `retrieve` does. - #[test] - fn sign_then_verify_over_payload() { - let signing = Ed25519SigningKey::generate(); - let pubkey: [u8; 32] = signing.verifying_key().as_ref().try_into().unwrap(); - let payload = encode_payload(1_700_000_000_000, b"fake-mls-keypackage-bytes"); - let signature = signing.sign(&payload); - - // retrieve side: recover key from device_id (hex of pubkey), verify payload. - let device_id = hex::encode(pubkey); - let recovered: [u8; 32] = hex::decode(&device_id) - .unwrap() - .as_slice() - .try_into() - .unwrap(); - Ed25519VerifyingKey::from_bytes(&recovered) - .unwrap() - .verify(&payload, &signature) - .expect("recovered key must verify the register-time signature"); - } - - /// Only 5xx and 429 are retried; 2xx/4xx are returned to the caller as-is. - #[test] - fn only_5xx_and_429_are_transient() { - use reqwest::StatusCode; - for s in [500u16, 502, 503, 504, 429] { - assert!( - is_transient_status(StatusCode::from_u16(s).unwrap()), - "{s} should be retried" - ); - } - for s in [200u16, 201, 400, 401, 404, 409] { - assert!( - !is_transient_status(StatusCode::from_u16(s).unwrap()), - "{s} should not be retried" - ); - } - } - - /// Backoff never exceeds the exponential ceiling for its attempt, nor the - /// absolute cap — and the exponent shift can't overflow at high attempts. - #[test] - fn backoff_stays_within_the_cap() { - for attempt in 0..40u32 { - let ceiling = RETRY_BASE_MS - .saturating_mul(1u64 << attempt.min(16)) - .min(RETRY_MAX_BACKOFF_MS); - let delay = backoff_with_jitter(attempt).as_millis() as u64; - assert!(delay <= ceiling, "attempt {attempt}: {delay} > {ceiling}"); - } - } - - #[test] - fn jitter_is_bounded() { - assert_eq!(jitter_below(0), 0); - for _ in 0..200 { - assert!(jitter_below(50) <= 50); - } - } -} diff --git a/extensions/components/src/contact_registry/store.rs b/extensions/components/src/contact_registry/store.rs new file mode 100644 index 0000000..251cf39 --- /dev/null +++ b/extensions/components/src/contact_registry/store.rs @@ -0,0 +1,636 @@ +use std::fmt::{self, Debug}; +use std::time::{Duration, SystemTime, UNIX_EPOCH}; + +use base64::Engine; +use base64::engine::general_purpose::STANDARD as BASE64; +use chat_proto::logoschat::store::{AccountSubmissionV1, KeyPackageSubmissionV1}; +use crypto::{Ed25519Signature, Ed25519VerifyingKey}; +use libchat::{AddressedEnvelope, DeliveryService, IdentityProvider, RegistrationService}; +use logos_account::{AccountDirectory, BundleError, DeviceSet, SignedDeviceBundle, verify_bundle}; +use prost::Message; +use prost::bytes::Bytes; +use serde::{Deserialize, Serialize}; + +/// Delivery address the store listens on for keypackage submissions. The +/// transport maps it to its content topic (e.g. +/// `/logos-chat/1/store-keypackage-v0/proto` on logos-delivery); the store +/// subscribes to the same topic. +pub const KEYPACKAGE_SUBMIT_ADDRESS: &str = "store-keypackage-v0"; + +/// Delivery address the store listens on for account device-list bundles. +pub const ACCOUNT_SUBMIT_ADDRESS: &str = "store-account-v0"; + +/// Request timeout for the store's HTTP API (queries, and submissions in +/// [`RegistryPublishMode::Http`]). +const HTTP_TIMEOUT: Duration = Duration::from_secs(10); + +/// How a [`ContactRegistry`] submits bundles to the store. Reads always use +/// the store's HTTP query API; only the write half switches. +#[derive(Clone, Copy, Debug, Default, PartialEq, Eq)] +pub enum RegistryPublishMode { + /// Submit via the store's HTTP POST endpoints (synchronous, acknowledged). + #[default] + Http, + /// Publish over the delivery network on the well-known store addresses; + /// the store subscribes and persists what verifies. Fire-and-forget — + /// there is no per-submission acknowledgement, which the registry can + /// afford because consumers verify every bundle on retrieval anyway. + Delivery, +} + +/// The keypackage store and account → device directory. +/// +/// Reads (keypackage retrieve, account fetch) always go over the store's HTTP +/// query API. Writes (register, publish) go over whichever wire +/// [`RegistryPublishMode`] selects: the store's HTTP POST endpoints (a JSON +/// body with hex + base64 fields) or a protobuf submission +/// ([`KeyPackageSubmissionV1`] / [`AccountSubmissionV1`]) published on the +/// well-known store addresses — matching the `/proto` content topics those +/// addresses map to, and carrying the keys, payload and signature as raw bytes. +/// +/// A single registry serves both wires so it can be used behind one +/// `ChatClient` registry type; the delivery transport `D` is unused in +/// [`RegistryPublishMode::Http`]. +#[derive(Clone)] +pub struct ContactRegistry { + base_url: String, + http: reqwest::blocking::Client, + delivery: D, + publish_mode: RegistryPublishMode, +} + +impl Debug for ContactRegistry { + fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { + f.debug_struct("ContactRegistry") + .field("base_url", &self.base_url) + .field("publish_mode", &self.publish_mode) + .field("delivery", &self.delivery) + .finish() + } +} + +#[derive(Debug, thiserror::Error)] +pub enum ContactRegistryError { + #[error("http: {0}")] + Http(#[from] reqwest::Error), + #[error("server returned status {0}: {1}")] + Server(u16, String), + #[error("decode: {0}")] + Decode(String), + #[error("clock before unix epoch")] + Clock, + #[error("signature verification failed")] + SignatureInvalid, + #[error("bundle: {0}")] + Bundle(#[from] BundleError), + #[error("publish over delivery: {0}")] + Publish(String), +} + +impl ContactRegistry { + /// A registry that queries the store's HTTP API at `base_url` and submits + /// per `publish_mode` — over `delivery` or over the same HTTP API. + pub fn new( + delivery: D, + base_url: impl Into, + publish_mode: RegistryPublishMode, + ) -> Self { + let http = reqwest::blocking::Client::builder() + .timeout(HTTP_TIMEOUT) + .build() + .expect("reqwest client builder is infallible with these options"); + Self { + base_url: base_url.into().trim_end_matches('/').to_string(), + http, + delivery, + publish_mode, + } + } +} + +impl ContactRegistry { + /// POST `body` as JSON to `path` on the store, mapping a non-success status + /// to [`ContactRegistryError::Server`] with the server's own message. + fn http_post(&self, path: &str, body: &S) -> Result<(), ContactRegistryError> { + let url = format!("{}{}", self.base_url, path); + let resp = send_retrying(|| self.http.post(&url).json(body))?; + if !resp.status().is_success() { + let status = resp.status().as_u16(); + return Err(ContactRegistryError::Server( + status, + resp.text().unwrap_or_default(), + )); + } + Ok(()) + } + + /// GET `url`, returning `None` on 404 (never published) and the decoded, + /// still-unverified bundle on success. The caller verifies the signature. + fn http_fetch(&self, url: &str) -> Result, ContactRegistryError> { + let resp = send_retrying(|| self.http.get(url))?; + if resp.status().as_u16() == 404 { + return Ok(None); + } + if !resp.status().is_success() { + let status = resp.status().as_u16(); + return Err(ContactRegistryError::Server( + status, + resp.text().unwrap_or_default(), + )); + } + let body: FetchResponse = resp.json()?; + let payload = BASE64 + .decode(&body.payload) + .map_err(|e| ContactRegistryError::Decode(e.to_string()))?; + let signature: [u8; 64] = BASE64 + .decode(&body.signature) + .map_err(|e| ContactRegistryError::Decode(e.to_string()))? + .as_slice() + .try_into() + .map_err(|_| ContactRegistryError::Decode("signature not 64 bytes".into()))?; + Ok(Some(FetchedBundle { payload, signature })) + } +} + +impl ContactRegistry { + /// Encode `submission` as protobuf and publish it on `delivery_address`. + /// Protobuf encoding into a `Vec` cannot fail, so the only error here is the + /// transport's. + fn publish_submission( + &mut self, + delivery_address: &str, + submission: &M, + ) -> Result<(), ContactRegistryError> { + self.delivery + .publish(AddressedEnvelope { + delivery_address: delivery_address.to_string(), + data: submission.encode_to_vec(), + }) + .map_err(|e| ContactRegistryError::Publish(e.to_string())) + } +} + +impl RegistrationService for ContactRegistry { + type Error = ContactRegistryError; + + fn register( + &mut self, + identity: &dyn IdentityProvider, + key_bundle: Vec, + ) -> Result<(), Self::Error> { + let timestamp_ms = SystemTime::now() + .duration_since(UNIX_EPOCH) + .map_err(|_| ContactRegistryError::Clock)? + .as_millis() as u64; + + // The signed bytes are the same on both wires; only the encoding of the + // submission around them differs. Sign once, then branch on transport. + let payload = encode_payload(timestamp_ms, &key_bundle); + let signature = identity.sign(&payload); + let device_id = identity.public_key().as_ref(); + + match self.publish_mode { + RegistryPublishMode::Http => self.http_post( + "/v0/keypackage", + &SubmitRequest { + device_id: hex::encode(device_id), + payload: BASE64.encode(&payload), + signature: BASE64.encode(signature.as_ref()), + }, + ), + RegistryPublishMode::Delivery => { + let req = KeyPackageSubmissionV1 { + device_id: Bytes::copy_from_slice(device_id), + payload: Bytes::from(payload), + signature: Bytes::copy_from_slice(signature.as_ref()), + }; + self.publish_submission(KEYPACKAGE_SUBMIT_ADDRESS, &req) + } + } + } + + fn retrieve(&self, device_id: &str) -> Result>, Self::Error> { + let url = format!("{}/v0/keypackage/{}", self.base_url, device_id); + let Some(FetchedBundle { payload, signature }) = self.http_fetch(&url)? else { + return Ok(None); + }; + + // Verify over the received payload bytes, using the key we asked for + // (`device_id`). A bundle the requested device didn't sign won't verify. + let device_pubkey: [u8; 32] = hex::decode(device_id) + .map_err(|e| ContactRegistryError::Decode(e.to_string()))? + .as_slice() + .try_into() + .map_err(|_| ContactRegistryError::Decode("device_id not a 32-byte key".into()))?; + let verifying_key = Ed25519VerifyingKey::from_bytes(&device_pubkey) + .map_err(|_| ContactRegistryError::Decode("device_id not a valid ed25519 vk".into()))?; + verifying_key + .verify(&payload, &Ed25519Signature::from(signature)) + .map_err(|_| ContactRegistryError::SignatureInvalid)?; + + let (_timestamp_ms, key_package) = decode_payload(&payload) + .ok_or_else(|| ContactRegistryError::Decode("short payload".into()))?; + Ok(Some(key_package.to_vec())) + } +} + +impl AccountDirectory for ContactRegistry { + type Error = ContactRegistryError; + + fn publish(&mut self, bundle: &SignedDeviceBundle) -> Result<(), Self::Error> { + // The bundle is already signed; both wires carry its exact bytes. + match self.publish_mode { + RegistryPublishMode::Http => self.http_post( + "/v0/account", + &SubmitAccountRequest { + account_pub: hex::encode(bundle.account_pub.as_ref()), + payload: BASE64.encode(&bundle.payload), + signature: BASE64.encode(bundle.signature.as_ref()), + }, + ), + RegistryPublishMode::Delivery => { + let req = AccountSubmissionV1 { + account_pub: Bytes::copy_from_slice(bundle.account_pub.as_ref()), + payload: Bytes::copy_from_slice(&bundle.payload), + signature: Bytes::copy_from_slice(bundle.signature.as_ref()), + }; + self.publish_submission(ACCOUNT_SUBMIT_ADDRESS, &req) + } + } + } + + fn fetch(&self, account: &Ed25519VerifyingKey) -> Result, Self::Error> { + let url = format!( + "{}/v0/account/{}", + self.base_url, + hex::encode(account.as_ref()) + ); + let Some(FetchedBundle { payload, signature }) = self.http_fetch(&url)? else { + return Ok(None); + }; + + // The directory service is untrusted: verify the account signature over + // the exact received bytes, and that the bundle is bound to the account + // we asked for, before handing back any device keys. + let bundle = SignedDeviceBundle { + account_pub: account.clone(), + payload, + signature: Ed25519Signature::from(signature), + }; + let device_set = verify_bundle(account, &bundle)?; + Ok(Some(device_set)) + } +} + +/// Keypackage submission as the HTTP POST body. The delivery path carries the +/// same fields as protobuf; this JSON shape is the store's HTTP endpoint only. +#[derive(Debug, Serialize)] +struct SubmitRequest { + /// hex of the 32-byte device verifying key — the verification + storage key. + device_id: String, + /// base64 of the canonical signed payload (see [`encode_payload`]). + payload: String, + /// base64 of the 64-byte Ed25519 signature over `payload`. + signature: String, +} + +/// Account device-list submission as the HTTP POST body, like [`SubmitRequest`]. +#[derive(Debug, Serialize)] +struct SubmitAccountRequest { + /// hex of the 32-byte account verifying key — verification + storage key. + account_pub: String, + /// base64 of the canonical signed device-list payload. + payload: String, + /// base64 of the 64-byte account signature over `payload`. + signature: String, +} + +/// The `payload` + `signature` of a store fetch response; both keypackage and +/// account queries return this shape. +#[derive(Debug, Deserialize)] +struct FetchResponse { + payload: String, + signature: String, +} + +/// A fetch response with its base64 fields decoded but not yet verified. +struct FetchedBundle { + payload: Vec, + signature: [u8; 64], +} + +/// Canonical binary payload — the bytes that are both signed and transmitted +/// verbatim. Opaque to the server; decoded only by consumers: +/// +/// ```text +/// timestamp_ms : u64 little-endian (8 bytes) +/// key_package : remaining bytes (variable, last → no length prefix needed) +/// ``` +/// +/// The fixed-width field first with the one variable field last makes every +/// byte string parse exactly one way — no delimiter, no ambiguity, even though +/// `key_package` is arbitrary bytes. The device verifying key is carried +/// alongside as `device_id`, not embedded here. +fn encode_payload(timestamp_ms: u64, key_package: &[u8]) -> Vec { + let mut out = Vec::with_capacity(8 + key_package.len()); + out.extend_from_slice(×tamp_ms.to_le_bytes()); + out.extend_from_slice(key_package); + out +} + +/// Inverse of [`encode_payload`]. Returns `None` if the payload is shorter than +/// the fixed header (`8`). +fn decode_payload(payload: &[u8]) -> Option<(u64, &[u8])> { + if payload.len() < 8 { + return None; + } + let timestamp_ms = u64::from_le_bytes(payload[..8].try_into().ok()?); + Some((timestamp_ms, &payload[8..])) +} + +/// Retry budget for the registry's transient, load-induced 5xx/429 responses. +/// The service is reliable request-by-request but sheds concurrent bursts, so a +/// few backed-off retries let a request land once the burst clears. On that path +/// each retry returns fast, so the added cost is the ~3s worst-case backoff sum, +/// well inside chat_module's ~20s init IPC budget. A fully unreachable registry +/// instead costs up to MAX_RETRIES times the reqwest timeout, which no retry +/// budget can rescue. +const MAX_RETRIES: u32 = 4; +const RETRY_BASE_MS: u64 = 200; +const RETRY_MAX_BACKOFF_MS: u64 = 2000; + +/// Send a request built by `build`, retrying transient failures — network errors +/// and 5xx/429 responses — with exponential backoff and full jitter. The +/// registry is reliable request-by-request but sheds concurrent bursts with a +/// 5xx, so a backed-off retry lands once the burst clears; a 4xx (and any other +/// final response) is returned to the caller unchanged. `build` is re-invoked per +/// attempt because sending consumes the builder. +fn send_retrying( + build: impl Fn() -> reqwest::blocking::RequestBuilder, +) -> Result { + let mut attempt = 0; + loop { + let outcome = build().send(); + let transient = match &outcome { + Err(_) => true, // network error / timeout: worth another try + Ok(resp) => is_transient_status(resp.status()), + }; + if !transient || attempt >= MAX_RETRIES { + return Ok(outcome?); + } + std::thread::sleep(backoff_with_jitter(attempt)); + attempt += 1; + } +} + +/// Whether a response status is worth retrying: 5xx (the registry sheds +/// concurrent load with these) or 429 (explicit backpressure). A 4xx is the +/// caller's fault and won't change on retry. +fn is_transient_status(status: reqwest::StatusCode) -> bool { + status.is_server_error() || status == reqwest::StatusCode::TOO_MANY_REQUESTS +} + +/// Full-jitter exponential backoff: a random delay in +/// `[0, min(RETRY_MAX_BACKOFF_MS, RETRY_BASE_MS * 2^attempt)]`. The jitter +/// decorrelates concurrent publishers so their retries don't collide into the +/// same burst that failed them. +fn backoff_with_jitter(attempt: u32) -> Duration { + let exp = RETRY_BASE_MS.saturating_mul(1u64 << attempt.min(16)); + Duration::from_millis(jitter_below(exp.min(RETRY_MAX_BACKOFF_MS))) +} + +/// A value in `[0, max]`, seeded from the wall clock's sub-second nanos — enough +/// entropy to spread retries across processes without pulling in an RNG crate. +fn jitter_below(max: u64) -> u64 { + if max == 0 { + return 0; + } + let nanos = SystemTime::now() + .duration_since(UNIX_EPOCH) + .map(|d| d.subsec_nanos() as u64) + .unwrap_or(0); + nanos % (max + 1) +} + +#[cfg(test)] +mod tests { + use super::*; + use crypto::Ed25519SigningKey; + use libchat::{IdentId, IdentIdRef}; + + #[derive(Debug, Default)] + struct CapturingDelivery { + published: Vec, + } + + impl DeliveryService for CapturingDelivery { + type Error = std::convert::Infallible; + fn publish(&mut self, envelope: AddressedEnvelope) -> Result<(), Self::Error> { + self.published.push(envelope); + Ok(()) + } + fn subscribe(&mut self, _delivery_address: &str) -> Result<(), Self::Error> { + Ok(()) + } + } + + struct TestIdent { + id: IdentId, + key: Ed25519SigningKey, + verifying: Ed25519VerifyingKey, + } + + impl TestIdent { + fn new() -> Self { + let key = Ed25519SigningKey::generate(); + let verifying = key.verifying_key(); + Self { + id: IdentId::new("test"), + key, + verifying, + } + } + } + + impl IdentityProvider for TestIdent { + fn id(&self) -> IdentIdRef<'_> { + &self.id + } + fn display_name(&self) -> String { + self.id.to_string() + } + fn sign(&self, payload: &[u8]) -> Ed25519Signature { + self.key.sign(payload) + } + fn public_key(&self) -> &Ed25519VerifyingKey { + &self.verifying + } + } + + #[test] + fn register_publishes_store_submission_on_keypackage_address() { + let mut registry = ContactRegistry::new( + CapturingDelivery::default(), + "http://unused.invalid", + RegistryPublishMode::Delivery, + ); + let ident = TestIdent::new(); + let key_bundle = b"kp-bytes".to_vec(); + registry.register(&ident, key_bundle.clone()).unwrap(); + + let [envelope] = ®istry.delivery.published[..] else { + panic!("expected exactly one published envelope"); + }; + assert_eq!(envelope.delivery_address, KEYPACKAGE_SUBMIT_ADDRESS); + + // Decode as the store does: the bytes on the wire are a protobuf + // submission, so a field-number or type change here breaks ingestion. + let wire = KeyPackageSubmissionV1::decode(&envelope.data[..]).unwrap(); + assert_eq!(wire.device_id.as_ref(), ident.verifying.as_ref()); + // The store verifies the signature over the payload bytes under the + // device key before persisting — the submission must pass that check. + assert!(wire.payload.ends_with(&key_bundle)); + let signature: [u8; 64] = wire.signature.as_ref().try_into().unwrap(); + ident + .verifying + .verify(&wire.payload, &Ed25519Signature::from(signature)) + .expect("store-side verification must succeed"); + } + + #[test] + fn account_publish_targets_account_address_verbatim() { + let mut registry = ContactRegistry::new( + CapturingDelivery::default(), + "http://unused.invalid", + RegistryPublishMode::Delivery, + ); + let account = Ed25519SigningKey::generate(); + let payload = b"signed-device-list".to_vec(); + let bundle = SignedDeviceBundle { + account_pub: account.verifying_key(), + signature: account.sign(&payload), + payload: payload.clone(), + }; + registry.publish(&bundle).unwrap(); + + let [envelope] = ®istry.delivery.published[..] else { + panic!("expected exactly one published envelope"); + }; + assert_eq!(envelope.delivery_address, ACCOUNT_SUBMIT_ADDRESS); + + let wire = AccountSubmissionV1::decode(&envelope.data[..]).unwrap(); + assert_eq!(wire.account_pub.as_ref(), bundle.account_pub.as_ref()); + // Payload travels verbatim so the store and consumers verify the exact + // signed bytes. + assert_eq!(wire.payload.as_ref(), payload.as_slice()); + assert_eq!(wire.signature.as_ref(), bundle.signature.as_ref()); + } + + #[test] + fn http_mode_never_touches_the_delivery_service() { + // Port 9 (discard) refuses immediately; the point is only that the + // submission goes down the HTTP path, not over delivery. + let mut registry = ContactRegistry::new( + CapturingDelivery::default(), + "http://127.0.0.1:9", + RegistryPublishMode::Http, + ); + let err = registry.register(&TestIdent::new(), vec![1]).unwrap_err(); + assert!(matches!(err, ContactRegistryError::Http(_))); + assert!(registry.delivery.published.is_empty()); + } + + /// `encode_payload` / `decode_payload` round-trip, including a key_package + /// containing bytes that a delimiter scheme would choke on (`:`, `|`, NUL). + #[test] + fn payload_roundtrips_with_arbitrary_bytes() { + let ts = 1_700_000_000_000u64; + let key_package = b"mls:bytes|with\x00delimiters".to_vec(); + + let payload = encode_payload(ts, &key_package); + let (got_ts, got_kp) = decode_payload(&payload).unwrap(); + assert_eq!(got_ts, ts); + assert_eq!(got_kp, key_package.as_slice()); + } + + #[test] + fn decode_rejects_short_payload() { + assert!(decode_payload(&[0u8; 7]).is_none()); + } + + /// Tampering with any byte of the payload breaks verification. + #[test] + fn signature_binds_payload() { + let signing = Ed25519SigningKey::generate(); + let verifying = signing.verifying_key(); + + let payload = encode_payload(1_700_000_000_000, b"original-keypackage"); + let signature = signing.sign(&payload); + + let tampered = encode_payload(1_700_000_000_000, b"tampered-keypackage"); + verifying + .verify(&tampered, &signature) + .expect_err("signature must not verify against a different payload"); + } + + /// End-to-end of the wire crypto: verify over the received payload bytes + /// using the key recovered from device_id, exactly as `retrieve` does. + #[test] + fn sign_then_verify_over_payload() { + let signing = Ed25519SigningKey::generate(); + let pubkey: [u8; 32] = signing.verifying_key().as_ref().try_into().unwrap(); + let payload = encode_payload(1_700_000_000_000, b"fake-mls-keypackage-bytes"); + let signature = signing.sign(&payload); + + // retrieve side: recover key from device_id (hex of pubkey), verify payload. + let device_id = hex::encode(pubkey); + let recovered: [u8; 32] = hex::decode(&device_id) + .unwrap() + .as_slice() + .try_into() + .unwrap(); + Ed25519VerifyingKey::from_bytes(&recovered) + .unwrap() + .verify(&payload, &signature) + .expect("recovered key must verify the register-time signature"); + } + + /// Only 5xx and 429 are retried; 2xx/4xx are returned to the caller as-is. + #[test] + fn only_5xx_and_429_are_transient() { + use reqwest::StatusCode; + for s in [500u16, 502, 503, 504, 429] { + assert!( + is_transient_status(StatusCode::from_u16(s).unwrap()), + "{s} should be retried" + ); + } + for s in [200u16, 201, 400, 401, 404, 409] { + assert!( + !is_transient_status(StatusCode::from_u16(s).unwrap()), + "{s} should not be retried" + ); + } + } + + /// Backoff never exceeds the exponential ceiling for its attempt, nor the + /// absolute cap — and the exponent shift can't overflow at high attempts. + #[test] + fn backoff_stays_within_the_cap() { + for attempt in 0..40u32 { + let ceiling = RETRY_BASE_MS + .saturating_mul(1u64 << attempt.min(16)) + .min(RETRY_MAX_BACKOFF_MS); + let delay = backoff_with_jitter(attempt).as_millis() as u64; + assert!(delay <= ceiling, "attempt {attempt}: {delay} > {ceiling}"); + } + } + + #[test] + fn jitter_is_bounded() { + assert_eq!(jitter_below(0), 0); + for _ in 0..200 { + assert!(jitter_below(50) <= 50); + } + } +} diff --git a/extensions/components/src/lib.rs b/extensions/components/src/lib.rs index 92ad4a4..70e05f3 100644 --- a/extensions/components/src/lib.rs +++ b/extensions/components/src/lib.rs @@ -4,7 +4,10 @@ mod storage; mod wakeup; pub use contact_registry::ephemeral::EphemeralRegistry; -pub use contact_registry::http::{HttpRegistry, HttpRegistryError}; +pub use contact_registry::store::{ + ACCOUNT_SUBMIT_ADDRESS, ContactRegistry, ContactRegistryError, KEYPACKAGE_SUBMIT_ADDRESS, + RegistryPublishMode, +}; pub use delivery::*; pub use storage::*; pub use wakeup::*; diff --git a/extensions/embedded-logos-delivery/build.rs b/extensions/embedded-logos-delivery/build.rs deleted file mode 100644 index 2773451..0000000 --- a/extensions/embedded-logos-delivery/build.rs +++ /dev/null @@ -1,160 +0,0 @@ -use std::fs; -use std::path::{Path, PathBuf}; -use std::process::Command; - -fn main() { - println!("cargo:rerun-if-env-changed=LOGOS_DELIVERY_LIB_DIR"); - - let Some(lib_dir) = locate_lib_dir() else { - println!( - "cargo:warning=liblogosdelivery could not be located; `cargo check`/\ - `clippy` will pass, but building or testing will fail at link. Enter \ - the dev shell with `nix develop` or set LOGOS_DELIVERY_LIB_DIR to \ - the directory containing the library." - ); - return; - }; - - let out_dir = std::env::var("OUT_DIR").expect("OUT_DIR not set"); - let target_os = std::env::var("CARGO_CFG_TARGET_OS").unwrap_or_default(); - - // The shipped library carries a relocatable install name (@rpath on macOS, - // $ORIGIN soname on Linux), which would force every downstream BINARY to - // inject its own RPATH. Cargo propagates `rustc-link-search` and - // `rustc-link-lib` across crates, but NOT `rustc-link-arg` (the rpath) — so - // that relocatable name is exactly what makes consumers need their own - // build.rs. Instead, stamp a private copy with an ABSOLUTE install name; - // the propagating search + lib directives are then sufficient and consumers - // need zero build-script glue. - match target_os.as_str() { - "macos" => stamp_absolute_macos(&lib_dir, &out_dir), - "linux" => stamp_absolute_linux(&lib_dir, &out_dir), - other => panic!("unsupported OS for logos-delivery transport: {other}"), - } - - println!("cargo:rustc-link-search=native={out_dir}"); - println!("cargo:rustc-link-lib=dylib=logosdelivery"); -} - -/// Locate the native library directory as an ABSOLUTE, canonical path. Prefers -/// `LOGOS_DELIVERY_LIB_DIR`, then falls back to building it via nix. Returns -/// `None` when neither is available (e.g. `cargo check` without nix). -fn locate_lib_dir() -> Option { - if let Ok(dir) = std::env::var("LOGOS_DELIVERY_LIB_DIR") { - if let Some(resolved) = resolve_lib_dir(&dir) { - return Some(resolved); - } - println!( - "cargo:warning=LOGOS_DELIVERY_LIB_DIR='{dir}' could not be resolved; \ - falling back to `nix build`" - ); - } - resolve_lib_dir(&nix_build_logos_delivery()?) -} - -/// Resolve a lib dir to an absolute, canonical path. Cargo runs build scripts -/// with the cwd set to the crate dir, but a relative value (e.g. CI's -/// `./result/lib`) is anchored at the flake/workspace root where `nix build` -/// drops `result`. Canonicalizing also follows the `result` symlink to the -/// immutable store path, so the stamped install name / soname stays stable. -fn resolve_lib_dir(dir: &str) -> Option { - let path = Path::new(dir); - let anchored = if path.is_absolute() { - path.to_path_buf() - } else { - let manifest = std::env::var("CARGO_MANIFEST_DIR").ok()?; - Path::new(&find_flake_root(&manifest)?).join(path) - }; - anchored.canonicalize().ok() -} - -/// Copy `liblogosdelivery.dylib` into `OUT_DIR` and rewrite its install name to -/// the absolute store path. The consumer records that absolute path, so dyld -/// loads the original file directly — whose own `@loader_path` RPATH resolves -/// `librln.dylib` beside it — with no RPATH needed on the consumer. -fn stamp_absolute_macos(lib_dir: &Path, out_dir: &str) { - let src = lib_dir.join("liblogosdelivery.dylib"); - let dst = format!("{out_dir}/liblogosdelivery.dylib"); - copy_writable(&src, Path::new(&dst)); - run("install_name_tool", &["-id", path_str(&src), &dst]); - println!("cargo:rerun-if-changed={}", src.display()); -} - -/// Linux equivalent: an absolute `DT_SONAME` is recorded verbatim in the -/// consumer's `DT_NEEDED`, so `ld.so` loads it by path with no RPATH. Requires -/// `patchelf` at build time (provided by the nix devshell). -fn stamp_absolute_linux(lib_dir: &Path, out_dir: &str) { - let src = lib_dir.join("liblogosdelivery.so"); - let dst = format!("{out_dir}/liblogosdelivery.so"); - copy_writable(&src, Path::new(&dst)); - run("patchelf", &["--set-soname", path_str(&src), &dst]); - println!("cargo:rerun-if-changed={}", src.display()); -} - -fn path_str(p: &Path) -> &str { - p.to_str() - .unwrap_or_else(|| panic!("non-UTF-8 path: {}", p.display())) -} - -fn copy_writable(src: &Path, dst: &Path) { - use std::os::unix::fs::PermissionsExt; - - fs::copy(src, dst) - .unwrap_or_else(|e| panic!("copy {} -> {}: {e}", src.display(), dst.display())); - // Store-sourced files are read-only; restore owner write so the install - // name / soname can be rewritten. - fs::set_permissions(dst, fs::Permissions::from_mode(0o644)).unwrap(); -} - -fn run(cmd: &str, args: &[&str]) { - let status = Command::new(cmd) - .args(args) - .status() - .unwrap_or_else(|e| panic!("failed to run `{cmd}`: {e}")); - assert!(status.success(), "`{cmd} {args:?}` failed with {status}"); -} - -fn nix_build_logos_delivery() -> Option { - let manifest_dir = std::env::var("CARGO_MANIFEST_DIR").ok()?; - let flake_root = find_flake_root(&manifest_dir)?; - - println!("cargo:rerun-if-changed={flake_root}/flake.lock"); - - let output = Command::new("nix") - .args([ - "build", - ".#logos-delivery", - "--no-link", - "--print-out-paths", - ]) - .current_dir(&flake_root) - .output() - .ok()?; - - if !output.status.success() { - let stderr = String::from_utf8_lossy(&output.stderr); - println!("cargo:warning=nix build .#logos-delivery failed: {stderr}"); - return None; - } - - let store_path = String::from_utf8(output.stdout).ok()?; - let lib_dir = format!("{}/lib", store_path.trim()); - - if std::path::Path::new(&lib_dir).exists() { - Some(lib_dir) - } else { - None - } -} - -fn find_flake_root(start: &str) -> Option { - let mut path = std::path::PathBuf::from(start); - loop { - if path.join("flake.nix").exists() { - return Some(path.to_string_lossy().into_owned()); - } - if !path.pop() { - return None; - } - } -} diff --git a/extensions/logos-delivery-rust/Cargo.toml b/extensions/logos-delivery-rust/Cargo.toml index e9dc546..bae3f20 100644 --- a/extensions/logos-delivery-rust/Cargo.toml +++ b/extensions/logos-delivery-rust/Cargo.toml @@ -7,8 +7,6 @@ links = "logosdelivery" [dependencies] # Workspace dependencies (sorted) crossbeam-channel = { workspace = true } -libchat = { workspace = true } -logos-generic-chat = { workspace = true } # External dependencies (sorted) base64 = "0.22" diff --git a/extensions/logos-delivery-rust/build.rs b/extensions/logos-delivery-rust/build.rs index 2773451..e015061 100644 --- a/extensions/logos-delivery-rust/build.rs +++ b/extensions/logos-delivery-rust/build.rs @@ -4,6 +4,7 @@ use std::process::Command; fn main() { println!("cargo:rerun-if-env-changed=LOGOS_DELIVERY_LIB_DIR"); + println!("cargo:rerun-if-env-changed=LOGOS_DELIVERY_RELOCATABLE"); let Some(lib_dir) = locate_lib_dir() else { println!( @@ -18,22 +19,50 @@ fn main() { let out_dir = std::env::var("OUT_DIR").expect("OUT_DIR not set"); let target_os = std::env::var("CARGO_CFG_TARGET_OS").unwrap_or_default(); - // The shipped library carries a relocatable install name (@rpath on macOS, - // $ORIGIN soname on Linux), which would force every downstream BINARY to - // inject its own RPATH. Cargo propagates `rustc-link-search` and - // `rustc-link-lib` across crates, but NOT `rustc-link-arg` (the rpath) — so - // that relocatable name is exactly what makes consumers need their own - // build.rs. Instead, stamp a private copy with an ABSOLUTE install name; - // the propagating search + lib directives are then sufficient and consumers - // need zero build-script glue. match target_os.as_str() { - "macos" => stamp_absolute_macos(&lib_dir, &out_dir), - "linux" => stamp_absolute_linux(&lib_dir, &out_dir), + "macos" | "linux" => {} other => panic!("unsupported OS for logos-delivery transport: {other}"), } - println!("cargo:rustc-link-search=native={out_dir}"); + // Two linking modes, because dev builds and *distributable* builds want + // opposite things out of the library's install name / soname. + if relocatable() { + // Distribution: link the shipped library in place and leave its + // relocatable name (@rpath on macOS, $ORIGIN soname on Linux) intact, + // so the consumer can copy it into its own bundle and resolve it from + // there. The library's own @loader_path/$ORIGIN rpath then finds + // librln beside it. This costs the consumer some build-script glue -- + // on macOS it MUST add an rpath, since cargo does not propagate + // `rustc-link-arg` across crates -- which is exactly what the default + // mode below exists to avoid. `lib_dir` is published as + // DEP_LOGOSDELIVERY_LIB_DIR so direct dependents can locate the + // libraries to bundle. + println!("cargo:rustc-link-search=native={}", lib_dir.display()); + } else { + // Default (dev): stamp a private copy with an ABSOLUTE install name. + // The propagating search + lib directives are then sufficient and + // consumers need zero build-script glue -- but the resulting binary + // hardcodes a nix store path and only runs on this machine. + match target_os.as_str() { + "macos" => stamp_absolute_macos(&lib_dir, &out_dir), + "linux" => stamp_absolute_linux(&lib_dir, &out_dir), + _ => unreachable!("target OS validated above"), + } + println!("cargo:rustc-link-search=native={out_dir}"); + } + println!("cargo:rustc-link-lib=dylib=logosdelivery"); + println!("cargo:lib_dir={}", lib_dir.display()); +} + +/// Opt-in relocatable linking for builds that get shipped to other machines. +/// Off by default so existing consumers (and this repo's own tests) keep the +/// zero-glue absolute-path behaviour. +fn relocatable() -> bool { + matches!( + std::env::var("LOGOS_DELIVERY_RELOCATABLE").as_deref(), + Ok("1") | Ok("true") + ) } /// Locate the native library directory as an ABSOLUTE, canonical path. Prefers