diff --git a/README.md b/README.md index 673a8fe..369d530 100644 --- a/README.md +++ b/README.md @@ -534,3 +534,34 @@ tuple. The preserved server candidate accepts one proof per unique target; that alone cannot recheck a second provider which has since remapped elsewhere. Retained domain labels are necessary, but enforcing that condition remains unresolved before any future activation. + +### Retained migration outcome observations (Rust source only) + +`BackupReadSession::reconcile_migration` compares an opaque retained proposal with +fresh status reads through both original app and LDK clients, including when their +original physical tuple was shared. It checks each original principal/store and the +exact UUID, all-tuples digest, archive presence/hash, initial encrypted-head hash and +complete component set. Component wire order is irrelevant. It compares the immutable +initial receipt, never a later mutable head or a substitute wallet export. + +Only two matching committed receipts return `ObservedCommitted`; two expected legacy +receipts return `ObservedUncommitted`. Neither is an atomic snapshot or write/replay +permit. Read failures, malformed/unsupported responses, remapped providers, mismatched +receipts, mixed outcomes, superseding reads and owner revocation remain errors. +Each read obtains a new token with its captured original signer, even when the ordinary +provider holds a valid cached token. Fresh authentication failure cannot fall back to +that token. The ten-second deadline includes authentication; the VSS storage POST has +no automatic retry or redirect. The inherited LNURL GETs retain their own redirect +policy and uncapped response parsing within that deadline. Cancellation +abandons pending reads; revocation discards their results. Another session, even with +the same seed, cannot adopt the handle. Starting reconciliation invalidates preparation +access and does not restore it. The retained immutable content is used only for comparison. + +This is in-memory native source, without new UniFFI exports or durable restart adoption. +Sequential provider reads do not repair the coalesced-provider atomic enforcement gap +above. Server deployment, app lifecycle admission, legacy-reader exclusion and complete +coherent capture/restore remain separate prerequisites; no migration is dispatched. + +The candidate server may initialize an empty legacy policy row during a status read. +This client invokes only status and authentication routes; it does not assert that +server implementation performs zero database writes. diff --git a/src/backup_session.rs b/src/backup_session.rs index 7d563b0..2cd74a7 100644 --- a/src/backup_session.rs +++ b/src/backup_session.rs @@ -6,8 +6,9 @@ use tokio::sync::watch; mod preparation; pub use preparation::{ - MigrationHeadInput, MigrationPreparation, MigrationPreparationError, PreparedMigration, - PreparedMigrationContent, PreparedMigrationHead, PreparedMigrationStore, + MigrationHeadInput, MigrationPreparation, MigrationPreparationError, MigrationReconciliation, + MigrationReconciliationError, PreparedMigration, PreparedMigrationContent, + PreparedMigrationHead, PreparedMigrationStore, }; /// One immutable storage/auth configuration for both native derivation domains. @@ -175,18 +176,25 @@ impl BackupReadSession { &self, domain: BackupDomain, ) -> Result { - let mut cancellation = self.revoked.subscribe(); - self.ensure_active()?; - // A pending, failed or abandoned refresh must not leave earlier policy - // observations admissible for a new proposal. - let revision = Arc::new(()); - self.migration_revisions[domain.index()].send_replace(Arc::clone(&revision)); - let read = async { + self.migration_observation(domain, async { match domain { BackupDomain::App => self.app.read_migration_state().await, BackupDomain::Ldk => self.ldk.read_migration_state().await, } - }; + }) + .await + } + + async fn migration_observation( + &self, + domain: BackupDomain, + read: impl std::future::Future>, + ) -> Result { + let mut cancellation = self.revoked.subscribe(); + self.ensure_active()?; + // Pending, failed and abandoned reads invalidate earlier preparation. + let revision = Arc::new(()); + self.migration_revisions[domain.index()].send_replace(Arc::clone(&revision)); let state = tokio::select! { biased; _ = cancellation.changed() => return Err(BackupReadError::Revoked), diff --git a/src/backup_session/preparation.rs b/src/backup_session/preparation.rs index 23a40e7..2735097 100644 --- a/src/backup_session/preparation.rs +++ b/src/backup_session/preparation.rs @@ -6,6 +6,8 @@ use bitcoin::hashes::{sha256, Hash}; use prost::Message; mod canonical; +mod reconciliation; +pub use reconciliation::{MigrationReconciliation, MigrationReconciliationError}; const MAX_HEAD_BYTES: usize = 10 * 1024 * 1024; diff --git a/src/backup_session/preparation/reconciliation.rs b/src/backup_session/preparation/reconciliation.rs new file mode 100644 index 0000000..97133f1 --- /dev/null +++ b/src/backup_session/preparation/reconciliation.rs @@ -0,0 +1,167 @@ +//! Read-only comparison of a retained attempt with fresh original-provider receipts. + +use super::*; +use crate::CommittedMigration; + +/// Observations from two sequential reads, never an atomic snapshot or permission +/// to activate/replay a migration. Even `ObservedUncommitted` may already be stale. +#[derive(Debug, PartialEq, Eq)] +pub enum MigrationReconciliation { + ObservedCommitted, + ObservedUncommitted, +} + +#[derive(Debug, thiserror::Error)] +pub enum MigrationReconciliationError { + #[error(transparent)] + Ownership(#[from] BackupReadError), + #[error("Migration status could not be read through the original {domain:?} provider")] + Read { + domain: BackupDomain, + #[source] + source: BackupReadError, + }, + #[error("The original {0:?} provider now identifies a different participant")] + ParticipantChanged(BackupDomain), + #[error("The {0:?} receipt does not match the retained migration attempt")] + ReceiptMismatch(BackupDomain), + #[error("The original participants report mixed migration outcomes")] + MixedOutcome, + #[error("A newer status read superseded reconciliation")] + SupersededObservation, +} + +impl BackupReadSession { + /// Reads each original credential domain once, including coalesced domains. + /// Each read obtains fresh credentials with its captured original signer and + /// includes authentication in the ten-second deadline. No storage retry, + /// redirect, upload, deletion, activation or current-head read occurs here. + /// The inherited LNURL authentication GETs retain their redirect/body behavior. + /// + /// The opaque attempt must belong to this still-active in-memory owner. Its + /// preparation may already be stale: this method only compares immutable + /// content, and never restores preparation access. It supplies no persisted + /// UUID adoption or restart authority. Revocation discards pending results; + /// dropping this future abandons its reads without scheduling continuation. + pub async fn reconcile_migration( + &self, + prepared: &PreparedMigration, + ) -> Result { + if !Arc::ptr_eq(&self.revoked, &prepared.owner) { + return Err(BackupReadError::DifferentOwner.into()); + } + self.ensure_active()?; + let app = self.reconciliation_observation(BackupDomain::App).await?; + let ldk = self.reconciliation_observation(BackupDomain::Ldk).await?; + let app_committed = matches_receipt(&prepared.content, &app)?; + let ldk_committed = matches_receipt(&prepared.content, &ldk)?; + self.ensure_reconciliation_current(&[app, ldk])?; + match (app_committed, ldk_committed) { + (true, true) => Ok(MigrationReconciliation::ObservedCommitted), + (false, false) => Ok(MigrationReconciliation::ObservedUncommitted), + _ => Err(MigrationReconciliationError::MixedOutcome), + } + } + + async fn reconciliation_observation( + &self, + domain: BackupDomain, + ) -> Result { + self.migration_observation(domain, async { + match domain { + BackupDomain::App => self.app.read_fresh_migration_state().await, + BackupDomain::Ldk => self.ldk.read_fresh_migration_state().await, + } + }) + .await + .map_err(|source| { + if matches!(source, BackupReadError::Revoked) { + MigrationReconciliationError::Ownership(source) + } else { + MigrationReconciliationError::Read { domain, source } + } + }) + } + + fn ensure_reconciliation_current( + &self, + observations: &[MigrationObservation; 2], + ) -> Result<(), MigrationReconciliationError> { + self.ensure_active()?; + for observation in observations { + if !Arc::ptr_eq( + &self.migration_revisions[observation.domain.index()].borrow(), + &observation.revision, + ) { + return Err(MigrationReconciliationError::SupersededObservation); + } + } + Ok(()) + } +} + +fn matches_receipt( + prepared: &PreparedMigrationContent, + observation: &MigrationObservation, +) -> Result { + let domain = observation.domain; + let expected = &prepared.stores[store_index(&prepared.stores, domain)]; + let actual = &observation.state; + if actual.effective_principal != expected.effective_principal + || actual.store_id != expected.store_id + { + return Err(MigrationReconciliationError::ParticipantChanged(domain)); + } + match &actual.policy { + MigrationPolicy::Legacy => Ok(false), + MigrationPolicy::Committed(receipt) if receipt_matches(prepared, expected, receipt) => { + Ok(true) + } + MigrationPolicy::Committed(_) => Err(MigrationReconciliationError::ReceiptMismatch(domain)), + } +} + +fn receipt_matches( + prepared: &PreparedMigrationContent, + expected: &PreparedMigrationStore, + actual: &CommittedMigration, +) -> bool { + actual.migration_id == prepared.migration_id + && actual.proposal_sha256 == prepared.proposal_sha256 + && archives_match(expected.archive.as_ref(), actual.archive.as_ref()) + && heads_match(expected.initial_head.as_ref(), actual.initial_head.as_ref()) + && objects_match(&expected.objects, &actual.objects) +} + +fn archives_match(expected: Option<&MigrationArchive>, actual: Option<&MigrationArchive>) -> bool { + match (expected, actual) { + (None, None) => true, + (Some(expected), Some(actual)) => { + expected.key == actual.key && expected.ciphertext_sha256 == actual.ciphertext_sha256 + } + _ => false, + } +} + +fn heads_match(expected: Option<&PreparedMigrationHead>, actual: Option<&MigrationObject>) -> bool { + match (expected, actual) { + (None, None) => true, + (Some(expected), Some(actual)) => { + expected.key == actual.key + && sha256::Hash::hash(&expected.encrypted_value).to_byte_array() + == actual.ciphertext_sha256 + } + _ => false, + } +} + +fn objects_match(expected: &[MigrationObject], actual: &[MigrationObject]) -> bool { + // Both sets were validated as unique. Wire order is not part of the receipt. + expected.len() == actual.len() + && expected.iter().all(|object| { + actual.iter().any(|candidate| { + object.key == candidate.key + && object.ciphertext_sha256 == candidate.ciphertext_sha256 + }) + }) +} diff --git a/src/backup_session/tests.rs b/src/backup_session/tests.rs index 8292e86..ba163bb 100644 --- a/src/backup_session/tests.rs +++ b/src/backup_session/tests.rs @@ -1,5 +1,6 @@ use super::*; mod preparation; +mod reconciliation; use bitcoin::{ hashes::{sha256, Hash}, hex::FromHex, @@ -7,7 +8,7 @@ use bitcoin::{ }; use prost::Message as ProstMessage; use std::{ - collections::HashMap, + collections::{HashMap, VecDeque}, sync::atomic::{AtomicBool, AtomicUsize, Ordering}, }; use tokio::{ @@ -21,6 +22,7 @@ use vss_client_ng::types::{ }; type StatusResponse = Arc)>>>; +type StatusQueue = Arc)>>>; struct Fixture { url: String, @@ -28,8 +30,15 @@ struct Fixture { auth_keys: Arc>>, requests: Arc>>, status_response: StatusResponse, + status_queue: StatusQueue, + status_hold_at: Arc, + status_principals: Arc>>, + issued_principals: Arc>>, + default_principal: Arc>>, + status_uses_token: Arc, status_renewed: Arc>>, status_requests: Arc>>, + status_arrived: Arc, expire_tokens: Arc, refuse_auth: Arc, redirect_requests: Arc, @@ -54,6 +63,20 @@ impl Fixture { let requests = Arc::new(Mutex::new(Vec::new())); let status_response = Arc::new(Mutex::new(None::<(u16, Vec)>)); let status_requests = Arc::new(Mutex::new(Vec::new())); + let status_arrived = Arc::new(Notify::new()); + let status_notification = status_arrived.clone(); + let status_queue = Arc::new(Mutex::new(VecDeque::<(u16, Vec)>::new())); + let queued_status = status_queue.clone(); + let status_hold_at = Arc::new(AtomicUsize::new(0)); + let held_status_number = status_hold_at.clone(); + let status_principals = Arc::new(Mutex::new(Vec::new())); + let observed_principals = status_principals.clone(); + let issued_principals = Arc::new(Mutex::new(HashMap::::new())); + let issuer_map = issued_principals.clone(); + let default_principal = Arc::new(Mutex::new(None::)); + let issuer_default = default_principal.clone(); + let status_uses_token = Arc::new(AtomicBool::new(false)); + let use_token = status_uses_token.clone(); let expire_tokens = Arc::new(AtomicBool::new(false)); let refuse_auth = Arc::new(AtomicBool::new(false)); let redirect_requests = Arc::new(AtomicUsize::new(0)); @@ -158,6 +181,13 @@ impl Fixture { .verify_ecdsa(&Message::from_digest([1; 32]), &signature, &public) .unwrap(); keys.lock().await.push(query["key"].clone()); + let issued_principal = issuer_map + .lock() + .await + .get(&query["key"]) + .cloned() + .or(issuer_default.lock().await.clone()) + .unwrap_or_else(|| query["key"].clone()); ( 200, format!( @@ -167,7 +197,7 @@ impl Fixture { } else { "eyJleHAiOjQxMDI0NDQ4MDB9" }, - query["key"] + issued_principal ) .into_bytes(), ) @@ -178,18 +208,33 @@ impl Fixture { use crate::migration_status::wire; let req = wire::Request::decode(&data[end..]).unwrap(); status_reads.lock().await.push(req.store_id.clone()); + status_notification.notify_one(); renewed_reads.lock().await.push(renewed_proof); - status_reply.lock().await.clone().unwrap_or_else(|| { - ( - 200, - wire::State { - effective_principal: principal.unwrap(), - store_id: req.store_id, - ..Default::default() - } - .encode_to_vec(), - ) - }) + observed_principals + .lock() + .await + .push(principal.clone().unwrap()); + let mut queued = queued_status.lock().await.pop_front(); + if use_token.load(Ordering::SeqCst) { + if let Some((200, body)) = &mut queued { + let mut state = wire::State::decode(body.as_slice()).unwrap(); + state.effective_principal = principal.clone().unwrap(); + *body = state.encode_to_vec(); + } + } + queued + .or(status_reply.lock().await.clone()) + .unwrap_or_else(|| { + ( + 200, + wire::State { + effective_principal: principal.unwrap(), + store_id: req.store_id, + ..Default::default() + } + .encode_to_vec(), + ) + }) } else { assert_eq!(path, "/getObject"); let req = GetObjectRequest::decode(&data[end..]).unwrap(); @@ -210,7 +255,10 @@ impl Fixture { ), } }; - let should_hold = held.lock().await.as_deref() == Some(&path); + let should_hold = held.lock().await.as_deref() == Some(&path) + || (path == "/getStoreMigrationState" + && held_status_number.load(Ordering::SeqCst) + == status_reads.lock().await.len()); let storage_path = path == "/getObject" || path == "/getStoreMigrationState"; if should_hold && !storage_path { arrived.notify_one(); @@ -243,8 +291,15 @@ impl Fixture { auth_keys, requests, status_response, + status_queue, + status_hold_at, + status_principals, + issued_principals, + default_principal, + status_uses_token, status_renewed, status_requests, + status_arrived, expire_tokens, refuse_auth, redirect_requests, diff --git a/src/backup_session/tests/reconciliation.rs b/src/backup_session/tests/reconciliation.rs new file mode 100644 index 0000000..94cd9b7 --- /dev/null +++ b/src/backup_session/tests/reconciliation.rs @@ -0,0 +1,591 @@ +use super::*; +use crate::migration_status::wire; + +const ID: &str = "9b1b6d9f-4a9e-40ca-8e77-c1d8e6dd48fb"; + +async fn prepared_fixture(coalesced: bool) -> (Fixture, BackupReadSession, PreparedMigration) { + prepared_fixture_with_archive(coalesced, true).await +} + +async fn prepared_fixture_with_archive( + coalesced: bool, + present: bool, +) -> (Fixture, BackupReadSession, PreparedMigration) { + let fixture = Fixture::new().await; + let session = fixture.session([42; 64]).await; + if present { + fixture.app_object(&session).await; + } + if coalesced { + *fixture.default_principal.lock().await = Some("shared".into()); + } + let app_status = session + .read_migration_state(BackupDomain::App) + .await + .unwrap(); + let ldk_status = session + .read_migration_state(BackupDomain::Ldk) + .await + .unwrap(); + let archive = session.read(app(), 4096).await.unwrap(); + let mut components = Vec::new(); + for key in ["snapshot/a", "snapshot/b"] { + let storage_key = session.app.build_key(key); + fixture.objects.lock().await.insert( + storage_key.clone(), + KeyValue { + key: storage_key.clone(), + version: 1, + value: session + .app + .storable_builder + .build( + b"immutable component".to_vec(), + 1, + &session.app.app_data_encryption_key, + storage_key.as_bytes(), + ) + .encode_to_vec(), + }, + ); + components.push( + session + .read(BackupReadTarget::App { key: key.into() }, 4096) + .await + .unwrap(), + ); + } + let prepared = session + .prepare_migration(MigrationPreparation { + migration_id: ID, + app_status: &app_status, + ldk_status: &ldk_status, + archive: &archive, + initial_head: MigrationHeadInput { + key: "backup/head/v1", + value: b"complete snapshot", + }, + components: &components.iter().collect::>(), + }) + .unwrap(); + (fixture, session, prepared) +} + +fn legacy(principal: &str) -> wire::State { + wire::State { + effective_principal: principal.into(), + store_id: "owned-store".into(), + ..Default::default() + } +} + +fn receipts(session: &BackupReadSession, prepared: &PreparedMigration) -> [wire::State; 2] { + let content = session.prepared_migration(prepared).unwrap(); + [BackupDomain::App, BackupDomain::Ldk].map(|domain| { + let store = content + .stores() + .iter() + .find(|s| s.domains().contains(&domain)) + .unwrap(); + wire::State { + effective_principal: store.effective_principal().into(), + store_id: store.store_id().into(), + epoch: 1, + required_protocol: "bitkit-migration-v1".into(), + migration_id: content.migration_id().into(), + proposal_sha256: content.proposal_sha256().to_vec(), + archive: store.archive().map(|a| wire::Archive { + key: a.key.clone(), + ciphertext_sha256: a.ciphertext_sha256.map(|d| d.to_vec()), + }), + initial_head: store.initial_head().map(|h| wire::Object { + key: h.key().into(), + ciphertext_sha256: sha256::Hash::hash(h.encrypted_value()) + .to_byte_array() + .to_vec(), + }), + objects: store + .objects() + .iter() + .map(|o| wire::Object { + key: o.key.clone(), + ciphertext_sha256: o.ciphertext_sha256.to_vec(), + }) + .collect(), + } + }) +} + +async fn queue(fixture: &Fixture, states: &[wire::State; 2]) { + fixture + .status_queue + .lock() + .await + .extend(states.iter().map(|s| (200, s.encode_to_vec()))); +} + +#[tokio::test] +async fn exact_receipts_observed_through_both_original_domains() { + for coalesced in [false, true] { + let (fixture, session, prepared) = prepared_fixture(coalesced).await; + let states = receipts(&session, &prepared); + let object_reads = fixture.requests.lock().await.len(); + queue(&fixture, &states).await; + assert_eq!( + session.reconcile_migration(&prepared).await.unwrap(), + MigrationReconciliation::ObservedCommitted + ); + assert!(matches!( + session.prepared_migration(&prepared), + Err(MigrationPreparationError::StalePolicy) + )); + let principals = fixture.status_principals.lock().await; + assert_eq!(principals.len(), 4); + assert_eq!(principals[0] == principals[1], coalesced); + let signers = fixture.auth_keys.lock().await; + assert_eq!(signers.len(), 4); + assert_ne!(signers[0], signers[1]); + assert_eq!(signers[..2], signers[2..]); + drop(signers); + assert_eq!(principals[..2], principals[2..]); + assert_eq!(fixture.requests.lock().await.len(), object_reads); + // A retained stale attempt remains usable only for another read-only comparison. + drop(principals); + queue(&fixture, &states).await; + assert_eq!( + session.reconcile_migration(&prepared).await.unwrap(), + MigrationReconciliation::ObservedCommitted + ); + assert!(!fixture.task.is_finished()); + } +} + +#[tokio::test] +async fn both_legacy_observations_do_not_restore_preparation_access() { + let (fixture, session, prepared) = prepared_fixture(false).await; + assert_eq!( + session.reconcile_migration(&prepared).await.unwrap(), + MigrationReconciliation::ObservedUncommitted + ); + assert_eq!(fixture.status_requests.lock().await.len(), 4); + assert!(matches!( + session.prepared_migration(&prepared), + Err(MigrationPreparationError::StalePolicy) + )); +} + +#[tokio::test] +async fn mixed_committed_and_legacy_is_not_a_positive_outcome() { + for legacy_domain in 0..2 { + let (fixture, session, prepared) = prepared_fixture(false).await; + let mut states = receipts(&session, &prepared); + states[legacy_domain] = legacy(&states[legacy_domain].effective_principal); + queue(&fixture, &states).await; + assert!(matches!( + session.reconcile_migration(&prepared).await, + Err(MigrationReconciliationError::MixedOutcome) + )); + } +} + +#[tokio::test] +async fn remapped_original_provider_never_counts_as_the_old_participant() { + for coalesced in [false, true] { + for index in 0..2 { + let (fixture, session, prepared) = prepared_fixture(coalesced).await; + let mut states = receipts(&session, &prepared); + states[index].effective_principal = "replacement-principal".into(); + queue(&fixture, &states).await; + assert!(matches!(session.reconcile_migration(&prepared).await, + Err(MigrationReconciliationError::ParticipantChanged(domain)) if domain.index() == index)); + } + } +} + +#[tokio::test] +async fn every_initial_receipt_field_is_bound_to_the_retained_attempt() { + let (fixture, session, prepared) = prepared_fixture(false).await; + let original = receipts(&session, &prepared); + let mutations: [fn(&mut wire::State); 13] = [ + |s| s.migration_id = "1b1b6d9f-4a9e-40ca-8e77-c1d8e6dd48fb".into(), + |s| s.proposal_sha256[0] ^= 1, + |s| s.archive = None, + |s| s.archive.as_mut().unwrap().key = "foreign/archive".into(), + |s| s.archive.as_mut().unwrap().ciphertext_sha256 = None, + |s| { + s.archive + .as_mut() + .unwrap() + .ciphertext_sha256 + .as_mut() + .unwrap()[0] ^= 1 + }, + |s| s.initial_head = None, + |s| s.initial_head.as_mut().unwrap().key = "foreign/head".into(), + |s| s.initial_head.as_mut().unwrap().ciphertext_sha256[0] ^= 1, + |s| { + s.objects.pop(); + }, + |s| { + s.objects.push(wire::Object { + key: "extra/object".into(), + ciphertext_sha256: vec![7; 32], + }) + }, + |s| s.objects[0].key = "foreign/component".into(), + |s| s.objects[0].ciphertext_sha256[0] ^= 1, + ]; + for mutate in mutations { + let mut states = original.clone(); + mutate(&mut states[0]); + queue(&fixture, &states).await; + assert!(matches!( + session.reconcile_migration(&prepared).await, + Err(MigrationReconciliationError::ReceiptMismatch( + BackupDomain::App + )) + )); + } + let mut states = original.clone(); + states[1].proposal_sha256[0] ^= 1; + queue(&fixture, &states).await; + assert!(matches!( + session.reconcile_migration(&prepared).await, + Err(MigrationReconciliationError::ReceiptMismatch( + BackupDomain::Ldk + )) + )); +} + +#[tokio::test] +async fn component_order_and_later_mutable_head_do_not_change_initial_receipt() { + let (fixture, session, prepared) = prepared_fixture(false).await; + let mut states = receipts(&session, &prepared); + let head = session + .prepared_migration(&prepared) + .unwrap() + .stores() + .iter() + .find_map(|s| s.initial_head()) + .unwrap(); + fixture.objects.lock().await.insert( + head.key().into(), + KeyValue { + key: head.key().into(), + version: 99, + value: b"later head is not the immutable receipt".to_vec(), + }, + ); + states[0].objects.reverse(); + let reads = fixture.requests.lock().await.len(); + queue(&fixture, &states).await; + assert_eq!( + session.reconcile_migration(&prepared).await.unwrap(), + MigrationReconciliation::ObservedCommitted + ); + assert_eq!(fixture.requests.lock().await.len(), reads); +} + +#[tokio::test] +async fn unavailable_malformed_unauthorized_and_wrong_store_are_read_errors() { + for code in [302, 401, 404, 503, 200] { + let (fixture, session, prepared) = prepared_fixture(false).await; + let states = receipts(&session, &prepared); + fixture + .status_queue + .lock() + .await + .extend([(200, states[0].encode_to_vec()), (code, vec![0xff])]); + assert!(matches!( + session.reconcile_migration(&prepared).await, + Err(MigrationReconciliationError::Read { + domain: BackupDomain::Ldk, + .. + }) + )); + assert_eq!(fixture.status_requests.lock().await.len(), 4); + assert_eq!(fixture.redirect_requests.load(Ordering::SeqCst), 0); + assert!(matches!( + session.prepared_migration(&prepared), + Err(MigrationPreparationError::StalePolicy) + )); + } + let (fixture, session, prepared) = prepared_fixture(false).await; + let mut states = receipts(&session, &prepared); + states[1].store_id = "other-store".into(); + queue(&fixture, &states).await; + assert!(matches!( + session.reconcile_migration(&prepared).await, + Err(MigrationReconciliationError::Read { + domain: BackupDomain::Ldk, + .. + }) + )); +} + +#[tokio::test] +async fn replacement_and_dropped_owner_cannot_adopt_a_prepared_uuid() { + let (fixture, session, prepared) = prepared_fixture(false).await; + let replacement = fixture.session([42; 64]).await; + for dropped in [false, true] { + if dropped { + session.revoke(); + } + assert!(matches!( + replacement.reconcile_migration(&prepared).await, + Err(MigrationReconciliationError::Ownership( + BackupReadError::DifferentOwner + )) + )); + } + drop(session); + assert!(matches!( + replacement.reconcile_migration(&prepared).await, + Err(MigrationReconciliationError::Ownership( + BackupReadError::DifferentOwner + )) + )); + assert_eq!(fixture.status_requests.lock().await.len(), 2); +} + +#[tokio::test] +async fn revocation_discards_a_held_receipt_and_prevents_second_read() { + let (fixture, session, prepared) = prepared_fixture(false).await; + queue(&fixture, &receipts(&session, &prepared)).await; + *fixture.hold.lock().await = Some("/getStoreMigrationState".into()); + let session = Arc::new(session); + let owner = session.clone(); + let mut task = tokio::spawn(async move { owner.reconcile_migration(&prepared).await }); + tokio::time::timeout(std::time::Duration::from_secs(2), async { + tokio::select! { + _ = fixture.received.notified() => {}, + result = &mut task => panic!("operation ended before held boundary: {result:?}"), + } + }) + .await + .expect("held receipt boundary"); + session.revoke(); + assert!(matches!( + tokio::time::timeout(std::time::Duration::from_secs(1), task) + .await + .unwrap() + .unwrap(), + Err(MigrationReconciliationError::Ownership( + BackupReadError::Revoked + )) + )); + assert_eq!(fixture.status_requests.lock().await.len(), 3); + fixture.release.notify_one(); +} + +#[tokio::test] +async fn abandoned_reconciliation_does_not_restore_or_schedule_continuation() { + let (fixture, session, prepared) = prepared_fixture(false).await; + queue(&fixture, &receipts(&session, &prepared)).await; + *fixture.hold.lock().await = Some("/getStoreMigrationState".into()); + let mut future = Box::pin(session.reconcile_migration(&prepared)); + tokio::select! { + biased; + _ = fixture.received.notified() => {}, + _ = &mut future => panic!("held read completed"), + } + // Consume the held request's notification before watching for later work. + tokio::time::timeout( + std::time::Duration::from_secs(2), + fixture.status_arrived.notified(), + ) + .await + .expect("held request must have notified the observer"); + drop(future); + fixture.release.notify_one(); + assert!( + tokio::time::timeout( + std::time::Duration::from_millis(250), + fixture.status_arrived.notified(), + ) + .await + .is_err(), + "abandoned operation issued another status request" + ); + assert!(matches!( + session.prepared_migration(&prepared), + Err(MigrationPreparationError::StalePolicy) + )); + assert_eq!(fixture.status_requests.lock().await.len(), 3); +} + +#[tokio::test] +async fn overlapping_status_read_supersedes_the_reconciliation_pair() { + let (fixture, session, prepared) = prepared_fixture(false).await; + queue(&fixture, &receipts(&session, &prepared)).await; + fixture.status_hold_at.store(4, Ordering::SeqCst); + let mut reconciliation = Box::pin(session.reconcile_migration(&prepared)); + tokio::select! { + biased; + _ = fixture.received.notified() => {}, + _ = &mut reconciliation => panic!("second read did not hold"), + } + let mut newer = Box::pin(session.read_migration_state(BackupDomain::App)); + std::future::poll_fn(|cx| { + assert!(std::future::Future::poll(newer.as_mut(), cx).is_pending()); + std::task::Poll::Ready(()) + }) + .await; + drop(newer); + fixture.release.notify_one(); + assert!(matches!( + reconciliation.await, + Err(MigrationReconciliationError::SupersededObservation) + )); +} + +#[tokio::test] +async fn explicit_archive_absence_is_a_bound_condition() { + let (fixture, session, prepared) = prepared_fixture_with_archive(false, false).await; + let mut states = receipts(&session, &prepared); + assert!(states[0] + .archive + .as_ref() + .unwrap() + .ciphertext_sha256 + .is_none()); + queue(&fixture, &states).await; + assert_eq!( + session.reconcile_migration(&prepared).await.unwrap(), + MigrationReconciliation::ObservedCommitted + ); + states[0].archive.as_mut().unwrap().ciphertext_sha256 = Some(vec![7; 32]); + queue(&fixture, &states).await; + assert!(matches!( + session.reconcile_migration(&prepared).await, + Err(MigrationReconciliationError::ReceiptMismatch( + BackupDomain::App + )) + )); +} + +#[tokio::test] +async fn ldk_receipt_cannot_add_unprepared_archive_head_or_components() { + let (fixture, session, prepared) = prepared_fixture(false).await; + let original = receipts(&session, &prepared); + for field in 0..3 { + let mut states = original.clone(); + match field { + 0 => states[1].archive = states[0].archive.clone(), + 1 => states[1].initial_head = states[0].initial_head.clone(), + _ => states[1].objects = states[0].objects.clone(), + } + queue(&fixture, &states).await; + assert!(matches!( + session.reconcile_migration(&prepared).await, + Err(MigrationReconciliationError::ReceiptMismatch( + BackupDomain::Ldk + )) + )); + } +} + +#[tokio::test] +async fn revocation_during_second_read_discards_the_matching_first_receipt() { + let (fixture, session, prepared) = prepared_fixture(false).await; + queue(&fixture, &receipts(&session, &prepared)).await; + fixture.status_hold_at.store(4, Ordering::SeqCst); + let mut future = Box::pin(session.reconcile_migration(&prepared)); + tokio::select! { + biased; + _ = fixture.received.notified() => {}, + _ = &mut future => panic!("second receipt did not hold"), + } + session.revoke(); + assert!(matches!( + future.await, + Err(MigrationReconciliationError::Ownership( + BackupReadError::Revoked + )) + )); + fixture.release.notify_one(); +} + +#[tokio::test] +async fn remapped_legacy_participant_is_not_observed_uncommitted() { + for coalesced in [false, true] { + for index in 0..2 { + let (fixture, session, prepared) = prepared_fixture(coalesced).await; + let committed = receipts(&session, &prepared); + let mut states = committed.map(|s| legacy(&s.effective_principal)); + states[index].effective_principal = "remapped".into(); + queue(&fixture, &states).await; + assert!(matches!(session.reconcile_migration(&prepared).await, + Err(MigrationReconciliationError::ParticipantChanged(domain)) if domain.index() == index)); + } + } +} + +#[tokio::test] +async fn issuer_remap_is_observed_even_while_old_tokens_remain_valid() { + for coalesced in [false, true] { + for index in 0..2 { + let (fixture, session, prepared) = prepared_fixture(coalesced).await; + let states = receipts(&session, &prepared); + let signer = fixture.auth_keys.lock().await[index].clone(); + fixture + .issued_principals + .lock() + .await + .insert(signer, "remapped".into()); + fixture.status_uses_token.store(true, Ordering::SeqCst); + // Control: the normal provider still has the original unexpired JWT. + let domain = [BackupDomain::App, BackupDomain::Ldk][index]; + let cached = session.read_migration_state(domain).await.unwrap(); + assert_eq!( + session + .migration_state(&cached) + .unwrap() + .effective_principal, + states[index].effective_principal + ); + // Reconciliation must freshly exercise the same signer, not that cached JWT. + queue(&fixture, &states).await; + assert!(matches!(session.reconcile_migration(&prepared).await, + Err(MigrationReconciliationError::ParticipantChanged(observed)) if observed == domain)); + assert_eq!(fixture.auth_keys.lock().await.len(), 4); + } + } +} + +#[tokio::test] +async fn issuer_unavailable_does_not_fall_back_to_valid_cached_credentials() { + let (fixture, session, prepared) = prepared_fixture(false).await; + fixture.refuse_auth.store(true, Ordering::SeqCst); + let cached = session + .read_migration_state(BackupDomain::App) + .await + .unwrap(); + assert!(matches!( + session.migration_state(&cached).unwrap().policy, + crate::MigrationPolicy::Legacy + )); + assert!(matches!( + session.reconcile_migration(&prepared).await, + Err(MigrationReconciliationError::Read { + domain: BackupDomain::App, + .. + }) + )); + assert_eq!(fixture.status_requests.lock().await.len(), 3); +} + +#[tokio::test] +async fn continuation_observer_detects_an_explicitly_scheduled_status_read() { + let fixture = Fixture::new().await; + let session = fixture.session([42; 64]).await; + let read = tokio::spawn(async move { session.read_migration_state(BackupDomain::Ldk).await }); + tokio::time::timeout( + std::time::Duration::from_secs(2), + fixture.status_arrived.notified(), + ) + .await + .expect("positive control must detect the scheduled request"); + read.await.unwrap().unwrap(); + assert_eq!(fixture.status_requests.lock().await.len(), 1); +} diff --git a/src/implementation.rs b/src/implementation.rs index bc405f6..8fa89ef 100644 --- a/src/implementation.rs +++ b/src/implementation.rs @@ -113,6 +113,7 @@ pub struct VssClient { inner: Arc>, pub(crate) conditional_inner: Arc>, pub(crate) receipt_reader: Arc, + pub(crate) fresh_migration_reader: Option, pub(crate) store_id: String, pub(crate) storable_builder: Arc>, pub(crate) app_data_encryption_key: [u8; 32], @@ -182,6 +183,11 @@ impl VssClient { error_details: format!("Failed to derive LNURL-auth key: {}", e), })?; + let fresh_migration_reader = crate::migration_status::FreshMigrationReader::new( + base_url.clone(), + lnurl_auth_xprv, + lnurl_auth_server_url.clone(), + ); let lnurl_auth_jwt_provider = LnurlAuthToJwtProvider::new(lnurl_auth_xprv, lnurl_auth_server_url, HashMap::new()) .map_err(|e| VssError::ConnectionError { @@ -190,13 +196,15 @@ impl VssClient { let header_provider = Arc::new(lnurl_auth_jwt_provider); - Self::new_with_header_provider( + let mut client = Self::new_with_header_provider( base_url, store_id, header_provider, Some(app_vss_seed_bytes), ) - .await + .await?; + client.fresh_migration_reader = Some(fresh_migration_reader); + Ok(client) } /// Internal method to create a client with any header provider @@ -240,6 +248,7 @@ impl VssClient { inner: Arc::new(client), conditional_inner: Arc::new(conditional_inner), receipt_reader, + fresh_migration_reader: None, store_id, storable_builder, app_data_encryption_key, diff --git a/src/ldk_client.rs b/src/ldk_client.rs index f6a8a1a..c7b4ff7 100644 --- a/src/ldk_client.rs +++ b/src/ldk_client.rs @@ -52,6 +52,7 @@ type LdkRetryPolicy = FilteredRetryPolicy< pub struct LdkVssClient { inner: Arc>, pub(crate) receipt_reader: Arc, + pub(crate) fresh_migration_reader: crate::migration_status::FreshMigrationReader, pub(crate) store_id: String, pub(crate) storable_builder: Arc>, pub(crate) data_encryption_key: [u8; 32], @@ -94,6 +95,11 @@ impl LdkVssClient { error_details: format!("Failed to derive LNURL-auth key: {}", e), })?; + let fresh_migration_reader = crate::migration_status::FreshMigrationReader::new( + base_url.clone(), + lnurl_auth_xprv, + lnurl_auth_server_url.clone(), + ); let lnurl_auth_jwt_provider = LnurlAuthToJwtProvider::new(lnurl_auth_xprv, lnurl_auth_server_url, HashMap::new()) .map_err(|e| VssError::ConnectionError { @@ -145,6 +151,7 @@ impl LdkVssClient { Ok(LdkVssClient { inner: Arc::new(client), receipt_reader, + fresh_migration_reader, store_id, storable_builder: Arc::new(StorableBuilder::new(RandEntropySource)), data_encryption_key, diff --git a/src/lib.rs b/src/lib.rs index b756264..285b8af 100644 --- a/src/lib.rs +++ b/src/lib.rs @@ -13,8 +13,8 @@ mod types; pub use backup_session::{ BackupDomain, BackupReadConfig, BackupReadError, BackupReadObservation, BackupReadSession, BackupReadTarget, MigrationHeadInput, MigrationObservation, MigrationPreparation, - MigrationPreparationError, PreparedMigration, PreparedMigrationContent, PreparedMigrationHead, - PreparedMigrationStore, + MigrationPreparationError, MigrationReconciliation, MigrationReconciliationError, + PreparedMigration, PreparedMigrationContent, PreparedMigrationHead, PreparedMigrationStore, }; pub use errors::*; pub use implementation::{derive_vss_store_id, VssClient}; diff --git a/src/migration_status.rs b/src/migration_status.rs index 753296e..0027396 100644 --- a/src/migration_status.rs +++ b/src/migration_status.rs @@ -55,6 +55,55 @@ impl LdkVssClient { } } +/// Captured original signer/configuration, never global replacement or cached JWT adoption. +#[derive(Clone)] +pub(crate) struct FreshMigrationReader { + base_url: String, + signer: bitcoin::bip32::Xpriv, + auth_url: String, +} + +impl FreshMigrationReader { + pub(crate) fn new(base_url: String, signer: bitcoin::bip32::Xpriv, auth_url: String) -> Self { + Self { + base_url, + signer, + auth_url, + } + } + + async fn read(&self, store_id: &str) -> Result { + let provider = vss_client_ng::headers::LnurlAuthToJwtProvider::new( + self.signer, + self.auth_url.clone(), + std::collections::HashMap::new(), + ) + .map_err(|_| VssError::AuthError { + error_details: "Original migration credentials are unavailable".into(), + })?; + let reader = ReceiptReader::new(&self.base_url, std::sync::Arc::new(provider))?; + read(&reader, store_id).await + } +} + +impl VssClient { + pub(crate) async fn read_fresh_migration_state(&self) -> Result { + let reader = self + .fresh_migration_reader + .as_ref() + .ok_or_else(|| VssError::AuthError { + error_details: "Original migration signer is unavailable".into(), + })?; + reader.read(&self.store_id).await + } +} + +impl LdkVssClient { + pub(crate) async fn read_fresh_migration_state(&self) -> Result { + self.fresh_migration_reader.read(&self.store_id).await + } +} + async fn read(reader: &ReceiptReader, store_id: &str) -> Result { if !valid_identity(store_id) { return Err(invalid());