From 7a5f71d809df869c1517ce1070ba2851e3d0222f Mon Sep 17 00:00:00 2001 From: Tran Duc Anh Dang Date: Wed, 11 Nov 2026 21:58:21 +1100 Subject: [PATCH 1/3] feat(device-client): sync per zone, zone keys, notices stream and status --- Cargo.lock | 30 +- crates/orbyn-device-client/Cargo.toml | 4 + crates/orbyn-device-client/src/api.rs | 10 + crates/orbyn-device-client/src/client.rs | 40 +- crates/orbyn-device-client/src/error.rs | 10 + crates/orbyn-device-client/src/lib.rs | 6 + crates/orbyn-device-client/src/notices.rs | 216 ++++++ crates/orbyn-device-client/src/status.rs | 61 ++ crates/orbyn-device-client/src/sync.rs | 399 ++++++++++ crates/orbyn-device-client/tests/notices.rs | 179 +++++ crates/orbyn-server/Cargo.toml | 3 + crates/orbyn-server/tests/device_client.rs | 780 ++++++++++++++++++++ docs/device-client.md | 30 +- scripts/check_structure.py | 8 +- 14 files changed, 1760 insertions(+), 16 deletions(-) create mode 100644 crates/orbyn-device-client/src/notices.rs create mode 100644 crates/orbyn-device-client/src/status.rs create mode 100644 crates/orbyn-device-client/src/sync.rs create mode 100644 crates/orbyn-device-client/tests/notices.rs create mode 100644 crates/orbyn-server/tests/device_client.rs diff --git a/Cargo.lock b/Cargo.lock index cb85da9..f8e0a55 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -1283,7 +1283,7 @@ dependencies = [ "tokio", "tokio-rustls", "tower-service", - "webpki-roots", + "webpki-roots 1.0.11", ] [[package]] @@ -1553,7 +1553,7 @@ dependencies = [ "tokio", "tokio-rustls", "url", - "webpki-roots", + "webpki-roots 1.0.11", ] [[package]] @@ -1880,15 +1880,19 @@ name = "orbyn-device-client" version = "0.1.0" dependencies = [ "base64 0.22.1", + "futures-util", "getrandom 0.4.3", "orbyn-auth-client", + "orbyn-client", "orbyn-contracts", "orbyn-crypto", + "orbyn-sync", "reqwest", "serde", "serde_json", "thiserror", "tokio", + "tokio-tungstenite", "uuid", "zeroize", ] @@ -1994,9 +1998,11 @@ dependencies = [ "orbyn-contracts", "orbyn-core", "orbyn-crypto", + "orbyn-device-client", "orbyn-keyservice-client", "orbyn-planner", "orbyn-store", + "orbyn-sync", "serde", "serde_json", "tokio", @@ -2006,6 +2012,7 @@ dependencies = [ "tracing", "tracing-subscriber", "uuid", + "zeroize", ] [[package]] @@ -2514,7 +2521,7 @@ dependencies = [ "wasm-bindgen", "wasm-bindgen-futures", "web-sys", - "webpki-roots", + "webpki-roots 1.0.11", ] [[package]] @@ -2874,7 +2881,7 @@ dependencies = [ "tracing", "url", "uuid", - "webpki-roots", + "webpki-roots 1.0.11", ] [[package]] @@ -3229,8 +3236,12 @@ checksum = "8f72a05e828585856dacd553fba484c242c46e391fb0e58917c942ee9202915c" dependencies = [ "futures-util", "log", + "rustls", + "rustls-pki-types", "tokio", + "tokio-rustls", "tungstenite", + "webpki-roots 0.26.11", ] [[package]] @@ -3398,6 +3409,8 @@ dependencies = [ "httparse", "log", "rand 0.9.5", + "rustls", + "rustls-pki-types", "sha1 0.10.7", "thiserror", ] @@ -3631,6 +3644,15 @@ dependencies = [ "wasm-bindgen", ] +[[package]] +name = "webpki-roots" +version = "0.26.11" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "521bc38abb08001b01866da9f51eb7c5d647a19260e00054a8c7fd5f9e57f7a9" +dependencies = [ + "webpki-roots 1.0.11", +] + [[package]] name = "webpki-roots" version = "1.0.11" diff --git a/crates/orbyn-device-client/Cargo.toml b/crates/orbyn-device-client/Cargo.toml index 0f08f74..d91c4e3 100644 --- a/crates/orbyn-device-client/Cargo.toml +++ b/crates/orbyn-device-client/Cargo.toml @@ -10,14 +10,18 @@ publish.workspace = true [dependencies] base64.workspace = true getrandom.workspace = true +futures-util = { version = "0.3", default-features = false, features = ["sink"] } orbyn-auth-client.workspace = true +orbyn-client.workspace = true orbyn-contracts.workspace = true orbyn-crypto.workspace = true +orbyn-sync.workspace = true reqwest.workspace = true serde.workspace = true serde_json.workspace = true thiserror.workspace = true tokio.workspace = true +tokio-tungstenite = { version = "0.29", default-features = false, features = ["connect", "rustls-tls-webpki-roots"] } uuid.workspace = true zeroize.workspace = true diff --git a/crates/orbyn-device-client/src/api.rs b/crates/orbyn-device-client/src/api.rs index f7821ef..9147a08 100644 --- a/crates/orbyn-device-client/src/api.rs +++ b/crates/orbyn-device-client/src/api.rs @@ -56,6 +56,12 @@ impl std::fmt::Debug for Response { /// One call to the engine. A failure to get any answer is [`DeviceError::Transport`]. pub trait DeviceApi: Send + Sync { fn send(&self, request: Request) -> impl Future> + Send; + + /// The engine's address, for the live notices stream (a WebSocket). A transport that is not a + /// network (a test double) has none, and then there is no notices stream. + fn base_url(&self) -> Option<&str> { + None + } } /// Whether `address` is a URL whose host is a loopback **IP literal** (`127.0.0.1`, `[::1]`). @@ -127,6 +133,10 @@ impl HttpApi { } impl DeviceApi for HttpApi { + fn base_url(&self) -> Option<&str> { + Some(&self.base) + } + async fn send(&self, request: Request) -> Result { let url = format!("{}{}", self.base, request.path); let mut builder = match request.method { diff --git a/crates/orbyn-device-client/src/client.rs b/crates/orbyn-device-client/src/client.rs index f03943c..2150bb5 100644 --- a/crates/orbyn-device-client/src/client.rs +++ b/crates/orbyn-device-client/src/client.rs @@ -4,6 +4,7 @@ use crate::api::{DeviceApi, Method, Request, Response}; use crate::clock::Clock; use crate::error::DeviceError; +use crate::status::SyncStatus; use crate::store::{KeyStore, StoredIdentity, StoredSession}; use base64::Engine as _; use base64::engine::general_purpose::URL_SAFE_NO_PAD as B64; @@ -13,6 +14,7 @@ use orbyn_crypto::{DeviceKeys, DevicePublic}; use serde::Serialize; use serde::de::DeserializeOwned; use std::time::Duration; +use tokio::sync::watch; use zeroize::Zeroizing; /// A session as the server answers it. The token is wiped when this is dropped. @@ -102,6 +104,7 @@ pub struct DeviceClient { pub(crate) store: S, pub(crate) clock: C, pub(crate) config: ClientConfig, + pub(crate) status: watch::Sender, } pub(crate) fn encode(bytes: &[u8]) -> String { @@ -170,9 +173,15 @@ impl DeviceClient { store, clock, config, + status: watch::Sender::new(SyncStatus::initial()), } } + /// The transport, for a host that wraps it or reads from it. + pub fn api(&self) -> &A { + &self.api + } + /// The key store, for the host. pub fn store(&self) -> &S { &self.store @@ -227,10 +236,11 @@ impl DeviceClient { parse(&response.body) } - /// An authenticated call with this device's session. A 401 asks the server whether the - /// account still exists: [`DeviceError::AccountGone`] if it was deleted, else + /// An authenticated call with this device's session, whatever the status: the caller reads + /// the answer. A 401 `unauthorized` is asked about: [`DeviceError::AccountGone`] only if + /// `POST /v1/accounts/status` answers 200 with a deletion time, else /// [`DeviceError::SessionExpired`]. - pub(crate) async fn authed( + pub(crate) async fn authed_raw( &self, method: Method, path: &str, @@ -241,11 +251,7 @@ impl DeviceClient { let response = self .call(method, path, Some(session.token.clone()), body) .await?; - if is_ok(&response) { - return Ok(response); - } - let error = failure(&response); - if error == DeviceError::SessionExpired { + if failure(&response) == DeviceError::SessionExpired { let status: Result = self .post( "/v1/accounts/status", @@ -257,8 +263,24 @@ impl DeviceClient { if status.is_ok_and(|s| !s.deleted_at.is_empty()) { return Err(DeviceError::AccountGone); } + return Err(DeviceError::SessionExpired); + } + Ok(response) + } + + /// An authenticated call that must answer 2xx. + pub(crate) async fn authed( + &self, + method: Method, + path: &str, + body: Option>>, + ) -> Result { + let response = self.authed_raw(method, path, body).await?; + if is_ok(&response) { + Ok(response) + } else { + Err(failure(&response)) } - Err(error) } // ---- sign-in ---- diff --git a/crates/orbyn-device-client/src/error.rs b/crates/orbyn-device-client/src/error.rs index 13403d4..a414d48 100644 --- a/crates/orbyn-device-client/src/error.rs +++ b/crates/orbyn-device-client/src/error.rs @@ -56,6 +56,16 @@ pub enum DeviceError { /// A value the caller gave is not valid (an email, an address, a name). #[error("{0}")] InvalidInput(&'static str), + /// The zone was deleted, here or on another device. + #[error("the zone was deleted")] + ZoneDeleted, + /// This device has no key for the zone yet: another device sends it. + #[error("this device has no key for the zone yet")] + NoZoneKey, + /// The sync engine refused something (a record, a key, a snapshot). The message never holds + /// record content or keys. + #[error("sync engine: {0}")] + Engine(String), /// The server answered something this client does not handle. #[error("the server answered {status} {code}")] Server { status: u16, code: String }, diff --git a/crates/orbyn-device-client/src/lib.rs b/crates/orbyn-device-client/src/lib.rs index 4f39318..f4f6556 100644 --- a/crates/orbyn-device-client/src/lib.rs +++ b/crates/orbyn-device-client/src/lib.rs @@ -7,8 +7,11 @@ mod api; mod client; mod clock; mod error; +mod notices; mod signup; +mod status; mod store; +mod sync; pub use api::{DeviceApi, HttpApi, Method, Request, Response, is_loopback_literal}; pub use client::{ @@ -17,5 +20,8 @@ pub use client::{ }; pub use clock::{Clock, SystemClock}; pub use error::{DeviceError, StoreError}; +pub use notices::{Notice, NoticeConfig, Notices}; pub use signup::PendingSignUp; +pub use status::{SyncState, SyncStatus}; pub use store::{KeyStore, MemoryStore, StoredIdentity, StoredSession}; +pub use sync::{SyncReport, ZoneReport}; diff --git a/crates/orbyn-device-client/src/notices.rs b/crates/orbyn-device-client/src/notices.rs new file mode 100644 index 0000000..c3bcef0 --- /dev/null +++ b/crates/orbyn-device-client/src/notices.rs @@ -0,0 +1,216 @@ +//! The live notices stream (`/v1/sync`, a WebSocket): the server says a zone changed, and the host +//! runs a pass. A notice never carries record data. The stream reconnects with a growing, jittered +//! delay, answers the server's pings (the library does, when it reads) and treats a silent +//! connection as dead. + +use crate::client::DeviceClient; +use crate::error::DeviceError; +use crate::{Clock, DeviceApi, KeyStore}; +use futures_util::{SinkExt, StreamExt}; +use orbyn_contracts as wire; +use std::time::Duration; +use tokio::sync::mpsc; +use tokio::task::JoinHandle; +use tokio_tungstenite::tungstenite::Message; + +/// What the stream tells the host. +#[derive(Clone, Debug, PartialEq, Eq)] +pub enum Notice { + /// Connected and signed in. Compare every zone now: changes made while disconnected are not + /// replayed. + Connected, + /// The connection ended; the stream reconnects by itself. + Disconnected, + /// Another device changed the zone. + ZoneChanged { zone: String, token: String }, + /// Notices may have been missed. + Resync, + /// The server refused this connection (`unauthorized`, `too_many_connections`). The stream + /// stops for `unauthorized`: sign in again and open a new one. + Rejected { code: String }, +} + +/// How the stream reconnects. +#[derive(Clone, Copy, Debug)] +pub struct NoticeConfig { + /// The delay before the first reconnect; it doubles each time up to `max_delay`. + pub first_delay: Duration, + pub max_delay: Duration, + /// A connection that hears nothing for this long (the server pings) is dropped and redone. + pub idle_timeout: Duration, +} + +impl Default for NoticeConfig { + fn default() -> Self { + Self { + first_delay: Duration::from_secs(1), + max_delay: Duration::from_secs(60), + idle_timeout: Duration::from_secs(120), + } + } +} + +/// The open stream. Dropping it closes the connection and stops the task. +pub struct Notices { + rx: mpsc::Receiver, + task: JoinHandle<()>, +} + +impl Notices { + /// The next notice, or `None` once the stream has stopped. + pub async fn next(&mut self) -> Option { + self.rx.recv().await + } +} + +impl Drop for Notices { + fn drop(&mut self) { + self.task.abort(); + } +} + +/// `http` to `ws`, `https` to `wss`, and the sync path. +pub(crate) fn websocket_address(base: &str) -> Option { + let rest = base + .strip_prefix("https://") + .map(|r| ("wss", r)) + .or_else(|| base.strip_prefix("http://").map(|r| ("ws", r)))?; + Some(format!( + "{}://{}/v1/sync", + rest.0, + rest.1.trim_end_matches('/') + )) +} + +/// The delay before attempt number `attempt` (0 is the first), with up to a fifth of jitter. +pub(crate) fn backoff(config: &NoticeConfig, attempt: u32) -> Duration { + let base = config + .first_delay + .saturating_mul(2u32.saturating_pow(attempt.min(16))); + let mut random = [0u8; 2]; + let _ = getrandom::fill(&mut random); + let jitter = f64::from(u16::from_be_bytes(random)) / f64::from(u16::MAX); // 0..=1 + base.mul_f64(0.8 + 0.4 * jitter).min(config.max_delay) +} + +impl DeviceClient { + /// Opens the notices stream with this device's current session. It signs in with the token + /// it has; when the token stops working the stream says `Rejected` and stops, and the host + /// signs in again (`reauthenticate`) and opens a new one. + pub fn notices(&self, config: NoticeConfig) -> Result { + let base = self.api.base_url().ok_or(DeviceError::InvalidInput( + "this transport has no live stream", + ))?; + let address = websocket_address(base).ok_or(DeviceError::InvalidInput( + "the engine address is not http or https", + ))?; + let identity = self.store.load()?.ok_or(DeviceError::NotSignedIn)?; + let token = identity + .session + .as_ref() + .ok_or(DeviceError::NotSignedIn)? + .token + .clone(); + let (tx, rx) = mpsc::channel(64); + let task = tokio::spawn(run(address, token, config, tx)); + Ok(Notices { rx, task }) + } +} + +async fn run( + address: String, + token: zeroize::Zeroizing, + config: NoticeConfig, + tx: mpsc::Sender, +) { + let mut attempt = 0u32; + loop { + match tokio_tungstenite::connect_async(address.as_str()).await { + Ok((mut socket, _)) => { + let auth = serde_json::to_string(&wire::SyncClientMessage::Auth { + token: token.to_string(), + }) + .unwrap_or_default(); + if socket.send(Message::text(auth)).await.is_ok() { + loop { + let next = tokio::time::timeout(config.idle_timeout, socket.next()).await; + let Ok(Some(Ok(message))) = next else { break }; + let Message::Text(text) = message else { + // Pings are answered by the library as it reads; closes end the loop. + if matches!(message, Message::Close(_)) { + break; + } + continue; + }; + let Ok(parsed) = serde_json::from_str::(&text) + else { + continue; + }; + let (notice, stop) = match parsed { + wire::SyncServerMessage::Ready { .. } => { + attempt = 0; + (Notice::Connected, false) + } + wire::SyncServerMessage::ZoneChanged { zone, token } => { + (Notice::ZoneChanged { zone, token }, false) + } + wire::SyncServerMessage::Resync => (Notice::Resync, false), + wire::SyncServerMessage::Error { code, .. } => { + let stop = code == "unauthorized"; + (Notice::Rejected { code }, stop) + } + }; + if tx.send(notice).await.is_err() { + return; + } + if stop { + return; + } + } + } + if tx.send(Notice::Disconnected).await.is_err() { + return; + } + } + Err(_) => { + if tx.send(Notice::Disconnected).await.is_err() { + return; + } + } + } + tokio::time::sleep(backoff(&config, attempt)).await; + attempt = attempt.saturating_add(1); + } +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn the_address_follows_the_scheme() { + assert_eq!( + websocket_address("https://api.orbyn.dev").as_deref(), + Some("wss://api.orbyn.dev/v1/sync") + ); + assert_eq!( + websocket_address("http://127.0.0.1:8080/").as_deref(), + Some("ws://127.0.0.1:8080/v1/sync") + ); + assert_eq!(websocket_address("ftp://x"), None); + } + + #[test] + fn the_delay_grows_with_jitter_and_stops_at_the_cap() { + let config = NoticeConfig::default(); + for attempt in 0..30 { + let delay = backoff(&config, attempt); + let base = Duration::from_secs(1u64 << attempt.min(6)).min(config.max_delay); + assert!( + delay >= base.mul_f64(0.79) && delay <= config.max_delay, + "{attempt}: {delay:?}" + ); + } + assert!(backoff(&config, 40) <= config.max_delay); + } +} diff --git a/crates/orbyn-device-client/src/status.rs b/crates/orbyn-device-client/src/status.rs new file mode 100644 index 0000000..b4abaf5 --- /dev/null +++ b/crates/orbyn-device-client/src/status.rs @@ -0,0 +1,61 @@ +//! What the sync is doing, for the host to show. A `tokio::sync::watch`: the latest value is +//! always there, and a host that only looks now and then misses nothing it needs. + +use crate::error::DeviceError; + +/// Where the sync stands. +#[derive(Clone, Copy, Debug, PartialEq, Eq)] +pub enum SyncState { + /// Nothing has run yet, or the last pass finished. + Idle, + /// A pass is running. + Syncing, + /// The last pass could not reach the server. Work is never blocked by this. + Offline, + /// There is no session: sign in. + SignedOut, + /// A recent sign-in is needed: call `reauthenticate`. + ReauthenticationRequired, + /// The session ended (signed out elsewhere); the account is alive. + SessionExpired, + /// The account was deleted: this device's data on the server is gone. + AccountGone, + /// The last pass failed for another reason (see `last_error`). + Failed, +} + +/// The sync status. +#[derive(Clone, Debug, PartialEq, Eq)] +pub struct SyncStatus { + pub state: SyncState, + /// Unix seconds of the last pass that finished without an error. + pub last_synced: Option, + /// The error text of the last failed pass. Never holds a secret. + pub last_error: Option, +} + +impl SyncStatus { + pub(crate) fn initial() -> Self { + Self { + state: SyncState::Idle, + last_synced: None, + last_error: None, + } + } + + pub(crate) fn after(&self, error: &DeviceError, _now: i64) -> Self { + let state = match error { + DeviceError::Transport => SyncState::Offline, + DeviceError::NotSignedIn => SyncState::SignedOut, + DeviceError::ReauthenticationRequired => SyncState::ReauthenticationRequired, + DeviceError::SessionExpired => SyncState::SessionExpired, + DeviceError::AccountGone => SyncState::AccountGone, + _ => SyncState::Failed, + }; + Self { + state, + last_synced: self.last_synced, + last_error: Some(error.to_string()), + } + } +} diff --git a/crates/orbyn-device-client/src/sync.rs b/crates/orbyn-device-client/src/sync.rs new file mode 100644 index 0000000..65f8c90 --- /dev/null +++ b/crates/orbyn-device-client/src/sync.rs @@ -0,0 +1,399 @@ +//! Sync: open the engine, then push and pull per zone, hand out and take in zone keys, and keep +//! the status current. The records go in and out as plain fields (`orbyn_sync::Fields`); sealing, +//! opening and the zone keys are inside `orbyn_client::Engine`, whose `Snapshot` the host stores +//! sealed under the vault key. + +use crate::api::Method; +use crate::client::{DeviceClient, failure, is_ok, json, parse}; +use crate::error::DeviceError; +use crate::status::{SyncState, SyncStatus}; +use crate::{Clock, DeviceApi, KeyStore}; +use orbyn_client::{Applied, ClientError, Engine, RefusedEnvelope, Snapshot}; +use orbyn_contracts as wire; + +/// How many times one zone's upload is retried inside a pass (a conflict merges and goes again). +const MAX_UPLOAD_ROUNDS: usize = 8; +/// The most envelopes one request carries. +const ENVELOPES_PER_REQUEST: usize = 100; + +/// What a pass did to one zone. +#[derive(Clone, Debug, Default, PartialEq, Eq)] +pub struct ZoneReport { + pub zone: String, + /// What came in from the server. + pub applied: Applied, + /// Records this device uploaded. + pub pushed: usize, + /// The zone has changes to upload and this device has no key for it yet: another device sends + /// it. Nothing is lost; the next pass tries again. + pub waiting_for_key: bool, +} + +/// What a whole pass did. +#[derive(Clone, Debug, Default, PartialEq, Eq)] +pub struct SyncReport { + pub zones: Vec, + /// Zones that got a new key from another device. + pub keys_accepted: Vec, + /// Envelopes this device refused, and why. + pub keys_refused: Vec, +} + +pub(crate) fn engine_error(error: ClientError) -> DeviceError { + match error { + ClientError::RecordTooLarge => DeviceError::TooLarge, + ClientError::NoZoneKey => DeviceError::NoZoneKey, + other => DeviceError::Engine(other.to_string()), + } +} + +/// A path segment, percent-encoded: zone names and tokens may hold `:` or `.`. +fn segment(text: &str) -> String { + let mut out = String::with_capacity(text.len()); + for byte in text.bytes() { + match byte { + b'A'..=b'Z' | b'a'..=b'z' | b'0'..=b'9' | b'-' | b'.' | b'_' | b'~' => { + out.push(char::from(byte)); + } + other => out.push_str(&format!("%{other:02X}")), + } + } + out +} + +fn merge(into: &mut Applied, from: Applied) { + into.changed.extend(from.changed); + into.removed.extend(from.removed); + into.trashed.extend(from.trashed); + into.quarantined.extend(from.quarantined); + into.unverified_deletes.extend(from.unverified_deletes); + into.evicted.extend(from.evicted); + into.rejected.extend(from.rejected); + into.more = from.more; +} + +impl DeviceClient { + fn now_millis(&self) -> u64 { + u64::try_from(self.clock.now_secs()).unwrap_or(0) * 1000 + } + + /// The status: `Idle`, `Syncing`, `Offline`, and the states that need the host to act. The + /// latest value is always there. + pub fn status(&self) -> tokio::sync::watch::Receiver { + self.status.subscribe() + } + + fn set_state(&self, state: SyncState) { + self.status.send_modify(|s| s.state = state); + } + + /// Opens the sync engine for this device: from the host's sealed snapshot, or empty the + /// first time. The snapshot holds the zone keys and the cache, and belongs to this device and + /// account only. + pub fn open_engine(&self, snapshot: Option) -> Result { + let identity = self.store.load()?.ok_or(DeviceError::NotSignedIn)?; + let keys = identity.keys()?; + match snapshot { + Some(snapshot) => { + if snapshot.account_id != identity.account_id + || snapshot.device_id != identity.device_id + { + return Err(DeviceError::InvalidInput( + "the snapshot belongs to another account or device", + )); + } + Engine::restore(keys, snapshot, self.now_millis()) + } + None => Engine::new(keys, &identity.account_id, &identity.device_id), + } + .map_err(engine_error) + } + + /// The account's devices, as the server lists them. The engine checks the approvals itself. + pub async fn devices(&self) -> Result { + let response = self.authed(Method::Get, "/v1/devices", None).await?; + parse(&response.body) + } + + /// One pass over the whole account: the device list, the zone keys other devices sent, then + /// every zone that changed on the server or has local changes, pulled and pushed. A failure + /// of the network is [`DeviceError::Transport`] and changes nothing: whatever was staged + /// stays staged, and the next pass sends it. The status reflects the result either way. + pub async fn sync_all(&self, engine: &mut Engine) -> Result { + self.set_state(SyncState::Syncing); + let result = self.sync_all_inner(engine).await; + let now = self.clock.now_secs(); + match &result { + Ok(_) => self.status.send_modify(|s| { + s.state = SyncState::Idle; + s.last_synced = Some(now); + s.last_error = None; + }), + Err(error) => self.status.send_modify(|s| *s = s.after(error, now)), + } + result + } + + async fn sync_all_inner(&self, engine: &mut Engine) -> Result { + let now = self.now_millis(); + // The engine verifies the devices itself when it takes keys in and hands them out. + let devices = self.devices().await?; + let mut report = SyncReport::default(); + + // Keys other devices sent come first, so what could not be opened can be. + let envelopes: wire::KeyEnvelopeList = parse( + &self + .authed(Method::Get, "/v1/keychain/envelopes", None) + .await? + .body, + )?; + let accepted = engine.accept_envelopes(&envelopes.envelopes, &devices); + for zone in &accepted.zones { + engine.retry_quarantined(zone, now); + } + report.keys_accepted = accepted.zones; + report.keys_refused = accepted.refused; + + // The zones on the server, and which differ from this device's view. + let mut on_server: Vec<(String, bool)> = Vec::new(); + let mut stale: Vec = Vec::new(); + let mut after: Option = None; + loop { + let path = match &after { + Some(after) => format!("/v1/zones?after={}", segment(after)), + None => "/v1/zones".to_owned(), + }; + let page: wire::ZoneList = parse(&self.authed(Method::Get, &path, None).await?.body)?; + let check = engine.check_zones(&page); + stale.extend(check.stale); + on_server.extend(page.zones.iter().map(|z| (z.zone.clone(), z.deleted))); + match check.next { + Some(next) => after = Some(next), + None => break, + } + } + + // Zones with local changes, and zones made here that the server does not have yet. + let mut work: Vec = stale; + for zone in engine.zones().map(str::to_owned).collect::>() { + if engine.has_pending(&zone) && !work.contains(&zone) { + work.push(zone); + } + } + work.sort(); + for zone in work { + let exists = on_server + .iter() + .any(|(name, deleted)| *name == zone && !deleted); + if !exists && engine.has_pending(&zone) { + self.make_zone(engine, &zone, &devices).await?; + } + report.zones.push(self.sync_zone(engine, &zone).await?); + } + Ok(report) + } + + /// Makes a zone on the server and, if this device has no key for it, a key; then hands the + /// key to the account's other trusted devices. + async fn make_zone( + &self, + engine: &mut Engine, + zone: &str, + devices: &wire::DeviceList, + ) -> Result<(), DeviceError> { + self.authed(Method::Put, &format!("/v1/zones/{}", segment(zone)), None) + .await?; + engine.create_zone_key(zone).map_err(engine_error)?; + self.send_keys(engine, zone, devices).await + } + + /// Seals every key of `zone` to each trusted device of the account that this device + /// verifies, and sends the envelopes. A zone nobody else needs yet sends nothing. + pub async fn send_keys( + &self, + engine: &mut Engine, + zone: &str, + devices: &wire::DeviceList, + ) -> Result<(), DeviceError> { + let envelopes = engine.envelopes_for(zone, devices).map_err(engine_error)?; + for chunk in envelopes.chunks(ENVELOPES_PER_REQUEST) { + let body = json(&wire::PutKeyEnvelopes { + envelopes: chunk.to_vec(), + service_envelopes: None, + })?; + self.authed(Method::Post, "/v1/keychain/envelopes", Some(body)) + .await?; + } + Ok(()) + } + + /// Pulls, then pushes, one zone. + pub async fn sync_zone( + &self, + engine: &mut Engine, + zone: &str, + ) -> Result { + let mut report = ZoneReport { + zone: zone.to_owned(), + ..ZoneReport::default() + }; + for _ in 0..3 { + self.pull(engine, zone, &mut report.applied).await?; + match self.push(engine, zone, &mut report).await { + Ok(false) => return Ok(report), + // An upload was told the token is too old: the zone starts again, then goes up. + Ok(true) => {} + Err(DeviceError::NoZoneKey) => { + report.waiting_for_key = true; + return Ok(report); + } + Err(error) => return Err(error), + } + } + Err(DeviceError::Server { + status: 410, + code: "token_expired".to_owned(), + }) + } + + async fn pull( + &self, + engine: &mut Engine, + zone: &str, + applied: &mut Applied, + ) -> Result<(), DeviceError> { + let mut restarts = 0; + loop { + let since = engine.token(zone).map(str::to_owned); + let mut path = format!("/v1/zones/{}/changes?limit=200", segment(zone)); + if let Some(token) = &since { + path.push_str(&format!("&since={}", segment(token))); + } + let response = self.authed_raw(Method::Get, &path, None).await?; + match response.status { + 200 => {} + // The server purged history this device has not seen: fetch from the start. + 410 => { + restarts += 1; + if restarts > 3 { + return Err(failure(&response)); + } + engine.token_expired(zone).map_err(engine_error)?; + continue; + } + // Not made yet, or deleted: nothing to pull. + 404 | 409 => return Ok(()), + _ => return Err(failure(&response)), + } + let changes: wire::ZoneChanges = parse(&response.body)?; + // `apply_changes_from` also serves the pages of a fetch that starts again. + let page = engine + .apply_changes_from(zone, since.as_deref(), &changes, self.now_millis()) + .map_err(engine_error)?; + let more = changes.more; + merge(applied, page); + if !more { + return Ok(()); + } + } + } + + async fn push( + &self, + engine: &mut Engine, + zone: &str, + report: &mut ZoneReport, + ) -> Result { + for _ in 0..MAX_UPLOAD_ROUNDS { + let Some(upload) = engine.build_upload(zone).map_err(engine_error)? else { + return Ok(false); + }; + let response = self + .authed_raw( + Method::Post, + &format!("/v1/zones/{}/changes", segment(zone)), + Some(json(&upload)?), + ) + .await?; + if is_ok(&response) { + let result: wire::UploadResult = parse(&response.body)?; + report.pushed += result.records.len(); + engine + .upload_succeeded(zone, &result) + .map_err(engine_error)?; + continue; + } + // Records changed on the server: merge what came and go again. + if response.status == 409 + && let Ok(conflict) = serde_json::from_slice::(&response.body) + && conflict.code == "record_changed" + { + let applied = engine + .upload_conflicted(zone, &conflict, self.now_millis()) + .map_err(engine_error)?; + merge(&mut report.applied, applied); + continue; + } + if response.status == 410 { + engine.token_expired(zone).map_err(engine_error)?; + return Ok(true); + } + let error = failure(&response); + // A batch the server will never take is dropped, so the next pass does not loop on it. + if error == DeviceError::TooLarge { + engine.abandon_upload(zone).map_err(engine_error)?; + } + if response.status == 404 { + return Err(DeviceError::Server { + status: 404, + code: "zone_not_found".to_owned(), + }); + } + return Err(error); + } + Ok(false) + } +} + +#[cfg(test)] +mod tests { + use super::*; + use crate::status::{SyncState, SyncStatus}; + + #[test] + fn path_segments_are_percent_encoded_and_plain_names_are_not() { + assert_eq!(segment("engineering.settings"), "engineering.settings"); + assert_eq!(segment("list:3f2c"), "list%3A3f2c"); + assert_eq!(segment("a b/c?d#e%"), "a%20b%2Fc%3Fd%23e%25"); + assert_eq!(segment("t0ken-_~."), "t0ken-_~."); + } + + #[test] + fn each_error_puts_the_status_in_the_state_the_host_acts_on() { + let start = SyncStatus { + state: SyncState::Syncing, + last_synced: Some(5), + last_error: None, + }; + for (error, state) in [ + (DeviceError::Transport, SyncState::Offline), + (DeviceError::NotSignedIn, SyncState::SignedOut), + ( + DeviceError::ReauthenticationRequired, + SyncState::ReauthenticationRequired, + ), + (DeviceError::SessionExpired, SyncState::SessionExpired), + (DeviceError::AccountGone, SyncState::AccountGone), + (DeviceError::TooLarge, SyncState::Failed), + ] { + let after = start.after(&error, 9); + assert_eq!(after.state, state, "{error:?}"); + assert_eq!( + after.last_synced, + Some(5), + "a failure keeps the last good time" + ); + assert!(after.last_error.is_some()); + } + } +} diff --git a/crates/orbyn-device-client/tests/notices.rs b/crates/orbyn-device-client/tests/notices.rs new file mode 100644 index 0000000..3f560ae --- /dev/null +++ b/crates/orbyn-device-client/tests/notices.rs @@ -0,0 +1,179 @@ +//! The notices stream against a hand-made WebSocket server: it signs in first, answers pings, +//! reconnects after a drop and stops when the server rejects the session. + +use futures_util::{SinkExt, StreamExt}; +use orbyn_contracts::DevicePlatform; +use orbyn_device_client::{ + ClientConfig, DeviceClient, HttpApi, KeyStore, MemoryStore, Notice, NoticeConfig, + StoredIdentity, StoredSession, SystemClock, +}; +use std::time::Duration; +use tokio::net::TcpListener; +use tokio_tungstenite::tungstenite::Message; +use zeroize::Zeroizing; + +async fn client_for(address: &str) -> DeviceClient { + let store = MemoryStore::new(); + store + .save(&StoredIdentity { + device_keys: Zeroizing::new([3u8; 64]), + account_id: "account".to_owned(), + device_id: "device".to_owned(), + email: "a@example.test".to_owned(), + session: Some(StoredSession { + token: Zeroizing::new("the-token".to_owned()), + expires_at: "2027-01-01T00:00:00Z".to_owned(), + }), + vault_key: Zeroizing::new([4u8; 32]), + }) + .expect("save"); + DeviceClient::new( + HttpApi::new(address).expect("loopback"), + store, + SystemClock, + ClientConfig { + device_name: "Laptop".to_owned(), + platform: DevicePlatform::Linux, + }, + ) +} + +fn quick() -> NoticeConfig { + NoticeConfig { + first_delay: Duration::from_millis(20), + max_delay: Duration::from_millis(100), + idle_timeout: Duration::from_secs(5), + } +} + +async fn next(notices: &mut orbyn_device_client::Notices) -> Notice { + tokio::time::timeout(Duration::from_secs(5), notices.next()) + .await + .expect("a notice in time") + .expect("the stream is open") +} + +#[tokio::test(flavor = "multi_thread")] +async fn it_signs_in_first_answers_pings_and_reconnects_after_a_drop() { + let listener = TcpListener::bind("127.0.0.1:0").await.expect("bind"); + let address = format!("http://{}", listener.local_addr().expect("addr")); + let server = tokio::spawn(async move { + let mut auths = Vec::new(); + // First connection: sign-in, ready, a ping that must be answered, a notice, then a drop. + let (stream, _) = listener.accept().await.expect("accept"); + let mut socket = tokio_tungstenite::accept_async(stream) + .await + .expect("handshake"); + let Some(Ok(Message::Text(first))) = socket.next().await else { + panic!("no auth") + }; + auths.push(first.to_string()); + socket + .send(Message::text(r#"{"type":"ready","device_id":"device"}"#)) + .await + .expect("ready"); + socket + .send(Message::Ping(b"are you there".to_vec().into())) + .await + .expect("ping"); + let pong = tokio::time::timeout(Duration::from_secs(5), async { + while let Some(Ok(message)) = socket.next().await { + if let Message::Pong(data) = message { + return Some(data); + } + } + None + }) + .await + .expect("a pong in time"); + assert_eq!(pong.as_deref().map(|d| &d[..]), Some(&b"are you there"[..])); + socket + .send(Message::text( + r#"{"type":"zone_changed","zone":"engineering.settings","token":"t1"}"#, + )) + .await + .expect("notice"); + drop(socket); + // Second connection: the stream came back by itself, signed in again. + let (stream, _) = listener.accept().await.expect("accept again"); + let mut socket = tokio_tungstenite::accept_async(stream) + .await + .expect("handshake"); + let Some(Ok(Message::Text(second))) = socket.next().await else { + panic!("no auth") + }; + auths.push(second.to_string()); + socket + .send(Message::text(r#"{"type":"ready","device_id":"device"}"#)) + .await + .expect("ready"); + socket + .send(Message::text(r#"{"type":"resync"}"#)) + .await + .expect("resync"); + tokio::time::sleep(Duration::from_millis(300)).await; + auths + }); + let client = client_for(&address).await; + let mut notices = client.notices(quick()).expect("stream"); + assert_eq!(next(&mut notices).await, Notice::Connected); + assert_eq!( + next(&mut notices).await, + Notice::ZoneChanged { + zone: "engineering.settings".to_owned(), + token: "t1".to_owned() + } + ); + assert_eq!(next(&mut notices).await, Notice::Disconnected); + assert_eq!(next(&mut notices).await, Notice::Connected); + assert_eq!(next(&mut notices).await, Notice::Resync); + let auths = server.await.expect("server"); + for auth in auths { + assert_eq!(auth, r#"{"type":"auth","token":"the-token"}"#); + } +} + +#[tokio::test(flavor = "multi_thread")] +async fn a_rejected_session_stops_the_stream_and_a_dropped_stream_closes_it() { + let listener = TcpListener::bind("127.0.0.1:0").await.expect("bind"); + let address = format!("http://{}", listener.local_addr().expect("addr")); + tokio::spawn(async move { + let (stream, _) = listener.accept().await.expect("accept"); + let mut socket = tokio_tungstenite::accept_async(stream) + .await + .expect("handshake"); + let _ = socket.next().await; + socket + .send(Message::text( + r#"{"type":"error","code":"unauthorized","message":"no"}"#, + )) + .await + .expect("error"); + // Keep the socket open: the stream must stop on its own. + tokio::time::sleep(Duration::from_secs(2)).await; + }); + let client = client_for(&address).await; + let mut notices = client.notices(quick()).expect("stream"); + assert_eq!( + next(&mut notices).await, + Notice::Rejected { + code: "unauthorized".to_owned() + } + ); + assert_eq!( + tokio::time::timeout(Duration::from_secs(3), notices.next()) + .await + .expect("ends"), + None + ); +} + +#[tokio::test(flavor = "multi_thread")] +async fn without_a_session_or_an_address_there_is_no_stream() { + let client = client_for("http://127.0.0.1:9").await; + client.sign_out(false).await.expect("sign out"); + assert_eq!( + client.notices(quick()).err(), + Some(orbyn_device_client::DeviceError::NotSignedIn) + ); +} diff --git a/crates/orbyn-server/Cargo.toml b/crates/orbyn-server/Cargo.toml index 84b329e..2d9cb70 100644 --- a/crates/orbyn-server/Cargo.toml +++ b/crates/orbyn-server/Cargo.toml @@ -27,10 +27,13 @@ uuid.workspace = true futures-util = { version = "0.3", default-features = false, features = ["sink"] } orbyn-auth-client.workspace = true orbyn-client.workspace = true +orbyn-device-client.workspace = true orbyn-planner.workspace = true +orbyn-sync.workspace = true tokio-tungstenite = "0.29" tower.workspace = true tracing-subscriber.workspace = true +zeroize.workspace = true [lints] workspace = true diff --git a/crates/orbyn-server/tests/device_client.rs b/crates/orbyn-server/tests/device_client.rs new file mode 100644 index 0000000..c44d1ce --- /dev/null +++ b/crates/orbyn-server/tests/device_client.rs @@ -0,0 +1,780 @@ +//! The device client against the real server on Postgres, over a real loopback socket: sign-up, +//! two devices, zone keys, push and pull, a conflict, the snapshot, the notices stream and the +//! status. Ignored unless run with `--include-ignored` and `DATABASE_URL`. + +use axum::body::{Body, to_bytes}; +use axum::extract::ConnectInfo; +use axum::http::{Method, Request, StatusCode, header}; +use orbyn_auth_client as auth; +use orbyn_client::Engine; +use orbyn_contracts as wire; +use orbyn_core::accounts::AuthSecrets; +use orbyn_core::{Accounts, Database}; +use orbyn_device_client::{ + ClientConfig, DeviceClient, DeviceError, HttpApi, KeyStore, MemoryStore, Notice, NoticeConfig, + SignInOutcome, SignUpRequest, SyncState, SystemClock, +}; +use orbyn_server::{AppState, Readiness, router}; +use orbyn_store::{Hub, Store}; +use orbyn_sync::Fields; +use serde::Serialize; +use serde::de::DeserializeOwned; +use serde_json::{Value, json}; +use std::future::Future; +use std::net::{IpAddr, Ipv4Addr, SocketAddr}; +use std::pin::Pin; +use std::sync::{Arc, OnceLock}; +use std::time::Duration; +use tower::ServiceExt; +use uuid::Uuid; +use zeroize::Zeroizing; + +const ZONE: &str = "engineering.settings"; +const PASSWORD: &str = "correct horse battery staple"; + +struct Up; +impl Readiness for Up { + fn database_ready(&self) -> Pin + Send + '_>> { + Box::pin(async { true }) + } +} + +static SETUP: OnceLock = OnceLock::new(); +static MAILBOX: OnceLock = OnceLock::new(); + +/// The verification code sent to `email`: delivers the queue into the shared mailbox until it shows. +async fn code_sent_to(pool: &orbyn_core::sqlx::PgPool, email: &str) -> String { + let mailbox = MAILBOX.get_or_init(orbyn_core::mail::Capture::default); + let secrets = AuthSecrets::from_encoded(SETUP.get().expect("setup")).expect("setup"); + let to = auth::normalize_email(email).expect("email"); + for _ in 0..200 { + orbyn_core::mail::deliver_queued(pool, &secrets, mailbox) + .await + .expect("deliver"); + if let Some(code) = mailbox.code_for(&to) { + return code; + } + tokio::time::sleep(Duration::from_millis(25)).await; + } + panic!("no verification code was sent to {to}"); +} + +struct Server { + app: axum::Router, + db: Database, + base: String, +} + +impl Server { + /// The real app on Postgres, served on a loopback port. + async fn start() -> Self { + let url = + std::env::var("DATABASE_URL").expect("set DATABASE_URL to run the Postgres tests"); + let db = Database::connect(&url).await.expect("connect to Postgres"); + db.migrate().await.expect("migrate"); + let setup = SETUP.get_or_init(|| AuthSecrets::generate().to_string()); + let secrets = AuthSecrets::from_encoded(setup).expect("setup"); + let accounts = + Accounts::new(db.pool().clone(), secrets).with_signup(orbyn_core::SignupMode::Open); + let hub = Hub::start(db.pool().clone()); + let store = Store::new(db.pool().clone(), hub); + let state = AppState::new(Arc::new(Up)) + .with_accounts(accounts) + .with_store(store); + let app = router(state, &[]); + let listener = tokio::net::TcpListener::bind("127.0.0.1:0") + .await + .expect("bind"); + let address = listener.local_addr().expect("address"); + let service = app + .clone() + .into_make_service_with_connect_info::(); + tokio::spawn(async move { axum::serve(listener, service).await }); + Self { + app, + db, + base: format!("http://{address}"), + } + } + + fn device(&self, name: &str) -> DeviceClient { + DeviceClient::new( + HttpApi::new(&self.base).expect("loopback"), + MemoryStore::new(), + SystemClock, + ClientConfig { + device_name: name.to_owned(), + platform: wire::DevicePlatform::Linux, + }, + ) + } + + /// A call straight into the app (no socket), for the approval steps. + async fn call( + &self, + method: Method, + path: &str, + token: &str, + body: Option, + ) -> (StatusCode, Value) { + let mut request = Request::builder() + .method(method) + .uri(path) + .header(header::AUTHORIZATION, format!("Bearer {token}")); + let body = match body { + Some(value) => { + request = request.header(header::CONTENT_TYPE, "application/json"); + Body::from(value.to_string()) + } + None => Body::empty(), + }; + let mut request = request.body(body).expect("request"); + request.extensions_mut().insert(ConnectInfo(SocketAddr::new( + IpAddr::V4(Ipv4Addr::LOCALHOST), + 40_000, + ))); + let response = self.app.clone().oneshot(request).await.expect("response"); + let status = response.status(); + let bytes = to_bytes(response.into_body(), 16 << 20) + .await + .expect("body"); + let json = if bytes.is_empty() { + Value::Null + } else { + serde_json::from_slice(&bytes).expect("json") + }; + (status, json) + } + + async fn ok( + &self, + method: Method, + path: &str, + token: &str, + body: Option, + expected: StatusCode, + ) -> T { + let (status, value) = self.call(method, path, token, body).await; + assert_eq!(status, expected, "{path}: {value}"); + serde_json::from_value(value).expect("wire type") + } +} + +type Client = DeviceClient; + +fn password() -> Zeroizing { + Zeroizing::new(PASSWORD.to_owned()) +} + +fn email() -> String { + format!("device-{}@example.test", Uuid::new_v4().simple()) +} + +async fn sign_up(server: &Server, client: &Client, email: &str) { + let mut pending = client + .begin_sign_up(SignUpRequest { + email: email.to_owned(), + password: password(), + invite: None, + }) + .await + .expect("begin"); + let code = code_sent_to(server.db.pool(), email).await; + client + .verify_sign_up(&mut pending, &code) + .await + .expect("verify"); +} + +fn token_of(client: &Client) -> String { + let identity = client.store().load().expect("store").expect("signed in"); + identity.session.expect("a session").token.to_string() +} + +fn device_id_of(client: &Client) -> String { + client + .store() + .load() + .expect("store") + .expect("signed in") + .device_id +} + +fn now_secs() -> i64 { + std::time::SystemTime::now() + .duration_since(std::time::UNIX_EPOCH) + .map_or(0, |d| i64::try_from(d.as_secs()).unwrap_or(0)) +} + +fn value(v: &T) -> Value { + serde_json::to_value(v).expect("json") +} + +/// The whole two-way approval over HTTP, with the person confirming that both screens show the +/// same digits (the real client API for this is the next change). +async fn approve( + server: &Server, + (approver_token, approver): (&str, &mut Engine), + (new_token, new, new_id): (&str, &mut Engine, &str), +) { + let list: wire::DeviceList = server + .ok( + Method::GET, + "/v1/devices", + approver_token, + None, + StatusCode::OK, + ) + .await; + let pending = list + .devices + .iter() + .find(|d| d.id == new_id) + .expect("the new device is listed") + .clone(); + let (mut pa, create) = approver + .start_approval(&pending, now_secs()) + .expect("start"); + let created: wire::ApprovalCreated = server + .ok( + Method::POST, + "/v1/devices/approvals", + approver_token, + Some(value(&create)), + StatusCode::CREATED, + ) + .await; + let step = |name: &str| format!("/v1/devices/approvals/{}/{name}", created.id); + let view = current_approval(server, new_token).await; + let (pd, nonce) = new.join_approval(&view, now_secs()).expect("join"); + let (status, body) = server + .call(Method::POST, &step("nonce"), new_token, Some(value(&nonce))) + .await; + assert_eq!(status, StatusCode::NO_CONTENT, "{body}"); + let on_approver: wire::PendingApproval = server + .ok( + Method::GET, + &format!("/v1/devices/approvals/{}", created.id), + approver_token, + None, + StatusCode::OK, + ) + .await; + let reveal = pa.reveal(&on_approver).expect("reveal"); + let (status, body) = server + .call( + Method::POST, + &step("reveal"), + approver_token, + Some(value(&reveal)), + ) + .await; + assert_eq!(status, StatusCode::NO_CONTENT, "{body}"); + let view = current_approval(server, new_token).await; + let on_new = pd.sas(&view).expect("sas"); + assert_eq!(pa.sas(), Some(on_new), "both screens show the same digits"); + let confirm = approver.confirm_approval(pa, &list).expect("confirm"); + let (status, body) = server + .call( + Method::POST, + &step("confirm"), + approver_token, + Some(value(&confirm)), + ) + .await; + assert_eq!(status, StatusCode::NO_CONTENT, "{body}"); + let view = current_approval(server, new_token).await; + let code = new + .finish_approval(pd, &view, None, now_secs()) + .expect("finish"); + let (status, body) = server + .call( + Method::POST, + &step("code"), + new_token, + Some(json!({"code": code})), + ) + .await; + assert_eq!(status, StatusCode::OK, "{body}"); +} + +async fn current_approval(server: &Server, token: &str) -> wire::PendingApproval { + let current: wire::CurrentDevice = server + .ok( + Method::GET, + "/v1/devices/current", + token, + None, + StatusCode::OK, + ) + .await; + current.approval.expect("an open approval") +} + +/// One field set on a record, stamped by the engine's clock. +fn set(engine: &mut Engine, zone: &str, id: &str, name: &str, to: Value) { + let stamp = engine.tick(u64::try_from(now_secs()).unwrap_or(0) * 1000); + let mut fields = Fields::default(); + fields.set(name, to, stamp); + engine + .stage(zone, "person_settings", id, &fields) + .expect("stage"); +} + +fn field(engine: &Engine, zone: &str, id: &str, name: &str) -> Option { + engine.record(zone, id)?.get(name).cloned() +} + +#[tokio::test] +#[ignore = "needs DATABASE_URL"] +async fn two_devices_sync_records_keys_and_a_conflict_through_the_real_server() { + let server = Server::start().await; + let email = email(); + let a = server.device("Laptop"); + sign_up(&server, &a, &email).await; + let b = server.device("Desktop"); + let outcome = b.sign_in(&email, &password()).await.expect("sign in"); + assert!( + matches!(outcome, SignInOutcome::NeedsApproval(_)), + "{outcome:?}" + ); + + // A pending device cannot sync. + let mut engine_b = b.open_engine(None).expect("engine"); + assert_eq!( + b.sync_all(&mut engine_b).await.err(), + Some(DeviceError::DeviceNotTrusted) + ); + assert_eq!(b.status().borrow().state, SyncState::Failed); + + // A stages a record and syncs: the zone, its key and the sealed record reach the server. + let mut engine_a = a.open_engine(None).expect("engine"); + set( + &mut engine_a, + ZONE, + "person", + "default_theme", + json!("dark"), + ); + let report = a.sync_all(&mut engine_a).await.expect("pass"); + assert_eq!(report.zones.len(), 1); + assert_eq!( + (report.zones[0].zone.as_str(), report.zones[0].pushed), + (ZONE, 1) + ); + assert!(!engine_a.has_pending(ZONE)); + { + let status = a.status(); + let status = status.borrow(); + assert_eq!(status.state, SyncState::Idle); + assert!(status.last_synced.is_some()); + } + // The server holds ciphertext, never the value. + let (length, at): (i32, i32) = orbyn_core::sqlx::query_as( + "SELECT length(r.ciphertext), position('dark'::bytea in r.ciphertext) + FROM records r JOIN zones z ON z.id = r.zone_id + WHERE z.name = $1 AND r.record_id = 'person'", + ) + .bind(ZONE) + .fetch_one(server.db.pool()) + .await + .expect("the record is on the server"); + assert!( + length > 0 && at == 0, + "ciphertext of {length} bytes, plaintext found at {at}" + ); + + // A approves B and hands it the zone key; B's pass takes the key and pulls the record. + approve( + &server, + (&token_of(&a), &mut engine_a), + (&token_of(&b), &mut engine_b, &device_id_of(&b)), + ) + .await; + // Trusted but without the key yet, B stages work: its pass reports it is waiting for the key + // and fails nothing; the work stays staged. + set( + &mut engine_b, + ZONE, + "extra", + "quit_confirmation", + json!("hold"), + ); + let waiting = b + .sync_all(&mut engine_b) + .await + .expect("a pass without the key"); + assert!( + waiting + .zones + .iter() + .any(|z| z.zone == ZONE && z.waiting_for_key), + "{waiting:?}" + ); + assert!(engine_b.has_pending(ZONE)); + let devices = a.devices().await.expect("devices"); + a.send_keys(&mut engine_a, ZONE, &devices) + .await + .expect("keys"); + let report = b.sync_all(&mut engine_b).await.expect("pass"); + assert_eq!(report.keys_accepted, vec![ZONE.to_owned()]); + assert!( + !engine_b.has_pending(ZONE), + "the staged work went up once the key arrived" + ); + assert_eq!( + field(&engine_b, ZONE, "person", "default_theme"), + Some(json!("dark")) + ); + + // Both edit the same record while apart, different fields; the passes merge them. + set( + &mut engine_a, + ZONE, + "person", + "branch_name_prefix", + json!("feat"), + ); + set(&mut engine_b, ZONE, "person", "preview_zoom", json!(2)); + set( + &mut engine_b, + ZONE, + "person", + "default_theme", + json!("light"), + ); + b.sync_all(&mut engine_b).await.expect("b first"); + a.sync_all(&mut engine_a) + .await + .expect("a second: a conflict, merged"); + b.sync_all(&mut engine_b).await.expect("b again"); + for engine in [&engine_a, &engine_b] { + assert_eq!( + field(engine, ZONE, "person", "branch_name_prefix"), + Some(json!("feat")) + ); + assert_eq!( + field(engine, ZONE, "person", "preview_zoom"), + Some(json!(2)) + ); + assert_eq!( + field(engine, ZONE, "person", "default_theme"), + Some(json!("light")) + ); + } + + // The snapshot is all a host keeps: a device restored from it sees the same record. + let saved = serde_json::to_vec(&engine_b.snapshot()).expect("snapshot"); + let restored = b + .open_engine(Some(serde_json::from_slice(&saved).expect("snapshot"))) + .expect("restore"); + assert_eq!( + field(&restored, ZONE, "person", "preview_zoom"), + Some(json!(2)) + ); + // A snapshot of another device is refused. + assert!(matches!( + a.open_engine(Some(serde_json::from_slice(&saved).expect("snapshot"))), + Err(DeviceError::InvalidInput(_)) + )); +} + +#[tokio::test] +#[ignore = "needs DATABASE_URL"] +async fn purged_history_makes_a_zone_start_again_instead_of_failing() { + let server = Server::start().await; + let email = email(); + let a = server.device("Laptop"); + sign_up(&server, &a, &email).await; + let mut engine = a.open_engine(None).expect("engine"); + set(&mut engine, ZONE, "person", "default_theme", json!("dark")); + a.sync_all(&mut engine).await.expect("pass"); + // The server purges history this device has not seen, so its change token is too old. + orbyn_core::sqlx::query("UPDATE zones SET purge_floor = 1000000 WHERE name = $1") + .bind(ZONE) + .execute(server.db.pool()) + .await + .expect("purge"); + set(&mut engine, ZONE, "person", "preview_zoom", json!(2)); + a.sync_all(&mut engine) + .await + .expect("a 410 starts the zone again"); + assert!(!engine.has_pending(ZONE)); + assert_eq!( + field(&engine, ZONE, "person", "default_theme"), + Some(json!("dark")) + ); + assert_eq!( + field(&engine, ZONE, "person", "preview_zoom"), + Some(json!(2)) + ); +} + +#[tokio::test] +#[ignore = "needs DATABASE_URL"] +async fn a_failed_network_keeps_the_work_and_says_offline() { + let server = Server::start().await; + let email = email(); + let a = server.device("Laptop"); + sign_up(&server, &a, &email).await; + let mut engine = a.open_engine(None).expect("engine"); + set(&mut engine, ZONE, "person", "default_theme", json!("dark")); + // The same identity, aimed at an address where nothing listens. + let lost: Client = DeviceClient::new( + HttpApi::new("http://127.0.0.1:1").expect("loopback"), + MemoryStore::new(), + SystemClock, + ClientConfig { + device_name: "Laptop".to_owned(), + platform: wire::DevicePlatform::Linux, + }, + ); + lost.store() + .save(&a.store().load().expect("store").expect("saved")) + .expect("save"); + assert_eq!( + lost.sync_all(&mut engine).await.err(), + Some(DeviceError::Transport) + ); + assert_eq!(lost.status().borrow().state, SyncState::Offline); + assert!(engine.has_pending(ZONE), "the work is still staged"); + // Back on the real address the next pass sends it. + a.sync_all(&mut engine).await.expect("pass"); + assert!(!engine.has_pending(ZONE)); +} + +#[tokio::test] +#[ignore = "needs DATABASE_URL"] +async fn the_notices_stream_tells_a_device_that_another_changed_a_zone() { + let server = Server::start().await; + let email = email(); + let a = server.device("Laptop"); + sign_up(&server, &a, &email).await; + let b = server.device("Desktop"); + b.sign_in(&email, &password()).await.expect("sign in"); + let mut engine_a = a.open_engine(None).expect("engine"); + let mut engine_b = b.open_engine(None).expect("engine"); + set( + &mut engine_a, + ZONE, + "person", + "default_theme", + json!("dark"), + ); + a.sync_all(&mut engine_a).await.expect("pass"); + approve( + &server, + (&token_of(&a), &mut engine_a), + (&token_of(&b), &mut engine_b, &device_id_of(&b)), + ) + .await; + let devices = a.devices().await.expect("devices"); + a.send_keys(&mut engine_a, ZONE, &devices) + .await + .expect("keys"); + b.sync_all(&mut engine_b).await.expect("pass"); + + let mut notices = a.notices(NoticeConfig::default()).expect("stream"); + assert_eq!(next(&mut notices).await, Notice::Connected); + set(&mut engine_b, ZONE, "person", "preview_zoom", json!(3)); + b.sync_all(&mut engine_b).await.expect("pass"); + let mut heard = None; + for _ in 0..10 { + if let Notice::ZoneChanged { zone, .. } = next(&mut notices).await { + heard = Some(zone); + break; + } + } + assert_eq!(heard.as_deref(), Some(ZONE)); + drop(notices); + + // A session the server no longer knows is rejected, and the stream stops. + orbyn_core::sqlx::query("UPDATE sessions SET revoked_at = now() WHERE device_id = $1::uuid") + .bind(device_id_of(&b)) + .execute(server.db.pool()) + .await + .expect("revoke"); + let mut notices = b.notices(NoticeConfig::default()).expect("stream"); + assert_eq!( + next(&mut notices).await, + Notice::Rejected { + code: "unauthorized".to_owned() + } + ); + assert_eq!( + tokio::time::timeout(Duration::from_secs(5), notices.next()) + .await + .expect("ends"), + None + ); +} + +async fn next(notices: &mut orbyn_device_client::Notices) -> Notice { + tokio::time::timeout(Duration::from_secs(10), notices.next()) + .await + .expect("a notice in time") + .expect("the stream is open") +} + +/// Two trusted devices of one account, each with an engine, and the zone's key shared. +async fn two_trusted_devices(server: &Server) -> (Client, Engine, Client, Engine) { + let email = email(); + let a = server.device("Laptop"); + sign_up(server, &a, &email).await; + let b = server.device("Desktop"); + b.sign_in(&email, &password()).await.expect("sign in"); + let mut engine_a = a.open_engine(None).expect("engine"); + let mut engine_b = b.open_engine(None).expect("engine"); + set( + &mut engine_a, + ZONE, + "person", + "default_theme", + json!("dark"), + ); + a.sync_all(&mut engine_a).await.expect("pass"); + approve( + server, + (&token_of(&a), &mut engine_a), + (&token_of(&b), &mut engine_b, &device_id_of(&b)), + ) + .await; + let devices = a.devices().await.expect("devices"); + a.send_keys(&mut engine_a, ZONE, &devices) + .await + .expect("keys"); + b.sync_all(&mut engine_b).await.expect("pass"); + (a, engine_a, b, engine_b) +} + +/// The real transport, except that just before it forwards the first upload it sends another +/// device's upload, so the server's records have changed under the first one: a real conflict. +struct RacingApi { + inner: HttpApi, + first: std::sync::Mutex>, + answer: std::sync::Mutex>, +} + +impl orbyn_device_client::DeviceApi for RacingApi { + fn base_url(&self) -> Option<&str> { + self.inner.base_url() + } + + async fn send( + &self, + request: orbyn_device_client::Request, + ) -> Result { + let rival = if request.path.ends_with("/changes") + && request.method == orbyn_device_client::Method::Post + { + self.first.lock().expect("lock").take() + } else { + None + }; + if let Some(rival) = rival { + let answer = self.inner.send(rival).await?; + *self.answer.lock().expect("lock") = Some(answer); + } + self.inner.send(request).await + } +} + +#[tokio::test] +#[ignore = "needs DATABASE_URL"] +async fn a_conflict_during_an_upload_is_merged_and_the_upload_goes_again() { + let server = Server::start().await; + let (a, mut engine_a, b, mut engine_b) = two_trusted_devices(&server).await; + // B's change is built and held back; A's client sends it just before its own upload. + set(&mut engine_b, ZONE, "person", "preview_zoom", json!(2)); + let rival = engine_b + .build_upload(ZONE) + .expect("build") + .expect("something to send"); + let racing: DeviceClient = DeviceClient::new( + RacingApi { + inner: HttpApi::new(&server.base).expect("loopback"), + first: std::sync::Mutex::new(Some(orbyn_device_client::Request { + method: orbyn_device_client::Method::Post, + path: format!("/v1/zones/{ZONE}/changes"), + bearer: Some(Zeroizing::new(token_of(&b))), + body: Some(Zeroizing::new(serde_json::to_vec(&rival).expect("json"))), + })), + answer: std::sync::Mutex::new(None), + }, + MemoryStore::new(), + SystemClock, + ClientConfig { + device_name: "Laptop".to_owned(), + platform: wire::DevicePlatform::Linux, + }, + ); + racing + .store() + .save(&a.store().load().expect("store").expect("saved")) + .expect("save"); + set( + &mut engine_a, + ZONE, + "person", + "branch_name_prefix", + json!("feat"), + ); + let report = racing + .sync_all(&mut engine_a) + .await + .expect("the conflict merges"); + assert_eq!( + report.zones[0].pushed, 1, + "A's change went up on the second try" + ); + // B's held-back upload did land, and B finishes it. + let answer = racing + .api() + .answer + .lock() + .expect("lock") + .take() + .expect("B's answer"); + assert_eq!(answer.status, 200); + let result: wire::UploadResult = serde_json::from_slice(&answer.body).expect("result"); + engine_b.upload_succeeded(ZONE, &result).expect("saved"); + a.sync_all(&mut engine_a).await.expect("pass"); + b.sync_all(&mut engine_b).await.expect("pass"); + for engine in [&engine_a, &engine_b] { + assert_eq!( + field(engine, ZONE, "person", "preview_zoom"), + Some(json!(2)) + ); + assert_eq!( + field(engine, ZONE, "person", "branch_name_prefix"), + Some(json!("feat")) + ); + } +} + +#[tokio::test] +#[ignore = "needs DATABASE_URL"] +async fn a_zone_made_after_the_approval_reaches_the_other_device_with_its_key() { + let server = Server::start().await; + let (a, mut engine_a, b, mut engine_b) = two_trusted_devices(&server).await; + // A makes a second zone: its key is sealed to B in the same pass, with no explicit call. + set( + &mut engine_a, + "engineering.profiles", + "p1", + "name", + json!("fast"), + ); + a.sync_all(&mut engine_a).await.expect("pass"); + let report = b.sync_all(&mut engine_b).await.expect("pass"); + assert!( + report + .keys_accepted + .contains(&"engineering.profiles".to_owned()), + "{report:?}" + ); + assert_eq!( + field(&engine_b, "engineering.profiles", "p1", "name"), + Some(json!("fast")) + ); +} diff --git a/docs/device-client.md b/docs/device-client.md index 65d1b86..31bc953 100644 --- a/docs/device-client.md +++ b/docs/device-client.md @@ -4,8 +4,8 @@ It ships in three pull requests: -1. **This one.** The design, the crate skeleton and sign-up, sign-in, the re-login and sign-out. -2. **Sync.** Push and pull per zone with plain records in and out (sealing, opening and zone keys inside), the notices stream and the status watch. +1. **Sign-up and sign-in** (merged). The design, the crate skeleton and sign-up, sign-in, the re-login and sign-out. +2. **Sync** (this document's second half). Push and pull per zone, zone keys in and out, the notices stream and the status watch. 3. **Approval and the label.** Approval digits, approve and deny, `devices()`, `remove_device`, and the device `app` label bound into the signed register message (engine#73). ## What the host supplies @@ -76,3 +76,29 @@ An error never carries a password, a token, a code, a key or a URL. `Request` an - `tests/client.rs` runs the client against an in-memory engine (`tests/support`) that speaks the **real OPAQUE server side** and checks the real signatures: proof of work, invite, code, trusted start, second factor, re-login after eleven minutes, account gone, rate limit, and that no request body holds the password. The same file serves a raw HTTP reply to check that redirects are not followed and `Retry-After` is read. - `tests/local_stack.rs` (ignored by default, and skipped with a printed note unless `ORBYN_LOCAL_STACK=1`, so `cargo test -- --include-ignored` in CI passes without a stack) runs sign-up, a second pending device, a wrong password, the re-login and sign-out against the real engine on `scripts/local.sh up`, using `scripts/local.sh invite-code` and `scripts/local.sh latest-code `. + +## Sync + +The records go in and out as plain fields (`orbyn_sync::Fields`: a value and a stamp for each field, merged field by field). Sealing, opening and the zone keys are inside `orbyn_client::Engine`, which the client opens and the host holds. + +- `open_engine(snapshot)` makes the engine for this device: from the host's snapshot, or empty the first time. The snapshot (`engine.snapshot()`, serialisable) holds the zone keys and the cache, so **the host seals it under the vault key** before it touches disk. A snapshot of another account or device is refused. +- The host stages with the engine (`stage`, `delete`, `tick`) and reads with it (`record`, `records`); both work offline and never block. +- `sync_all(&mut engine)` is one pass: the device list (the engine verifies the approvals itself), the zone keys other devices sent (accepted, then the records that could not be opened are tried again), the zone list compared with this device's tokens, and every zone that changed on the server or has local changes, **pulled then pushed**. A zone made here is created on the server with a new key, which is sealed to the account's other trusted devices. `sync_zone` does one zone; `send_keys` hands a zone's keys to the trusted devices (the approver calls it for every zone after an approval). +- A pull follows the change pages; a 410 `token_expired` starts the zone again (three times at most). A push uploads in batches; a 409 `record_changed` merges what the server has and goes again. A zone this device has no key for yet is reported (`waiting_for_key`), not failed: another device sends the key. `TooLarge` drops the batch the server will not take. +- A pass that cannot reach the server is `Transport`, changes nothing and leaves everything staged. Nothing here ever blocks the host's own work. + +### Status + +`status()` is a `tokio::sync::watch` of `SyncStatus { state, last_synced, last_error }`. The states are `Idle`, `Syncing`, `Offline`, and the ones that need the host to act: `SignedOut`, `ReauthenticationRequired` (call `reauthenticate`), `SessionExpired`, `AccountGone` (the app destroys key material) and `Failed`. The latest value is always there; a failure keeps the last good time. + +### Notices + +`notices(config)` opens the live stream (`/v1/sync`, a WebSocket over the same address; `ws` for a loopback `http`, `wss` otherwise, rustls with ring and the web roots). It signs in first with the session token, then tells the host: `Connected` (compare every zone now: changes made while disconnected are not replayed), `ZoneChanged { zone, token }`, `Resync`, `Disconnected`, `Rejected { code }`. A notice never carries record data; the host answers one with a pass. The library answers the server's pings as it reads; a connection that hears nothing for the idle timeout (two minutes) is dropped. The stream **reconnects by itself** with a delay that doubles from one second to one minute, with up to a fifth of jitter, and starts again from the first delay once the server says `Ready`. `Rejected { code: "unauthorized" }` stops it: the session is gone, so `reauthenticate` and open a new one. Dropping the handle closes the connection. + +### Not here yet + +The approval flow itself (digits, approve and deny, `remove_device`) and the device `app` label are the next change; the integration test does the approval over HTTP with the engine's own pairing calls until then. The account's protection chain is not signed by the client; the host's first protection setting follows the rules in [architecture.md](architecture.md). + +### Tests + +`orbyn-server/tests/device_client.rs` runs the client over a real loopback socket against the real server on Postgres: two devices, the zone key, a sealed record that the server holds only as ciphertext, a conflict merged both ways, the snapshot round trip, a pending device refused, offline work kept and sent later, and the notices stream hearing another device's change and being rejected when the session is revoked. `tests/notices.rs` serves the stream by hand to check the sign-in message, the pong, the reconnect and the stop. diff --git a/scripts/check_structure.py b/scripts/check_structure.py index 3b8b24e..782700b 100644 --- a/scripts/check_structure.py +++ b/scripts/check_structure.py @@ -50,7 +50,13 @@ "orbyn-contracts": set(), "orbyn-core": {"orbyn-auth-client", "orbyn-crypto"}, "orbyn-crypto": set(), - "orbyn-device-client": {"orbyn-auth-client", "orbyn-contracts", "orbyn-crypto"}, + "orbyn-device-client": { + "orbyn-auth-client", + "orbyn-client", + "orbyn-contracts", + "orbyn-crypto", + "orbyn-sync", + }, "orbyn-engineering": {"orbyn-sync"}, "orbyn-keyservice": {"orbyn-crypto", "orbyn-translog"}, "orbyn-keyservice-client": set(), From c78227ae0260afceb54c77fa99be6b43dac9b1c2 Mon Sep 17 00:00:00 2001 From: Tran Duc Anh Dang Date: Wed, 11 Nov 2026 22:16:28 +1100 Subject: [PATCH 2/3] fix(device-client): bound the retries, test the limits, loopback-only ws, frame limits and connect timeout --- crates/orbyn-device-client/src/notices.rs | 70 +++++-- crates/orbyn-device-client/src/sync.rs | 28 ++- crates/orbyn-device-client/tests/limits.rs | 211 ++++++++++++++++++++ crates/orbyn-device-client/tests/notices.rs | 140 +++++++++++++ docs/device-client.md | 4 +- 5 files changed, 427 insertions(+), 26 deletions(-) create mode 100644 crates/orbyn-device-client/tests/limits.rs diff --git a/crates/orbyn-device-client/src/notices.rs b/crates/orbyn-device-client/src/notices.rs index c3bcef0..0d5f921 100644 --- a/crates/orbyn-device-client/src/notices.rs +++ b/crates/orbyn-device-client/src/notices.rs @@ -12,6 +12,7 @@ use std::time::Duration; use tokio::sync::mpsc; use tokio::task::JoinHandle; use tokio_tungstenite::tungstenite::Message; +use tokio_tungstenite::tungstenite::protocol::WebSocketConfig; /// What the stream tells the host. #[derive(Clone, Debug, PartialEq, Eq)] @@ -38,6 +39,8 @@ pub struct NoticeConfig { pub max_delay: Duration, /// A connection that hears nothing for this long (the server pings) is dropped and redone. pub idle_timeout: Duration, + /// The most time to connect and finish the WebSocket handshake. + pub connect_timeout: Duration, } impl Default for NoticeConfig { @@ -46,6 +49,7 @@ impl Default for NoticeConfig { first_delay: Duration::from_secs(1), max_delay: Duration::from_secs(60), idle_timeout: Duration::from_secs(120), + connect_timeout: Duration::from_secs(10), } } } @@ -69,19 +73,24 @@ impl Drop for Notices { } } -/// `http` to `ws`, `https` to `wss`, and the sync path. +/// `https` to `wss`, and `http` to `ws` **only for a loopback IP literal**, plus the sync path. +/// Any other plain-text address is refused, whatever a transport reports as its base. pub(crate) fn websocket_address(base: &str) -> Option { - let rest = base - .strip_prefix("https://") - .map(|r| ("wss", r)) - .or_else(|| base.strip_prefix("http://").map(|r| ("ws", r)))?; - Some(format!( - "{}://{}/v1/sync", - rest.0, - rest.1.trim_end_matches('/') - )) + let (scheme, rest) = if let Some(rest) = base.strip_prefix("https://") { + ("wss", rest) + } else if let Some(rest) = base.strip_prefix("http://") + && crate::is_loopback_literal(base) + { + ("ws", rest) + } else { + return None; + }; + Some(format!("{scheme}://{}/v1/sync", rest.trim_end_matches('/'))) } +/// The most a notice frame or message may hold. Notices are a few hundred bytes. +const MAX_FRAME: usize = 16 * 1024; + /// The delay before attempt number `attempt` (0 is the first), with up to a fifth of jitter. pub(crate) fn backoff(config: &NoticeConfig, attempt: u32) -> Duration { let base = config @@ -125,13 +134,28 @@ async fn run( ) { let mut attempt = 0u32; loop { - match tokio_tungstenite::connect_async(address.as_str()).await { - Ok((mut socket, _)) => { - let auth = serde_json::to_string(&wire::SyncClientMessage::Auth { - token: token.to_string(), - }) - .unwrap_or_default(); - if socket.send(Message::text(auth)).await.is_ok() { + let limits = WebSocketConfig::default() + .max_message_size(Some(MAX_FRAME)) + .max_frame_size(Some(MAX_FRAME)); + let connected = tokio::time::timeout( + config.connect_timeout, + tokio_tungstenite::connect_async_with_config(address.as_str(), Some(limits), false), + ) + .await; + match connected { + Ok(Ok((mut socket, _))) => { + // The sign-in message is written by hand into a wiped buffer, so the token is not + // copied into a string that is dropped without being wiped. A token has only + // URL-safe characters; anything else is not sent. + let safe = token + .chars() + .all(|c| c.is_ascii_alphanumeric() || matches!(c, '-' | '_' | '.' | '~')); + let auth = zeroize::Zeroizing::new(if safe { + format!(r#"{{"type":"auth","token":"{}"}}"#, token.as_str()) + } else { + String::new() + }); + if safe && socket.send(Message::text(auth.as_str())).await.is_ok() { loop { let next = tokio::time::timeout(config.idle_timeout, socket.next()).await; let Ok(Some(Ok(message))) = next else { break }; @@ -172,7 +196,7 @@ async fn run( return; } } - Err(_) => { + Ok(Err(_)) | Err(_) => { if tx.send(Notice::Disconnected).await.is_err() { return; } @@ -198,6 +222,16 @@ mod tests { Some("ws://127.0.0.1:8080/v1/sync") ); assert_eq!(websocket_address("ftp://x"), None); + // Plain text only to a loopback IP literal: a transport that reports http://anything gets + // no stream, and a name such as localhost is not a literal. + for refused in [ + "http://example.com", + "http://localhost:8080", + "http://192.168.1.2", + "http://127.0.0.1.evil.example", + ] { + assert_eq!(websocket_address(refused), None, "{refused}"); + } } #[test] diff --git a/crates/orbyn-device-client/src/sync.rs b/crates/orbyn-device-client/src/sync.rs index 65f8c90..417a7c5 100644 --- a/crates/orbyn-device-client/src/sync.rs +++ b/crates/orbyn-device-client/src/sync.rs @@ -11,8 +11,12 @@ use crate::{Clock, DeviceApi, KeyStore}; use orbyn_client::{Applied, ClientError, Engine, RefusedEnvelope, Snapshot}; use orbyn_contracts as wire; -/// How many times one zone's upload is retried inside a pass (a conflict merges and goes again). -const MAX_UPLOAD_ROUNDS: usize = 8; +/// How many conflicts one zone's upload merges inside a pass before the pass gives up on it. +const MAX_CONFLICT_ROUNDS: usize = 8; +/// How many times one zone may start again (a 410) inside a pass, on a pull or an upload. +const MAX_RESTARTS: usize = 3; +/// A safety stop on the batches one pass sends for one zone (each holds up to 200 records). +const MAX_BATCHES: usize = 1000; /// The most envelopes one request carries. const ENVELOPES_PER_REQUEST: usize = 100; @@ -237,7 +241,7 @@ impl DeviceClient { zone: zone.to_owned(), ..ZoneReport::default() }; - for _ in 0..3 { + for _ in 0..MAX_RESTARTS { self.pull(engine, zone, &mut report.applied).await?; match self.push(engine, zone, &mut report).await { Ok(false) => return Ok(report), @@ -275,7 +279,7 @@ impl DeviceClient { // The server purged history this device has not seen: fetch from the start. 410 => { restarts += 1; - if restarts > 3 { + if restarts > MAX_RESTARTS { return Err(failure(&response)); } engine.token_expired(zone).map_err(engine_error)?; @@ -304,7 +308,8 @@ impl DeviceClient { zone: &str, report: &mut ZoneReport, ) -> Result { - for _ in 0..MAX_UPLOAD_ROUNDS { + let mut conflicts = 0; + for _ in 0..MAX_BATCHES { let Some(upload) = engine.build_upload(zone).map_err(engine_error)? else { return Ok(false); }; @@ -332,6 +337,15 @@ impl DeviceClient { .upload_conflicted(zone, &conflict, self.now_millis()) .map_err(engine_error)?; merge(&mut report.applied, applied); + conflicts += 1; + // Something keeps changing the same records: stop. Everything stays staged and + // the next pass tries again. + if conflicts >= MAX_CONFLICT_ROUNDS { + return Err(DeviceError::Server { + status: 409, + code: "record_changed".to_owned(), + }); + } continue; } if response.status == 410 { @@ -339,7 +353,9 @@ impl DeviceClient { return Ok(true); } let error = failure(&response); - // A batch the server will never take is dropped, so the next pass does not loop on it. + // The batch in flight is cleared, but its records stay dirty: the next pass builds the + // same batch and gets `TooLarge` again until the host makes a record smaller or + // deletes it. The error is how the host learns of it. if error == DeviceError::TooLarge { engine.abandon_upload(zone).map_err(engine_error)?; } diff --git a/crates/orbyn-device-client/tests/limits.rs b/crates/orbyn-device-client/tests/limits.rs new file mode 100644 index 0000000..971138c --- /dev/null +++ b/crates/orbyn-device-client/tests/limits.rs @@ -0,0 +1,211 @@ +//! The limits inside one pass: a server that always answers 409 `record_changed`, or always 410 +//! `token_expired`, must not hold a pass in a loop. The pass stops, says why and keeps the work +//! staged. The transport is scripted; no engine is needed. + +use base64::Engine as _; +use orbyn_contracts as wire; +use orbyn_crypto::DeviceKeys; +use orbyn_device_client::{ + ClientConfig, DeviceApi, DeviceClient, DeviceError, KeyStore, MemoryStore, Method, Request, + Response, StoredIdentity, StoredSession, SystemClock, +}; +use orbyn_sync::Fields; +use serde_json::{Value, json}; +use std::sync::Mutex; +use std::sync::atomic::{AtomicUsize, Ordering}; +use std::time::Duration; +use zeroize::Zeroizing; + +const ZONE: &str = "engineering.settings"; + +#[derive(Clone, Copy, PartialEq, Eq)] +enum Upload { + AlwaysConflicts, + AlwaysExpired, +} + +#[derive(Clone, Copy, PartialEq, Eq)] +enum Fetch { + Empty, + AlwaysExpired, +} + +struct Script { + devices: Value, + upload: Upload, + fetch: Fetch, + gets: AtomicUsize, + posts: AtomicUsize, + seen: Mutex>, +} + +fn answer(status: u16, body: Value) -> Result { + Ok(Response { + status, + retry_after: None, + body: Zeroizing::new(serde_json::to_vec(&body).expect("json")), + }) +} + +impl DeviceApi for Script { + async fn send(&self, request: Request) -> Result { + self.seen + .lock() + .expect("lock") + .push(format!("{:?} {}", request.method, request.path)); + let path = request.path.as_str(); + match (request.method, path) { + (Method::Get, "/v1/devices") => answer(200, self.devices.clone()), + (Method::Get, "/v1/keychain/envelopes") => answer(200, json!({"envelopes": []})), + (Method::Get, "/v1/zones") => answer( + 200, + json!({"zones": [{"zone": ZONE, "token": "t0", "deleted": false, "created_at": "2026-01-01T00:00:00Z"}], "next": null}), + ), + (Method::Get, p) if p.contains("/changes") => { + self.gets.fetch_add(1, Ordering::SeqCst); + match self.fetch { + Fetch::Empty => answer( + 200, + json!({"records": [], "token": "t1", "more": false, "zone_deleted": false}), + ), + Fetch::AlwaysExpired => { + answer(410, json!({"code": "token_expired", "message": "x"})) + } + } + } + (Method::Post, p) if p.ends_with("/changes") => { + self.posts.fetch_add(1, Ordering::SeqCst); + match self.upload { + Upload::AlwaysConflicts => answer( + 409, + json!({"code": "record_changed", "message": "x", "conflicts": []}), + ), + Upload::AlwaysExpired => { + answer(410, json!({"code": "token_expired", "message": "x"})) + } + } + } + _ => answer(404, json!({"code": "not_found", "message": "x"})), + } + } +} + +type Client = DeviceClient; + +fn client(upload: Upload, fetch: Fetch) -> Client { + let keys = DeviceKeys::generate().expect("keys"); + let public = keys.public(); + let b64 = base64::engine::general_purpose::URL_SAFE_NO_PAD; + let device_id = uuid::Uuid::new_v4().to_string(); + let account_id = uuid::Uuid::new_v4().to_string(); + let devices = json!({"devices": [{ + "id": device_id, "name": "Laptop", "platform": "linux", "state": "trusted", + "signing_key": b64.encode(public.signing), "agreement_key": b64.encode(public.agreement), + "created_at": "2026-01-01T00:00:00Z", "approved_by": null, "approval_signature": null, + "current": true, + }], "removals": []}); + let store = MemoryStore::new(); + store + .save(&StoredIdentity { + device_keys: keys.to_bytes(), + account_id, + device_id, + email: "a@example.test".to_owned(), + session: Some(StoredSession { + token: Zeroizing::new("token".to_owned()), + expires_at: "2027-01-01T00:00:00Z".to_owned(), + }), + vault_key: Zeroizing::new([5u8; 32]), + }) + .expect("save"); + DeviceClient::new( + Script { + devices, + upload, + fetch, + gets: AtomicUsize::new(0), + posts: AtomicUsize::new(0), + seen: Mutex::new(Vec::new()), + }, + store, + SystemClock, + ClientConfig { + device_name: "Laptop".to_owned(), + platform: wire::DevicePlatform::Linux, + }, + ) +} + +/// An engine with a key for the zone and one record waiting to go up. +fn staged(client: &Client) -> orbyn_client::Engine { + let mut engine = client.open_engine(None).expect("engine"); + engine.create_zone_key(ZONE).expect("key"); + let stamp = engine.tick(1_800_000_000_000); + let mut fields = Fields::default(); + fields.set("default_theme", json!("dark"), stamp); + engine + .stage(ZONE, "person_settings", "person", &fields) + .expect("stage"); + engine +} + +/// A pass that is stuck in a loop would hang the test, so every pass has a deadline. +async fn pass(client: &Client, engine: &mut orbyn_client::Engine) -> Result<(), DeviceError> { + tokio::time::timeout(Duration::from_secs(10), client.sync_all(engine)) + .await + .expect("the pass ended (it is not stuck in a loop)") + .map(|_| ()) +} + +#[tokio::test(flavor = "multi_thread")] +async fn a_server_that_always_answers_a_conflict_stops_the_pass_after_eight_rounds() { + let client = client(Upload::AlwaysConflicts, Fetch::Empty); + let mut engine = staged(&client); + let error = pass(&client, &mut engine).await.err(); + assert_eq!( + error, + Some(DeviceError::Server { + status: 409, + code: "record_changed".to_owned() + }) + ); + assert_eq!( + client.api().posts.load(Ordering::SeqCst), + 8, + "eight uploads, no more" + ); + assert!(engine.has_pending(ZONE), "the work is still staged"); +} + +#[tokio::test(flavor = "multi_thread")] +async fn a_server_that_always_says_the_token_expired_on_a_pull_stops_after_three_restarts() { + let client = client(Upload::AlwaysConflicts, Fetch::AlwaysExpired); + let mut engine = staged(&client); + let error = pass(&client, &mut engine).await.err(); + assert!( + matches!(error, Some(DeviceError::Server { status: 410, .. })), + "{error:?}" + ); + // The first fetch and three restarts. + assert_eq!(client.api().gets.load(Ordering::SeqCst), 4); + assert_eq!( + client.api().posts.load(Ordering::SeqCst), + 0, + "nothing was uploaded" + ); + assert!(engine.has_pending(ZONE)); +} + +#[tokio::test(flavor = "multi_thread")] +async fn a_server_that_always_says_the_token_expired_on_an_upload_stops_after_three_rounds() { + let client = client(Upload::AlwaysExpired, Fetch::Empty); + let mut engine = staged(&client); + let error = pass(&client, &mut engine).await.err(); + assert!( + matches!(error, Some(DeviceError::Server { status: 410, .. })), + "{error:?}" + ); + assert_eq!(client.api().posts.load(Ordering::SeqCst), 3); + assert_eq!(client.api().gets.load(Ordering::SeqCst), 3); + assert!(engine.has_pending(ZONE)); +} diff --git a/crates/orbyn-device-client/tests/notices.rs b/crates/orbyn-device-client/tests/notices.rs index 3f560ae..c546b6d 100644 --- a/crates/orbyn-device-client/tests/notices.rs +++ b/crates/orbyn-device-client/tests/notices.rs @@ -43,6 +43,7 @@ fn quick() -> NoticeConfig { first_delay: Duration::from_millis(20), max_delay: Duration::from_millis(100), idle_timeout: Duration::from_secs(5), + connect_timeout: Duration::from_secs(2), } } @@ -177,3 +178,142 @@ async fn without_a_session_or_an_address_there_is_no_stream() { Some(orbyn_device_client::DeviceError::NotSignedIn) ); } + +#[tokio::test(flavor = "multi_thread")] +async fn a_connection_that_reaches_ready_starts_the_delay_over() { + // The server accepts three connections, each says ready and drops. With the delay reset by + // each ready, the pause before the third is as short as the pause before the second (about + // 200 ms); without it, the second pause has doubled (about 400 ms). + let listener = TcpListener::bind("127.0.0.1:0").await.expect("bind"); + let address = format!("http://{}", listener.local_addr().expect("addr")); + let server = tokio::spawn(async move { + let mut at = Vec::new(); + for _ in 0..3 { + let (stream, _) = listener.accept().await.expect("accept"); + at.push(std::time::Instant::now()); + let mut socket = tokio_tungstenite::accept_async(stream) + .await + .expect("handshake"); + let _ = socket.next().await; + socket + .send(Message::text(r#"{"type":"ready","device_id":"device"}"#)) + .await + .expect("ready"); + tokio::time::sleep(Duration::from_millis(30)).await; + drop(socket); + } + at + }); + let config = NoticeConfig { + first_delay: Duration::from_millis(200), + max_delay: Duration::from_secs(5), + idle_timeout: Duration::from_secs(5), + connect_timeout: Duration::from_secs(2), + }; + let client = client_for(&address).await; + let _notices = client.notices(config).expect("stream"); + let at = server.await.expect("server"); + let first = at[1].duration_since(at[0]); + let second = at[2].duration_since(at[1]); + assert!(first >= Duration::from_millis(150), "{first:?}"); + assert!( + second < Duration::from_millis(330), + "the delay was not reset: {second:?}" + ); +} + +#[tokio::test(flavor = "multi_thread")] +async fn a_server_that_never_answers_the_handshake_is_given_up_on() { + // Accepts the TCP connection and says nothing: without a connect timeout the stream hangs. + let listener = TcpListener::bind("127.0.0.1:0").await.expect("bind"); + let address = format!("http://{}", listener.local_addr().expect("addr")); + tokio::spawn(async move { + let mut held = Vec::new(); + while let Ok((stream, _)) = listener.accept().await { + held.push(stream); + } + }); + let config = NoticeConfig { + connect_timeout: Duration::from_millis(150), + ..quick() + }; + let client = client_for(&address).await; + let mut notices = client.notices(config).expect("stream"); + assert_eq!(next(&mut notices).await, Notice::Disconnected); +} + +#[tokio::test(flavor = "multi_thread")] +async fn a_transport_that_reports_a_plain_text_address_off_loopback_gets_no_stream() { + use orbyn_device_client::{DeviceApi, DeviceError, Request, Response}; + struct Remote; + impl DeviceApi for Remote { + async fn send(&self, _: Request) -> Result { + Err(DeviceError::Transport) + } + fn base_url(&self) -> Option<&str> { + Some("http://example.com") + } + } + let store = MemoryStore::new(); + store + .save(&StoredIdentity { + device_keys: Zeroizing::new([3u8; 64]), + account_id: "a".to_owned(), + device_id: "d".to_owned(), + email: "a@example.test".to_owned(), + session: Some(StoredSession { + token: Zeroizing::new("t".to_owned()), + expires_at: String::new(), + }), + vault_key: Zeroizing::new([4u8; 32]), + }) + .expect("save"); + let client = DeviceClient::new( + Remote, + store, + SystemClock, + ClientConfig { + device_name: "x".to_owned(), + platform: DevicePlatform::Linux, + }, + ); + assert!(matches!( + client.notices(quick()), + Err(DeviceError::InvalidInput(_)) + )); +} + +#[tokio::test(flavor = "multi_thread")] +async fn an_oversized_message_ends_the_connection_instead_of_being_read() { + let listener = TcpListener::bind("127.0.0.1:0").await.expect("bind"); + let address = format!("http://{}", listener.local_addr().expect("addr")); + tokio::spawn(async move { + let (stream, _) = listener.accept().await.expect("accept"); + let mut socket = tokio_tungstenite::accept_async(stream) + .await + .expect("handshake"); + let _ = socket.next().await; + socket + .send(Message::text(r#"{"type":"ready","device_id":"device"}"#)) + .await + .expect("ready"); + let big = format!( + r#"{{"type":"zone_changed","zone":"{}","token":"t"}}"#, + "z".repeat(200_000) + ); + let _ = socket.send(Message::text(big)).await; + tokio::time::sleep(Duration::from_secs(1)).await; + }); + let config = NoticeConfig { + first_delay: Duration::from_secs(30), + ..quick() + }; + let client = client_for(&address).await; + let mut notices = client.notices(config).expect("stream"); + assert_eq!(next(&mut notices).await, Notice::Connected); + assert_eq!( + next(&mut notices).await, + Notice::Disconnected, + "no notice came from the huge frame" + ); +} diff --git a/docs/device-client.md b/docs/device-client.md index 31bc953..34822d6 100644 --- a/docs/device-client.md +++ b/docs/device-client.md @@ -84,7 +84,7 @@ The records go in and out as plain fields (`orbyn_sync::Fields`: a value and a s - `open_engine(snapshot)` makes the engine for this device: from the host's snapshot, or empty the first time. The snapshot (`engine.snapshot()`, serialisable) holds the zone keys and the cache, so **the host seals it under the vault key** before it touches disk. A snapshot of another account or device is refused. - The host stages with the engine (`stage`, `delete`, `tick`) and reads with it (`record`, `records`); both work offline and never block. - `sync_all(&mut engine)` is one pass: the device list (the engine verifies the approvals itself), the zone keys other devices sent (accepted, then the records that could not be opened are tried again), the zone list compared with this device's tokens, and every zone that changed on the server or has local changes, **pulled then pushed**. A zone made here is created on the server with a new key, which is sealed to the account's other trusted devices. `sync_zone` does one zone; `send_keys` hands a zone's keys to the trusted devices (the approver calls it for every zone after an approval). -- A pull follows the change pages; a 410 `token_expired` starts the zone again (three times at most). A push uploads in batches; a 409 `record_changed` merges what the server has and goes again. A zone this device has no key for yet is reported (`waiting_for_key`), not failed: another device sends the key. `TooLarge` drops the batch the server will not take. +- A pull follows the change pages; a 410 `token_expired` (on a pull or an upload) starts the zone again, **three times at most**, then the pass fails with `Server { 410, token_expired }`. A push uploads in batches; a 409 `record_changed` merges what the server has and goes again, **eight conflicts at most**, then the pass fails with `Server { 409, record_changed }` and everything stays staged for the next pass. A zone this device has no key for yet is reported (`waiting_for_key`), not failed: another device sends the key. `TooLarge` clears the batch in flight but its records stay dirty, so the next pass builds the same batch and gets `TooLarge` again until the host makes a record smaller or deletes it; the error is how the host learns of it. - A pass that cannot reach the server is `Transport`, changes nothing and leaves everything staged. Nothing here ever blocks the host's own work. ### Status @@ -93,7 +93,7 @@ The records go in and out as plain fields (`orbyn_sync::Fields`: a value and a s ### Notices -`notices(config)` opens the live stream (`/v1/sync`, a WebSocket over the same address; `ws` for a loopback `http`, `wss` otherwise, rustls with ring and the web roots). It signs in first with the session token, then tells the host: `Connected` (compare every zone now: changes made while disconnected are not replayed), `ZoneChanged { zone, token }`, `Resync`, `Disconnected`, `Rejected { code }`. A notice never carries record data; the host answers one with a pass. The library answers the server's pings as it reads; a connection that hears nothing for the idle timeout (two minutes) is dropped. The stream **reconnects by itself** with a delay that doubles from one second to one minute, with up to a fifth of jitter, and starts again from the first delay once the server says `Ready`. `Rejected { code: "unauthorized" }` stops it: the session is gone, so `reauthenticate` and open a new one. Dropping the handle closes the connection. +`notices(config)` opens the live stream (`/v1/sync`, a WebSocket over the same address: `wss` for `https`, and `ws` **only for a loopback IP literal**, so a transport that reports some other `http://` address gets no stream; rustls with ring and the web roots). Frames and messages are capped at 16 KiB (a notice is a few hundred bytes), the connect and handshake have a timeout (ten seconds), and the token is written into a wiped buffer for the sign-in message. It signs in first with the session token, then tells the host: `Connected` (compare every zone now: changes made while disconnected are not replayed), `ZoneChanged { zone, token }`, `Resync`, `Disconnected`, `Rejected { code }`. A notice never carries record data; the host answers one with a pass. The library answers the server's pings as it reads; a connection that hears nothing for the idle timeout (two minutes) is dropped. The stream **reconnects by itself** with a delay that doubles from one second to one minute, with up to a fifth of jitter, and starts again from the first delay once the server says `Ready`. `Rejected { code: "unauthorized" }` stops it: the session is gone, so `reauthenticate` and open a new one. Dropping the handle closes the connection. ### Not here yet From ef967426470cfb45f0a39d7609a328701c78dce5 Mon Sep 17 00:00:00 2001 From: Tran Duc Anh Dang Date: Wed, 11 Nov 2026 22:47:24 +1100 Subject: [PATCH 3/3] test(device-client): fix a clippy lint and let the scripted transport yield --- crates/orbyn-device-client/tests/limits.rs | 2 ++ crates/orbyn-device-client/tests/notices.rs | 2 +- 2 files changed, 3 insertions(+), 1 deletion(-) diff --git a/crates/orbyn-device-client/tests/limits.rs b/crates/orbyn-device-client/tests/limits.rs index 971138c..d488f39 100644 --- a/crates/orbyn-device-client/tests/limits.rs +++ b/crates/orbyn-device-client/tests/limits.rs @@ -49,6 +49,8 @@ fn answer(status: u16, body: Value) -> Result { impl DeviceApi for Script { async fn send(&self, request: Request) -> Result { + // Lets the runtime run its timers, so a pass stuck in a loop is stopped by the deadline. + tokio::task::yield_now().await; self.seen .lock() .expect("lock") diff --git a/crates/orbyn-device-client/tests/notices.rs b/crates/orbyn-device-client/tests/notices.rs index c546b6d..8ea5d48 100644 --- a/crates/orbyn-device-client/tests/notices.rs +++ b/crates/orbyn-device-client/tests/notices.rs @@ -87,7 +87,7 @@ async fn it_signs_in_first_answers_pings_and_reconnects_after_a_drop() { }) .await .expect("a pong in time"); - assert_eq!(pong.as_deref().map(|d| &d[..]), Some(&b"are you there"[..])); + assert_eq!(pong.as_deref(), Some(&b"are you there"[..])); socket .send(Message::text( r#"{"type":"zone_changed","zone":"engineering.settings","token":"t1"}"#,