ZacharyZhang-NY/ELY-Browser
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

fix(auth): revoke desktop sessions safely

Browse Source
This commit is contained in:
ZacharyZhang-NY
2026-07-10 08:57:00 -04:00
parent 94afa23a69
commit 2f346abaea
21 changed files with 1375 additions and 455 deletions
Download Patch File Download Diff File Expand all files Collapse all files
PRD.md
+1
View File
@@ -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/*` 路由。
crates/ely_app/src/shell/auth.rs
+198 -196
View File
@@ -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<Self>) {
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<Self>) {
/// 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<Self>) {
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 {
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<ProfileId> {
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<SyncStateUpdate>) {
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<BearerToken>,
) -> 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<SyncStateUpda
});
}
}
})
.map(|_| ())
.unwrap_or_else(|error| {
tracing::warn!(target: "ely::sync", error = %error, "spawn ely-sync-auth-send failed");
});
if let Err(error) = spawn_result {
tracing::warn!(target: "ely::sync", error = %error, "spawn ely-sync-auth-send failed");
let _ = failure_tx.send(SyncStateUpdate::AuthError {
profile_id: failure_profile_id,
email: failure_email,
message: "Unable to start sign-in. Retry.".to_string(),
});
}
}
fn spawn_verify_otp(
profile_id: ProfileId,
email: String,
otp: String,
profile_dir: std::path::PathBuf,
tx: Sender<SyncStateUpdate>,
) {
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;
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;
};
if let Err(error) = engine.install_bearer(token.as_str()) {
let _ = tx.send(SyncStateUpdate::AuthError {
profile_id,
email,
message: error.to_string(),
});
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<dyn std::error::Error>> {
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<dyn std::error::Error>> {
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<dyn std::error::Error>> {
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<BearerToken> {
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;
crates/ely_app/src/shell/auth_operation_gate.rs
+149
View File
@@ -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<GateState>, Condvar)>,
}
#[derive(Debug)]
pub(super) struct AuthenticatedOperationLease {
inner: Arc<(Mutex<GateState>, Condvar)>,
}
#[derive(Debug)]
pub(super) struct AuthenticatedOperationBarrier {
inner: Option<Arc<(Mutex<GateState>, 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<AuthenticatedOperationLease> {
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<AuthenticatedOperationLease> {
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<dyn std::error::Error>> {
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());
}
}
crates/ely_app/src/shell/auth_tests.rs
+118
View File
@@ -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<dyn std::error::Error>> {
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<dyn std::error::Error>> {
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<dyn std::error::Error>> {
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<dyn std::error::Error>> {
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(())
}
crates/ely_app/src/shell/internal_pages/sync.rs
+42 -50
View File
@@ -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<gpui_component::input::InputState>) -> 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;
crates/ely_app/src/shell/internal_pages/sync_controls.rs
+55 -3
View File
@@ -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<ElyShell>) -> AnyElement {
render_secondary_button(shell, "sign-out", "Sign out", false, cx, |shell, cx| {
pub(super) fn render_sign_out_button(
_shell: &ElyShell,
label: &'static str,
disabled: bool,
cx: &mut Context<ElyShell>,
) -> 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<InputState>) -> 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<ElyShell>) -> AnyElement {
crates/ely_app/src/shell/internal_pages/sync_tests.rs
+9
View File
@@ -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));
}
crates/ely_app/src/shell/mod.rs
+42 -13
View File
@@ -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<SyncStateUpdate>,
pub(crate) sync_inbox_tx: std::sync::mpsc::Sender<SyncStateUpdate>,
sync_inbox_rx: std::sync::mpsc::Receiver<SyncStateMessage>,
sync_inbox_tx: std::sync::mpsc::Sender<SyncStateMessage>,
sync_generation: SyncWorkGeneration,
sync_profile_id: Option<ProfileId>,
authenticated_operation_gate: AuthenticatedOperationGate,
auth_flow_barrier: Option<AuthenticatedOperationBarrier>,
sign_out_phases: std::collections::HashMap<ProfileId, auth::SignOutPhase>,
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<Self>,
) {
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<Self>) {
crates/ely_app/src/shell/settings_actions.rs
+7 -2
View File
@@ -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<u8>,
merge: Option<PendingMergeUpload>,
inbox: std::sync::mpsc::Sender<SyncStateUpdate>,
inbox: SyncStateSender,
) {
let mut engine = match SyncEngine::for_profile_dir(
&profile_id,
crates/ely_app/src/shell/sync_devices.rs
+46 -14
View File
@@ -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<Self>) {
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,8 +143,13 @@ 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 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,
@@ -149,11 +158,14 @@ impl ElyShell {
)?;
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<Self>) {
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,8 +177,13 @@ 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 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,
@@ -175,11 +192,14 @@ impl ElyShell {
)?;
engine.revoke_cloud_device(&device_id)?;
load_devices(profile_id, engine)
});
},
);
}
fn load_sync_devices(&mut self, force: bool, _cx: &mut Context<Self>) {
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,8 +207,13 @@ 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 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,
@@ -196,10 +221,13 @@ impl ElyShell {
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<F>(
name: &str,
profile_id: ProfileId,
tx: std::sync::mpsc::Sender<SyncStateUpdate>,
tx: SyncStateSender,
operation_lease: AuthenticatedOperationLease,
task: F,
) where
F: FnOnce() -> Result<SyncStateUpdate, ely_sync_client::SyncClientError> + Send + 'static,
@@ -237,6 +268,7 @@ fn spawn_device_task<F>(
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);
});
crates/ely_app/src/shell/sync_inbox.rs
+88
View File
@@ -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<SyncStateMessage>,
}
impl SyncStateSender {
pub(super) fn send(
&self,
update: SyncStateUpdate,
) -> Result<(), Box<SendError<SyncStateMessage>>> {
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<dyn std::error::Error>> {
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<dyn std::error::Error>> {
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(())
}
}
crates/ely_app/src/shell/sync_state.rs
+108 -140
View File
@@ -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<DeviceRecord>, 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>) {
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<SyncConnectionState> = 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) {
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;
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;
}
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<ProfileId> {
@@ -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<dyn std::error::Error>> {
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<dyn std::error::Error>> {
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<dyn std::error::Error>>
{
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;
crates/ely_app/src/shell/sync_state/failure.rs
+42
View File
@@ -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,
}
}
crates/ely_app/src/shell/sync_state/sign_out.rs
+167
View File
@@ -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<ProfileId, SignOutPhase>,
) -> 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<ProfileId, SignOutPhase>,
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<dyn std::error::Error>> {
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(())
}
}
crates/ely_app/src/shell/sync_state/sync_state_tests.rs
+87
View File
@@ -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<dyn std::error::Error>> {
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<dyn std::error::Error>> {
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<dyn std::error::Error>> {
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
));
}
crates/ely_browser_core/tests/sync.rs
+11 -3
View File
@@ -39,13 +39,21 @@ fn default_sync_status_reflects_local_browser_state() -> Result<(), Box<dyn Erro
}
#[test]
fn unavailable_credentials_disable_cloud_uploads() -> Result<(), Box<dyn Error>> {
fn auth_transition_states_disable_cloud_uploads() -> Result<(), Box<dyn Error>> {
let mut core = BrowserCore::new(InitialBrowserConfig::ely_defaults()?)?;
core.set_sync_connection_state(SyncConnectionState::CredentialUnavailable {
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());
}
Ok(())
}
crates/ely_domain/src/sync.rs
+3
View File
@@ -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<SyncObjectStatus>) -> Self {
let failed_objects = match &connection {
SyncConnectionState::CredentialUnavailable { .. }
| SyncConnectionState::SignOutError { .. }
| SyncConnectionState::SyncError { .. } => 1,
_ => 0,
};
crates/ely_sync_client/src/client.rs
+2
View File
@@ -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"));
crates/ely_sync_client/src/client/session.rs
+52
View File
@@ -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::<SessionLogoutDocument>(&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::<AuthErrorDocument>(body).is_ok_and(|document| {
matches!(document.error.as_str(), "session_not_found" | "session_expired")
})
}
crates/ely_sync_client/src/client_tests.rs
+111
View File
@@ -12,6 +12,83 @@ use crate::{
type TestServer = JoinHandle<std::io::Result<()>>;
#[test]
fn sign_out_posts_the_bearer_and_validates_success() -> Result<(), Box<dyn Error>> {
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<dyn Error>> {
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<dyn Error>> {
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<dyn Error>> {
let (base_url, server) = spawn_conflict_server()?;
@@ -99,6 +176,40 @@ fn spawn_conflict_server() -> Result<(String, TestServer), Box<dyn Error>> {
Ok((format!("http://{address}"), server))
}
fn spawn_logout_server(
status_line: &'static str,
body: &str,
) -> Result<(String, TestServer), Box<dyn Error>> {
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<dyn Error>> {
match server.join() {
Ok(result) => result.map_err(Into::into),
crates/ely_sync_client/src/error.rs
+3
View File
@@ -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,