diff --git a/app/desktop/src-tauri/src/commands/request.rs b/app/desktop/src-tauri/src/commands/request.rs index f082bbe..4f5e5f5 100644 --- a/app/desktop/src-tauri/src/commands/request.rs +++ b/app/desktop/src-tauri/src/commands/request.rs @@ -128,11 +128,11 @@ async fn act_on_message_request_impl( ) .await .map_err(|error| error.to_string())?; - if sender_bundle.user_id != preview.author_user_id { - return Err( - "fetched identity bundle user_id does not match the Welcome author".to_string(), - ); - } + tapchat_core::contact_workflows::ensure_welcome_signed_by_bundle( + &preview, + &sender_bundle, + ) + .map_err(|error| error.to_string())?; let imported_sender_user_id = sender_bundle.user_id.clone(); drive_core_with_handle( diff --git a/contracts/emission-ledger.json b/contracts/emission-ledger.json index 3e53f4b..c34ad26 100644 --- a/contracts/emission-ledger.json +++ b/contracts/emission-ledger.json @@ -24,7 +24,8 @@ "enqueue_protected_app", "enqueue_or_create_direct_app", "flush_pending_direct_app", - "enqueue_lane_rotation" + "enqueue_lane_rotation", + "announce_lane_after_rebuild" ], "sites": [ { @@ -60,12 +61,12 @@ { "fn": "handle_inbox_records_batch", "trigger": "reaction", - "emits": "the contact-accepted reply that ends establishment, on applying the initiator's welcome" + "emits": "the contact-accepted reply, on applying the initiator's welcome, which ends establishment, or a race loser's rebuild welcome, where it is the first record of the new session the loser processes and so ends the loser's recovery; after a rebuild welcome, also a lane notice" }, { "fn": "finalize_reconcile_membership_rebootstrap", "trigger": "reaction", - "emits": "the welcome of a re-entry, on losing a race; or of a rebuild the user confirmed" + "emits": "the welcome of a re-entry, on losing a race; or of a rebuild the user confirmed; then the frames kept for a resend and a lane notice" }, { "fn": "resend_unconfirmed_frames", diff --git a/src/contact_workflows.rs b/src/contact_workflows.rs index 703a1ba..30799bc 100644 --- a/src/contact_workflows.rs +++ b/src/contact_workflows.rs @@ -6,9 +6,9 @@ use crate::cli::util::to_snake_case_json_string; use crate::contact_share::contact_share_url_user_id; use crate::error::{CoreError, CoreResult}; use crate::external_fetch::{fetch_external_json, ExternalResourceKind}; -use crate::ffi_api::{CoreCommand, CoreOutput, MessageRequestActionSummary}; -use crate::identity::IdentityManager; -use crate::model::{IdentityBundle, Validate}; +use crate::ffi_api::{CoreCommand, CoreOutput, MessageRequestActionSummary, WelcomePreview}; +use crate::identity::{parse_verifying_key, IdentityManager}; +use crate::model::{DeviceStatusKind, IdentityBundle, Validate}; use crate::transport_contract::{MessageRequestAction, MessageRequestItem}; pub async fn import_identity_bundle_into_profile( @@ -64,7 +64,7 @@ pub async fn accept_message_request_with_bundle_import( .filter(|value| !value.trim().is_empty()) .ok_or_else(|| anyhow!("the Welcome did not include an importable identity bundle"))?; let bundle = fetch_identity_bundle_from_url(&sender_bundle_share_url).await?; - ensure_fetched_bundle_matches_expected_sender(&bundle, &preview.author_user_id)?; + ensure_welcome_signed_by_bundle(&preview, &bundle)?; import_identity_bundle_into_profile(profile, driver, bundle).await?; let output = driver .run_command_until_idle(CoreCommand::ActOnMessageRequest { @@ -80,13 +80,36 @@ pub async fn accept_message_request_with_bundle_import( Ok(summary) } -fn ensure_fetched_bundle_matches_expected_sender( +/// Whether `bundle` belongs to whoever signed the Welcome that `preview` +/// describes. A request names its sender only inside the Welcome, and the +/// bundle comes from wherever the Welcome points, so neither is evidence on +/// its own: the Welcome is the bundle owner's only if the device key in the +/// bundle signed it. Checked before the bundle is imported, so a Welcome that +/// merely names someone leaves no contact behind. +pub fn ensure_welcome_signed_by_bundle( + preview: &WelcomePreview, bundle: &IdentityBundle, - expected_sender_user_id: &str, -) -> Result<()> { - if bundle.user_id != expected_sender_user_id { - return Err(anyhow!( - "fetched identity bundle user_id does not match the Welcome author" +) -> CoreResult<()> { + if bundle.user_id != preview.author_user_id { + return Err(CoreError::invalid_input( + "fetched identity bundle user_id does not match the Welcome author", + )); + } + let device = bundle + .devices + .iter() + .find(|device| { + device.device_id == preview.author_device_id + && matches!(device.status, DeviceStatusKind::Active) + }) + .ok_or_else(|| { + CoreError::invalid_input("the Welcome author is not an active device of the bundle") + })?; + let trusted = parse_verifying_key(&device.device_public_key)?; + let signer = parse_verifying_key(&preview.author_device_public_key)?; + if trusted.as_bytes() != signer.as_bytes() { + return Err(CoreError::invalid_input( + "the Welcome was not signed by the device key in the sender's bundle", )); } Ok(()) @@ -163,30 +186,41 @@ mod tests { use std::io::{Read, Write}; use std::net::TcpListener; - use super::{ensure_fetched_bundle_matches_expected_sender, fetch_identity_bundle_from_url}; + use super::{ensure_welcome_signed_by_bundle, fetch_identity_bundle_from_url}; use crate::contact_share::encode_contact_share_url; - use crate::identity::IdentityManager; - use crate::mls_adapter::MlsAdapter; + use crate::ffi_api::WelcomePreview; + use crate::identity::{IdentityManager, LocalIdentityState}; + use crate::mls_adapter::{MlsAdapter, PeerDeviceKeyPackage}; use crate::model::IdentityBundle; const NAMED_MNEMONIC: &str = "abandon abandon abandon abandon abandon abandon abandon abandon abandon abandon abandon about"; const IMPOSTOR_MNEMONIC: &str = "legal winner thank year wave sausage worth useful legal winner thank yellow"; + const RECEIVER_MNEMONIC: &str = + "letter advice cage absurd amount doctor acoustic avoid letter advice cage above"; const SECRET: &str = "0123456789abcdef0123456789abcdef"; + fn identity(mnemonic: &str) -> LocalIdentityState { + IdentityManager::create_or_recover(Some(mnemonic), None).expect("root") + } + /// A bundle signed by the root key `mnemonic` recovers, as its host serves it. fn signed_bundle(mnemonic: &str) -> IdentityBundle { - let identity = IdentityManager::create_or_recover(Some(mnemonic), None).expect("root"); + signed_bundle_of(&identity(mnemonic)) + } + + /// `identity`'s bundle, as its host serves it. + fn signed_bundle_of(identity: &LocalIdentityState) -> IdentityBundle { let now_ms = std::time::SystemTime::now() .duration_since(std::time::UNIX_EPOCH) .expect("test clock") .as_millis() as u64; - let package = MlsAdapter::generate_key_package(&identity, now_ms).expect("key package"); + let package = MlsAdapter::generate_key_package(identity, now_ms).expect("key package"); let mut deployment = crate::ffi_api::tests::tests::sample_deployment(); deployment.runtime_config.identity_bundle_ref = Some("https://runtime.example/v1/shared-state/{userId}/identity-bundle".into()); let bundle = IdentityManager::export_identity_bundle( - &identity, + identity, &deployment, package.key_package_b64, package.expires_at, @@ -283,30 +317,69 @@ mod tests { } } - fn bundle_with_user_id(user_id: &str) -> IdentityBundle { - serde_json::from_value(serde_json::json!({ - "version": "v1", - "user_id": user_id, - "user_public_key": "aa", - "devices": [], - "updated_at": 0, - "signature": "sig", - })) - .expect("minimal identity bundle") + /// The preview the receiver's core shows for a Welcome `author` made to it. + fn preview_of_welcome_from(author: &LocalIdentityState) -> WelcomePreview { + let receiver = identity(RECEIVER_MNEMONIC); + let (mut author_adapter, _) = MlsAdapter::bootstrap(author).expect("author adapter"); + let (receiver_adapter, receiver_package) = + MlsAdapter::bootstrap(&receiver).expect("receiver adapter"); + let artifacts = author_adapter + .create_conversation( + &crate::model::random_opaque_id(), + &[PeerDeviceKeyPackage { + user_id: receiver.user_identity.user_id.clone(), + device_id: receiver.device_identity.device_id.clone(), + device_public_key: receiver.device_identity.device_public_key.clone(), + key_package_b64: receiver_package.key_package_b64, + }], + ) + .expect("welcome"); + let inspection = receiver_adapter + .inspect_welcome(&artifacts.welcomes[0].payload_b64) + .expect("inspect"); + WelcomePreview { + conversation_id: inspection.conversation_id, + author_user_id: inspection.author_user_id, + author_device_id: inspection.author_device_id, + author_device_public_key: inspection.author_signature_key, + identity_bundle_ref: inspection.identity_bundle_ref, + } + } + + #[test] + fn a_welcome_is_accepted_with_the_bundle_whose_device_key_signed_it() { + let named = identity(NAMED_MNEMONIC); + let preview = preview_of_welcome_from(&named); + ensure_welcome_signed_by_bundle(&preview, &signed_bundle_of(&named)) + .expect("the author's own bundle is accepted"); } #[test] - fn rejects_bundle_whose_user_id_does_not_match_expected_sender() { - let bundle = bundle_with_user_id("user:alice"); - let error = ensure_fetched_bundle_matches_expected_sender(&bundle, "user:mallory") - .expect_err("mismatched sender should be rejected"); - assert!(error.to_string().contains("does not match")); + fn a_welcome_that_only_names_a_user_is_refused_with_that_users_bundle() { + // The credential is a string the Welcome's maker chooses; the key that + // signs the Welcome is not. + let named = identity(NAMED_MNEMONIC); + let mut preview = preview_of_welcome_from(&identity(IMPOSTOR_MNEMONIC)); + preview.author_user_id = named.user_identity.user_id.clone(); + preview.author_device_id = named.device_identity.device_id.clone(); + let error = ensure_welcome_signed_by_bundle(&preview, &signed_bundle_of(&named)) + .expect_err("a Welcome the named device did not sign must be refused"); + assert!( + error.to_string().contains("not signed by the device key"), + "{error}" + ); } #[test] - fn accepts_bundle_whose_user_id_matches_expected_sender() { - let bundle = bundle_with_user_id("user:alice"); - ensure_fetched_bundle_matches_expected_sender(&bundle, "user:alice") - .expect("matching sender should be accepted"); + fn a_bundle_of_another_user_than_the_welcome_names_is_refused() { + let preview = preview_of_welcome_from(&identity(NAMED_MNEMONIC)); + let error = ensure_welcome_signed_by_bundle(&preview, &signed_bundle(IMPOSTOR_MNEMONIC)) + .expect_err("another user's bundle must be refused"); + assert!( + error + .to_string() + .contains("does not match the Welcome author"), + "{error}" + ); } } diff --git a/src/ffi_api/engine/direct.rs b/src/ffi_api/engine/direct.rs index e5f4372..7c45376 100644 --- a/src/ffi_api/engine/direct.rs +++ b/src/ffi_api/engine/direct.rs @@ -57,6 +57,23 @@ impl CoreEngine { .effects .extend(self.local_shared_state_publish_effects()?); if inbox_moved { + // The new host numbers its records from 1. A cursor carried over + // would wait for that host's head to pass the old ack before + // fetching anything, and the host refuses an ack above its head. + // Records held back for a commit cannot resolve either: whatever + // they wait for is at the old host, which this device no longer + // reads. + if let Ok(device_id) = self.local_device_id_required() { + self.state + .sync_states + .insert(device_id.clone(), SyncEngine::new_device_state(&device_id)); + self.state.pending_acks.remove(&device_id); + } + // The new host's KeyPackage pool starts empty, and until it is + // filled every claim there falls back to the last-resort package. + if let Ok(pool_check) = self.key_package_pool_count_check_effect() { + output.effects.push(pool_check); + } output = merge_outputs(output, self.relocate_inbound_lanes()?); output = self.merge_with_transport_flush(output)?; } @@ -441,13 +458,51 @@ impl CoreEngine { pub(super) fn maintain_local_credentials(&mut self, now_ms: u64) -> CoreResult { let swept = self.sweep_expired_key_packages(now_ms)?; + let quiet = self.expire_quiet_direct_state(); let mut output = self.maintain_local_credentials_after_sweep(now_ms)?; if !swept.is_empty() { output.effects.push(persist_effect(&self.state, swept)); } + if !quiet.is_empty() { + output.effects.push(persist_effect(&self.state, quiet)); + } Ok(output) } + /// Expire what each direct session keeps for a race or as fork evidence: + /// pending commits with their keys and KeyPackage secrets, the frames + /// kept for a resend, and witnesses. A send or a receipt does this too, + /// but a session with no traffic, or one marked compromised, would + /// otherwise keep them for as long as it stays that way, and a snapshot + /// would find them. + fn expire_quiet_direct_state(&mut self) -> Vec { + let now_ms = current_unix_millis(self.state.message_nonce); + let conversation_ids: Vec = self.state.conversations.keys().cloned().collect(); + let mut ops = Vec::new(); + for conversation_id in conversation_ids { + let before = self.kept_for_races(&conversation_id); + self.expire_direct_pending(&conversation_id); + if let Some(state) = self.state.conversations.get_mut(&conversation_id) { + state.fork.prune(now_ms); + } + if self.kept_for_races(&conversation_id) != before { + ops.push(PersistOp::SaveConversation { conversation_id }); + } + } + ops + } + + /// How much a session keeps for a race or as fork evidence, by count. + fn kept_for_races(&self, conversation_id: &str) -> Option<[usize; 4]> { + let state = self.state.conversations.get(conversation_id)?; + Some([ + state.pcs.pending_commits.len(), + state.pcs.unconfirmed.len(), + state.fork.witnesses.len(), + state.fork.rebuild_welcomes.len(), + ]) + } + /// Delete the private keys of every inventoried KeyPackage that has /// passed its lifetime and the clock-skew margin, and return what must be /// persisted for the deletion to reach disk. @@ -3728,6 +3783,22 @@ impl CoreEngine { Ok(()) } + /// Announce the inbound lane again once a rebuild has settled a race. A + /// notice sent in the epoch the race took was never opened: the loser's + /// went out after its losing commit, and the winner's after a commit the + /// loser abandoned. Every refresh repeats the notice, but the next one may + /// be days away, and after a move the peer writes to the host this device + /// left until then. + pub(super) fn announce_lane_after_rebuild(&mut self, conversation_id: &str) { + if let Err(error) = self.enqueue_lane_rotation(conversation_id) { + log::warn!( + "lane announcement after rebuild failed conversation_id={}: {}", + redact_id("conv", conversation_id), + error.message() + ); + } + } + fn bundle_device_ids(bundle: &IdentityBundle) -> BTreeSet { bundle .devices diff --git a/src/ffi_api/engine/keypackage_claim.rs b/src/ffi_api/engine/keypackage_claim.rs index e115524..460dee0 100644 --- a/src/ffi_api/engine/keypackage_claim.rs +++ b/src/ffi_api/engine/keypackage_claim.rs @@ -436,9 +436,13 @@ impl CoreEngine { /// Builds the effect that checks the remaining count of the local /// device's own one-time KeyPackage pool. Called once per maintenance - /// cycle, right after the existing last-resort rotation check. + /// cycle, right after the existing last-resort rotation check, and once + /// after a move, since the new host's pool starts empty. + /// + /// The host serves the count at the pool's own path; it has no `/count` + /// route, and a request there is refused, which left the pool unfilled. pub(super) fn key_package_pool_count_check_effect(&mut self) -> CoreResult { - let url = format!("{}/count", self.own_key_package_pool_url()?); + let url = self.own_key_package_pool_url()?; let auth = self.device_runtime_auth_requirement()?; let request_id = self.next_request_id("keypackage_pool_count"); self.state diff --git a/src/ffi_api/engine/lanes.rs b/src/ffi_api/engine/lanes.rs index 7414787..51097d5 100644 --- a/src/ffi_api/engine/lanes.rs +++ b/src/ffi_api/engine/lanes.rs @@ -138,6 +138,16 @@ impl CoreEngine { Ok(()) } + /// Whether `lane` is the inbound lane of a conversation this device holds. + pub(super) fn is_current_inbound_lane(&self, lane: &str) -> bool { + self.state.conversations.values().any(|conversation| { + conversation + .lanes + .as_ref() + .is_some_and(|lanes| lanes.inbound_lane == lane) + }) + } + pub(super) fn register_accepted_lane(&mut self, lane: String) -> CoreResult { if !is_opaque_id(&lane) { return Err(CoreError::invalid_input("lane must be a 128-bit hex id")); @@ -795,6 +805,7 @@ impl CoreEngine { conversation_id: inspection.conversation_id, author_user_id: inspection.author_user_id, author_device_id: inspection.author_device_id, + author_device_public_key: inspection.author_signature_key, identity_bundle_ref: inspection.identity_bundle_ref, }), ..CoreViewModel::default() diff --git a/src/ffi_api/engine/mod.rs b/src/ffi_api/engine/mod.rs index 6e92730..a357bb9 100644 --- a/src/ffi_api/engine/mod.rs +++ b/src/ffi_api/engine/mod.rs @@ -1517,19 +1517,35 @@ impl CoreEngine { view_model: None, }), CoreEvent::AcceptedLaneRegistered { .. } => Ok(CoreOutput::default()), - CoreEvent::AcceptedLaneRegisterFailed { lane: _, failure: _ } => Ok(CoreOutput { - state_update: CoreStateUpdate { - system_statuses_changed: vec![SystemStatus::TemporaryNetworkFailure], - ..CoreStateUpdate::default() - }, - effects: vec![CoreEffect::EmitUserNotification { + CoreEvent::AcceptedLaneRegisterFailed { lane, failure } => { + let mut effects = vec![CoreEffect::EmitUserNotification { notification: UserNotificationEffect { status: SystemStatus::TemporaryNetworkFailure, message: "TapChat couldn't register the inbox lane. Try again.".into(), }, - }], - view_model: None, - }), + }]; + // A lane its owner has not admitted turns every later append + // on it into a message request, so the registration is tried + // again while this device still reads the lane. The host is + // this device's own, so the retry keeps the slowest pace. + if failure.retryable && self.is_current_inbound_lane(&lane) { + let timer_id = format!("retry_register_lane:{lane}"); + effects.push(CoreEffect::ScheduleTimer { + timer: TimerEffect { + delay_ms: retry_delay_ms(&timer_id, u32::MAX), + timer_id, + }, + }); + } + Ok(CoreOutput { + state_update: CoreStateUpdate { + system_statuses_changed: vec![SystemStatus::TemporaryNetworkFailure], + ..CoreStateUpdate::default() + }, + effects, + view_model: None, + }) + } CoreEvent::AcceptedLanesRevoked { .. } => Ok(CoreOutput::default()), CoreEvent::AcceptedLanesRevokeFailed { lanes: _, failure: _ } => Ok(CoreOutput { state_update: CoreStateUpdate { diff --git a/src/ffi_api/engine/recovery.rs b/src/ffi_api/engine/recovery.rs index eaf61d3..b117cfc 100644 --- a/src/ffi_api/engine/recovery.rs +++ b/src/ffi_api/engine/recovery.rs @@ -355,11 +355,10 @@ impl CoreEngine { peer_active_device_ids.iter().cloned().collect(); } - let mut generated = self.commit_envelopes_for_artifacts( - &conversation_id, - &peer_active_device_ids, - &artifacts, - )?; + // Everyone in the new group joins it from a Welcome, so the commit + // that created it has no reader: a peer could only drop it as + // superseded. A rebuild therefore sends the Welcome and the resends, + // as a first contact sends only its Welcome. let mut welcomes = self.welcome_envelopes_for_artifacts(&conversation_id, &artifacts)?; let wrap_out = self .state @@ -394,9 +393,10 @@ impl CoreEngine { .iter() .map(|envelope| envelope.mid.clone()) .collect(); - generated.extend(welcomes); + let generated = welcomes; self.enqueue_envelopes(peer_user_id, generated.clone()); self.resend_unconfirmed_frames(&conversation_id)?; + self.announce_lane_after_rebuild(&conversation_id); self.mark_recovery_needed(&conversation_id, RecoveryReason::MembershipChanged); self.merge_with_transport_flush(CoreOutput { state_update: CoreStateUpdate { diff --git a/src/ffi_api/engine/transport.rs b/src/ffi_api/engine/transport.rs index 22816ca..1a1f5a4 100644 --- a/src/ffi_api/engine/transport.rs +++ b/src/ffi_api/engine/transport.rs @@ -871,6 +871,12 @@ impl CoreEngine { } return self.flush_pending_transport(); } + if let Some(lane) = timer_id.strip_prefix("retry_register_lane:") { + if self.is_current_inbound_lane(lane) { + return self.register_accepted_lane(lane.to_string()); + } + return Ok(CoreOutput::default()); + } if let Some(device_id) = timer_id.strip_prefix("retry_ack:") { if let Some(ack) = self.state.pending_acks.get_mut(device_id) { ack.in_flight = false; @@ -4071,6 +4077,9 @@ impl CoreEngine { ProtectedPayloadKind::ContactAccepted, "{}".to_string(), )?; + if !first_contact { + self.announce_lane_after_rebuild(&conversation_id); + } } } unapplied @ (IngestResult::Rejected(_) diff --git a/src/ffi_api/tests.rs b/src/ffi_api/tests.rs index eb70ef0..3ac9bab 100644 --- a/src/ffi_api/tests.rs +++ b/src/ffi_api/tests.rs @@ -5549,14 +5549,16 @@ pub(crate) mod tests { }) .expect("claim response rebuilds the conversation"); + // A rebuild sends its Welcome and not the commit that made the group: + // everyone joins from the Welcome, so the commit has no reader. assert!(completed.view_model.as_ref().is_some_and(|view| { view.messages .iter() - .any(|message| message.message_type == MessageType::MlsCommit) - && view + .any(|message| message.message_type == MessageType::MlsWelcome) + && !view .messages .iter() - .any(|message| message.message_type == MessageType::MlsWelcome) + .any(|message| message.message_type == MessageType::MlsCommit) })); assert_eq!( alice @@ -5627,6 +5629,46 @@ pub(crate) mod tests { let _ = laptop_keypackage_b64; } + /// The pool count is read where the host serves it, `GET + /// /v1/keypackage-pool/{deviceId}`. The host has no `/count` route, so a + /// request there was refused and the pool was never topped up. + #[test] + fn credential_maintenance_reads_the_pool_count_at_the_hosts_route() { + let mut engine = local_engine(ALICE_MNEMONIC, "phone"); + let now_ms = engine + .state + .published_key_package + .as_ref() + .expect("published key package") + .created_at; + let output = engine + .handle_event(CoreEvent::CredentialMaintenanceRequested { now_ms }) + .expect("credential maintenance"); + let request = output + .effects + .iter() + .find_map(|effect| match effect { + CoreEffect::ExecuteHttpRequest { request } + if request.url.contains("/keypackage-pool/") => + { + Some(request.clone()) + } + _ => None, + }) + .expect("pool count request"); + let device_id = engine + .state + .local_identity + .as_ref() + .expect("identity") + .device_identity + .device_id + .clone(); + let route = format!("/v1/keypackage-pool/{}", urlencoding::encode(&device_id)); + assert!(request.method == crate::ffi_api::HttpMethod::Get); + assert!(request.url.ends_with(&route), "{}", request.url); + } + #[test] fn credential_maintenance_replenishes_pool_when_below_low_water_mark() { let mut engine = local_engine(ALICE_MNEMONIC, "phone"); @@ -10964,13 +11006,13 @@ pub(crate) mod tests { assert!(output.view_model.as_ref().is_some_and(|view| { view.messages .iter() - .any(|message| message.message_type == MessageType::MlsCommit) - && view + .any(|message| message.message_type == MessageType::MlsWelcome) + && !view .messages .iter() - .any(|message| message.message_type == MessageType::MlsWelcome) + .any(|message| message.message_type == MessageType::MlsCommit) })); - assert!(restored.state.pending_outbox[pending_before..] + assert!(!restored.state.pending_outbox[pending_before..] .iter() .any(|item| outbox_item_matches_type(item, MessageType::MlsCommit))); assert!(restored.state.pending_outbox[pending_before..] @@ -12582,6 +12624,93 @@ pub(crate) mod tests { ); } + /// **A lane whose registration failed is registered again.** Until the + /// owner's host admits a lane, every append on it becomes a message + /// request, so a retryable failure is retried while the lane is still + /// one this device reads, and not for a lane it no longer reads. + #[test] + fn a_failed_lane_registration_is_retried_while_the_lane_is_read() { + let mut chat = paired_direct_chat(); + let lane = chat.bob.state.conversations[&chat.conversation_id] + .lanes + .as_ref() + .expect("lanes") + .inbound_lane + .clone(); + let failed = chat + .bob + .handle_event(CoreEvent::AcceptedLaneRegisterFailed { + lane: lane.clone(), + failure: test_failure("request_timeout", true, None), + }) + .expect("failure handled"); + let timer_id = format!("retry_register_lane:{lane}"); + assert!(failed.effects.iter().any(|effect| matches!( + effect, + CoreEffect::ScheduleTimer { timer } if timer.timer_id == timer_id + ))); + + let retried = chat + .bob + .handle_event(CoreEvent::TimerTriggered { timer_id }) + .expect("retry handled"); + assert!(retried.effects.iter().any(|effect| matches!( + effect, + CoreEffect::RegisterAcceptedLane { register } if register.lane == lane + ))); + + let gone = crate::model::random_opaque_id(); + let ignored = chat + .bob + .handle_event(CoreEvent::AcceptedLaneRegisterFailed { + lane: gone, + failure: test_failure("request_timeout", true, None), + }) + .expect("failure handled"); + assert!(!ignored + .effects + .iter() + .any(|effect| matches!(effect, CoreEffect::ScheduleTimer { .. }))); + } + + #[test] + fn a_move_reads_the_new_host_from_its_first_record() { + use crate::sync_engine::SyncEngine; + + let mut chat = paired_direct_chat(); + let device_id = chat.bob_device_id.clone(); + let sync = chat + .bob + .state + .sync_states + .entry(device_id.clone()) + .or_insert_with(|| SyncEngine::new_device_state(&device_id)); + sync.checkpoint.last_fetched_seq = 9; + sync.checkpoint.last_acked_seq = 9; + sync.last_head_seq = 9; + + let mut relocated = sample_deployment(); + relocated.inbox_http_endpoint = "https://bob-new.example.test".into(); + relocated.inbox_websocket_endpoint = "wss://bob-new.example.test/ws".into(); + relocated.runtime_id = "runtime:bob-new".into(); + relocated.runtime_config.identity_bundle_ref = + Some("https://bob-new.example.test/state/identity.json".into()); + chat.bob + .handle_command(CoreCommand::ImportDeploymentBundle { bundle: relocated }) + .expect("bob relocates"); + + let sync = &chat.bob.state.sync_states[&device_id]; + assert_eq!(sync.checkpoint.last_fetched_seq, 0); + assert_eq!(sync.checkpoint.last_acked_seq, 0); + assert_eq!(sync.last_head_seq, 0); + let mut sync = sync.clone(); + SyncEngine::register_head(&mut sync, 1); + assert_eq!( + SyncEngine::next_fetch(&sync).map(|next| next.from_seq), + Some(1) + ); + } + #[test] fn relocation_to_empty_runtime_keeps_the_conversation() { let mut chat = paired_direct_chat(); @@ -16700,6 +16829,37 @@ pub(crate) mod tests { ); } + /// **A quiet session drops what no race can reach, too.** A send or a + /// receipt expires old pending commits, but a session with no traffic + /// has neither, so the daily upkeep does it. + #[test] + fn upkeep_drops_a_pending_commit_no_race_can_reach_in_a_quiet_session() { + let mut chat = paired_direct_chat(); + let conversation_id = chat.conversation_id.clone(); + let (rotating, [older, latest]) = two_pending_commits(&mut chat); + let rotator = rotator_engine_mut(&mut chat, rotating); + age_pending_commits( + rotator, + &conversation_id, + crate::direct_pcs::PENDING_COMMIT_TTL_MS, + ); + rotator + .handle_event(CoreEvent::CredentialMaintenanceRequested { + now_ms: wall_clock_ms(), + }) + .expect("upkeep"); + + let rotator = rotator_engine(&chat, rotating); + let pending = rotator.state.conversations[&conversation_id] + .pcs + .pending_commits + .iter() + .filter_map(|commit| commit.key_package_b64.clone()) + .collect::>(); + assert_eq!(pending, vec![latest.clone()], "only the latest stays"); + assert!(!holds_key_package(rotator, &older)); + } + /// **A pending commit stays for the whole window a race can take.** Past /// the host's retention `Δ` but inside `2Δ`, its rival may still be on its /// way. Past `2Δ`, the commit whose rebuild we await stays too: the diff --git a/src/ffi_api/types.rs b/src/ffi_api/types.rs index 8fb590c..493da83 100644 --- a/src/ffi_api/types.rs +++ b/src/ffi_api/types.rs @@ -1109,6 +1109,10 @@ pub struct WelcomePreview { pub conversation_id: String, pub author_user_id: String, pub author_device_id: String, + /// Hex. The key that signed the Welcome; see + /// [`crate::contact_workflows::ensure_welcome_signed_by_bundle`]. + #[serde(default)] + pub author_device_public_key: String, #[serde(default, skip_serializing_if = "Option::is_none")] pub identity_bundle_ref: Option, } diff --git a/src/mls_adapter/mod.rs b/src/mls_adapter/mod.rs index ba05d9e..8208444 100644 --- a/src/mls_adapter/mod.rs +++ b/src/mls_adapter/mod.rs @@ -121,6 +121,10 @@ pub struct WelcomeInspection { pub conversation_id: String, pub author_user_id: String, pub author_device_id: String, + /// The author's leaf signature key, hex. Staging verified the GroupInfo + /// signature under it, so a caller that finds it equal to a device key it + /// trusts knows who made the Welcome before anything is stored. + pub author_signature_key: String, pub identity_bundle_ref: Option, } @@ -1632,6 +1636,7 @@ impl MlsAdapter { return None; } let sender = staged.welcome_sender().ok()?; + let author_signature_key = crate::identity::encode_hex(sender.signature_key().as_slice()); let identity = extract_sender_identity(sender.credential()).ok()?; let mut parts = identity.split('|'); let author_user_id = parts.next()?.to_string(); @@ -1649,6 +1654,7 @@ impl MlsAdapter { conversation_id, author_user_id, author_device_id, + author_signature_key, identity_bundle_ref, }) } @@ -3185,6 +3191,14 @@ mod tests { assert_eq!(inspection.conversation_id, conversation_id); assert_eq!(inspection.author_user_id, alice.user_identity.user_id); assert_eq!(inspection.author_device_id, alice.device_identity.device_id); + assert_eq!( + crate::identity::parse_verifying_key(&inspection.author_signature_key) + .expect("signature key") + .as_bytes(), + crate::identity::parse_verifying_key(&alice.device_identity.device_public_key) + .expect("device key") + .as_bytes() + ); assert_eq!( inspection.identity_bundle_ref.as_deref(), Some("https://example.test/v1/contact-share/alice")