fix(auth): reconcile expired desktop sessions

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