From 435bb2a48459a556f3504981c4dc471c21166d02 Mon Sep 17 00:00:00 2001 From: benk10 Date: Wed, 4 Nov 2026 00:19:06 -0500 Subject: [PATCH] feat: observe owner-bound migration status --- README.md | 24 +++ src/backup_session.rs | 64 +++++- src/backup_session/tests.rs | 374 +++++++++++++++++++++++++++++++++- src/lib.rs | 7 +- src/migration_status.rs | 294 ++++++++++++++++++++++++++ src/migration_status/tests.rs | 203 ++++++++++++++++++ src/receipt.rs | 29 ++- 7 files changed, 975 insertions(+), 20 deletions(-) create mode 100644 src/migration_status.rs create mode 100644 src/migration_status/tests.rs diff --git a/README.md b/README.md index 245e475..52d6060 100644 --- a/README.md +++ b/README.md @@ -464,3 +464,27 @@ they are in-crate Rust source preparation: this package still emits only static and dynamic libraries, not a downstream Cargo `rlib`. Released Swift/Kotlin interfaces, binary tags and checksums are unchanged. No migration or writer is activated by creating or reading a session. Do not log observations or plaintext. + + +### Original-wallet migration status (Rust source preparation) + +`BackupReadSession::read_migration_state` observes the candidate VSS status contract +through the original app or LDK credential provider. Each result retains its session +and domain, including the server-reported physical principal/store. Different +principals may share the same textual store ID. A renewed proof can return a changed +mapping; old observations remain unchanged and never authorize a later mutation. + +An explicit legacy tuple is distinct from committed `bitkit-migration-v1` metadata. +Unsupported routes, authorization failures, incomplete or conflicting fields, +unknown protocols/epochs and over-limit responses remain errors. The storage POST +has no automatic retry or redirect, a 64 KiB body limit, and a ten-second deadline +covering credentials and body. The inherited credential provider keeps its own +challenge/callback redirect and body policies. Dropping a read does not replay it; +session revocation rejects pending and later results. + +This is unpublished-contract, in-crate source preparation: no new UniFFI export, +released native API, server deployment or app activation. A single response cannot +prove all-tuples coherence, deployed principal mapping or old reader/writer +exclusion. Before any mutation an enforcing server must atomically recheck fresh +authorization against the observed original principal/store. Do not serialize +observations, log identities/proofs or treat status as permission to migrate. diff --git a/src/backup_session.rs b/src/backup_session.rs index a2b1636..60c15a1 100644 --- a/src/backup_session.rs +++ b/src/backup_session.rs @@ -1,6 +1,6 @@ //! Read-only, in-memory ownership across app and LDK backup observations. -use crate::{LdkNamespace, LdkVssClient, VssClient, VssError, VssReadReceipt}; +use crate::{LdkNamespace, LdkVssClient, MigrationState, VssClient, VssError, VssReadReceipt}; use std::sync::Arc; use tokio::sync::watch; @@ -34,6 +34,27 @@ pub enum BackupReadError { Read(#[from] VssError), } +/// Credential/derivation domain, independent of textual store identity. +#[derive(Clone, Copy, Debug, PartialEq, Eq)] +pub enum BackupDomain { + App, + Ldk, +} + +/// Server-reported policy bound to one original session and credential domain. +/// Not serializable or Debug; it carries physical storage identity, not authority. +pub struct MigrationObservation { + owner: Arc>, + domain: BackupDomain, + state: MigrationState, +} + +impl MigrationObservation { + pub fn domain(&self) -> BackupDomain { + self.domain + } +} + /// An observation bound to one native session. Deliberately not serializable or /// Debug: it can contain a complete plaintext wallet and is not remote authority. pub struct BackupReadObservation { @@ -138,6 +159,47 @@ impl BackupReadSession { Ok(observation.receipt.as_ref()) } + /// One read of the candidate VSS migration-status contract. Unsupported routes, + /// policy errors and missing/malformed tuples are errors, never legacy absence. + /// The ten-second deadline includes credentials; the storage body is capped at 64 KiB. + pub async fn read_migration_state( + &self, + domain: BackupDomain, + ) -> Result { + let mut cancellation = self.revoked.subscribe(); + self.ensure_active()?; + let read = async { + match domain { + BackupDomain::App => self.app.read_migration_state().await, + BackupDomain::Ldk => self.ldk.read_migration_state().await, + } + }; + let state = tokio::select! { + biased; + _ = cancellation.changed() => return Err(BackupReadError::Revoked), + result = read => result, + }; + self.ensure_active()?; + Ok(MigrationObservation { + owner: Arc::clone(&self.revoked), + domain, + state: state?, + }) + } + + /// Local observation access only. A future mutation must freshly and atomically + /// bind its authorization to the observed principal/store on an enforcing server. + pub fn migration_state<'a>( + &self, + observation: &'a MigrationObservation, + ) -> Result<&'a MigrationState, BackupReadError> { + if !Arc::ptr_eq(&self.revoked, &observation.owner) { + return Err(BackupReadError::DifferentOwner); + } + self.ensure_active()?; + Ok(&observation.state) + } + fn ensure_active(&self) -> Result<(), BackupReadError> { if *self.revoked.borrow() { return Err(BackupReadError::Revoked); diff --git a/src/backup_session/tests.rs b/src/backup_session/tests.rs index 8cef39d..a140205 100644 --- a/src/backup_session/tests.rs +++ b/src/backup_session/tests.rs @@ -5,7 +5,10 @@ use bitcoin::{ secp256k1::{ecdsa::Signature, Message, PublicKey, Secp256k1}, }; use prost::Message as ProstMessage; -use std::collections::HashMap; +use std::{ + collections::HashMap, + sync::atomic::{AtomicBool, AtomicUsize, Ordering}, +}; use tokio::{ io::{AsyncReadExt, AsyncWriteExt}, net::TcpListener, @@ -16,11 +19,19 @@ use vss_client_ng::types::{ ErrorCode, ErrorResponse, GetObjectRequest, GetObjectResponse, KeyValue, }; +type StatusResponse = Arc)>>>; + struct Fixture { url: String, objects: Arc>>, auth_keys: Arc>>, requests: Arc>>, + status_response: StatusResponse, + status_renewed: Arc>>, + status_requests: Arc>>, + expire_tokens: Arc, + refuse_auth: Arc, + redirect_requests: Arc, hold: Arc>>, received: Arc, release: Arc, @@ -40,6 +51,18 @@ impl Fixture { let objects = Arc::new(Mutex::new(HashMap::::new())); let auth_keys = Arc::new(Mutex::new(Vec::new())); 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 expire_tokens = Arc::new(AtomicBool::new(false)); + let refuse_auth = Arc::new(AtomicBool::new(false)); + let redirect_requests = Arc::new(AtomicUsize::new(0)); + let expired = expire_tokens.clone(); + let refused = refuse_auth.clone(); + let redirected = redirect_requests.clone(); + let status_reply = status_response.clone(); + let status_reads = status_requests.clone(); + let status_renewed = Arc::new(Mutex::new(Vec::new())); + let renewed_reads = status_renewed.clone(); let hold = Arc::new(Mutex::new(None::)); let received = Arc::new(Notify::new()); let release = Arc::new(Notify::new()); @@ -86,7 +109,24 @@ impl Fixture { .then(|| value.trim().parse().unwrap()) }) .unwrap_or(0); - if path == "/getObject" { + let renewed_proof = headers.lines().any(|line| { + line.to_ascii_lowercase().starts_with("authorization:") + && line.contains(".eyJleHAiOjQxMDI0NDQ4MDB9.") + }); + let principal = headers.lines().find_map(|line| { + let (name, value) = line.split_once(':')?; + name.eq_ignore_ascii_case("authorization").then(|| { + value + .trim() + .rsplit('.') + .next() + .unwrap() + .strip_suffix("AA") + .unwrap() + .to_owned() + }) + }); + if path == "/getObject" || path == "/getStoreMigrationState" { assert!(headers.to_lowercase().contains("authorization: bearer ")); } while data.len() < end + length { @@ -96,10 +136,14 @@ impl Fixture { data.extend_from_slice(&chunk[..n]); } let (status, body) = if path == "/auth" { - ( - 200, - format!("{base}/callback?k1={}", "01".repeat(32)).into_bytes(), - ) + if refused.load(Ordering::SeqCst) { + (401, b"auth unavailable".to_vec()) + } else { + ( + 200, + format!("{base}/callback?k1={}", "01".repeat(32)).into_bytes(), + ) + } } else if path.starts_with("/callback?") { let query: HashMap<_, _> = reqwest::Url::parse(&format!("{base}{path}")) .unwrap() @@ -115,8 +159,36 @@ impl Fixture { keys.lock().await.push(query["key"].clone()); ( 200, - br#"{"token":"e30.eyJleHAiOjQxMDI0NDQ4MDB9.test"}"#.to_vec(), + format!( + r#"{{"token":"e30.{}.{}AA"}}"#, + if expired.load(Ordering::SeqCst) { + "eyJleHAiOjB9" + } else { + "eyJleHAiOjQxMDI0NDQ4MDB9" + }, + query["key"] + ) + .into_bytes(), ) + } else if path == "/status-redirect" { + redirected.fetch_add(1, Ordering::SeqCst); + (200, vec![]) + } else if path == "/getStoreMigrationState" { + use crate::migration_status::wire; + let req = wire::Request::decode(&data[end..]).unwrap(); + status_reads.lock().await.push(req.store_id.clone()); + 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(), + ) + }) } else { assert_eq!(path, "/getObject"); let req = GetObjectRequest::decode(&data[end..]).unwrap(); @@ -138,13 +210,19 @@ impl Fixture { } }; let should_hold = held.lock().await.as_deref() == Some(&path); - if should_hold && path != "/getObject" { + let storage_path = path == "/getObject" || path == "/getStoreMigrationState"; + if should_hold && !storage_path { arrived.notify_one(); resume.notified().await; } - let response = format!("HTTP/1.1 {status} Test\r\nContent-Type: application/json\r\nContent-Length: {}\r\nConnection: close\r\n\r\n",body.len()); + let location = if status == 302 { + format!("Location: {base}/status-redirect\r\n") + } else { + String::new() + }; + let response = format!("HTTP/1.1 {status} Test\r\n{location}Content-Type: application/json\r\nContent-Length: {}\r\nConnection: close\r\n\r\n",body.len()); let header_written = stream.write_all(response.as_bytes()).await; - let prefix = if should_hold && path == "/getObject" { + let prefix = if should_hold && storage_path { body.len() / 2 } else { 0 @@ -163,6 +241,12 @@ impl Fixture { objects, auth_keys, requests, + status_response, + status_renewed, + status_requests, + expire_tokens, + refuse_auth, + redirect_requests, hold, received, release, @@ -483,3 +567,273 @@ async fn ldk_receipt_refuses_cross_namespace_aad_wrong_key_and_response_limit() )); assert_eq!(fixture.requests.lock().await.len(), 3); } + +#[tokio::test] +async fn migration_observations_bind_original_domain_and_owner() { + let fixture = Fixture::new().await; + let session = fixture.session([7; 64]).await; + let app = session + .read_migration_state(BackupDomain::App) + .await + .unwrap(); + let ldk = session + .read_migration_state(BackupDomain::Ldk) + .await + .unwrap(); + assert_eq!(app.domain(), BackupDomain::App); + assert_eq!(ldk.domain(), BackupDomain::Ldk); + let a = session.migration_state(&app).unwrap(); + let l = session.migration_state(&ldk).unwrap(); + assert_eq!(a.store_id, l.store_id); + assert_ne!(a.effective_principal, l.effective_principal); + let keys = fixture.auth_keys.lock().await; + assert_eq!(a.effective_principal, keys[0]); + assert_eq!(l.effective_principal, keys[1]); + assert!(matches!(a.policy, crate::MigrationPolicy::Legacy)); + assert!(matches!(l.policy, crate::MigrationPolicy::Legacy)); + drop(keys); + let replacement = fixture.session([7; 64]).await; + assert!(matches!( + replacement.migration_state(&app), + Err(BackupReadError::DifferentOwner) + )); + session.revoke(); + assert!(matches!( + session.migration_state(&ldk), + Err(BackupReadError::Revoked) + )); + assert!(matches!( + session.read_migration_state(BackupDomain::App).await, + Err(BackupReadError::Revoked) + )); + assert_eq!(fixture.status_requests.lock().await.len(), 2); +} + +#[tokio::test] +async fn migration_unavailable_or_wrong_store_is_never_legacy() { + use crate::migration_status::wire; + let fixture = Fixture::new().await; + let session = fixture.session([8; 64]).await; + for (status, body) in [ + ( + 404, + ErrorResponse { + error_code: ErrorCode::NoSuchKeyException as i32, + message: "missing".into(), + } + .encode_to_vec(), + ), + (401, vec![]), + (403, vec![]), + (502, vec![]), + (302, vec![]), + (204, vec![]), + (200, vec![]), + ( + 200, + wire::State { + effective_principal: "principal".into(), + store_id: "wrong".into(), + ..Default::default() + } + .encode_to_vec(), + ), + (200, vec![0; 65537]), + ] { + *fixture.status_response.lock().await = Some((status, body)); + assert!(session + .read_migration_state(BackupDomain::App) + .await + .is_err()); + } + assert_eq!(fixture.status_requests.lock().await.len(), 9); + assert_eq!(fixture.redirect_requests.load(Ordering::SeqCst), 0); +} + +#[tokio::test] +async fn migration_revoke_abandons_credentials_and_partial_body() { + for path in ["/auth", "/getStoreMigrationState"] { + let fixture = Fixture::new().await; + *fixture.hold.lock().await = Some(path.into()); + let session = fixture.session([9; 64]).await; + let read = session.read_migration_state(BackupDomain::Ldk); + tokio::pin!(read); + tokio::select! { + _ = fixture.received.notified() => {}, + _ = &mut read => panic!("read completed before controlled boundary"), + _ = tokio::time::sleep(std::time::Duration::from_secs(3)) => panic!("fixture boundary not reached"), + } + session.revoke(); + let result = tokio::time::timeout(std::time::Duration::from_secs(1), &mut read) + .await + .unwrap(); + assert!(matches!(result, Err(BackupReadError::Revoked))); + fixture.release.notify_one(); + assert_eq!( + fixture.status_requests.lock().await.len(), + usize::from(path != "/auth") + ); + } +} + +#[tokio::test] +async fn migration_reads_retain_original_clients_during_global_replacement() { + let _guard = crate::FFI_TEST_LOCK.lock().await; + let original = Fixture::new().await; + let replacement = Fixture::new().await; + let session = original.session([10; 64]).await; + *original.hold.lock().await = Some("/auth".into()); + let read = session.read_migration_state(BackupDomain::App); + tokio::pin!(read); + tokio::select! { + _ = original.received.notified() => {}, + _ = &mut read => panic!("original credential challenge must remain held"), + _ = tokio::time::sleep(std::time::Duration::from_secs(3)) => panic!("fixture boundary not reached"), + } + let other = replacement.session([11; 64]).await; + let old_app = crate::get_vss_client() + .lock() + .unwrap() + .replace(other.app.clone()); + let old_ldk = crate::get_ldk_client() + .lock() + .unwrap() + .replace(other.ldk.clone()); + *original.hold.lock().await = None; + original.release.notify_one(); + let app = read.await.unwrap(); + let ldk = session + .read_migration_state(BackupDomain::Ldk) + .await + .unwrap(); + assert_eq!( + session.migration_state(&app).unwrap().effective_principal, + original.auth_keys.lock().await[0] + ); + assert_eq!( + session.migration_state(&ldk).unwrap().effective_principal, + original.auth_keys.lock().await[1] + ); + assert_eq!(original.status_requests.lock().await.len(), 2); + assert!(replacement.auth_keys.lock().await.is_empty()); + assert!(replacement.status_requests.lock().await.is_empty()); + *crate::get_vss_client().lock().unwrap() = old_app; + *crate::get_ldk_client().lock().unwrap() = old_ldk; +} + +#[tokio::test] +async fn migration_credential_renewal_keeps_signer_and_reports_mapping_change() { + use crate::migration_status::wire; + let fixture = Fixture::new().await; + fixture.expire_tokens.store(true, Ordering::SeqCst); + let session = fixture.session([12; 64]).await; + let original = session + .read_migration_state(BackupDomain::App) + .await + .unwrap(); + let original_principal = session + .migration_state(&original) + .unwrap() + .effective_principal + .clone(); + fixture.expire_tokens.store(false, Ordering::SeqCst); + *fixture.status_response.lock().await = Some(( + 200, + wire::State { + effective_principal: "changed-mapping".into(), + store_id: "owned-store".into(), + ..Default::default() + } + .encode_to_vec(), + )); + let renewed = session + .read_migration_state(BackupDomain::App) + .await + .unwrap(); + let keys = fixture.auth_keys.lock().await; + assert_eq!(keys.len(), 2); + assert_eq!(keys[0], keys[1]); + assert_eq!(*fixture.status_renewed.lock().await, vec![false, true]); + assert_eq!( + session + .migration_state(&original) + .unwrap() + .effective_principal, + original_principal + ); + assert_eq!( + session + .migration_state(&renewed) + .unwrap() + .effective_principal, + "changed-mapping" + ); + assert_eq!(fixture.status_requests.lock().await.len(), 2); +} + +#[tokio::test] +async fn migration_credentials_refused_never_dispatch_storage() { + let fixture = Fixture::new().await; + fixture.refuse_auth.store(true, Ordering::SeqCst); + let session = fixture.session([13; 64]).await; + assert!(session + .read_migration_state(BackupDomain::App) + .await + .is_err()); + assert!(fixture.status_requests.lock().await.is_empty()); +} + +#[tokio::test] +async fn migration_deadline_covers_held_credentials_and_body_without_replay() { + for path in ["/auth", "/getStoreMigrationState"] { + let fixture = Fixture::new().await; + *fixture.hold.lock().await = Some(path.into()); + let session = fixture.session([14; 64]).await; + let read = session.read_migration_state(BackupDomain::App); + tokio::pin!(read); + tokio::select! { + _ = fixture.received.notified() => {}, + _ = &mut read => panic!("deadline boundary not reached"), + _ = tokio::time::sleep(std::time::Duration::from_secs(3)) => panic!("fixture boundary not reached"), + } + let result = tokio::time::timeout(std::time::Duration::from_secs(12), &mut read) + .await + .unwrap(); + assert!(matches!( + result, + Err(BackupReadError::Read(VssError::NetworkError { .. })) + )); + assert_eq!( + fixture.status_requests.lock().await.len(), + usize::from(path != "/auth") + ); + } +} + +#[tokio::test] +async fn migration_caller_drop_does_not_revoke_or_replay() { + let fixture = Fixture::new().await; + *fixture.hold.lock().await = Some("/getStoreMigrationState".into()); + let session = fixture.session([15; 64]).await; + { + let read = session.read_migration_state(BackupDomain::App); + tokio::pin!(read); + tokio::select! { + _ = fixture.received.notified() => {}, + _ = &mut read => panic!("body not held"), + _ = tokio::time::sleep(std::time::Duration::from_secs(3)) => panic!("fixture boundary not reached"), + } + } + *fixture.hold.lock().await = None; + fixture.release.notify_one(); + // Only this explicit second operation may issue another request. + let observed = session + .read_migration_state(BackupDomain::App) + .await + .unwrap(); + assert!(matches!( + session.migration_state(&observed).unwrap().policy, + crate::MigrationPolicy::Legacy + )); + assert_eq!(fixture.status_requests.lock().await.len(), 2); +} diff --git a/src/lib.rs b/src/lib.rs index 2dccd27..1db9752 100644 --- a/src/lib.rs +++ b/src/lib.rs @@ -5,16 +5,21 @@ mod errors; mod ffi_tests; mod implementation; mod ldk_client; +mod migration_status; mod receipt; mod tests; mod types; pub use backup_session::{ - BackupReadConfig, BackupReadError, BackupReadObservation, BackupReadSession, BackupReadTarget, + BackupDomain, BackupReadConfig, BackupReadError, BackupReadObservation, BackupReadSession, + BackupReadTarget, MigrationObservation, }; pub use errors::*; pub use implementation::{derive_vss_store_id, VssClient}; pub use ldk_client::LdkVssClient; +pub use migration_status::{ + CommittedMigration, MigrationArchive, MigrationObject, MigrationPolicy, MigrationState, +}; pub use types::*; uniffi::setup_scaffolding!(); diff --git a/src/migration_status.rs b/src/migration_status.rs new file mode 100644 index 0000000..753296e --- /dev/null +++ b/src/migration_status.rs @@ -0,0 +1,294 @@ +//! Strict observations of the candidate VSS migration-status contract. +//! This module does not activate migration or assert deployed server support. + +use crate::{receipt::ReceiptReader, LdkVssClient, VssClient, VssError}; +use prost::{ + encoding::{decode_key, decode_varint, WireType}, + Message, +}; +use std::collections::HashSet; + +const STATUS_LIMIT: u32 = 64 * 1024; +const PROTOCOL: &str = "bitkit-migration-v1"; + +/// Physical identity reported by this authenticated status read. Equal store IDs +/// do not imply equal principals. No Debug/Serialize or persisted replay authority. +pub struct MigrationState { + pub effective_principal: String, + pub store_id: String, + pub policy: MigrationPolicy, +} + +pub enum MigrationPolicy { + Legacy, + Committed(CommittedMigration), +} + +pub struct CommittedMigration { + pub migration_id: String, + pub proposal_sha256: [u8; 32], + pub archive: Option, + pub initial_head: Option, + pub objects: Vec, +} + +pub struct MigrationArchive { + pub key: String, + /// None is an explicit protected-absence condition, not a failed read. + pub ciphertext_sha256: Option<[u8; 32]>, +} + +pub struct MigrationObject { + pub key: String, + pub ciphertext_sha256: [u8; 32], +} + +impl VssClient { + pub(crate) async fn read_migration_state(&self) -> Result { + read(&self.receipt_reader, &self.store_id).await + } +} + +impl LdkVssClient { + pub(crate) async fn read_migration_state(&self) -> Result { + read(&self.receipt_reader, &self.store_id).await + } +} + +async fn read(reader: &ReceiptReader, store_id: &str) -> Result { + if !valid_identity(store_id) { + return Err(invalid()); + } + let request = wire::Request { + store_id: store_id.to_owned(), + } + .encode_to_vec(); + let (status, body) = reader + .post("getStoreMigrationState", request, STATUS_LIMIT) + .await?; + match status { + reqwest::StatusCode::OK => decode(&body, store_id), + reqwest::StatusCode::UNAUTHORIZED | reqwest::StatusCode::FORBIDDEN => { + Err(VssError::AuthError { + error_details: "Migration status was not authorized".into(), + }) + } + // Even a typed missing-key 404 is not legacy policy. + _ => Err(VssError::GetError { + error_details: "Migration status is unavailable".into(), + }), + } +} + +fn decode(bytes: &[u8], store_id: &str) -> Result { + validate_wire(bytes, true)?; + let state = wire::State::decode(bytes).map_err(|_| invalid())?; + if state.store_id != store_id + || !valid_identity(&state.store_id) + || !valid_identity(&state.effective_principal) + { + return Err(invalid()); + } + let policy = match state.epoch { + 0 if state.required_protocol.is_empty() + && state.migration_id.is_empty() + && state.proposal_sha256.is_empty() + && state.archive.is_none() + && state.initial_head.is_none() + && state.objects.is_empty() => + { + MigrationPolicy::Legacy + } + 1 if state.required_protocol == PROTOCOL && valid_uuid(&state.migration_id) => { + MigrationPolicy::Committed(committed(&state)?) + } + _ => return Err(invalid()), + }; + Ok(MigrationState { + effective_principal: state.effective_principal, + store_id: state.store_id, + policy, + }) +} + +fn committed(state: &wire::State) -> Result { + // Activation bounds bytes; existing legacy PostgreSQL identities are varchar(120). + if state.store_id.len() > 120 || state.effective_principal.len() > 120 { + return Err(invalid()); + } + let archive = state + .archive + .as_ref() + .map(|a| { + valid_key(&a.key)?; + Ok(MigrationArchive { + key: a.key.clone(), + ciphertext_sha256: a.ciphertext_sha256.as_deref().map(digest).transpose()?, + }) + }) + .transpose()?; + let initial_head = state.initial_head.as_ref().map(object).transpose()?; + if archive + .as_ref() + .zip(initial_head.as_ref()) + .is_some_and(|(a, h)| a.key == h.key) + { + return Err(invalid()); + } + let mut keys = HashSet::new(); + let mut objects = Vec::with_capacity(state.objects.len()); + for row in &state.objects { + let row = object(row)?; + if !keys.insert(row.key.clone()) || initial_head.as_ref().is_some_and(|h| h.key == row.key) + { + return Err(invalid()); + } + // The server permits an archive to also be a referenced component. The + // observations must agree on its existence and exact encrypted digest. + if archive + .as_ref() + .is_some_and(|a| a.key == row.key && a.ciphertext_sha256 != Some(row.ciphertext_sha256)) + { + return Err(invalid()); + } + objects.push(row); + } + // An LDK-only tuple can omit archive and head. The all-tuples digest cannot be + // recomputed from one status response; it is an observation, not proof of coherence. + Ok(CommittedMigration { + migration_id: state.migration_id.clone(), + proposal_sha256: digest(&state.proposal_sha256)?, + archive, + initial_head, + objects, + }) +} + +fn object(row: &wire::Object) -> Result { + valid_key(&row.key)?; + Ok(MigrationObject { + key: row.key.clone(), + ciphertext_sha256: digest(&row.ciphertext_sha256)?, + }) +} +fn digest(value: &[u8]) -> Result<[u8; 32], VssError> { + value.try_into().map_err(|_| invalid()) +} +fn valid_identity(value: &str) -> bool { + !value.is_empty() && value.chars().count() <= 120 +} +fn valid_key(value: &str) -> Result<(), VssError> { + if value.is_empty() + || value.len() > 600 + || value == "global_version" + || value == "vss_global_version" + { + return Err(invalid()); + } + Ok(()) +} +fn valid_uuid(value: &str) -> bool { + value.len() == 36 + && value.bytes().enumerate().all(|(i, b)| { + if [8, 13, 18, 23].contains(&i) { + b == b'-' + } else { + b.is_ascii_digit() || (b'a'..=b'f').contains(&b) + } + }) +} +fn invalid() -> VssError { + VssError::InvalidData { + error_details: "Invalid migration status".into(), + } +} + +/// Validate wire values before prost narrows integer types or merges duplicate +/// fields. Unknown fields fail closed until their semantics are supported. +fn validate_wire(mut bytes: &[u8], state: bool) -> Result<(), VssError> { + let mut seen = 0u16; + let mut objects = 0; + while !bytes.is_empty() { + let (tag, wire) = decode_key(&mut bytes).map_err(|_| invalid())?; + if tag == 0 || tag > if state { 9 } else { 2 } { + return Err(invalid()); + } + if state && tag == 7 { + objects += 1; + if objects > 32 { + return Err(invalid()); + } + } else if seen & (1 << tag) != 0 { + return Err(invalid()); + } + seen |= 1 << tag; + if state && tag == 1 { + if wire != WireType::Varint || decode_varint(&mut bytes).map_err(|_| invalid())? > 1 { + return Err(invalid()); + } + } else { + if wire != WireType::LengthDelimited { + return Err(invalid()); + } + let length = usize::try_from(decode_varint(&mut bytes).map_err(|_| invalid())?) + .map_err(|_| invalid())?; + if length > bytes.len() { + return Err(invalid()); + } + let (field, tail) = bytes.split_at(length); + if state && (5..=7).contains(&tag) { + validate_wire(field, false)?; + } + bytes = tail; + } + } + Ok(()) +} + +// Pinned to the preserved VSS server candidate proto/vss.proto. These private +// messages intentionally add no UniFFI API or dependency on unpublished packages. +pub(crate) mod wire { + #[derive(Clone, PartialEq, prost::Message)] + pub struct Request { + #[prost(string, tag = "1")] + pub store_id: String, + } + #[derive(Clone, PartialEq, prost::Message)] + pub struct Archive { + #[prost(string, tag = "1")] + pub key: String, + #[prost(bytes = "vec", optional, tag = "2")] + pub ciphertext_sha256: Option>, + } + #[derive(Clone, PartialEq, prost::Message)] + pub struct Object { + #[prost(string, tag = "1")] + pub key: String, + #[prost(bytes = "vec", tag = "2")] + pub ciphertext_sha256: Vec, + } + #[derive(Clone, PartialEq, prost::Message)] + pub struct State { + #[prost(int64, tag = "1")] + pub epoch: i64, + #[prost(string, tag = "2")] + pub required_protocol: String, + #[prost(string, tag = "3")] + pub migration_id: String, + #[prost(bytes = "vec", tag = "4")] + pub proposal_sha256: Vec, + #[prost(message, optional, tag = "5")] + pub archive: Option, + #[prost(message, optional, tag = "6")] + pub initial_head: Option, + #[prost(message, repeated, tag = "7")] + pub objects: Vec, + #[prost(string, tag = "8")] + pub effective_principal: String, + #[prost(string, tag = "9")] + pub store_id: String, + } +} + +#[cfg(test)] +mod tests; diff --git a/src/migration_status/tests.rs b/src/migration_status/tests.rs new file mode 100644 index 0000000..f628570 --- /dev/null +++ b/src/migration_status/tests.rs @@ -0,0 +1,203 @@ +use super::*; + +fn legacy() -> wire::State { + wire::State { + effective_principal: "principal".into(), + store_id: "store".into(), + ..Default::default() + } +} +fn committed_state() -> wire::State { + wire::State { + epoch: 1, + required_protocol: PROTOCOL.into(), + migration_id: "00000000-0000-4000-8000-000000000001".into(), + proposal_sha256: vec![1; 32], + archive: Some(wire::Archive { + key: "old-wallet".into(), + ciphertext_sha256: Some(vec![2; 32]), + }), + initial_head: Some(wire::Object { + key: "head".into(), + ciphertext_sha256: vec![3; 32], + }), + objects: vec![wire::Object { + key: "part".into(), + ciphertext_sha256: vec![4; 32], + }], + ..legacy() + } +} +fn read(state: wire::State) -> Result { + decode(&state.encode_to_vec(), "store") +} + +#[test] +fn explicit_legacy_identity_is_required() { + let state = read(legacy()).unwrap(); + assert_eq!(state.effective_principal, "principal"); + assert!(matches!(state.policy, MigrationPolicy::Legacy)); + assert!(read(wire::State::default()).is_err()); + for s in [ + wire::State { + store_id: "another".into(), + ..legacy() + }, + wire::State { + effective_principal: String::new(), + ..legacy() + }, + wire::State { + effective_principal: "x".repeat(121), + ..legacy() + }, + wire::State { + migration_id: "anything".into(), + ..legacy() + }, + wire::State { + required_protocol: PROTOCOL.into(), + ..legacy() + }, + wire::State { + proposal_sha256: vec![1; 32], + ..legacy() + }, + wire::State { + archive: committed_state().archive, + ..legacy() + }, + wire::State { + initial_head: committed_state().initial_head, + ..legacy() + }, + wire::State { + objects: committed_state().objects, + ..legacy() + }, + ] { + assert!(read(s).is_err()); + } +} + +#[test] +fn committed_state_retains_exact_conditions_and_allows_ldk_only_tuple() { + let MigrationPolicy::Committed(state) = read(committed_state()).unwrap().policy else { + panic!("expected committed"); + }; + assert_eq!(state.proposal_sha256, [1; 32]); + assert_eq!(state.archive.unwrap().ciphertext_sha256, Some([2; 32])); + assert_eq!(state.initial_head.unwrap().ciphertext_sha256, [3; 32]); + assert_eq!(state.objects[0].ciphertext_sha256, [4; 32]); + let mut absent = committed_state(); + absent.archive.as_mut().unwrap().ciphertext_sha256 = None; + let MigrationPolicy::Committed(absent) = read(absent).unwrap().policy else { + panic!("expected committed"); + }; + assert!(absent.archive.unwrap().ciphertext_sha256.is_none()); + let mut ldk = committed_state(); + ldk.archive = None; + ldk.initial_head = None; + ldk.objects.clear(); + assert!(read(ldk).is_ok()); +} + +#[test] +fn ambiguous_committed_shapes_are_rejected() { + let mutations: &[fn(&mut wire::State)] = &[ + (|s| s.epoch = -1), + (|s| s.epoch = 2), + (|s| s.required_protocol = "future-v2".into()), + (|s| s.required_protocol.clear()), + (|s| s.migration_id = "A".repeat(36)), + (|s| s.migration_id.clear()), + (|s| s.proposal_sha256.pop().map(|_| ()).unwrap()), + (|s| s.archive.as_mut().unwrap().ciphertext_sha256 = Some(vec![])), + (|s| s.initial_head.as_mut().unwrap().ciphertext_sha256.clear()), + (|s| s.objects[0].ciphertext_sha256.push(1)), + (|s| s.objects.push(s.objects[0].clone())), + (|s| { + s.objects = (0..33) + .map(|i| wire::Object { + key: format!("part-{i}"), + ciphertext_sha256: vec![4; 32], + }) + .collect() + }), + (|s| s.archive.as_mut().unwrap().key = "head".into()), + (|s| s.objects[0].key = "head".into()), + (|s| s.objects[0].key = "old-wallet".into()), + (|s| s.objects[0].key = "global_version".into()), + (|s| s.objects[0].key = "vss_global_version".into()), + (|s| s.objects[0].key.clear()), + (|s| s.objects[0].key = "x".repeat(601)), + ]; + for (i, mutate) in mutations.iter().enumerate() { + let mut s = committed_state(); + mutate(&mut s); + assert!(read(s).is_err(), "mutation {i} admitted"); + } + let mut same = committed_state(); + same.objects[0] = wire::Object { + key: "old-wallet".into(), + ciphertext_sha256: vec![2; 32], + }; + assert!(read(same).is_ok()); +} + +#[test] +fn duplicate_unknown_overflow_and_truncated_wire_fields_fail_closed() { + let base = legacy().encode_to_vec(); + for suffix in [ + vec![0x42, 1, b'x'], + vec![0x4a, 5, b's', b't', b'o', b'r', b'e'], + vec![0x08, 0x00, 0x08, 0x00], + vec![0x50, 0], + vec![0x08, 0x80, 0x80, 0x80, 0x80, 0x10], + vec![ + 0x08, 0xff, 0xff, 0xff, 0xff, 0xff, 0xff, 0xff, 0xff, 0xff, 0x01, + ], + vec![0x42, 0xff], + vec![0x08, 0x80], + vec![0x0a, 0], + ] { + let mut bytes = base.clone(); + bytes.extend(suffix); + assert!(decode(&bytes, "store").is_err()); + } + let good = committed_state().encode_to_vec(); + for n in 0..good.len() { + assert!(decode(&good[..n], "store").is_err(), "truncated at {n}"); + } + // Prost would merge these nested singular keys and keep the last digest. + let mut state = committed_state(); + state.archive = None; + let mut control = state.encode_to_vec(); + control.extend([0x2a, 3, 0x0a, 1, b'b']); + assert!(decode(&control, "store").is_ok()); + let mut bytes = state.encode_to_vec(); + bytes.extend([0x2a, 6, 0x0a, 1, b'a', 0x0a, 1, b'b']); + assert!(decode(&bytes, "store").is_err()); + let mut max = committed_state(); + max.objects = (0..32) + .map(|i| wire::Object { + key: format!("part-{i}"), + ciphertext_sha256: vec![4; 32], + }) + .collect(); + assert!(read(max).is_ok()); +} + +#[test] +fn legacy_identities_preserve_character_bounded_storage_names() { + let identity = "é".repeat(120); + let state = wire::State { + effective_principal: identity.clone(), + store_id: identity.clone(), + ..Default::default() + }; + assert!(decode(&state.encode_to_vec(), &identity).is_ok()); + let mut committed = committed_state(); + committed.effective_principal = identity; + assert!(read(committed).is_err()); +} diff --git a/src/receipt.rs b/src/receipt.rs index 98e8135..b42bd75 100644 --- a/src/receipt.rs +++ b/src/receipt.rs @@ -19,7 +19,7 @@ const MAX_RESPONSE_BYTES: u32 = 64 * 1024 * 1024; pub(crate) struct ReceiptReader { http: reqwest::Client, - url: String, + base_url: String, headers: Arc, } @@ -36,7 +36,7 @@ impl ReceiptReader { .map_err(|_| network_error())?; Ok(Self { http, - url: format!("{}/getObject", base_url.trim_end_matches('/')), + base_url: base_url.trim_end_matches('/').to_owned(), headers, }) } @@ -46,22 +46,35 @@ impl ReceiptReader { request: GetObjectRequest, limit: u32, ) -> Result, VssError> { + let (status, bytes) = self + .post("getObject", request.encode_to_vec(), limit) + .await?; + decode_response(status, &bytes) + } + + /// One bounded storage request under this client's original provider. + pub(crate) async fn post( + &self, + route: &str, + body: Vec, + limit: u32, + ) -> Result<(reqwest::StatusCode, Vec), VssError> { if limit == 0 || limit > MAX_RESPONSE_BYTES { return Err(invalid( "Response limit must be between 1 and 67108864 bytes", )); } - tokio::time::timeout(TIMEOUT, self.request(request, limit as usize)) + tokio::time::timeout(TIMEOUT, self.request(route, body, limit as usize)) .await .map_err(|_| network_error())? } async fn request( &self, - request: GetObjectRequest, + route: &str, + body: Vec, limit: usize, - ) -> Result, VssError> { - let body = request.encode_to_vec(); + ) -> Result<(reqwest::StatusCode, Vec), VssError> { let headers = self .headers .get_headers(&body) @@ -69,7 +82,7 @@ impl ReceiptReader { .map_err(credential_error)?; let mut request = self .http - .post(&self.url) + .post(format!("{}/{route}", self.base_url)) .header("Content-Type", "application/octet-stream"); for (name, value) in headers { request = request.header(name, value); @@ -81,7 +94,7 @@ impl ReceiptReader { .map_err(|_| network_error())?; let status = response.status(); let bytes = bounded_body(response, limit).await?; - decode_response(status, &bytes) + Ok((status, bytes)) } }