diff --git a/CHANGELOG.md b/CHANGELOG.md index c901cd13446..41499f171dc 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -6,6 +6,10 @@ - Timestamps outside the range a JS `Date` can represent are now reported as 0 wherever they enter or leave the store: on the metadata an application supplies to `add_with_meta()`, on every read out of the local database, and on incoming sync payloads, similar to what Desktop does. ([bug 2066257](https://bugzilla.mozilla.org/show_bug.cgi?id=2066257)) +### Remote Settings + +- Keep synchronization locked until collection signature verification finishes. + [Full Changelog](In progress) ## ✨ What's Changed ✨ diff --git a/components/remote_settings/src/client.rs b/components/remote_settings/src/client.rs index 84398b22ed9..dc42f38bfb7 100644 --- a/components/remote_settings/src/client.rs +++ b/components/remote_settings/src/client.rs @@ -360,8 +360,7 @@ impl RemoteSettingsClient { /// 1. Fetches the last modified timestamp of the collection from local storage. /// 2. Fetches the changeset from the remote server based on the last modified timestamp. /// 3. Inserts the fetched changeset into local storage. - fn perform_sync_operation(&self) -> Result<()> { - let mut inner = self.lock_inner()?; + fn perform_sync_operation(&self, inner: &mut RemoteSettingsClientInner) -> Result<()> { let collection_url = inner.api_client.collection_url(); let timestamp = inner.storage.get_last_modified_timestamp(&collection_url)?; let changeset = inner.api_client.fetch_changeset(timestamp)?; @@ -379,23 +378,29 @@ impl RemoteSettingsClient { } pub fn sync(&self) -> Result<()> { + // Keep the client locked until the fetched collection has either been verified or reset. + // Otherwise, another thread can observe records after they are inserted but before their + // content signature is checked. + let mut inner = self.lock_inner()?; // First attempt - self.perform_sync_operation()?; + self.perform_sync_operation(&mut inner)?; // Verify that inserted data has valid signature - if self.verify_signature().is_err() { + if self.verify_signature(&mut inner).is_err() { debug!( "{0}: signature verification failed. Reset and retry.", self.collection_name ); // Retry with packaged dataset as base - self.reset_storage()?; - self.perform_sync_operation()?; + self.reset_storage_inner(&mut inner)?; + self.perform_sync_operation(&mut inner)?; // Verify signature again - self.verify_signature().inspect_err(|_| { + let verification_result = self.verify_signature(&mut inner); + if verification_result.is_err() { // And reset with packaged data if it fails again. - self.reset_storage() + self.reset_storage_inner(&mut inner) .expect("Failed to reset storage after verification failure"); - })?; + } + verification_result?; } trace!("{0}: sync done.", self.collection_name); Ok(()) @@ -407,8 +412,12 @@ impl RemoteSettingsClient { } pub fn reset_storage(&self) -> Result<()> { - trace!("{0}: reset local storage.", self.collection_name); let mut inner = self.lock_inner()?; + self.reset_storage_inner(&mut inner) + } + + fn reset_storage_inner(&self, inner: &mut RemoteSettingsClientInner) -> Result<()> { + trace!("{0}: reset local storage.", self.collection_name); let collection_url = inner.api_client.collection_url(); // Clear existing storage inner.storage.empty()?; @@ -432,14 +441,13 @@ impl RemoteSettingsClient { } #[cfg(not(feature = "signatures"))] - fn verify_signature(&self) -> Result<()> { + fn verify_signature(&self, _inner: &mut RemoteSettingsClientInner) -> Result<()> { debug!("{0}: signature verification skipped.", self.collection_name); Ok(()) } #[cfg(feature = "signatures")] - fn verify_signature(&self) -> Result<()> { - let mut inner = self.lock_inner()?; + fn verify_signature(&self, inner: &mut RemoteSettingsClientInner) -> Result<()> { let collection_url = inner.api_client.collection_url(); let timestamp = inner.storage.get_last_modified_timestamp(&collection_url)?; let records = inner.storage.get_records(&collection_url)?; @@ -1245,6 +1253,9 @@ mod jexl_tests { #[cfg(test)] mod test_signatures { use core::assert_eq; + use std::sync::{mpsc, Arc}; + use std::thread; + use std::time::Duration; use crate::RemoteSettingsContext; @@ -1705,6 +1716,104 @@ IKdcFKAt3fFrpyMhlfIKkLfmm0iDjmfmIXbDGBJw9SE= Ok(()) } + + #[test] + fn test_records_are_not_visible_until_sync_finishes() -> Result<()> { + ensure_initialized(); + + let attachment = b"test attachment".to_vec(); + let attachment_location = "test/attachment.bin"; + let record = RemoteSettingsRecord { + id: "test-record".into(), + last_modified: 100, + deleted: false, + attachment: Some(Attachment { + filename: "attachment.bin".into(), + mimetype: "application/octet-stream".into(), + location: attachment_location.into(), + hash: format!("{:x}", Sha256::digest(&attachment)), + size: attachment.len() as u64, + }), + fields: serde_json::json!({ + "name": "test-record", + "version": "1.0" + }) + .as_object() + .unwrap() + .clone(), + }; + let changeset = ChangesetResponse { + changes: vec![record], + timestamp: 100, + metadata: CollectionMetadata { + bucket: "main".into(), + signatures: vec![], + }, + }; + + let (fetch_entered_tx, fetch_entered_rx) = mpsc::channel(); + let (release_fetch_tx, release_fetch_rx) = mpsc::channel(); + let mut first_fetch = true; + let mut api_client = MockApiClient::new(); + api_client + .expect_collection_url() + .returning(|| "http://server/test-collection".into()); + api_client.expect_is_prod_server().returning(|| Ok(false)); + let expected_attachment = attachment.clone(); + api_client + .expect_fetch_attachment() + .with(mockall::predicate::eq(attachment_location)) + .returning(move |_| Ok(expected_attachment.clone())); + api_client + .expect_fetch_changeset() + .times(2) + .returning(move |_| { + if first_fetch { + first_fetch = false; + fetch_entered_tx.send(()).unwrap(); + release_fetch_rx.recv().unwrap(); + } + Ok(changeset.clone()) + }); + + let client = Arc::new(RemoteSettingsClient::new_from_parts( + "test-collection".into(), + Storage::new(":memory:".into()), + JexlFilter::new(None), + api_client, + )); + + let sync_client = Arc::clone(&client); + let sync_thread = thread::spawn(move || sync_client.sync()); + fetch_entered_rx.recv().unwrap(); + + let (reader_started_tx, reader_started_rx) = mpsc::channel(); + let reader_client = Arc::clone(&client); + let reader_thread = thread::spawn(move || { + reader_started_tx.send(()).unwrap(); + let records = reader_client.get_records(false)?; + records + .and_then(|records| records.into_iter().next()) + .map(|record| reader_client.get_attachment(&record)) + .transpose() + }); + reader_started_rx.recv().unwrap(); + + // Keep the sync lock held long enough for the waiting reader to queue. parking_lot + // hands a contended mutex to a waiter after a long critical section. + thread::sleep(Duration::from_millis(25)); + release_fetch_tx.send(()).unwrap(); + + let result = reader_thread.join().unwrap()?; + assert_ne!( + result, + Some(attachment), + "records should not be visible until synchronization succeeds" + ); + assert!(sync_thread.join().unwrap().is_err()); + + Ok(()) + } } #[cfg(test)]