diff --git a/PRD.md b/PRD.md index 843a106..ad0992c 100644 --- a/PRD.md +++ b/PRD.md @@ -962,6 +962,7 @@ Better Auth 在 Cloudflare Workers 中初始化，D1 binding 作为 database 传 - Bearer logout 精确删除当前 D1 session，并级联清理 session device context 与 rebind challenge。 - Desktop bearer 以 stable `ProfileId` 作为系统凭据 account，Windows 使用 Local persistence；旧明文文件在 credential read-back 与 durable marker 提交后清理，系统凭据不可用时阻断设备加载与 Sync upload。 - Desktop sign-out closes the authenticated-operation gate, drains active leases, revokes the exact server session, and conditionally clears the captured native credential; generation-stamped async results cannot restore stale auth or Sync state. +- Runtime `session_not_found` and `session_expired` responses conditionally clear the exact captured bearer inside `SyncEngine`; replacement credentials survive, stale Profile work converges through credential reprobe, and active Cloud Sync/device work resets before session-state reconciliation. - 设备注册、rebind、批准、撤销与 Vault rotation。 - Signed Sync reset 和 signed account deletion。 - 管理所有 `/api/auth/*` 路由。 diff --git a/crates/ely_app/src/shell/settings_actions.rs b/crates/ely_app/src/shell/settings_actions.rs index 109558c..58f0ebf 100644 --- a/crates/ely_app/src/shell/settings_actions.rs +++ b/crates/ely_app/src/shell/settings_actions.rs @@ -380,7 +380,7 @@ fn run_sync_upload( match outcome { Ok(ely_browser_core::SyncOutcome::SignedOut) => { tracing::info!(target: "ely::sync", "no bearer token on disk; sync skipped"); - let _ = inbox.send(SyncStateUpdate::SignedOut { profile_id }); + let _ = inbox.send(SyncStateUpdate::AuthenticationExpired { profile_id }); } Ok(ely_browser_core::SyncOutcome::AwaitingDeviceApproval { device_id }) => { tracing::info!( diff --git a/crates/ely_app/src/shell/sync_devices.rs b/crates/ely_app/src/shell/sync_devices.rs index b0af2c7..657155b 100644 --- a/crates/ely_app/src/shell/sync_devices.rs +++ b/crates/ely_app/src/shell/sync_devices.rs @@ -300,7 +300,7 @@ mod tests { } #[test] - fn credential_failures_do_not_finish_an_upload() { + fn credential_failures_use_the_unavailable_update() { let profile_id = ProfileId::new(); let update = device_failure_update( profile_id.clone(), @@ -311,12 +311,27 @@ mod tests { update, SyncStateUpdate::CredentialUnavailable { profile_id: owner, - finishes_upload: false, .. } if owner == profile_id )); } + #[test] + fn terminal_sessions_expire_profile_authentication() { + let profile_id = ProfileId::new(); + for error in [ + ely_sync_client::SyncClientError::SessionEnded, + ely_sync_client::SyncClientError::SessionChanged, + ] { + let update = device_failure_update(profile_id.clone(), error); + + assert!(matches!( + update, + SyncStateUpdate::AuthenticationExpired { profile_id: owner } if owner == profile_id + )); + } + } + #[test] fn approval_failures_do_not_finish_an_upload() { let profile_id = ProfileId::new(); diff --git a/crates/ely_app/src/shell/sync_inbox.rs b/crates/ely_app/src/shell/sync_inbox.rs index 8609a9f..e98b683 100644 --- a/crates/ely_app/src/shell/sync_inbox.rs +++ b/crates/ely_app/src/shell/sync_inbox.rs @@ -23,6 +23,24 @@ impl SyncStateMessage { } } +pub(super) fn early_reconciliation_priority(update: &SyncStateUpdate) -> Option { + match update { + SyncStateUpdate::AuthenticationExpired { .. } => Some(0), + SyncStateUpdate::CredentialUnavailable { .. } => Some(1), + _ => None, + } +} + +pub(super) fn early_reconciliation_messages( + messages: &[SyncStateMessage], +) -> impl Iterator { + [0_u8, 1].into_iter().flat_map(move |priority| { + messages + .iter() + .filter(move |message| early_reconciliation_priority(&message.update) == Some(priority)) + }) +} + #[derive(Clone, Debug)] pub(super) struct SyncStateSender { generation: SyncWorkGeneration, @@ -46,12 +64,20 @@ impl ElyShell { pub(super) fn invalidate_sync_work(&mut self) { self.release_auth_flow_barrier(); self.sync_generation.advance(); + self.reset_sync_work_state(); + } + + pub(super) fn reset_sync_work_state(&mut self) { self.sync_upload_scheduled = false; self.sync_upload_in_flight = false; self.sync_retry_at = None; self.clear_pending_cloud_sync_upload(); self.sync_devices.reset(); } + + pub(super) fn expire_authenticated_work(&mut self) { + self.invalidate_sync_work(); + } } #[cfg(test)] @@ -65,7 +91,7 @@ mod tests { let generation = SyncWorkGeneration(7); let sender = SyncStateSender { generation, sender }; - sender.send(SyncStateUpdate::SignedOut { profile_id: ProfileId::new() })?; + sender.send(SyncStateUpdate::AuthenticationExpired { profile_id: ProfileId::new() })?; assert_eq!(receiver.recv()?.generation, generation); Ok(()) @@ -80,9 +106,63 @@ mod tests { current.advance(); current.advance(); - old_sender.send(SyncStateUpdate::SignedOut { profile_id: ProfileId::new() })?; + old_sender.send(SyncStateUpdate::AuthenticationExpired { profile_id: ProfileId::new() })?; assert!(!receiver.recv()?.belongs_to(current)); Ok(()) } + + #[test] + fn early_reconciliation_order_is_queue_independent() { + let profile_id = ProfileId::new(); + let generation = SyncWorkGeneration::default(); + let expired = SyncStateMessage { + generation, + update: SyncStateUpdate::AuthenticationExpired { profile_id: profile_id.clone() }, + }; + let unavailable = SyncStateMessage { + generation, + update: SyncStateUpdate::CredentialUnavailable { + profile_id, + message: "locked".to_string(), + }, + }; + + for messages in + [vec![expired.clone(), unavailable.clone()], vec![unavailable.clone(), expired.clone()]] + { + let ordered = early_reconciliation_messages(&messages) + .map(|message| early_reconciliation_priority(&message.update)) + .collect::>(); + assert_eq!(ordered, [Some(0), Some(1)]); + } + } + + #[gpui::test] + async fn authentication_expiry_resets_scoped_work(cx: &mut gpui::TestAppContext) { + cx.update(gpui_component::init); + let (shell, cx) = cx.add_window_view(ElyShell::new); + + cx.update(|_, app_cx| { + shell.update(app_cx, |shell, _| { + let generation = shell.sync_generation; + shell.sync_upload_scheduled = true; + shell.sync_upload_in_flight = true; + shell.sync_upload_pending = true; + shell.sync_retry_at = Some(std::time::Instant::now()); + shell + .sync_devices + .set_error(ProfileId::new(), "expired device session".to_string()); + + shell.expire_authenticated_work(); + + assert_ne!(shell.sync_generation, generation); + assert!(!shell.sync_upload_scheduled); + assert!(!shell.sync_upload_in_flight); + assert!(!shell.sync_upload_pending); + assert!(shell.sync_retry_at.is_none()); + assert!(shell.sync_devices.error().is_none()); + }); + }); + } } diff --git a/crates/ely_app/src/shell/sync_state.rs b/crates/ely_app/src/shell/sync_state.rs index ac5b062..dd9ac44 100644 --- a/crates/ely_app/src/shell/sync_state.rs +++ b/crates/ely_app/src/shell/sync_state.rs @@ -4,10 +4,14 @@ use ely_domain::{ProfileId, ProfileKind, SyncConnectionState}; use ely_sync_client::{AuthenticatedSnapshotHead, BearerToken, BearerTokenStore, DeviceRecord}; use gpui::{Context, Timer}; -use super::{ElyShell, ShellState, auth}; +use super::{ + ElyShell, ShellState, auth, + sync_inbox::{early_reconciliation_messages, early_reconciliation_priority}, +}; mod failure; mod legacy_sync_migration; +mod session_expiry; mod sign_out; pub(super) use failure::{device_failure_update, sync_failure_update}; @@ -26,17 +30,16 @@ pub(crate) struct PendingMergeUpload { /// Messages the off-thread sync workers push back to the shell so /// `SyncConnectionState` on `BrowserCore` and the in-flight auth /// form reflect live state without the UI thread ever touching the -/// network. `SignedIn` is the initial-probe state set synchronously -/// on shell startup and does not flow through this channel. +/// network. Startup seeds `SignedIn` outside this channel. #[derive(Clone, Debug)] pub(crate) enum SyncStateUpdate { - SignedOut { profile_id: ProfileId }, + AuthenticationExpired { profile_id: ProfileId }, AwaitingDeviceApproval { profile_id: ProfileId, finishes_upload: bool }, RemoteSnapshot { profile_id: ProfileId, bytes: Vec, merge: PendingMergeUpload }, SyncReady { profile_id: ProfileId, last_synced_at_secs: u64 }, SyncBusy { profile_id: ProfileId }, SyncError { profile_id: ProfileId, message: String }, - CredentialUnavailable { profile_id: ProfileId, message: String, finishes_upload: bool }, + CredentialUnavailable { profile_id: ProfileId, message: String }, DevicesLoaded { profile_id: ProfileId, devices: Vec, current_code: String }, DevicesError { profile_id: ProfileId, message: String }, SignOutSucceeded { profile_id: ProfileId }, @@ -172,7 +175,48 @@ impl ElyShell { let mut trigger_merged_upload = None; let mut upload_finished = false; let mut devices_changed = false; - while let Ok(message) = self.sync_inbox_rx.try_recv() { + let messages = self.sync_inbox_rx.try_iter().collect::>(); + for state_message in early_reconciliation_messages(&messages) { + match &state_message.update { + SyncStateUpdate::AuthenticationExpired { profile_id } => { + let current = state_message.belongs_to(self.sync_generation); + match self.reconcile_authentication_expiry(profile_id, current) { + session_expiry::AuthenticationExpiryResolution::Ignored => {} + session_expiry::AuthenticationExpiryResolution::ReplacementCredential => { + latest_connection = Some(SyncConnectionState::SignedIn); + trigger_initial_sync = true; + devices_changed = true; + auth_changed = true; + } + session_expiry::AuthenticationExpiryResolution::Connection(connection) => { + latest_connection = Some(connection); + trigger_initial_sync = false; + trigger_merged_upload = None; + devices_changed = true; + auth_changed = true; + } + } + } + SyncStateUpdate::CredentialUnavailable { profile_id, message } + if state_message.belongs_to(self.sync_generation) + && active_profile_id(&self.state).as_ref() == Some(profile_id) => + { + self.reconcile_credential_unavailable(profile_id); + latest_connection = Some(SyncConnectionState::CredentialUnavailable { + message: message.clone(), + }); + trigger_initial_sync = false; + trigger_merged_upload = None; + devices_changed = true; + auth_changed = true; + } + _ => {} + } + } + for message in messages { + if early_reconciliation_priority(&message.update).is_some() { + continue; + } if let Some((profile_id, connection)) = self.reconcile_sign_out_update(message.generation, &message.update) { @@ -191,12 +235,8 @@ impl ElyShell { } let update = message.update; match update { - SyncStateUpdate::SignedOut { profile_id } => { - upload_finished = true; - if active_profile_id(&self.state).as_ref() == Some(&profile_id) { - latest_connection = Some(SyncConnectionState::SignedOut); - } - } + SyncStateUpdate::AuthenticationExpired { .. } + | SyncStateUpdate::CredentialUnavailable { .. } => {} SyncStateUpdate::AwaitingDeviceApproval { profile_id, finishes_upload } => { upload_finished |= finishes_upload; if active_profile_id(&self.state).as_ref() == Some(&profile_id) { @@ -258,20 +298,6 @@ impl ElyShell { latest_connection = Some(SyncConnectionState::SyncError { message }); } } - SyncStateUpdate::CredentialUnavailable { profile_id, message, finishes_upload } => { - upload_finished |= finishes_upload; - if active_profile_id(&self.state).as_ref() == Some(&profile_id) { - latest_connection = - Some(SyncConnectionState::CredentialUnavailable { message }); - self.sync_devices.reset(); - self.sync_upload_scheduled = false; - self.sync_retry_at = None; - self.clear_pending_cloud_sync_upload(); - trigger_initial_sync = false; - trigger_merged_upload = None; - devices_changed = true; - } - } SyncStateUpdate::DevicesLoaded { profile_id, devices, current_code } => { if active_profile_id(&self.state).as_ref() == Some(&profile_id) { self.sync_devices.set_ready(profile_id, devices, current_code); diff --git a/crates/ely_app/src/shell/sync_state/failure.rs b/crates/ely_app/src/shell/sync_state/failure.rs index b7ddfab..21bcb29 100644 --- a/crates/ely_app/src/shell/sync_state/failure.rs +++ b/crates/ely_app/src/shell/sync_state/failure.rs @@ -8,10 +8,14 @@ pub(in crate::shell) fn sync_failure_update( ) -> SyncStateUpdate { let message = error.to_string(); match error { + ely_sync_client::SyncClientError::SessionEnded + | ely_sync_client::SyncClientError::SessionChanged => { + SyncStateUpdate::AuthenticationExpired { profile_id } + } ely_sync_client::SyncClientError::BearerCredentialStorage(_) | ely_sync_client::SyncClientError::AccountKeyStorage(_) | ely_sync_client::SyncClientError::DeviceKeyStorage(_) => { - SyncStateUpdate::CredentialUnavailable { profile_id, message, finishes_upload: true } + SyncStateUpdate::CredentialUnavailable { profile_id, message } } ely_sync_client::SyncClientError::DeviceApprovalStatus { .. } => { SyncStateUpdate::AwaitingDeviceApproval { profile_id, finishes_upload: true } @@ -28,9 +32,6 @@ pub(in crate::shell) fn device_failure_update( error: ely_sync_client::SyncClientError, ) -> SyncStateUpdate { match sync_failure_update(profile_id, error) { - SyncStateUpdate::CredentialUnavailable { profile_id, message, .. } => { - SyncStateUpdate::CredentialUnavailable { profile_id, message, finishes_upload: false } - } SyncStateUpdate::AwaitingDeviceApproval { profile_id, .. } => { SyncStateUpdate::AwaitingDeviceApproval { profile_id, finishes_upload: false } } diff --git a/crates/ely_app/src/shell/sync_state/session_expiry.rs b/crates/ely_app/src/shell/sync_state/session_expiry.rs new file mode 100644 index 0000000..c840df7 --- /dev/null +++ b/crates/ely_app/src/shell/sync_state/session_expiry.rs @@ -0,0 +1,214 @@ +use ely_domain::{ProfileId, SyncConnectionState}; + +use super::super::{ElyShell, ShellState, auth}; +use crate::services::servo_profile_data::{default_profile_data_root, sync_profile_data_dir}; + +#[derive(Clone, Copy, Debug, Eq, PartialEq)] +enum CredentialState { + Present, + Missing, + Unavailable, +} + +pub(super) enum AuthenticationExpiryResolution { + Ignored, + ReplacementCredential, + Connection(SyncConnectionState), +} + +#[derive(Debug, Eq, PartialEq)] +enum ExpiryDecision { + Ignore, + RefreshReplacement { preserve_generation: bool }, + Reconcile { connection: SyncConnectionState, preserve_generation: bool, finish_sign_out: bool }, +} + +impl ElyShell { + pub(super) fn reconcile_authentication_expiry( + &mut self, + profile_id: &ProfileId, + is_current_generation: bool, + ) -> AuthenticationExpiryResolution { + let ShellState::Ready(core) = &self.state else { + return AuthenticationExpiryResolution::Ignored; + }; + let active_profile_id = match core.snapshot() { + Ok(snapshot) => snapshot.active_profile_id, + Err(_) => return AuthenticationExpiryResolution::Ignored, + }; + let active_matches = &active_profile_id == profile_id; + if !active_matches { + return AuthenticationExpiryResolution::Ignored; + } + + let decision = expiry_decision( + active_matches, + is_current_generation, + self.sign_out_phases.contains_key(profile_id), + self.credential_state(profile_id), + preserves_auth_attempt(&self.auth_flow_phase, profile_id), + ); + match decision { + ExpiryDecision::Ignore => AuthenticationExpiryResolution::Ignored, + ExpiryDecision::RefreshReplacement { preserve_generation } => { + if preserve_generation { + self.reset_sync_work_state(); + } else { + self.invalidate_sync_work(); + } + AuthenticationExpiryResolution::ReplacementCredential + } + ExpiryDecision::Reconcile { connection, preserve_generation, finish_sign_out } => { + if finish_sign_out { + self.sign_out_phases.remove(profile_id); + } + self.reset_after_credential_loss(preserve_generation); + AuthenticationExpiryResolution::Connection(connection) + } + } + } + + pub(super) fn reconcile_credential_unavailable(&mut self, profile_id: &ProfileId) { + let preserve_generation = preserves_auth_attempt(&self.auth_flow_phase, profile_id); + self.reset_after_credential_loss(preserve_generation); + } + + fn reset_after_credential_loss(&mut self, preserve_generation: bool) { + if preserve_generation { + self.reset_sync_work_state(); + } else { + self.expire_authenticated_work(); + self.auth_flow_phase = auth::AuthFlowPhase::Idle; + } + } + + fn credential_state(&self, profile_id: &ProfileId) -> CredentialState { + let profile_root = match default_profile_data_root() { + Some(root) => root, + None => return CredentialState::Unavailable, + }; + let profile_dir = sync_profile_data_dir(&profile_root, profile_id); + let store = auth::bearer_store_for_profile( + profile_id, + &profile_dir, + &profile_root, + self.default_profile_id.as_ref(), + ); + match store.load() { + Ok(Some(_)) => CredentialState::Present, + Ok(None) => CredentialState::Missing, + Err(error) => { + tracing::warn!(target: "ely::sync", error = %error, "expired session credential probe failed"); + CredentialState::Unavailable + } + } + } +} + +fn expiry_decision( + active_matches: bool, + is_current_generation: bool, + sign_out_active: bool, + credential: CredentialState, + preserve_auth_attempt: bool, +) -> ExpiryDecision { + if !active_matches { + return ExpiryDecision::Ignore; + } + if credential == CredentialState::Present { + return if is_current_generation && !sign_out_active { + ExpiryDecision::RefreshReplacement { preserve_generation: preserve_auth_attempt } + } else { + ExpiryDecision::Ignore + }; + } + if sign_out_active && credential == CredentialState::Unavailable { + return ExpiryDecision::Ignore; + } + let connection = match credential { + CredentialState::Present => return ExpiryDecision::Ignore, + CredentialState::Missing => SyncConnectionState::SignedOut, + CredentialState::Unavailable => SyncConnectionState::CredentialUnavailable { + message: "System credential access failed.".to_string(), + }, + }; + ExpiryDecision::Reconcile { + connection, + preserve_generation: preserve_auth_attempt, + finish_sign_out: sign_out_active, + } +} + +fn preserves_auth_attempt(phase: &auth::AuthFlowPhase, profile_id: &ProfileId) -> bool { + !matches!(phase, auth::AuthFlowPhase::Idle) && phase.belongs_to(profile_id) +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn expiry_decision_covers_profile_aba_replacement_and_sign_out() { + assert_eq!( + expiry_decision(false, false, false, CredentialState::Missing, false), + ExpiryDecision::Ignore + ); + assert_eq!( + expiry_decision(true, true, false, CredentialState::Present, false), + ExpiryDecision::RefreshReplacement { preserve_generation: false } + ); + assert_eq!( + expiry_decision(true, false, false, CredentialState::Present, false), + ExpiryDecision::Ignore + ); + assert_eq!( + expiry_decision(true, true, true, CredentialState::Unavailable, false), + ExpiryDecision::Ignore + ); + assert_eq!( + expiry_decision(true, false, true, CredentialState::Missing, false), + ExpiryDecision::Reconcile { + connection: SyncConnectionState::SignedOut, + preserve_generation: false, + finish_sign_out: true, + } + ); + } + + #[test] + fn missing_credential_preserves_a_current_otp_generation() { + assert_eq!( + expiry_decision(true, true, false, CredentialState::Present, true), + ExpiryDecision::RefreshReplacement { preserve_generation: true } + ); + assert_eq!( + expiry_decision(true, false, false, CredentialState::Missing, true), + ExpiryDecision::Reconcile { + connection: SyncConnectionState::SignedOut, + preserve_generation: true, + finish_sign_out: false, + } + ); + assert!(matches!( + expiry_decision(true, false, false, CredentialState::Unavailable, true), + ExpiryDecision::Reconcile { + connection: SyncConnectionState::CredentialUnavailable { .. }, + preserve_generation: true, + finish_sign_out: false, + } + )); + } + + #[test] + fn current_otp_attempt_is_profile_scoped() { + let profile_id = ProfileId::new(); + let phase = auth::AuthFlowPhase::Verifying { + profile_id: profile_id.clone(), + email: "user@example.com".to_string(), + }; + + assert!(preserves_auth_attempt(&phase, &profile_id)); + assert!(!preserves_auth_attempt(&phase, &ProfileId::new())); + assert!(!preserves_auth_attempt(&auth::AuthFlowPhase::Idle, &profile_id)); + } +} diff --git a/crates/ely_app/src/shell/sync_state/sync_state_tests.rs b/crates/ely_app/src/shell/sync_state/sync_state_tests.rs index 132355c..7c2990e 100644 --- a/crates/ely_app/src/shell/sync_state/sync_state_tests.rs +++ b/crates/ely_app/src/shell/sync_state/sync_state_tests.rs @@ -80,8 +80,23 @@ fn credential_storage_failures_use_the_unavailable_update() { update, SyncStateUpdate::CredentialUnavailable { profile_id: owner, - finishes_upload: true, .. } if owner == profile_id )); } + +#[test] +fn terminal_sessions_expire_upload_authentication() { + let profile_id = ely_domain::ProfileId::new(); + for error in [ + ely_sync_client::SyncClientError::SessionEnded, + ely_sync_client::SyncClientError::SessionChanged, + ] { + let update = sync_failure_update(profile_id.clone(), error); + + assert!(matches!( + update, + SyncStateUpdate::AuthenticationExpired { profile_id: owner } if owner == profile_id + )); + } +} diff --git a/crates/ely_browser_core/src/sync_engine.rs b/crates/ely_browser_core/src/sync_engine.rs index 5161d5e..3fb656f 100644 --- a/crates/ely_browser_core/src/sync_engine.rs +++ b/crates/ely_browser_core/src/sync_engine.rs @@ -16,6 +16,7 @@ use crate::sync_records::{SNAPSHOT_SCHEMA_REV, SyncSnapshotBody}; mod concurrency; mod device_management; +mod session; mod vault_management; use concurrency::{conflict_head, ensure_remote_generation_is_available}; @@ -84,73 +85,6 @@ impl SyncEngine { self.bearer_store.load().map(|token| token.is_some()) } - /// Reconcile a pre-serialised local payload with the authenticated global snapshot head. - pub fn sync_bytes(&mut self, bytes: Vec) -> Result { - let Some(bearer) = self.bearer_store.load()? else { - let outcome = SyncOutcome::SignedOut; - self.last_outcome = Some(outcome.clone()); - return Ok(outcome); - }; - let client = SyncApiClient::new(self.api_config.clone(), bearer)?; - let Some((client, user_id)) = self.approved_client(client)? else { - let outcome = - SyncOutcome::AwaitingDeviceApproval { device_id: self.identity.device_id.clone() }; - self.last_outcome = Some(outcome.clone()); - return Ok(outcome); - }; - - let vault = self.resolve_vault(&client, &user_id)?; - let status = client.sync_status()?; - validate_sync_status(&status, &user_id, &self.identity.device_id)?; - let outcome = match status.snapshots.head { - Some(head) => { - let remote = self.download_remote_snapshot(&client, &user_id, &vault, head)?; - if remote.bytes == bytes && remote.merge_base.vault_generation() == vault.generation - { - SyncOutcome::AlreadyCurrent { - snapshot_id: remote.merge_base.snapshot_id().to_string(), - logical_clock: remote.merge_base.logical_clock(), - payload_bytes: remote.merge_base.size_bytes(), - device_id: remote.merge_base.device_id().to_string(), - } - } else if remote.bytes == bytes { - self.upload_payload(&client, &user_id, &vault, bytes, Some(&remote.merge_base))? - } else { - remote.into_outcome(false) - } - } - None => self.upload_payload(&client, &user_id, &vault, bytes, None)?, - }; - self.last_outcome = Some(outcome.clone()); - Ok(outcome) - } - - /// Upload a local payload after the UI thread has applied a remote - /// snapshot. The caller passes the remote logical clock so the new - /// merged snapshot is ordered after the downloaded one. - pub fn upload_merged_bytes( - &mut self, - bytes: Vec, - merge_base: AuthenticatedSnapshotHead, - ) -> Result { - let Some(bearer) = self.bearer_store.load()? else { - let outcome = SyncOutcome::SignedOut; - self.last_outcome = Some(outcome.clone()); - return Ok(outcome); - }; - let client = SyncApiClient::new(self.api_config.clone(), bearer)?; - let Some((client, user_id)) = self.approved_client(client)? else { - let outcome = - SyncOutcome::AwaitingDeviceApproval { device_id: self.identity.device_id.clone() }; - self.last_outcome = Some(outcome.clone()); - return Ok(outcome); - }; - let vault = self.resolve_vault(&client, &user_id)?; - let outcome = self.upload_payload(&client, &user_id, &vault, bytes, Some(&merge_base))?; - self.last_outcome = Some(outcome.clone()); - Ok(outcome) - } - fn approved_client( &self, client: SyncApiClient, diff --git a/crates/ely_browser_core/src/sync_engine/device_management.rs b/crates/ely_browser_core/src/sync_engine/device_management.rs index a5642e6..e5b64c7 100644 --- a/crates/ely_browser_core/src/sync_engine/device_management.rs +++ b/crates/ely_browser_core/src/sync_engine/device_management.rs @@ -9,7 +9,13 @@ use super::SyncEngine; impl SyncEngine { pub fn cloud_devices(&self) -> Result { - let client = self.authenticated_client()?; + self.with_required_authenticated_client(|client| self.cloud_devices_with_client(client)) + } + + fn cloud_devices_with_client( + &self, + client: SyncApiClient, + ) -> Result { let (client, registration) = self.registered_client(client)?; let devices = client.list_devices()?; validate_device_list( @@ -26,7 +32,18 @@ impl SyncEngine { target_device_id: &str, verification_code: &str, ) -> Result { - let (client, user_id) = self.approved_device_client()?; + self.with_required_authenticated_client(|client| { + self.approve_cloud_device_with_client(client, target_device_id, verification_code) + }) + } + + fn approve_cloud_device_with_client( + &self, + client: SyncApiClient, + target_device_id: &str, + verification_code: &str, + ) -> Result { + let (client, user_id) = self.approved_device_client(client)?; let devices = client.list_devices()?; validate_device_list(&devices, &user_id, &self.identity.device_id, true)?; let target = pending_device(&devices.devices, target_device_id)?; @@ -72,7 +89,17 @@ impl SyncEngine { &self, target_device_id: &str, ) -> Result { - let (client, user_id) = self.approved_device_client()?; + self.with_required_authenticated_client(|client| { + self.revoke_cloud_device_with_client(client, target_device_id) + }) + } + + fn revoke_cloud_device_with_client( + &self, + client: SyncApiClient, + target_device_id: &str, + ) -> Result { + let (client, user_id) = self.approved_device_client(client)?; let devices = client.list_devices()?; validate_device_list(&devices, &user_id, &self.identity.device_id, true)?; let target = revocable_device(&devices.devices, target_device_id)?; @@ -134,20 +161,15 @@ impl SyncEngine { Ok(document) } - fn approved_device_client(&self) -> Result<(SyncApiClient, String), SyncClientError> { - let client = self.authenticated_client()?; + fn approved_device_client( + &self, + client: SyncApiClient, + ) -> Result<(SyncApiClient, String), SyncClientError> { self.approved_client(client)?.ok_or_else(|| SyncClientError::DeviceApprovalStatus { device_id: self.identity.device_id.clone(), status: "pending".to_string(), }) } - - fn authenticated_client(&self) -> Result { - let bearer = self.bearer_store.load()?.ok_or(SyncClientError::DeviceTrust { - reason: "device management requires an authenticated session", - })?; - SyncApiClient::new(self.api_config.clone(), bearer) - } } fn validate_device_list( diff --git a/crates/ely_browser_core/src/sync_engine/session.rs b/crates/ely_browser_core/src/sync_engine/session.rs new file mode 100644 index 0000000..d9eed9b --- /dev/null +++ b/crates/ely_browser_core/src/sync_engine/session.rs @@ -0,0 +1,159 @@ +use ely_sync_client::{SyncApiClient, SyncClientError}; + +use super::{SyncEngine, SyncOutcome, validate_sync_status}; + +impl SyncEngine { + pub fn sync_bytes(&mut self, bytes: Vec) -> Result { + let outcome = self + .with_authenticated_client(|client| { + let Some((client, user_id)) = self.approved_client(client)? else { + return Ok(SyncOutcome::AwaitingDeviceApproval { + device_id: self.identity.device_id.clone(), + }); + }; + let vault = self.resolve_vault(&client, &user_id)?; + let status = client.sync_status()?; + validate_sync_status(&status, &user_id, &self.identity.device_id)?; + match status.snapshots.head { + Some(head) => { + let remote = + self.download_remote_snapshot(&client, &user_id, &vault, head)?; + if remote.bytes == bytes + && remote.merge_base.vault_generation() == vault.generation + { + Ok(SyncOutcome::AlreadyCurrent { + snapshot_id: remote.merge_base.snapshot_id().to_string(), + logical_clock: remote.merge_base.logical_clock(), + payload_bytes: remote.merge_base.size_bytes(), + device_id: remote.merge_base.device_id().to_string(), + }) + } else if remote.bytes == bytes { + self.upload_payload( + &client, + &user_id, + &vault, + bytes, + Some(&remote.merge_base), + ) + } else { + Ok(remote.into_outcome(false)) + } + } + None => self.upload_payload(&client, &user_id, &vault, bytes, None), + } + })? + .unwrap_or(SyncOutcome::SignedOut); + self.last_outcome = Some(outcome.clone()); + Ok(outcome) + } + + pub fn upload_merged_bytes( + &mut self, + bytes: Vec, + merge_base: ely_sync_client::AuthenticatedSnapshotHead, + ) -> Result { + let outcome = self + .with_authenticated_client(|client| { + let Some((client, user_id)) = self.approved_client(client)? else { + return Ok(SyncOutcome::AwaitingDeviceApproval { + device_id: self.identity.device_id.clone(), + }); + }; + let vault = self.resolve_vault(&client, &user_id)?; + self.upload_payload(&client, &user_id, &vault, bytes, Some(&merge_base)) + })? + .unwrap_or(SyncOutcome::SignedOut); + self.last_outcome = Some(outcome.clone()); + Ok(outcome) + } + + pub(super) fn with_required_authenticated_client( + &self, + operation: impl FnOnce(SyncApiClient) -> Result, + ) -> Result { + self.with_authenticated_client(operation)?.ok_or(SyncClientError::SessionEnded) + } + + fn with_authenticated_client( + &self, + operation: impl FnOnce(SyncApiClient) -> Result, + ) -> Result, SyncClientError> { + let Some(bearer) = self.bearer_store.load()? else { + return Ok(None); + }; + let client = SyncApiClient::new(self.api_config.clone(), bearer.clone())?; + reconcile_authenticated_result(operation(client), || { + self.bearer_store.clear_if_matches(&bearer) + }) + } +} + +fn reconcile_authenticated_result( + result: Result, + clear_if_matches: impl FnOnce() -> Result, +) -> Result, SyncClientError> { + match result { + Ok(value) => Ok(Some(value)), + Err(SyncClientError::SessionEnded) => { + if clear_if_matches()? { + Ok(None) + } else { + Err(SyncClientError::SessionChanged) + } + } + Err(error) => Err(error), + } +} + +#[cfg(test)] +mod tests { + use std::cell::Cell; + + use super::reconcile_authenticated_result; + use ely_sync_client::SyncClientError; + + #[test] + fn terminal_session_clears_the_captured_credential() -> Result<(), SyncClientError> { + let cleared = Cell::new(false); + let result = + reconcile_authenticated_result::<()>(Err(SyncClientError::SessionEnded), || { + cleared.set(true); + Ok(true) + })?; + + assert!(result.is_none()); + assert!(cleared.get()); + Ok(()) + } + + #[test] + fn replacement_credential_stays_signed_in() { + let result = + reconcile_authenticated_result::<()>(Err(SyncClientError::SessionEnded), || Ok(false)); + + assert!(matches!(result, Err(SyncClientError::SessionChanged))); + } + + #[test] + fn credential_clear_failures_are_preserved() { + let result = + reconcile_authenticated_result::<()>(Err(SyncClientError::SessionEnded), || { + Err(SyncClientError::BearerCredentialStorage("locked".to_string())) + }); + + assert!(matches!(result, Err(SyncClientError::BearerCredentialStorage(_)))); + } + + #[test] + fn ordinary_failures_preserve_the_credential() { + let clear_called = Cell::new(false); + let result = + reconcile_authenticated_result::<()>(Err(SyncClientError::SnapshotBusy), || { + clear_called.set(true); + Ok(true) + }); + + assert!(matches!(result, Err(SyncClientError::SnapshotBusy))); + assert!(!clear_called.get()); + } +} diff --git a/crates/ely_sync_client/src/client.rs b/crates/ely_sync_client/src/client.rs index 6b8bf55..41453f8 100644 --- a/crates/ely_sync_client/src/client.rs +++ b/crates/ely_sync_client/src/client.rs @@ -429,6 +429,9 @@ fn read_json_response( Ok(ok) => read_json_from_response(endpoint, ok), Err(ureq::Error::Status(status, raw)) => { let body = raw.into_string().unwrap_or_default(); + if session::response_ends_session(status, &body) { + return Err(SyncClientError::SessionEnded); + } Err(SyncClientError::HttpStatus { endpoint: endpoint.to_string(), status, body }) } Err(other) => { diff --git a/crates/ely_sync_client/src/client/session.rs b/crates/ely_sync_client/src/client/session.rs index 054eee5..4093eda 100644 --- a/crates/ely_sync_client/src/client/session.rs +++ b/crates/ely_sync_client/src/client/session.rs @@ -35,7 +35,7 @@ impl SyncApiClient { } Err(ureq::Error::Status(status, response)) => { let body = response.into_string().unwrap_or_default(); - if status == 401 && logout_is_already_complete(&body) { + if response_ends_session(status, &body) { return Ok(()); } Err(SyncClientError::HttpStatus { endpoint, status, body }) @@ -45,8 +45,32 @@ impl SyncApiClient { } } -fn logout_is_already_complete(body: &str) -> bool { - serde_json::from_str::(body).is_ok_and(|document| { - matches!(document.error.as_str(), "session_not_found" | "session_expired") - }) +pub(super) fn response_ends_session(status: u16, body: &str) -> bool { + status == 401 + && serde_json::from_str::(body).is_ok_and(|document| { + matches!(document.error.as_str(), "session_not_found" | "session_expired") + }) +} + +#[cfg(test)] +mod tests { + use super::response_ends_session; + + #[test] + fn terminal_session_parser_is_exact() { + for body in [r#"{"error":"session_not_found"}"#, r#"{"error":"session_expired"}"#] { + assert!(response_ends_session(401, body)); + } + for (status, body) in [ + (401, r#"{"error":"authorization_missing"}"#), + (401, r#"{"error":"authorization_invalid"}"#), + (401, r#"{"error":"unknown"}"#), + (401, r#"{"error":"session_not_found","extra":true}"#), + (401, "{"), + (403, r#"{"error":"session_not_found"}"#), + (500, r#"{"error":"session_expired"}"#), + ] { + assert!(!response_ends_session(status, body)); + } + } } diff --git a/crates/ely_sync_client/src/client_tests.rs b/crates/ely_sync_client/src/client_tests.rs index d061266..6fae0da 100644 --- a/crates/ely_sync_client/src/client_tests.rs +++ b/crates/ely_sync_client/src/client_tests.rs @@ -89,6 +89,46 @@ fn sign_out_preserves_retryable_failures() -> Result<(), Box> { Ok(()) } +#[test] +fn runtime_requests_classify_only_strict_terminal_sessions() -> Result<(), Box> { + for code in ["session_not_found", "session_expired"] { + let body = format!(r#"{{"error":"{code}"}}"#); + let (base_url, server) = + spawn_authenticated_server("GET /api/devices HTTP/1.1\r\n", "401 Unauthorized", &body)?; + let client = SyncApiClient::new( + ApiClientConfig::custom(base_url, "auto"), + BearerToken::new("a".repeat(64))?, + )?; + + assert!(matches!(client.list_devices(), Err(crate::SyncClientError::SessionEnded))); + join_server(server)?; + } + + for (status_line, body, expected_status) in [ + ("401 Unauthorized", r#"{"error":"authorization_missing"}"#, 401), + ("401 Unauthorized", r#"{"error":"authorization_invalid"}"#, 401), + ("401 Unauthorized", r#"{"error":"unknown"}"#, 401), + ("401 Unauthorized", r#"{"error":"session_not_found","extra":true}"#, 401), + ("401 Unauthorized", "{", 401), + ("403 Forbidden", r#"{"error":"session_not_found"}"#, 403), + ("500 Internal Server Error", r#"{"error":"session_expired"}"#, 500), + ] { + let (base_url, server) = + spawn_authenticated_server("GET /api/devices HTTP/1.1\r\n", status_line, body)?; + let client = SyncApiClient::new( + ApiClientConfig::custom(base_url, "auto"), + BearerToken::new("a".repeat(64))?, + )?; + + assert!(matches!( + client.list_devices(), + Err(crate::SyncClientError::HttpStatus { status, .. }) if status == expected_status + )); + join_server(server)?; + } + Ok(()) +} + #[test] fn upload_parses_structured_snapshot_head_conflict() -> Result<(), Box> { let (base_url, server) = spawn_conflict_server()?; @@ -179,6 +219,14 @@ fn spawn_conflict_server() -> Result<(String, TestServer), Box> { fn spawn_logout_server( status_line: &'static str, body: &str, +) -> Result<(String, TestServer), Box> { + spawn_authenticated_server("POST /api/session/logout HTTP/1.1\r\n", status_line, body) +} + +fn spawn_authenticated_server( + expected_request_line: &'static str, + status_line: &'static str, + body: &str, ) -> Result<(String, TestServer), Box> { let listener = TcpListener::bind("127.0.0.1:0")?; let address = listener.local_addr()?; @@ -195,7 +243,7 @@ fn spawn_logout_server( request.extend_from_slice(&chunk[..read]); } let request = String::from_utf8_lossy(&request); - if !request.starts_with("POST /api/session/logout HTTP/1.1\r\n") + if !request.starts_with(expected_request_line) || !request.contains(&format!("Authorization: Bearer {}\r\n", "a".repeat(64))) { return Err(std::io::Error::other("logout request contract mismatch")); diff --git a/crates/ely_sync_client/src/error.rs b/crates/ely_sync_client/src/error.rs index 2ed41d2..c550dcb 100644 --- a/crates/ely_sync_client/src/error.rs +++ b/crates/ely_sync_client/src/error.rs @@ -14,6 +14,12 @@ pub enum SyncClientError { #[error("Session logout response is invalid")] SessionLogoutInvalid, + #[error("Authenticated session ended")] + SessionEnded, + + #[error("Authenticated session changed during reconciliation")] + SessionChanged, + #[error("HTTP request failed for {endpoint}: {source}")] Http { endpoint: String,