diff --git a/PRD.md b/PRD.md index 834e8fa..843a106 100644 --- a/PRD.md +++ b/PRD.md @@ -961,6 +961,7 @@ Better Auth 在 Cloudflare Workers 中初始化，D1 binding 作为 database 传 - D1 session validation 与自定义 device/session binding。 - 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. - 设备注册、rebind、批准、撤销与 Vault rotation。 - Signed Sync reset 和 signed account deletion。 - 管理所有 `/api/auth/*` 路由。 diff --git a/crates/ely_app/src/shell/auth.rs b/crates/ely_app/src/shell/auth.rs index e3ee33b..9315239 100644 --- a/crates/ely_app/src/shell/auth.rs +++ b/crates/ely_app/src/shell/auth.rs @@ -7,19 +7,20 @@ //! so the existing 8 ms tick is the single point that reconciles //! background-task state with `BrowserCore`. -use std::{path::Path, sync::mpsc::Sender}; +use std::path::Path; -use ely_browser_core::SyncEngine; use ely_domain::ProfileId; use ely_sync_client::{ - ApiClientConfig, BearerToken, BearerTokenStore, SyncClientError, send_email_otp, + ApiClientConfig, BearerToken, BearerTokenStore, SyncApiClient, SyncClientError, send_email_otp, verify_email_otp, }; use gpui::Context; use crate::services::servo_profile_data::{default_profile_data_root, sync_profile_data_dir}; -use super::sync_state::{SyncStateUpdate, sync_platform_label}; +use super::auth_operation_gate::AuthenticatedOperationBarrier; +use super::sync_inbox::{SyncStateSender, SyncWorkGeneration}; +use super::sync_state::SyncStateUpdate; use super::{ElyShell, ShellState}; /// Where the user is in the email OTP form. Tracked on `ElyShell` so @@ -47,6 +48,20 @@ pub(crate) enum AuthFlowPhase { Error { profile_id: ProfileId, email: String, message: String }, } +#[derive(Clone, Debug, Eq, PartialEq)] +pub(super) enum SignOutPhase { + SigningOut { generation: SyncWorkGeneration }, + Error { generation: SyncWorkGeneration, message: String }, +} + +impl SignOutPhase { + pub(super) fn generation(&self) -> SyncWorkGeneration { + match self { + Self::SigningOut { generation } | Self::Error { generation, .. } => *generation, + } + } +} + impl AuthFlowPhase { pub(crate) fn error_message(&self) -> Option<&str> { match self { @@ -89,20 +104,21 @@ impl ElyShell { return; }; let profile_id = active_profile.id; + self.invalidate_sync_work(); + self.hold_auth_flow_barrier(); self.auth_flow_phase = AuthFlowPhase::SendingCode { profile_id: profile_id.clone(), email: email.clone() }; - let tx = self.sync_inbox_tx.clone(); + let tx = self.sync_state_sender(); spawn_send_otp(profile_id, email, tx); } /// Hand the typed OTP to the worker thread that calls - /// `verify_email_otp`, persists the bearer token, and triggers - /// the first snapshot upload on success. + /// `verify_email_otp`; the UI generation validates its result + /// before persisting the bearer and starting the first sync. pub(crate) fn submit_email_otp_verify(&mut self, cx: &mut Context) { let (profile_id, email) = match self.auth_flow_phase.clone() { AuthFlowPhase::AwaitingOtp { profile_id, email } - | AuthFlowPhase::Error { profile_id, email, .. } - | AuthFlowPhase::Verifying { profile_id, email } => (profile_id, email), + | AuthFlowPhase::Error { profile_id, email, .. } => (profile_id, email), _ => return, }; let otp = self.read_auth_otp_input(cx); @@ -120,10 +136,12 @@ impl ElyShell { None => return, }; if active_profile.id != profile_id { + self.release_auth_flow_barrier(); self.auth_flow_phase = AuthFlowPhase::Idle; return; } - let Some(profile_root) = default_profile_data_root() else { + if default_profile_data_root().is_none() { + self.release_auth_flow_barrier(); self.auth_flow_phase = AuthFlowPhase::Error { profile_id, email, @@ -131,59 +149,69 @@ impl ElyShell { }; return; }; - let profile_dir = sync_profile_data_dir(&profile_root, &profile_id); + self.invalidate_sync_work(); + self.hold_auth_flow_barrier(); self.auth_flow_phase = AuthFlowPhase::Verifying { profile_id: profile_id.clone(), email: email.clone() }; - let tx = self.sync_inbox_tx.clone(); - spawn_verify_otp(profile_id, email, normalized_otp, profile_dir, tx); + let tx = self.sync_state_sender(); + spawn_verify_otp(profile_id, email, normalized_otp, tx); } - /// Drop the persisted bearer token and reset the local form. - pub(crate) fn submit_sign_out(&mut self, _cx: &mut Context) { + /// Revoke the remote bearer session, then conditionally clear the + /// exact local credential used by that request. + pub(crate) fn submit_sign_out(&mut self, cx: &mut Context) { let active_profile_id = match active_profile_id_for(&self.state) { Some(profile_id) => profile_id, None => return, }; + if matches!( + self.sign_out_phases.get(&active_profile_id), + Some(SignOutPhase::SigningOut { .. }) + ) { + return; + } + self.invalidate_sync_work(); + let generation = self.sync_generation; + let draining_barrier = self.block_authenticated_operations(); + self.sign_out_phases + .insert(active_profile_id.clone(), SignOutPhase::SigningOut { generation }); self.auth_flow_phase = AuthFlowPhase::Idle; - self.sync_devices.reset(); - self.sync_upload_scheduled = false; - self.sync_retry_at = None; - self.clear_pending_cloud_sync_upload(); + if let ShellState::Ready(core) = &mut self.state { + core.set_sync_connection_state(ely_domain::SyncConnectionState::SigningOut); + } + cx.notify(); let Some(profile_root) = default_profile_data_root() else { self.set_sign_out_error( active_profile_id, + generation, "Profile data root is unavailable. Retry sign out.", ); return; }; let profile_dir = sync_profile_data_dir(&profile_root, &active_profile_id); - if let Err(error) = clear_persisted_bearer( + let store = bearer_store_for_profile( &active_profile_id, &profile_dir, &profile_root, self.default_profile_id.as_ref(), - ) { - tracing::warn!(target: "ely::sync", error = %error, "sign-out failed to clear bearer"); - self.set_sign_out_error( - active_profile_id, - "System credential access failed. Retry sign out.", - ); - return; - } - if let ShellState::Ready(core) = &mut self.state { - core.set_sync_connection_state(ely_domain::SyncConnectionState::SignedOut); - } + ); + let tx = self.sync_state_sender(); + spawn_sign_out(active_profile_id, store, draining_barrier, tx); } - fn set_sign_out_error(&mut self, profile_id: ProfileId, message: &str) { - self.auth_flow_phase = - AuthFlowPhase::Error { profile_id, email: String::new(), message: message.to_string() }; + fn set_sign_out_error( + &mut self, + profile_id: ProfileId, + generation: SyncWorkGeneration, + message: &str, + ) { + self.sign_out_phases + .insert(profile_id, SignOutPhase::Error { generation, message: message.to_string() }); + self.auth_flow_phase = AuthFlowPhase::Idle; if let ShellState::Ready(core) = &mut self.state { - core.set_sync_connection_state( - ely_domain::SyncConnectionState::CredentialUnavailable { - message: message.to_string(), - }, - ); + core.set_sync_connection_state(ely_domain::SyncConnectionState::SignOutError { + message: message.to_string(), + }); } } @@ -226,15 +254,6 @@ fn active_profile_id_for(state: &ShellState) -> Option { core.snapshot().ok().map(|snapshot| snapshot.active_profile_id) } -pub(super) fn clear_persisted_bearer( - profile_id: &ProfileId, - profile_dir: &Path, - profile_root: &Path, - default_profile_id: Option<&ProfileId>, -) -> Result<(), SyncClientError> { - bearer_store_for_profile(profile_id, profile_dir, profile_root, default_profile_id).clear() -} - pub(super) fn bearer_store_for_profile( profile_id: &ProfileId, profile_dir: &Path, @@ -250,10 +269,81 @@ pub(super) fn bearer_store_for_profile( ) } -fn spawn_send_otp(profile_id: ProfileId, email: String, tx: Sender) { - std::thread::Builder::new() - .name("ely-sync-auth-send".to_string()) +fn spawn_sign_out( + profile_id: ProfileId, + store: BearerTokenStore, + draining_barrier: AuthenticatedOperationBarrier, + tx: SyncStateSender, +) { + let worker_tx = tx.clone(); + let worker_profile_id = profile_id.clone(); + let spawn_result = std::thread::Builder::new() + .name("ely-sync-auth-sign-out".to_string()) .spawn(move || { + let update = match store.load() { + Ok(token) => { + draining_barrier.wait_until_idle(); + sign_out_loaded_token(worker_profile_id, store, token) + } + Err(error) => { + tracing::warn!(target: "ely::sync", error = %error, "sign-out credential load failed"); + SyncStateUpdate::SignOutFailed { + profile_id: worker_profile_id, + message: "System credential access failed. Retry sign out.".to_string(), + } + } + }; + draining_barrier.release(); + let _ = worker_tx.send(update); + }); + if let Err(error) = spawn_result { + tracing::warn!(target: "ely::sync", error = %error, "spawn sign-out worker failed"); + let _ = tx.send(SyncStateUpdate::SignOutFailed { + profile_id, + message: "Session revoke failed.".to_string(), + }); + } +} + +fn sign_out_loaded_token( + profile_id: ProfileId, + store: BearerTokenStore, + token: Option, +) -> SyncStateUpdate { + let Some(token) = token else { + return SyncStateUpdate::SignOutSucceeded { profile_id }; + }; + if let Err(error) = SyncApiClient::new(ApiClientConfig::production(), token.clone()) + .and_then(|client| client.sign_out()) + { + tracing::warn!(target: "ely::sync", error = %error, "remote sign-out failed"); + return SyncStateUpdate::SignOutFailed { + profile_id, + message: "Session revoke failed.".to_string(), + }; + } + match store.clear_if_matches(&token) { + Ok(true) => SyncStateUpdate::SignOutSucceeded { profile_id }, + Ok(false) => SyncStateUpdate::SignOutFailed { + profile_id, + message: "Session changed. Retry sign out.".to_string(), + }, + Err(error) => { + tracing::warn!(target: "ely::sync", error = %error, "sign-out credential clear failed"); + SyncStateUpdate::SignOutFailed { + profile_id, + message: "System credential access failed. Retry sign out.".to_string(), + } + } + } +} + +fn spawn_send_otp(profile_id: ProfileId, email: String, tx: SyncStateSender) { + let failure_profile_id = profile_id.clone(); + let failure_email = email.clone(); + let failure_tx = tx.clone(); + let spawn_result = + std::thread::Builder::new().name("ely-sync-auth-send".to_string()).spawn(move || { let config = ApiClientConfig::production(); match send_email_otp(&config, &email) { Ok(()) => { @@ -267,23 +357,23 @@ fn spawn_send_otp(profile_id: ProfileId, email: String, tx: Sender, -) { - std::thread::Builder::new() - .name("ely-sync-auth-verify".to_string()) - .spawn(move || { +fn spawn_verify_otp(profile_id: ProfileId, email: String, otp: String, tx: SyncStateSender) { + let failure_profile_id = profile_id.clone(); + let failure_email = email.clone(); + let failure_tx = tx.clone(); + let spawn_result = + std::thread::Builder::new().name("ely-sync-auth-verify".to_string()).spawn(move || { let config = ApiClientConfig::production(); let token: BearerToken = match verify_email_otp(&config, &email, &otp) { Ok(token) => token, @@ -296,147 +386,59 @@ fn spawn_verify_otp( return; } }; - let mut engine = - match SyncEngine::for_profile_dir( - &profile_id, - &profile_dir, - "ELY", - sync_platform_label(), - ) { - Ok(engine) => engine, - Err(error) => { - let _ = tx.send(SyncStateUpdate::AuthError { - profile_id, - email, - message: error.to_string(), - }); - return; - } - }; - if let Err(error) = engine.install_bearer(token.as_str()) { - let _ = tx.send(SyncStateUpdate::AuthError { - profile_id, - email, - message: error.to_string(), - }); - return; + let update = SyncStateUpdate::AuthVerified { profile_id, email, token }; + if let Err(error) = tx.send(update) { + retire_stale_auth_update(error.0.update); + } + }); + if let Err(error) = spawn_result { + tracing::warn!(target: "ely::sync", error = %error, "spawn ely-sync-auth-verify failed"); + let _ = failure_tx.send(SyncStateUpdate::AuthError { + profile_id: failure_profile_id, + email: failure_email, + message: "Unable to start sign-in. Retry.".to_string(), + }); + } +} + +pub(super) fn retire_stale_auth_update(update: SyncStateUpdate) { + let Some(token) = verified_bearer_from(update) else { + return; + }; + std::thread::Builder::new() + .name("ely-sync-auth-retire".to_string()) + .spawn(move || { + let client = SyncApiClient::new(ApiClientConfig::production(), token); + if let Ok(client) = client { + let _ = client.sign_out(); } - let _ = tx.send(SyncStateUpdate::AuthSucceeded { profile_id, email }); }) .map(|_| ()) .unwrap_or_else(|error| { - tracing::warn!(target: "ely::sync", error = %error, "spawn ely-sync-auth-verify failed"); + tracing::warn!(target: "ely::sync", error = %error, "spawn auth retire failed"); }); } -#[cfg(test)] -mod tests { - use ely_browser_core::{BrowserCore, InitialBrowserConfig}; - use ely_domain::ProfileId; - - use super::{ - AuthFlowPhase, active_profile_sync_context_for, bearer_store_for_profile, normalize_email, - }; - use crate::shell::ShellState; - - #[test] - fn normalize_lowercases_and_trims() { - assert_eq!(normalize_email(" User@Example.COM "), Some("user@example.com".to_string())); - } - - #[test] - fn normalize_rejects_obviously_broken() { - assert_eq!(normalize_email("noatsign"), None); - assert_eq!(normalize_email("@no-local-part"), None); - assert_eq!(normalize_email("missing-domain@"), None); - assert_eq!(normalize_email(""), None); - } - - #[test] - fn auth_phase_helpers() { - let profile_id = ProfileId::new(); - let phase = AuthFlowPhase::Verifying { - profile_id: profile_id.clone(), - email: "you@there".to_string(), - }; - assert!(phase.is_busy()); - assert!(phase.belongs_to(&profile_id)); - assert!(!phase.belongs_to(&ProfileId::new())); - assert_eq!(phase.error_message(), None); - - let phase = AuthFlowPhase::Error { - profile_id, - email: "you@there".to_string(), - message: "rate limited".to_string(), - }; - assert_eq!(phase.error_message(), Some("rate limited")); - assert!(!phase.is_busy()); - } - - #[test] - fn private_profile_has_no_sync_auth_context() -> Result<(), Box> { - let state = - ShellState::Ready(Box::new(BrowserCore::new(InitialBrowserConfig::private_window()?)?)); - - assert_eq!(active_profile_sync_context_for(&state), None); - - let state = - ShellState::Ready(Box::new(BrowserCore::new(InitialBrowserConfig::ely_defaults()?)?)); - assert!(active_profile_sync_context_for(&state).is_some()); - - Ok(()) - } - - #[test] - fn default_profile_store_cleans_stable_and_old_legacy_paths() - -> Result<(), Box> { - let directory = tempfile::tempdir()?; - let profile_id = ProfileId::new(); - let profile_dir = directory.path().join(profile_id.as_str()).join("servo"); - let stable = profile_dir.join("sync/bearer.token"); - let old = directory.path().join("default/servo/sync/bearer.token"); - std::fs::create_dir_all(stable.parent().ok_or("missing stable parent")?)?; - std::fs::create_dir_all(old.parent().ok_or("missing old parent")?)?; - std::fs::write(&stable, "a".repeat(64))?; - std::fs::write(&old, "b".repeat(64))?; - let store = bearer_store_for_profile( - &profile_id, - &profile_dir, - directory.path(), - Some(&profile_id), - ); - - store.clear_legacy_files()?; - - assert!(!stable.exists()); - assert!(!old.exists()); - Ok(()) - } - - #[test] - fn custom_profile_store_leaves_default_legacy_credentials_untouched() - -> Result<(), Box> { - let directory = tempfile::tempdir()?; - let profile_id = ProfileId::new(); - let default_profile_id = ProfileId::new(); - let profile_dir = directory.path().join(profile_id.as_str()).join("servo"); - let stable = profile_dir.join("sync/bearer.token"); - let old_default = directory.path().join("default/servo/sync/bearer.token"); - std::fs::create_dir_all(stable.parent().ok_or("missing stable parent")?)?; - std::fs::create_dir_all(old_default.parent().ok_or("missing default parent")?)?; - std::fs::write(&stable, "a".repeat(64))?; - std::fs::write(&old_default, "b".repeat(64))?; - let store = bearer_store_for_profile( - &profile_id, - &profile_dir, - directory.path(), - Some(&default_profile_id), - ); - - store.clear_legacy_files()?; - - assert!(!stable.exists()); - assert!(old_default.exists()); - Ok(()) +fn verified_bearer_from(update: SyncStateUpdate) -> Option { + match update { + SyncStateUpdate::AuthVerified { token, .. } => Some(token), + _ => None, } } + +pub(super) fn save_verified_bearer( + profile_id: &ProfileId, + token: &BearerToken, + default_profile_id: Option<&ProfileId>, +) -> Result<(), SyncClientError> { + let profile_root = default_profile_data_root().ok_or_else(|| { + SyncClientError::BearerCredentialStorage("profile data root is unavailable".to_string()) + })?; + let profile_dir = sync_profile_data_dir(&profile_root, profile_id); + bearer_store_for_profile(profile_id, &profile_dir, &profile_root, default_profile_id) + .save(token) +} + +#[cfg(test)] +#[path = "auth_tests.rs"] +mod tests; diff --git a/crates/ely_app/src/shell/auth_operation_gate.rs b/crates/ely_app/src/shell/auth_operation_gate.rs new file mode 100644 index 0000000..243caa3 --- /dev/null +++ b/crates/ely_app/src/shell/auth_operation_gate.rs @@ -0,0 +1,149 @@ +use std::sync::{Arc, Condvar, Mutex}; + +use super::ElyShell; + +#[derive(Debug)] +struct GateState { + blockers: usize, + active: usize, +} + +#[derive(Clone, Debug)] +pub(super) struct AuthenticatedOperationGate { + inner: Arc<(Mutex, Condvar)>, +} + +#[derive(Debug)] +pub(super) struct AuthenticatedOperationLease { + inner: Arc<(Mutex, Condvar)>, +} + +#[derive(Debug)] +pub(super) struct AuthenticatedOperationBarrier { + inner: Option, Condvar)>>, +} + +impl AuthenticatedOperationGate { + pub(super) fn open() -> Self { + Self { inner: Arc::new((Mutex::new(GateState { blockers: 0, active: 0 }), Condvar::new())) } + } + + pub(super) fn try_acquire(&self) -> Option { + let (mutex, _) = self.inner.as_ref(); + let mut state = mutex.lock().unwrap_or_else(std::sync::PoisonError::into_inner); + if state.blockers > 0 { + return None; + } + state.active += 1; + Some(AuthenticatedOperationLease { inner: self.inner.clone() }) + } + + fn block_new_operations(&self) -> AuthenticatedOperationBarrier { + let (mutex, _) = self.inner.as_ref(); + let mut state = mutex.lock().unwrap_or_else(std::sync::PoisonError::into_inner); + state.blockers += 1; + AuthenticatedOperationBarrier { inner: Some(self.inner.clone()) } + } +} + +impl AuthenticatedOperationBarrier { + pub(super) fn wait_until_idle(&self) { + let Some(inner) = &self.inner else { + return; + }; + let (mutex, condition) = inner.as_ref(); + let mut state = mutex.lock().unwrap_or_else(std::sync::PoisonError::into_inner); + while state.active > 0 { + state = condition.wait(state).unwrap_or_else(std::sync::PoisonError::into_inner); + } + } + + pub(super) fn release(mut self) { + self.release_inner(); + } + + fn release_inner(&mut self) { + let Some(inner) = self.inner.take() else { + return; + }; + let (mutex, condition) = inner.as_ref(); + let mut state = mutex.lock().unwrap_or_else(std::sync::PoisonError::into_inner); + state.blockers = state.blockers.saturating_sub(1); + condition.notify_all(); + } +} + +impl Drop for AuthenticatedOperationBarrier { + fn drop(&mut self) { + self.release_inner(); + } +} + +impl Drop for AuthenticatedOperationLease { + fn drop(&mut self) { + let (mutex, condition) = self.inner.as_ref(); + let mut state = mutex.lock().unwrap_or_else(std::sync::PoisonError::into_inner); + state.active = state.active.saturating_sub(1); + if state.active == 0 { + condition.notify_all(); + } + } +} + +impl ElyShell { + pub(super) fn begin_authenticated_operation(&self) -> Option { + self.authenticated_operation_gate.try_acquire() + } + + pub(super) fn block_authenticated_operations(&self) -> AuthenticatedOperationBarrier { + self.authenticated_operation_gate.block_new_operations() + } + + pub(super) fn hold_auth_flow_barrier(&mut self) { + let barrier = self.block_authenticated_operations(); + self.auth_flow_barrier = Some(barrier); + } + + pub(super) fn release_auth_flow_barrier(&mut self) { + self.auth_flow_barrier.take(); + } +} + +#[cfg(test)] +mod tests { + use super::AuthenticatedOperationGate; + + #[test] + fn closed_gate_rejects_new_work_and_waits_for_active_work() + -> Result<(), Box> { + let gate = AuthenticatedOperationGate::open(); + let lease = gate.try_acquire().ok_or("open gate rejected work")?; + let barrier = gate.block_new_operations(); + assert!(gate.try_acquire().is_none()); + + let (sender, receiver) = std::sync::mpsc::channel(); + let waiter = std::thread::spawn(move || { + barrier.wait_until_idle(); + let _ = sender.send(()); + }); + assert!(receiver.recv_timeout(std::time::Duration::from_millis(20)).is_err()); + drop(lease); + receiver.recv_timeout(std::time::Duration::from_secs(1))?; + waiter.join().map_err(|_| "gate waiter panicked")?; + Ok(()) + } + + #[test] + fn overlapping_barriers_release_independently() { + let gate = AuthenticatedOperationGate::open(); + let first = gate.block_new_operations(); + let second = gate.block_new_operations(); + assert!(gate.try_acquire().is_none()); + + first.release(); + assert!(gate.try_acquire().is_none()); + + second.release(); + assert!(gate.try_acquire().is_some()); + } +} diff --git a/crates/ely_app/src/shell/auth_tests.rs b/crates/ely_app/src/shell/auth_tests.rs new file mode 100644 index 0000000..de6e2bf --- /dev/null +++ b/crates/ely_app/src/shell/auth_tests.rs @@ -0,0 +1,118 @@ +use ely_browser_core::{BrowserCore, InitialBrowserConfig}; +use ely_domain::ProfileId; + +use super::super::{ShellState, sync_state::SyncStateUpdate}; +use super::{ + AuthFlowPhase, active_profile_sync_context_for, bearer_store_for_profile, normalize_email, + verified_bearer_from, +}; + +#[test] +fn normalize_lowercases_and_trims() { + assert_eq!(normalize_email(" User@Example.COM "), Some("user@example.com".to_string())); +} + +#[test] +fn normalize_rejects_obviously_broken() { + assert_eq!(normalize_email("noatsign"), None); + assert_eq!(normalize_email("@no-local-part"), None); + assert_eq!(normalize_email("missing-domain@"), None); + assert_eq!(normalize_email(""), None); +} + +#[test] +fn auth_phase_helpers() { + let profile_id = ProfileId::new(); + let phase = + AuthFlowPhase::Verifying { profile_id: profile_id.clone(), email: "you@there".to_string() }; + assert!(phase.is_busy()); + assert!(phase.belongs_to(&profile_id)); + assert!(!phase.belongs_to(&ProfileId::new())); + assert_eq!(phase.error_message(), None); + + let phase = AuthFlowPhase::Error { + profile_id, + email: "you@there".to_string(), + message: "rate limited".to_string(), + }; + assert_eq!(phase.error_message(), Some("rate limited")); + assert!(!phase.is_busy()); +} + +#[test] +fn stale_verified_updates_retain_the_token_for_server_retirement() +-> Result<(), Box> { + let profile_id = ProfileId::new(); + let token = ely_sync_client::BearerToken::new("a".repeat(64))?; + let update = SyncStateUpdate::AuthVerified { + profile_id, + email: "user@example.com".to_string(), + token: token.clone(), + }; + + assert_eq!(verified_bearer_from(update), Some(token)); + Ok(()) +} + +#[test] +fn private_profile_has_no_sync_auth_context() -> Result<(), Box> { + let state = + ShellState::Ready(Box::new(BrowserCore::new(InitialBrowserConfig::private_window()?)?)); + + assert_eq!(active_profile_sync_context_for(&state), None); + + let state = + ShellState::Ready(Box::new(BrowserCore::new(InitialBrowserConfig::ely_defaults()?)?)); + assert!(active_profile_sync_context_for(&state).is_some()); + + Ok(()) +} + +#[test] +fn default_profile_store_cleans_stable_and_old_legacy_paths() +-> Result<(), Box> { + let directory = tempfile::tempdir()?; + let profile_id = ProfileId::new(); + let profile_dir = directory.path().join(profile_id.as_str()).join("servo"); + let stable = profile_dir.join("sync/bearer.token"); + let old = directory.path().join("default/servo/sync/bearer.token"); + std::fs::create_dir_all(stable.parent().ok_or("missing stable parent")?)?; + std::fs::create_dir_all(old.parent().ok_or("missing old parent")?)?; + std::fs::write(&stable, "a".repeat(64))?; + std::fs::write(&old, "b".repeat(64))?; + let store = + bearer_store_for_profile(&profile_id, &profile_dir, directory.path(), Some(&profile_id)); + + store.clear_legacy_files()?; + + assert!(!stable.exists()); + assert!(!old.exists()); + Ok(()) +} + +#[test] +fn custom_profile_store_leaves_default_legacy_credentials_untouched() +-> Result<(), Box> { + let directory = tempfile::tempdir()?; + let profile_id = ProfileId::new(); + let default_profile_id = ProfileId::new(); + let profile_dir = directory.path().join(profile_id.as_str()).join("servo"); + let stable = profile_dir.join("sync/bearer.token"); + let old_default = directory.path().join("default/servo/sync/bearer.token"); + std::fs::create_dir_all(stable.parent().ok_or("missing stable parent")?)?; + std::fs::create_dir_all(old_default.parent().ok_or("missing default parent")?)?; + std::fs::write(&stable, "a".repeat(64))?; + std::fs::write(&old_default, "b".repeat(64))?; + let store = bearer_store_for_profile( + &profile_id, + &profile_dir, + directory.path(), + Some(&default_profile_id), + ); + + store.clear_legacy_files()?; + + assert!(!stable.exists()); + assert!(old_default.exists()); + Ok(()) +} diff --git a/crates/ely_app/src/shell/internal_pages/sync.rs b/crates/ely_app/src/shell/internal_pages/sync.rs index a03bde8..bc7bf3e 100644 --- a/crates/ely_app/src/shell/internal_pages/sync.rs +++ b/crates/ely_app/src/shell/internal_pages/sync.rs @@ -11,7 +11,8 @@ use gpui_component::{input::Input, scroll::ScrollableElement}; use crate::shell::auth::AuthFlowPhase; use super::sync_controls::{ - button_bg, render_dual_button_row, render_policy_toggle, render_primary_button, + button_bg, render_card_heading, render_dual_button_row, render_field_label, + render_inline_error, render_input, render_policy_toggle, render_primary_button, render_reset_button, render_secondary_button, render_sign_out_button, }; use super::{ElyShell, render_canvas_surface}; @@ -24,7 +25,10 @@ impl ElyShell { if profile_allows_sync_controls(&snapshot.active_profile_kind) && !matches!( snapshot.sync_status.connection(), - SyncConnectionState::SignedOut | SyncConnectionState::CredentialUnavailable { .. } + SyncConnectionState::SignedOut + | SyncConnectionState::CredentialUnavailable { .. } + | SyncConnectionState::SigningOut + | SyncConnectionState::SignOutError { .. } ) { self.ensure_sync_devices_loaded(cx); @@ -105,6 +109,39 @@ fn render_account_card( .child(render_inline_error(message)) .children(account_form(shell, &snapshot.active_profile_id, cx)) .into_any_element(), + SyncConnectionState::SigningOut => card + .child( + div() + .flex() + .items_center() + .justify_between() + .child(render_card_heading("Account")) + .child(render_sign_out_button(shell, "Signing out...", true, cx)), + ) + .child( + div() + .text_size(px(12.0)) + .text_color(rgb(colors::ink_3())) + .child("End-to-end encrypted"), + ) + .into_any_element(), + SyncConnectionState::SignOutError { message } => card + .child( + div() + .flex() + .items_center() + .justify_between() + .child(render_card_heading("Account")) + .child(render_sign_out_button(shell, "Retry sign out", false, cx)), + ) + .child(render_inline_error(message)) + .child( + div() + .text_size(px(12.0)) + .text_color(rgb(colors::ink_3())) + .child("End-to-end encrypted"), + ) + .into_any_element(), SyncConnectionState::SignedIn | SyncConnectionState::AwaitingDeviceApproval | SyncConnectionState::SyncReady { .. } @@ -115,7 +152,7 @@ fn render_account_card( .items_center() .justify_between() .child(render_card_heading("Account")) - .child(render_sign_out_button(shell, cx)), + .child(render_sign_out_button(shell, "Sign out", false, cx)), ) .child( div() @@ -431,42 +468,6 @@ fn render_sync_object_row( .into_any_element() } -fn render_card_heading(label: &'static str) -> AnyElement { - div() - .text_size(px(13.0)) - .font_weight(FontWeight(500.0)) - .text_color(rgb(colors::ink())) - .child(label) - .into_any_element() -} - -fn render_field_label(label: &'static str) -> AnyElement { - div() - .text_size(px(10.5)) - .font_weight(FontWeight(500.0)) - .text_color(rgb(colors::ink_4())) - .child(label) - .into_any_element() -} - -fn render_input(state: &gpui::Entity) -> AnyElement { - div() - .px(px(10.0)) - .py(px(8.0)) - .rounded(px(8.0)) - .bg(rgba(button_bg())) - .child(Input::new(state).appearance(false).cleanable(false)) - .into_any_element() -} - -fn render_inline_error(message: &str) -> AnyElement { - div() - .text_size(px(11.5)) - .text_color(rgb(colors::error())) - .child(message.to_string()) - .into_any_element() -} - fn sync_object_kind_label(kind: SyncObjectKind) -> &'static str { match kind { SyncObjectKind::Spaces => "Spaces", @@ -486,14 +487,5 @@ fn card_bg() -> u32 { } #[cfg(test)] -mod tests { - use ely_domain::ProfileKind; - - use super::profile_allows_sync_controls; - - #[test] - fn private_profile_hides_sync_controls() { - assert!(!profile_allows_sync_controls(&ProfileKind::Private)); - assert!(profile_allows_sync_controls(&ProfileKind::Standard)); - } -} +#[path = "sync_tests.rs"] +mod tests; diff --git a/crates/ely_app/src/shell/internal_pages/sync_controls.rs b/crates/ely_app/src/shell/internal_pages/sync_controls.rs index 15ee041..51fa145 100644 --- a/crates/ely_app/src/shell/internal_pages/sync_controls.rs +++ b/crates/ely_app/src/shell/internal_pages/sync_controls.rs @@ -4,6 +4,11 @@ use gpui::{ AnyElement, Context, FontWeight, InteractiveElement, IntoElement, ParentElement, SharedString, StatefulInteractiveElement, Styled, div, prelude::FluentBuilder, px, rgb, rgba, }; +use gpui_component::{ + Disableable, Sizable, + button::{Button, ButtonVariants}, + input::{Input, InputState}, +}; use crate::shell::ElyShell; use crate::shell::chrome::animations::{chrome_motion_feedback, toggle_thumb_motion}; @@ -70,10 +75,57 @@ where .into_any_element() } -pub(super) fn render_sign_out_button(shell: &ElyShell, cx: &mut Context) -> AnyElement { - render_secondary_button(shell, "sign-out", "Sign out", false, cx, |shell, cx| { - shell.submit_sign_out(cx); - }) +pub(super) fn render_sign_out_button( + _shell: &ElyShell, + label: &'static str, + disabled: bool, + cx: &mut Context, +) -> AnyElement { + Button::new("sign-out") + .ghost() + .xsmall() + .label(label) + .disabled(disabled) + .on_click(cx.listener(|shell, _, _, cx| { + shell.submit_sign_out(cx); + })) + .into_any_element() +} + +pub(super) fn render_card_heading(label: &'static str) -> AnyElement { + div() + .text_size(px(13.0)) + .font_weight(FontWeight(500.0)) + .text_color(rgb(colors::ink())) + .child(label) + .into_any_element() +} + +pub(super) fn render_field_label(label: &'static str) -> AnyElement { + div() + .text_size(px(10.5)) + .font_weight(FontWeight(500.0)) + .text_color(rgb(colors::ink_4())) + .child(label) + .into_any_element() +} + +pub(super) fn render_input(state: &gpui::Entity) -> AnyElement { + div() + .px(px(10.0)) + .py(px(8.0)) + .rounded(px(8.0)) + .bg(rgba(button_bg())) + .child(Input::new(state).appearance(false).cleanable(false)) + .into_any_element() +} + +pub(super) fn render_inline_error(message: &str) -> AnyElement { + div() + .text_size(px(11.5)) + .text_color(rgb(colors::error())) + .child(message.to_string()) + .into_any_element() } pub(super) fn render_reset_button(shell: &ElyShell, cx: &mut Context) -> AnyElement { diff --git a/crates/ely_app/src/shell/internal_pages/sync_tests.rs b/crates/ely_app/src/shell/internal_pages/sync_tests.rs new file mode 100644 index 0000000..7026428 --- /dev/null +++ b/crates/ely_app/src/shell/internal_pages/sync_tests.rs @@ -0,0 +1,9 @@ +use ely_domain::ProfileKind; + +use super::profile_allows_sync_controls; + +#[test] +fn private_profile_hides_sync_controls() { + assert!(!profile_allows_sync_controls(&ProfileKind::Private)); + assert!(profile_allows_sync_controls(&ProfileKind::Standard)); +} diff --git a/crates/ely_app/src/shell/mod.rs b/crates/ely_app/src/shell/mod.rs index 49f908e..67efe3d 100644 --- a/crates/ely_app/src/shell/mod.rs +++ b/crates/ely_app/src/shell/mod.rs @@ -1,5 +1,6 @@ mod archive_labels; mod auth; +mod auth_operation_gate; mod bookmark_files; mod bookmarks; pub(crate) mod chrome; @@ -25,6 +26,7 @@ mod space_files; mod spaces; mod splits; mod sync_devices; +mod sync_inbox; mod sync_state; mod tab_groups; mod tab_lifecycle; @@ -54,12 +56,14 @@ use gpui_component::input::{InputEvent, InputState}; use gpui_component::slider::{SliderEvent, SliderState, SliderValue}; use crate::shortcuts::ShortcutProfile; +use auth_operation_gate::{AuthenticatedOperationBarrier, AuthenticatedOperationGate}; use bookmarks::PendingBookmarkEdit; use downloads::PendingDownloadFileAction; use history::{PendingHistoryDomainClear, PendingHistoryTimeClear}; use plugins::{PendingPluginInstall, PendingPluginUninstall}; use sync_devices::SyncDeviceUiState; -use sync_state::{PendingMergeUpload, SyncStateUpdate}; +use sync_inbox::{SyncStateMessage, SyncWorkGeneration}; +use sync_state::PendingMergeUpload; use web_surface::WebSurfaceStore; enum ShellState { @@ -108,8 +112,13 @@ pub struct ElyShell { /// shell can refresh the connection state without blocking. Sender /// is cloned for each spawned upload; receiver is drained on every /// `tick_external_web_surfaces`. - sync_inbox_rx: std::sync::mpsc::Receiver, - pub(crate) sync_inbox_tx: std::sync::mpsc::Sender, + sync_inbox_rx: std::sync::mpsc::Receiver, + sync_inbox_tx: std::sync::mpsc::Sender, + sync_generation: SyncWorkGeneration, + sync_profile_id: Option, + authenticated_operation_gate: AuthenticatedOperationGate, + auth_flow_barrier: Option, + sign_out_phases: std::collections::HashMap, sync_upload_scheduled: bool, sync_upload_in_flight: bool, sync_upload_pending: bool, @@ -224,6 +233,12 @@ impl ElyShell { }; let (sync_inbox_tx, sync_inbox_rx) = std::sync::mpsc::channel(); + let sync_profile_id = match &state { + ShellState::Ready(core) => { + core.snapshot().ok().map(|snapshot| snapshot.active_profile_id) + } + ShellState::StartupError(_) => None, + }; let mut shell = Self { state, focus_handle: cx.focus_handle(), @@ -260,6 +275,11 @@ impl ElyShell { web_surfaces: WebSurfaceStore::new(), sync_inbox_rx, sync_inbox_tx, + sync_generation: SyncWorkGeneration::default(), + sync_profile_id, + authenticated_operation_gate: AuthenticatedOperationGate::open(), + auth_flow_barrier: None, + sign_out_phases: std::collections::HashMap::new(), sync_upload_scheduled: false, sync_upload_in_flight: false, sync_upload_pending: false, @@ -320,17 +340,26 @@ impl ElyShell { window: &mut Window, cx: &mut Context, ) { - if let ShellState::Ready(core) = &mut self.state - && core.select_profile(profile_id).is_ok() - { - self.auth_flow_phase = auth::AuthFlowPhase::Idle; - self.sync_devices.reset(); - self.sync_retry_at = None; - self.clear_pending_cloud_sync_upload(); - self.sync_address_input(window, cx); - self.schedule_cloud_sync_upload(cx); - cx.notify(); + let selected = match &mut self.state { + ShellState::Ready(core) => core.select_profile(profile_id).is_ok(), + ShellState::StartupError(_) => false, + }; + if !selected { + return; } + self.invalidate_sync_work(); + self.sync_profile_id = match &self.state { + ShellState::Ready(core) => { + core.snapshot().ok().map(|snapshot| snapshot.active_profile_id) + } + ShellState::StartupError(_) => None, + }; + self.auth_flow_phase = auth::AuthFlowPhase::Idle; + self.sync_address_input(window, cx); + if self.probe_initial_sync_state() { + self.schedule_cloud_sync_upload(cx); + } + cx.notify(); } fn select_next_tab(&mut self, window: &mut Window, cx: &mut Context) { diff --git a/crates/ely_app/src/shell/settings_actions.rs b/crates/ely_app/src/shell/settings_actions.rs index 4493fd5..109558c 100644 --- a/crates/ely_app/src/shell/settings_actions.rs +++ b/crates/ely_app/src/shell/settings_actions.rs @@ -9,6 +9,7 @@ use gpui_component::slider::SliderValue; use crate::services::servo_profile_data::{default_profile_data_root, sync_profile_data_dir}; +use super::sync_inbox::SyncStateSender; use super::sync_state::{ PendingMergeUpload, SyncStateUpdate, sync_failure_update, sync_platform_label, }; @@ -300,12 +301,16 @@ impl ElyShell { return; } }; - let tx = self.sync_inbox_tx.clone(); + let tx = self.sync_state_sender(); let worker_profile_id = active_profile_id.clone(); let thread_name = if merge.is_some() { "ely-sync-merge-upload" } else { "ely-sync-upload" }; + let Some(operation_lease) = self.begin_authenticated_operation() else { + return; + }; self.sync_upload_in_flight = true; if let Err(error) = std::thread::Builder::new().name(thread_name.to_string()).spawn(move || { + let _operation_lease = operation_lease; run_sync_upload(worker_profile_id, profile_dir, device_name, bytes, merge, tx) }) { @@ -351,7 +356,7 @@ fn run_sync_upload( device_name: String, bytes: Vec, merge: Option, - inbox: std::sync::mpsc::Sender, + inbox: SyncStateSender, ) { let mut engine = match SyncEngine::for_profile_dir( &profile_id, diff --git a/crates/ely_app/src/shell/sync_devices.rs b/crates/ely_app/src/shell/sync_devices.rs index 7ba9b07..b0af2c7 100644 --- a/crates/ely_app/src/shell/sync_devices.rs +++ b/crates/ely_app/src/shell/sync_devices.rs @@ -7,6 +7,8 @@ use gpui::Context; use crate::services::servo_profile_data::{default_profile_data_root, sync_profile_data_dir}; +use super::auth_operation_gate::AuthenticatedOperationLease; +use super::sync_inbox::SyncStateSender; use super::sync_state::{device_failure_update, sync_platform_label}; use super::{ElyShell, ShellState, sync_state::SyncStateUpdate}; @@ -131,7 +133,9 @@ impl ElyShell { pub(crate) fn approve_sync_device(&mut self, device_id: String, cx: &mut Context) { let verification_code = self.sync_verification_input.read(cx).value().to_string(); - let Some((profile_id, profile_dir, device_name)) = self.sync_device_context() else { + let Some((profile_id, profile_dir, device_name, operation_lease)) = + self.sync_device_context() + else { self.sync_devices.reset(); return; }; @@ -139,21 +143,29 @@ impl ElyShell { if !self.sync_devices.begin_action(device_id.clone()) { return; } - let tx = self.sync_inbox_tx.clone(); - spawn_device_task("ely-sync-device-approve", profile_id.clone(), tx, move || { - let engine = SyncEngine::for_profile_dir( - &profile_id, - &profile_dir, - device_name, - sync_platform_label(), - )?; - engine.approve_cloud_device(&device_id, &verification_code)?; - load_devices(profile_id, engine) - }); + let tx = self.sync_state_sender(); + spawn_device_task( + "ely-sync-device-approve", + profile_id.clone(), + tx, + operation_lease, + move || { + let engine = SyncEngine::for_profile_dir( + &profile_id, + &profile_dir, + device_name, + sync_platform_label(), + )?; + engine.approve_cloud_device(&device_id, &verification_code)?; + load_devices(profile_id, engine) + }, + ); } pub(crate) fn revoke_sync_device(&mut self, device_id: String, cx: &mut Context) { - let Some((profile_id, profile_dir, device_name)) = self.sync_device_context() else { + let Some((profile_id, profile_dir, device_name, operation_lease)) = + self.sync_device_context() + else { self.sync_devices.reset(); return; }; @@ -165,21 +177,29 @@ impl ElyShell { if !self.sync_devices.begin_action(device_id.clone()) { return; } - let tx = self.sync_inbox_tx.clone(); - spawn_device_task("ely-sync-device-revoke", profile_id.clone(), tx, move || { - let engine = SyncEngine::for_profile_dir( - &profile_id, - &profile_dir, - device_name, - sync_platform_label(), - )?; - engine.revoke_cloud_device(&device_id)?; - load_devices(profile_id, engine) - }); + let tx = self.sync_state_sender(); + spawn_device_task( + "ely-sync-device-revoke", + profile_id.clone(), + tx, + operation_lease, + move || { + let engine = SyncEngine::for_profile_dir( + &profile_id, + &profile_dir, + device_name, + sync_platform_label(), + )?; + engine.revoke_cloud_device(&device_id)?; + load_devices(profile_id, engine) + }, + ); } fn load_sync_devices(&mut self, force: bool, _cx: &mut Context) { - let Some((profile_id, profile_dir, device_name)) = self.sync_device_context() else { + let Some((profile_id, profile_dir, device_name, operation_lease)) = + self.sync_device_context() + else { self.sync_devices.reset(); return; }; @@ -187,19 +207,27 @@ impl ElyShell { if !self.sync_devices.begin_load(force) { return; } - let tx = self.sync_inbox_tx.clone(); - spawn_device_task("ely-sync-device-list", profile_id.clone(), tx, move || { - let engine = SyncEngine::for_profile_dir( - &profile_id, - &profile_dir, - device_name, - sync_platform_label(), - )?; - load_devices(profile_id, engine) - }); + let tx = self.sync_state_sender(); + spawn_device_task( + "ely-sync-device-list", + profile_id.clone(), + tx, + operation_lease, + move || { + let engine = SyncEngine::for_profile_dir( + &profile_id, + &profile_dir, + device_name, + sync_platform_label(), + )?; + load_devices(profile_id, engine) + }, + ); } - fn sync_device_context(&self) -> Option<(ProfileId, PathBuf, String)> { + fn sync_device_context( + &self, + ) -> Option<(ProfileId, PathBuf, String, AuthenticatedOperationLease)> { let ShellState::Ready(core) = &self.state else { return None; }; @@ -208,10 +236,12 @@ impl ElyShell { } let snapshot = core.snapshot().ok()?; let root = default_profile_data_root()?; + let operation_lease = self.begin_authenticated_operation()?; Some(( snapshot.active_profile_id.clone(), sync_profile_data_dir(&root, &snapshot.active_profile_id), format!("ELY · {}", snapshot.active_profile_name), + operation_lease, )) } } @@ -228,7 +258,8 @@ fn load_devices( fn spawn_device_task( name: &str, profile_id: ProfileId, - tx: std::sync::mpsc::Sender, + tx: SyncStateSender, + operation_lease: AuthenticatedOperationLease, task: F, ) where F: FnOnce() -> Result + Send + 'static, @@ -237,6 +268,7 @@ fn spawn_device_task( let worker_tx = tx.clone(); let worker_profile_id = profile_id.clone(); let spawn_result = std::thread::Builder::new().name(thread_name).spawn(move || { + let _operation_lease = operation_lease; let update = task().unwrap_or_else(|error| device_failure_update(worker_profile_id, error)); let _ = worker_tx.send(update); }); diff --git a/crates/ely_app/src/shell/sync_inbox.rs b/crates/ely_app/src/shell/sync_inbox.rs new file mode 100644 index 0000000..8609a9f --- /dev/null +++ b/crates/ely_app/src/shell/sync_inbox.rs @@ -0,0 +1,88 @@ +use std::sync::mpsc::{SendError, Sender}; + +use super::{ElyShell, sync_state::SyncStateUpdate}; + +#[derive(Clone, Copy, Debug, Default, Eq, PartialEq)] +pub(super) struct SyncWorkGeneration(u64); + +impl SyncWorkGeneration { + pub(super) fn advance(&mut self) { + self.0 = self.0.wrapping_add(1); + } +} + +#[derive(Clone, Debug)] +pub(super) struct SyncStateMessage { + pub(super) generation: SyncWorkGeneration, + pub(super) update: SyncStateUpdate, +} + +impl SyncStateMessage { + pub(super) fn belongs_to(&self, generation: SyncWorkGeneration) -> bool { + self.generation == generation + } +} + +#[derive(Clone, Debug)] +pub(super) struct SyncStateSender { + generation: SyncWorkGeneration, + sender: Sender, +} + +impl SyncStateSender { + pub(super) fn send( + &self, + update: SyncStateUpdate, + ) -> Result<(), Box>> { + self.sender.send(SyncStateMessage { generation: self.generation, update }).map_err(Box::new) + } +} + +impl ElyShell { + pub(super) fn sync_state_sender(&self) -> SyncStateSender { + SyncStateSender { generation: self.sync_generation, sender: self.sync_inbox_tx.clone() } + } + + pub(super) fn invalidate_sync_work(&mut self) { + self.release_auth_flow_barrier(); + self.sync_generation.advance(); + 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(); + } +} + +#[cfg(test)] +mod tests { + use super::*; + use ely_domain::ProfileId; + + #[test] + fn sender_stamps_the_captured_generation() -> Result<(), Box> { + let (sender, receiver) = std::sync::mpsc::channel(); + let generation = SyncWorkGeneration(7); + let sender = SyncStateSender { generation, sender }; + + sender.send(SyncStateUpdate::SignedOut { profile_id: ProfileId::new() })?; + + assert_eq!(receiver.recv()?.generation, generation); + Ok(()) + } + + #[test] + fn old_generation_stays_stale_after_an_aba_profile_sequence() + -> Result<(), Box> { + let (sender, receiver) = std::sync::mpsc::channel(); + let mut current = SyncWorkGeneration::default(); + let old_sender = SyncStateSender { generation: current, sender }; + current.advance(); + current.advance(); + + old_sender.send(SyncStateUpdate::SignedOut { profile_id: ProfileId::new() })?; + + assert!(!receiver.recv()?.belongs_to(current)); + Ok(()) + } +} diff --git a/crates/ely_app/src/shell/sync_state.rs b/crates/ely_app/src/shell/sync_state.rs index 7ba369c..ac5b062 100644 --- a/crates/ely_app/src/shell/sync_state.rs +++ b/crates/ely_app/src/shell/sync_state.rs @@ -1,12 +1,16 @@ use std::{path::Path, time::Duration}; use ely_domain::{ProfileId, ProfileKind, SyncConnectionState}; -use ely_sync_client::{AuthenticatedSnapshotHead, BearerTokenStore, DeviceRecord}; +use ely_sync_client::{AuthenticatedSnapshotHead, BearerToken, BearerTokenStore, DeviceRecord}; use gpui::{Context, Timer}; use super::{ElyShell, ShellState, auth}; +mod failure; mod legacy_sync_migration; +mod sign_out; + +pub(super) use failure::{device_failure_update, sync_failure_update}; const CLOUD_SYNC_UPLOAD_DEBOUNCE: Duration = Duration::from_millis(750); const CAS_RETRY_LIMIT: u8 = 3; @@ -35,50 +39,13 @@ pub(crate) enum SyncStateUpdate { CredentialUnavailable { profile_id: ProfileId, message: String, finishes_upload: bool }, DevicesLoaded { profile_id: ProfileId, devices: Vec, current_code: String }, DevicesError { profile_id: ProfileId, message: String }, + SignOutSucceeded { profile_id: ProfileId }, + SignOutFailed { profile_id: ProfileId, message: String }, AuthOtpSent { profile_id: ProfileId, email: String }, - AuthSucceeded { profile_id: ProfileId, email: String }, + AuthVerified { profile_id: ProfileId, email: String, token: BearerToken }, AuthError { profile_id: ProfileId, email: String, message: String }, } -pub(super) fn sync_failure_update( - profile_id: ProfileId, - error: ely_sync_client::SyncClientError, -) -> SyncStateUpdate { - let message = error.to_string(); - match error { - ely_sync_client::SyncClientError::BearerCredentialStorage(_) - | ely_sync_client::SyncClientError::AccountKeyStorage(_) - | ely_sync_client::SyncClientError::DeviceKeyStorage(_) => { - SyncStateUpdate::CredentialUnavailable { profile_id, message, finishes_upload: true } - } - ely_sync_client::SyncClientError::DeviceApprovalStatus { .. } => { - SyncStateUpdate::AwaitingDeviceApproval { profile_id, finishes_upload: true } - } - _ if message.contains("device_not_approved") => { - SyncStateUpdate::AwaitingDeviceApproval { profile_id, finishes_upload: true } - } - _ => SyncStateUpdate::SyncError { profile_id, message }, - } -} - -pub(super) fn device_failure_update( - profile_id: ProfileId, - 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 } - } - SyncStateUpdate::SyncError { profile_id, message } => { - SyncStateUpdate::DevicesError { profile_id, message } - } - update => update, - } -} - /// Stable label for the current OS used by the device registration /// payload. Defined once here so every off-thread call site agrees. pub(crate) const fn sync_platform_label() -> &'static str { @@ -95,6 +62,7 @@ pub(crate) const fn sync_platform_label() -> &'static str { impl ElyShell { pub(crate) fn schedule_cloud_sync_upload(&mut self, cx: &mut Context) { + self.reconcile_active_sync_profile(); if !self.can_schedule_cloud_sync_upload() { return; } @@ -167,10 +135,24 @@ impl ElyShell { else { return false; }; - let ShellState::Ready(core) = &mut self.state else { - return false; + let sign_out_phases = &self.sign_out_phases; + let (bearer_present, sign_out_active) = match &mut self.state { + ShellState::Ready(core) => { + let bearer_present = probe_initial_sync_state_at( + core, + &profile_root, + self.default_profile_id.as_ref(), + ); + let sign_out_active = + sign_out::overlay_active_sign_out_connection(core, sign_out_phases); + (bearer_present, sign_out_active) + } + ShellState::StartupError(_) => return false, }; - probe_initial_sync_state_at(core, &profile_root, self.default_profile_id.as_ref()) + if sign_out_active { + return false; + } + bearer_present } /// Drain any sync upload outcomes the off-thread worker pushed @@ -178,6 +160,7 @@ impl ElyShell { /// state on `BrowserCore`. Returns `true` when at least one /// update was applied so callers can `cx.notify()` accordingly. pub(super) fn drain_sync_updates(&mut self) -> bool { + let profile_needs_initial_sync = self.reconcile_active_sync_profile(); let retry_due = self.sync_retry_at.is_some_and(|deadline| deadline <= std::time::Instant::now()); if retry_due { @@ -185,11 +168,28 @@ impl ElyShell { } let mut latest_connection: Option = None; let mut auth_changed = false; - let mut trigger_initial_sync = false; + let mut trigger_initial_sync = profile_needs_initial_sync; let mut trigger_merged_upload = None; let mut upload_finished = false; let mut devices_changed = false; - while let Ok(update) = self.sync_inbox_rx.try_recv() { + while let Ok(message) = self.sync_inbox_rx.try_recv() { + if let Some((profile_id, connection)) = + self.reconcile_sign_out_update(message.generation, &message.update) + { + if active_profile_id(&self.state).as_ref() == Some(&profile_id) { + self.auth_flow_phase = auth::AuthFlowPhase::Idle; + latest_connection = Some(connection); + } else { + trigger_initial_sync |= self.can_schedule_cloud_sync_upload(); + } + auth_changed = true; + continue; + } + if !message.belongs_to(self.sync_generation) { + auth::retire_stale_auth_update(message.update); + continue; + } + let update = message.update; match update { SyncStateUpdate::SignedOut { profile_id } => { upload_finished = true; @@ -284,24 +284,69 @@ impl ElyShell { devices_changed = true; } } + SyncStateUpdate::SignOutSucceeded { .. } + | SyncStateUpdate::SignOutFailed { .. } => {} SyncStateUpdate::AuthOtpSent { profile_id, email } => { if active_profile_id(&self.state).as_ref() == Some(&profile_id) { self.auth_flow_phase = auth::AuthFlowPhase::AwaitingOtp { profile_id, email }; auth_changed = true; + } else { + self.release_auth_flow_barrier(); } } - SyncStateUpdate::AuthSucceeded { profile_id, email } => { - if active_profile_id(&self.state).as_ref() == Some(&profile_id) { - self.auth_flow_phase = auth::AuthFlowPhase::Idle; - self.sync_devices.reset(); - latest_connection = Some(SyncConnectionState::SignedIn); - trigger_initial_sync = true; - tracing::info!(target: "ely::sync", email = %email, "email OTP sign-in succeeded"); - auth_changed = true; + SyncStateUpdate::AuthVerified { profile_id, email, token } => { + let attempt_matches = active_profile_id(&self.state).as_ref() + == Some(&profile_id) + && matches!( + &self.auth_flow_phase, + auth::AuthFlowPhase::Verifying { + profile_id: owner, + email: expected_email, + } if owner == &profile_id && expected_email == &email + ); + self.release_auth_flow_barrier(); + if !attempt_matches { + auth::retire_stale_auth_update(SyncStateUpdate::AuthVerified { + profile_id, + email, + token, + }); + continue; + } + match auth::save_verified_bearer( + &profile_id, + &token, + self.default_profile_id.as_ref(), + ) { + Ok(()) => { + self.sign_out_phases.remove(&profile_id); + self.auth_flow_phase = auth::AuthFlowPhase::Idle; + latest_connection = Some(SyncConnectionState::SignedIn); + trigger_initial_sync = true; + auth_changed = true; + } + Err(error) => { + tracing::warn!(target: "ely::sync", error = %error, "bearer credential save failed"); + auth::retire_stale_auth_update(SyncStateUpdate::AuthVerified { + profile_id: profile_id.clone(), + email: email.clone(), + token, + }); + let message = "System credential access failed.".to_string(); + self.auth_flow_phase = auth::AuthFlowPhase::Error { + profile_id, + email, + message: message.clone(), + }; + latest_connection = + Some(SyncConnectionState::CredentialUnavailable { message }); + auth_changed = true; + } } } SyncStateUpdate::AuthError { profile_id, email, message } => { + self.release_auth_flow_barrier(); if active_profile_id(&self.state).as_ref() == Some(&profile_id) { self.auth_flow_phase = auth::AuthFlowPhase::Error { profile_id, email, message }; @@ -340,6 +385,17 @@ impl ElyShell { || retry_due || connection_changed } + + fn reconcile_active_sync_profile(&mut self) -> bool { + let current_profile_id = active_profile_id(&self.state); + if current_profile_id == self.sync_profile_id { + return false; + } + self.invalidate_sync_work(); + self.sync_profile_id = current_profile_id; + self.auth_flow_phase = auth::AuthFlowPhase::Idle; + self.probe_initial_sync_state() + } } fn active_profile_id(state: &ShellState) -> Option { @@ -405,93 +461,5 @@ fn probe_initial_sync_state_at( } #[cfg(test)] -mod tests { - use ely_browser_core::{BrowserCore, InitialBrowserConfig}; - use ely_domain::SyncConnectionState; - - use super::{SyncStateUpdate, probe_initial_sync_state_at, sync_failure_update}; - use crate::services::servo_profile_data::sync_profile_data_dir; - - #[test] - fn private_startup_clears_a_persisted_bearer() -> Result<(), Box> { - let directory = tempfile::tempdir()?; - let mut core = BrowserCore::new(InitialBrowserConfig::private_window()?)?; - let profile_id = core.snapshot()?.active_profile_id; - let profile_dir = sync_profile_data_dir(directory.path(), &profile_id); - let bearer_path = profile_dir.join("sync/bearer.token"); - std::fs::create_dir_all(bearer_path.parent().ok_or("missing bearer parent")?)?; - std::fs::write(&bearer_path, "a".repeat(64))?; - std::fs::write(bearer_path.with_extension("tmp"), "b".repeat(64))?; - core.set_sync_connection_state(SyncConnectionState::SignedIn); - - assert!(!probe_initial_sync_state_at(&mut core, directory.path(), None)); - assert!(!bearer_path.exists()); - assert!(!bearer_path.with_extension("tmp").exists()); - assert_eq!(core.snapshot()?.sync_status.connection(), &SyncConnectionState::SignedOut); - - Ok(()) - } - - #[test] - fn standard_startup_preserves_a_persisted_bearer() -> Result<(), Box> { - let directory = tempfile::tempdir()?; - let mut core = BrowserCore::new(InitialBrowserConfig::ely_defaults()?)?; - let profile_id = core.snapshot()?.active_profile_id; - let profile_dir = sync_profile_data_dir(directory.path(), &profile_id); - let bearer_path = profile_dir.join("sync/bearer.token"); - std::fs::create_dir_all(bearer_path.parent().ok_or("missing bearer parent")?)?; - std::fs::write(&bearer_path, "a".repeat(64))?; - - assert!(probe_initial_sync_state_at(&mut core, directory.path(), Some(&profile_id),)); - assert!(!bearer_path.exists()); - assert_eq!(core.snapshot()?.sync_status.connection(), &SyncConnectionState::SignedIn); - - let store = ely_sync_client::BearerTokenStore::new(&profile_id, &profile_dir); - assert!(store.load()?.is_some()); - store.clear()?; - - Ok(()) - } - - #[cfg(unix)] - #[test] - fn credential_probe_failure_enters_unavailable_state() -> Result<(), Box> - { - use std::os::unix::fs::symlink; - - let directory = tempfile::tempdir()?; - let mut core = BrowserCore::new(InitialBrowserConfig::ely_defaults()?)?; - let profile_id = core.snapshot()?.active_profile_id; - let profile_dir = sync_profile_data_dir(directory.path(), &profile_id); - let lock = profile_dir.join("sync/bearer.lock"); - let target = directory.path().join("untrusted.lock"); - std::fs::create_dir_all(lock.parent().ok_or("missing lock parent")?)?; - std::fs::write(&target, "target")?; - symlink(&target, &lock)?; - - assert!(!probe_initial_sync_state_at(&mut core, directory.path(), Some(&profile_id),)); - assert!(matches!( - core.snapshot()?.sync_status.connection(), - SyncConnectionState::CredentialUnavailable { .. } - )); - Ok(()) - } - - #[test] - fn credential_storage_failures_use_the_unavailable_update() { - let profile_id = ely_domain::ProfileId::new(); - let update = sync_failure_update( - profile_id.clone(), - ely_sync_client::SyncClientError::BearerCredentialStorage("locked".to_string()), - ); - - assert!(matches!( - update, - SyncStateUpdate::CredentialUnavailable { - profile_id: owner, - finishes_upload: true, - .. - } if owner == profile_id - )); - } -} +#[path = "sync_state/sync_state_tests.rs"] +mod tests; diff --git a/crates/ely_app/src/shell/sync_state/failure.rs b/crates/ely_app/src/shell/sync_state/failure.rs new file mode 100644 index 0000000..b7ddfab --- /dev/null +++ b/crates/ely_app/src/shell/sync_state/failure.rs @@ -0,0 +1,42 @@ +use ely_domain::ProfileId; + +use super::SyncStateUpdate; + +pub(in crate::shell) fn sync_failure_update( + profile_id: ProfileId, + error: ely_sync_client::SyncClientError, +) -> SyncStateUpdate { + let message = error.to_string(); + match error { + ely_sync_client::SyncClientError::BearerCredentialStorage(_) + | ely_sync_client::SyncClientError::AccountKeyStorage(_) + | ely_sync_client::SyncClientError::DeviceKeyStorage(_) => { + SyncStateUpdate::CredentialUnavailable { profile_id, message, finishes_upload: true } + } + ely_sync_client::SyncClientError::DeviceApprovalStatus { .. } => { + SyncStateUpdate::AwaitingDeviceApproval { profile_id, finishes_upload: true } + } + _ if message.contains("device_not_approved") => { + SyncStateUpdate::AwaitingDeviceApproval { profile_id, finishes_upload: true } + } + _ => SyncStateUpdate::SyncError { profile_id, message }, + } +} + +pub(in crate::shell) fn device_failure_update( + profile_id: ProfileId, + 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 } + } + SyncStateUpdate::SyncError { profile_id, message } => { + SyncStateUpdate::DevicesError { profile_id, message } + } + update => update, + } +} diff --git a/crates/ely_app/src/shell/sync_state/sign_out.rs b/crates/ely_app/src/shell/sync_state/sign_out.rs new file mode 100644 index 0000000..7fe969f --- /dev/null +++ b/crates/ely_app/src/shell/sync_state/sign_out.rs @@ -0,0 +1,167 @@ +use ely_domain::{ProfileId, SyncConnectionState}; +use std::collections::HashMap; + +use super::super::{ElyShell, auth::SignOutPhase}; +use crate::shell::{sync_inbox::SyncWorkGeneration, sync_state::SyncStateUpdate}; + +impl ElyShell { + pub(super) fn reconcile_sign_out_update( + &mut self, + generation: SyncWorkGeneration, + update: &SyncStateUpdate, + ) -> Option<(ProfileId, SyncConnectionState)> { + reconcile_phase(&mut self.sign_out_phases, generation, update) + } +} + +pub(super) fn overlay_active_sign_out_connection( + core: &mut ely_browser_core::BrowserCore, + phases: &HashMap, +) -> bool { + let Some(profile_id) = core.snapshot().ok().map(|snapshot| snapshot.active_profile_id) else { + return false; + }; + let Some(phase) = phases.get(&profile_id) else { + return false; + }; + core.set_sync_connection_state(connection_for_phase(phase)); + true +} + +fn reconcile_phase( + phases: &mut HashMap, + generation: SyncWorkGeneration, + update: &SyncStateUpdate, +) -> Option<(ProfileId, SyncConnectionState)> { + let (profile_id, failure) = match update { + SyncStateUpdate::SignOutSucceeded { profile_id } => (profile_id, None), + SyncStateUpdate::SignOutFailed { profile_id, message } => { + (profile_id, Some(message.clone())) + } + _ => return None, + }; + let owns_phase = phases.get(profile_id).is_some_and(|phase| phase.generation() == generation); + if !owns_phase { + return None; + } + + let connection = match failure { + Some(message) => { + phases.insert( + profile_id.clone(), + SignOutPhase::Error { generation, message: message.clone() }, + ); + SyncConnectionState::SignOutError { message } + } + None => { + phases.remove(profile_id); + SyncConnectionState::SignedOut + } + }; + Some((profile_id.clone(), connection)) +} + +fn connection_for_phase(phase: &SignOutPhase) -> SyncConnectionState { + match phase { + SignOutPhase::SigningOut { .. } => SyncConnectionState::SigningOut, + SignOutPhase::Error { message, .. } => { + SyncConnectionState::SignOutError { message: message.clone() } + } + } +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn completions_only_update_their_profile_and_generation() { + let profile_a = ProfileId::new(); + let profile_b = ProfileId::new(); + let generation_a = SyncWorkGeneration::default(); + let mut generation_b = generation_a; + generation_b.advance(); + let mut phases = HashMap::from([ + (profile_a.clone(), SignOutPhase::SigningOut { generation: generation_a }), + (profile_b.clone(), SignOutPhase::SigningOut { generation: generation_b }), + ]); + + let success = SyncStateUpdate::SignOutSucceeded { profile_id: profile_a.clone() }; + assert!(reconcile_phase(&mut phases, generation_a, &success).is_some()); + assert!(!phases.contains_key(&profile_a)); + assert_eq!(phases.get(&profile_b).map(SignOutPhase::generation), Some(generation_b)); + + let mut newer_generation_a = generation_b; + newer_generation_a.advance(); + phases + .insert(profile_a.clone(), SignOutPhase::SigningOut { generation: newer_generation_a }); + let stale_failure = SyncStateUpdate::SignOutFailed { + profile_id: profile_a.clone(), + message: "stale".to_string(), + }; + assert!(reconcile_phase(&mut phases, generation_a, &stale_failure).is_none()); + assert_eq!(phases.get(&profile_a).map(SignOutPhase::generation), Some(newer_generation_a)); + } + + #[test] + fn failure_phase_preserves_retry_state() -> Result<(), &'static str> { + let profile_id = ProfileId::new(); + let generation = SyncWorkGeneration::default(); + let mut phases = + HashMap::from([(profile_id.clone(), SignOutPhase::SigningOut { generation })]); + let failure = SyncStateUpdate::SignOutFailed { + profile_id: profile_id.clone(), + message: "Session revoke failed.".to_string(), + }; + + let (_, connection) = + reconcile_phase(&mut phases, generation, &failure).ok_or("missing owned result")?; + + assert!(matches!(connection, SyncConnectionState::SignOutError { .. })); + assert!(matches!( + phases.get(&profile_id), + Some(SignOutPhase::Error { message, .. }) if message == "Session revoke failed." + )); + Ok(()) + } + + #[test] + fn profile_switches_reapply_the_owned_sign_out_connection() + -> Result<(), Box> { + let mut core = ely_browser_core::BrowserCore::new( + ely_browser_core::InitialBrowserConfig::ely_defaults()?, + )?; + let profile_a = core.snapshot()?.active_profile_id; + let profile_b = + core.create_profile("Profile B", 0x26251e, ely_domain::ProfileKind::Standard)?; + let generation = SyncWorkGeneration::default(); + let mut phases = + HashMap::from([(profile_a.clone(), SignOutPhase::SigningOut { generation })]); + + core.select_profile(&profile_a)?; + core.set_sync_connection_state(SyncConnectionState::SignedIn); + assert!(overlay_active_sign_out_connection(&mut core, &phases)); + assert_eq!(core.snapshot()?.sync_status.connection(), &SyncConnectionState::SigningOut); + assert!(!core.cloud_sync_upload_enabled()); + + core.select_profile(&profile_b)?; + core.set_sync_connection_state(SyncConnectionState::SignedIn); + assert!(!overlay_active_sign_out_connection(&mut core, &phases)); + assert_eq!(core.snapshot()?.sync_status.connection(), &SyncConnectionState::SignedIn); + assert!(core.cloud_sync_upload_enabled()); + + phases.insert( + profile_a.clone(), + SignOutPhase::Error { generation, message: "Session revoke failed.".to_string() }, + ); + core.select_profile(&profile_a)?; + core.set_sync_connection_state(SyncConnectionState::SignedIn); + assert!(overlay_active_sign_out_connection(&mut core, &phases)); + assert!(matches!( + core.snapshot()?.sync_status.connection(), + SyncConnectionState::SignOutError { .. } + )); + assert!(!core.cloud_sync_upload_enabled()); + Ok(()) + } +} 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 new file mode 100644 index 0000000..132355c --- /dev/null +++ b/crates/ely_app/src/shell/sync_state/sync_state_tests.rs @@ -0,0 +1,87 @@ +use ely_browser_core::{BrowserCore, InitialBrowserConfig}; +use ely_domain::SyncConnectionState; + +use super::{SyncStateUpdate, probe_initial_sync_state_at, sync_failure_update}; +use crate::services::servo_profile_data::sync_profile_data_dir; + +#[test] +fn private_startup_clears_a_persisted_bearer() -> Result<(), Box> { + let directory = tempfile::tempdir()?; + let mut core = BrowserCore::new(InitialBrowserConfig::private_window()?)?; + let profile_id = core.snapshot()?.active_profile_id; + let profile_dir = sync_profile_data_dir(directory.path(), &profile_id); + let bearer_path = profile_dir.join("sync/bearer.token"); + std::fs::create_dir_all(bearer_path.parent().ok_or("missing bearer parent")?)?; + std::fs::write(&bearer_path, "a".repeat(64))?; + std::fs::write(bearer_path.with_extension("tmp"), "b".repeat(64))?; + core.set_sync_connection_state(SyncConnectionState::SignedIn); + + assert!(!probe_initial_sync_state_at(&mut core, directory.path(), None)); + assert!(!bearer_path.exists()); + assert!(!bearer_path.with_extension("tmp").exists()); + assert_eq!(core.snapshot()?.sync_status.connection(), &SyncConnectionState::SignedOut); + + Ok(()) +} + +#[test] +fn standard_startup_preserves_a_persisted_bearer() -> Result<(), Box> { + let directory = tempfile::tempdir()?; + let mut core = BrowserCore::new(InitialBrowserConfig::ely_defaults()?)?; + let profile_id = core.snapshot()?.active_profile_id; + let profile_dir = sync_profile_data_dir(directory.path(), &profile_id); + let bearer_path = profile_dir.join("sync/bearer.token"); + std::fs::create_dir_all(bearer_path.parent().ok_or("missing bearer parent")?)?; + std::fs::write(&bearer_path, "a".repeat(64))?; + + assert!(probe_initial_sync_state_at(&mut core, directory.path(), Some(&profile_id),)); + assert!(!bearer_path.exists()); + assert_eq!(core.snapshot()?.sync_status.connection(), &SyncConnectionState::SignedIn); + + let store = ely_sync_client::BearerTokenStore::new(&profile_id, &profile_dir); + assert!(store.load()?.is_some()); + store.clear()?; + + Ok(()) +} + +#[cfg(unix)] +#[test] +fn credential_probe_failure_enters_unavailable_state() -> Result<(), Box> { + use std::os::unix::fs::symlink; + + let directory = tempfile::tempdir()?; + let mut core = BrowserCore::new(InitialBrowserConfig::ely_defaults()?)?; + let profile_id = core.snapshot()?.active_profile_id; + let profile_dir = sync_profile_data_dir(directory.path(), &profile_id); + let lock = profile_dir.join("sync/bearer.lock"); + let target = directory.path().join("untrusted.lock"); + std::fs::create_dir_all(lock.parent().ok_or("missing lock parent")?)?; + std::fs::write(&target, "target")?; + symlink(&target, &lock)?; + + assert!(!probe_initial_sync_state_at(&mut core, directory.path(), Some(&profile_id),)); + assert!(matches!( + core.snapshot()?.sync_status.connection(), + SyncConnectionState::CredentialUnavailable { .. } + )); + Ok(()) +} + +#[test] +fn credential_storage_failures_use_the_unavailable_update() { + let profile_id = ely_domain::ProfileId::new(); + let update = sync_failure_update( + profile_id.clone(), + ely_sync_client::SyncClientError::BearerCredentialStorage("locked".to_string()), + ); + + assert!(matches!( + update, + SyncStateUpdate::CredentialUnavailable { + profile_id: owner, + finishes_upload: true, + .. + } if owner == profile_id + )); +} diff --git a/crates/ely_browser_core/tests/sync.rs b/crates/ely_browser_core/tests/sync.rs index 616033a..8f0b724 100644 --- a/crates/ely_browser_core/tests/sync.rs +++ b/crates/ely_browser_core/tests/sync.rs @@ -39,13 +39,21 @@ fn default_sync_status_reflects_local_browser_state() -> Result<(), Box Result<(), Box> { +fn auth_transition_states_disable_cloud_uploads() -> Result<(), Box> { let mut core = BrowserCore::new(InitialBrowserConfig::ely_defaults()?)?; - core.set_sync_connection_state(SyncConnectionState::CredentialUnavailable { - message: "credential unavailable".to_string(), - }); + let blocked_states = [ + SyncConnectionState::CredentialUnavailable { + message: "credential unavailable".to_string(), + }, + SyncConnectionState::SigningOut, + SyncConnectionState::SignOutError { message: "retry sign out".to_string() }, + ]; + + for state in blocked_states { + core.set_sync_connection_state(state); + assert!(!core.cloud_sync_upload_enabled()); + } - assert!(!core.cloud_sync_upload_enabled()); Ok(()) } diff --git a/crates/ely_domain/src/sync.rs b/crates/ely_domain/src/sync.rs index d2f73e7..3908569 100644 --- a/crates/ely_domain/src/sync.rs +++ b/crates/ely_domain/src/sync.rs @@ -14,6 +14,8 @@ pub enum SyncConnectionState { SignedOut, CredentialUnavailable { message: String }, + SigningOut, + SignOutError { message: String }, SignedIn, AwaitingDeviceApproval, SyncReady { last_synced_at_secs: u64 }, @@ -106,6 +108,7 @@ impl SyncStatus { pub fn new(connection: SyncConnectionState, objects: Vec) -> Self { let failed_objects = match &connection { SyncConnectionState::CredentialUnavailable { .. } + | SyncConnectionState::SignOutError { .. } | SyncConnectionState::SyncError { .. } => 1, _ => 0, }; diff --git a/crates/ely_sync_client/src/client.rs b/crates/ely_sync_client/src/client.rs index 6e28e1e..6b8bf55 100644 --- a/crates/ely_sync_client/src/client.rs +++ b/crates/ely_sync_client/src/client.rs @@ -18,6 +18,8 @@ use crate::{ vault_bootstrap::SyncVaultBootstrapRequest, }; +mod session; + const REQUEST_TIMEOUT: Duration = Duration::from_secs(30); const USER_AGENT: &str = concat!("ELY Browser/", env!("CARGO_PKG_VERSION")); diff --git a/crates/ely_sync_client/src/client/session.rs b/crates/ely_sync_client/src/client/session.rs new file mode 100644 index 0000000..054eee5 --- /dev/null +++ b/crates/ely_sync_client/src/client/session.rs @@ -0,0 +1,52 @@ +use serde::Deserialize; + +use super::{SyncApiClient, read_json_from_response}; +use crate::error::SyncClientError; + +#[derive(Deserialize)] +#[serde(deny_unknown_fields)] +struct SessionLogoutDocument { + version: u32, + signed_out: bool, +} + +#[derive(Deserialize)] +#[serde(deny_unknown_fields)] +struct AuthErrorDocument { + error: String, +} + +impl SyncApiClient { + pub fn sign_out(&self) -> Result<(), SyncClientError> { + let endpoint = self.endpoint("/api/session/logout"); + let response = self + .agent + .post(&endpoint) + .set("Authorization", &format!("Bearer {}", self.bearer.as_str())) + .call(); + match response { + Ok(response) => { + let document = + read_json_from_response::(&endpoint, response)?; + if document.version != 1 || !document.signed_out { + return Err(SyncClientError::SessionLogoutInvalid); + } + Ok(()) + } + Err(ureq::Error::Status(status, response)) => { + let body = response.into_string().unwrap_or_default(); + if status == 401 && logout_is_already_complete(&body) { + return Ok(()); + } + Err(SyncClientError::HttpStatus { endpoint, status, body }) + } + Err(source) => Err(SyncClientError::Http { endpoint, source: Box::new(source) }), + } + } +} + +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") + }) +} diff --git a/crates/ely_sync_client/src/client_tests.rs b/crates/ely_sync_client/src/client_tests.rs index b634b7d..d061266 100644 --- a/crates/ely_sync_client/src/client_tests.rs +++ b/crates/ely_sync_client/src/client_tests.rs @@ -12,6 +12,83 @@ use crate::{ type TestServer = JoinHandle>; +#[test] +fn sign_out_posts_the_bearer_and_validates_success() -> Result<(), Box> { + let (base_url, server) = spawn_logout_server("200 OK", r#"{"version":1,"signed_out":true}"#)?; + let client = SyncApiClient::new( + ApiClientConfig::custom(base_url, "auto"), + BearerToken::new("a".repeat(64))?, + )?; + + client.sign_out()?; + + join_server(server) +} + +#[test] +fn sign_out_accepts_already_ended_sessions() -> Result<(), Box> { + for code in ["session_not_found", "session_expired"] { + let body = format!(r#"{{"error":"{code}"}}"#); + let (base_url, server) = spawn_logout_server("401 Unauthorized", &body)?; + let client = SyncApiClient::new( + ApiClientConfig::custom(base_url, "auto"), + BearerToken::new("a".repeat(64))?, + )?; + + client.sign_out()?; + join_server(server)?; + } + Ok(()) +} + +#[test] +fn sign_out_preserves_retryable_failures() -> Result<(), Box> { + let failures = [ + ("401 Unauthorized", r#"{"error":"authorization_invalid"}"#, 401), + ("401 Unauthorized", r#"{"error":"session_not_found","extra":true}"#, 401), + ("403 Forbidden", r#"{"error":"session_not_found"}"#, 403), + ("500 Internal Server Error", r#"{"error":"session_logout_failed"}"#, 500), + ]; + for (status_line, body, expected_status) in failures { + let (base_url, server) = spawn_logout_server(status_line, body)?; + let client = SyncApiClient::new( + ApiClientConfig::custom(base_url, "auto"), + BearerToken::new("a".repeat(64))?, + )?; + + let error = match client.sign_out() { + Ok(()) => return Err("logout failure was treated as success".into()), + Err(error) => error, + }; + assert!(matches!( + error, + crate::SyncClientError::HttpStatus { status, .. } if status == expected_status + )); + join_server(server)?; + } + + for body in [r#"{"version":1,"signed_out":false}"#, r#"{"version":2,"signed_out":true}"#] { + let (base_url, server) = spawn_logout_server("200 OK", body)?; + let client = SyncApiClient::new( + ApiClientConfig::custom(base_url, "auto"), + BearerToken::new("a".repeat(64))?, + )?; + assert!(matches!(client.sign_out(), Err(crate::SyncClientError::SessionLogoutInvalid))); + join_server(server)?; + } + + for body in ["{", r#"{"version":1,"signed_out":true,"extra":true}"#, r#"{"version":1}"#] { + let (base_url, server) = spawn_logout_server("200 OK", body)?; + let client = SyncApiClient::new( + ApiClientConfig::custom(base_url, "auto"), + BearerToken::new("a".repeat(64))?, + )?; + assert!(matches!(client.sign_out(), Err(crate::SyncClientError::Json { .. }))); + join_server(server)?; + } + Ok(()) +} + #[test] fn upload_parses_structured_snapshot_head_conflict() -> Result<(), Box> { let (base_url, server) = spawn_conflict_server()?; @@ -99,6 +176,40 @@ fn spawn_conflict_server() -> Result<(String, TestServer), Box> { Ok((format!("http://{address}"), server)) } +fn spawn_logout_server( + status_line: &'static str, + body: &str, +) -> Result<(String, TestServer), Box> { + let listener = TcpListener::bind("127.0.0.1:0")?; + let address = listener.local_addr()?; + let body = body.to_string(); + let server = thread::spawn(move || -> std::io::Result<()> { + let (mut stream, _) = listener.accept()?; + let mut request = Vec::new(); + let mut chunk = [0_u8; 1024]; + while !request.windows(4).any(|window| window == b"\r\n\r\n") { + let read = stream.read(&mut chunk)?; + if read == 0 || request.len() + read > 8192 { + return Err(std::io::Error::other("logout request headers are incomplete")); + } + 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") + || !request.contains(&format!("Authorization: Bearer {}\r\n", "a".repeat(64))) + { + return Err(std::io::Error::other("logout request contract mismatch")); + } + let response = format!( + "HTTP/1.1 {status_line}\r\nContent-Type: application/json\r\nContent-Length: {}\r\nConnection: close\r\n\r\n{body}", + body.len() + ); + stream.write_all(response.as_bytes())?; + stream.flush() + }); + Ok((format!("http://{address}"), server)) +} + fn join_server(server: TestServer) -> Result<(), Box> { match server.join() { Ok(result) => result.map_err(Into::into), diff --git a/crates/ely_sync_client/src/error.rs b/crates/ely_sync_client/src/error.rs index 74c6cf1..2ed41d2 100644 --- a/crates/ely_sync_client/src/error.rs +++ b/crates/ely_sync_client/src/error.rs @@ -11,6 +11,9 @@ pub enum SyncClientError { #[error("Bearer credential storage is unavailable: {0}")] BearerCredentialStorage(String), + #[error("Session logout response is invalid")] + SessionLogoutInvalid, + #[error("HTTP request failed for {endpoint}: {source}")] Http { endpoint: String,