diff --git a/src/client.rs b/src/client.rs index e84749fc0..abf6132be 100644 --- a/src/client.rs +++ b/src/client.rs @@ -2506,10 +2506,16 @@ impl Client { } continue; } - let is_db_locked = e.downcast_ref::() - .is_some_and(|se| matches!(se, wacore::store::error::StoreError::Database(msg) if msg.contains("locked") || msg.contains("busy"))) + let is_db_locked = e + .downcast_ref::() + .is_some_and(|se| se.is_database_busy_or_locked()) || e.downcast_ref::() - .is_some_and(|ase| matches!(ase, crate::appstate_sync::AppStateSyncError::Store(wacore::store::error::StoreError::Database(msg)) if msg.contains("locked") || msg.contains("busy"))); + .is_some_and(|ase| match ase { + crate::appstate_sync::AppStateSyncError::Store(se) => { + se.is_database_busy_or_locked() + } + _ => false, + }); if is_db_locked && attempt < APP_STATE_RETRY_MAX_ATTEMPTS { let backoff = Duration::from_millis(200 * attempt as u64 + 150); warn!(target: "Client/AppState", "Attempt {} for {:?} failed due to locked DB; backing off {:?} and retrying", attempt, name, backoff); @@ -3608,7 +3614,7 @@ impl Client { let plaintext_buf = wacore_binary::marshal::marshal_auto(&node).map_err(|e| { error!("Failed to marshal node: {e:?}"); - SocketError::Crypto("Marshal error".to_string()) + SocketError::Marshal(e) })?; self.send_raw_bytes(plaintext_buf).await diff --git a/src/features/mex.rs b/src/features/mex.rs index 001b5068f..8cc72839f 100644 --- a/src/features/mex.rs +++ b/src/features/mex.rs @@ -7,6 +7,7 @@ use crate::request::IqError; use serde_json::Value; use thiserror::Error; use wacore::iq::mex::MexQuerySpec; +use wacore_binary::jid::JidError; // Re-export types from wacore pub use wacore::iq::mex::{MexDoc, MexErrorExtensions, MexGraphQLError, MexResponse}; @@ -14,16 +15,21 @@ pub use wacore::iq::mex::{MexDoc, MexErrorExtensions, MexGraphQLError, MexRespon /// Error types for MEX operations. #[derive(Debug, Error)] pub enum MexError { + /// Payload missing or otherwise malformed in a way that has no underlying + /// typed source (descriptive message only — e.g. "missing data"). #[error("MEX payload parsing error: {0}")] PayloadParsing(String), + #[error("MEX payload contained an invalid JID")] + InvalidJid(#[from] JidError), + #[error("MEX extension error: code={code}, message='{message}'")] ExtensionError { code: i32, message: String }, - #[error("IQ request failed: {0}")] + #[error("IQ request failed")] Request(#[from] IqError), - #[error("JSON error: {0}")] + #[error("JSON error")] Json(#[from] serde_json::Error), } @@ -237,4 +243,28 @@ mod tests { assert!(ext.is_retryable.is_none()); assert!(ext.severity.is_none()); } + + #[test] + fn invalid_jid_preserves_jid_error_source() { + let raw: Result = "not-a-valid-jid".parse(); + let jid_err = raw.unwrap_err(); + let me: MexError = jid_err.into(); + let src = std::error::Error::source(&me).expect("source preserved"); + let inner = src + .downcast_ref::() + .expect("downcasts to JidError"); + assert!(matches!(inner, JidError::InvalidFormat(_))); + } + + #[test] + fn request_preserves_iq_error_source() { + let iq = IqError::ServerError { + code: 404, + text: "not-found".into(), + }; + let me: MexError = iq.into(); + let src = std::error::Error::source(&me).expect("source preserved"); + let inner = src.downcast_ref::().expect("downcasts to IqError"); + assert!(matches!(inner, IqError::ServerError { code: 404, .. })); + } } diff --git a/src/features/newsletter.rs b/src/features/newsletter.rs index 421bac34a..eec257668 100644 --- a/src/features/newsletter.rs +++ b/src/features/newsletter.rs @@ -420,9 +420,7 @@ fn parse_newsletter_metadata(value: &serde_json::Value) -> Result KeepaliveResult { match e { IqError::Socket(_) + | IqError::EncryptSend(_) + | IqError::ClientState(_) | IqError::Disconnected(_) | IqError::NotConnected | IqError::InternalChannelClosed @@ -229,7 +231,7 @@ mod tests { #[test] fn test_classify_socket_error_is_fatal() { assert_eq!( - classify_keepalive_error(&IqError::Socket(SocketError::Crypto("test".to_string()))), + classify_keepalive_error(&IqError::Socket(SocketError::SocketClosed)), KeepaliveResult::FatalFailure, ); } diff --git a/src/pair_code.rs b/src/pair_code.rs index fe25662b7..fa8bcfa20 100644 --- a/src/pair_code.rs +++ b/src/pair_code.rs @@ -52,13 +52,25 @@ use log::{error, info, warn}; use std::sync::Arc; use wacore::libsignal::protocol::KeyPair; -use wacore::pair_code::{PairCodeError, PairCodeState, PairCodeUtils, resolve_companion_platform}; +use wacore::pair_code::{PairCodeState, PairCodeUtils, resolve_companion_platform}; use wacore_binary::Jid; use wacore_binary::{NodeContent, NodeContentRef, NodeRef}; -// Re-export types for user convenience pub use wacore::companion_reg::CompanionWebClientType; -pub use wacore::pair_code::PairCodeOptions; +pub use wacore::pair_code::{PairCodeError, PairCodeOptions}; + +/// Errors raised by the high-level pair-code flow. +/// +/// Wraps `wacore::pair_code::PairCodeError` (validation, key derivation, bundle +/// building) and adds the IQ transport layer via `RequestFailed`. +#[derive(Debug, thiserror::Error)] +pub enum PairError { + #[error(transparent)] + PairCode(#[from] PairCodeError), + + #[error("pair-code IQ request failed")] + RequestFailed(#[from] IqError), +} impl Client { /// Initiates pair code authentication as an alternative to QR code pairing. @@ -99,7 +111,7 @@ impl Client { pub async fn pair_with_code( self: &Arc, options: PairCodeOptions, - ) -> Result { + ) -> Result { // Strip non-digit characters from phone number (allows "+1-555-123-4567" format) let phone_number: String = options .phone_number @@ -109,20 +121,20 @@ impl Client { // Validate phone number if phone_number.is_empty() { - return Err(PairCodeError::PhoneNumberRequired); + return Err(PairCodeError::PhoneNumberRequired.into()); } if phone_number.len() < 7 { - return Err(PairCodeError::PhoneNumberTooShort); + return Err(PairCodeError::PhoneNumberTooShort.into()); } if phone_number.starts_with('0') { - return Err(PairCodeError::PhoneNumberNotInternational); + return Err(PairCodeError::PhoneNumberNotInternational.into()); } // Generate or validate code let code = match &options.custom_code { Some(custom) => { if !PairCodeUtils::validate_code(custom) { - return Err(PairCodeError::InvalidCustomCode); + return Err(PairCodeError::InvalidCustomCode.into()); } custom.to_uppercase() } @@ -192,12 +204,8 @@ impl Client { timeout: Some(std::time::Duration::from_secs(30)), }; - let response = self - .send_iq(query) - .await - .map_err(|e: IqError| PairCodeError::RequestFailed(e.to_string()))?; + let response = self.send_iq(query).await?; - // Extract pairing ref from response let pairing_ref = PairCodeUtils::parse_companion_hello_response(response.get()) .ok_or(PairCodeError::MissingPairingRef)?; @@ -373,3 +381,36 @@ pub(crate) async fn handle_pair_code_notification( true } + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn pair_error_request_failed_preserves_iq_source() { + let iq = IqError::ServerError { + code: 400, + text: "bad-request".into(), + }; + let pe: PairError = iq.into(); + let src = std::error::Error::source(&pe).expect("source preserved"); + let downcast = src.downcast_ref::().expect("downcasts to IqError"); + assert!(matches!(downcast, IqError::ServerError { code: 400, .. })); + } + + #[test] + fn pair_error_paircode_transparent_walks_to_curve_error() { + use wacore::libsignal::protocol::CurveError; + // Wrap a wacore PairCodeError that itself carries a CurveError source. + // Because PairError::PairCode is `transparent`, walking source() once + // skips the transparent layer and lands directly on the CurveError. + let pe: PairError = + PairCodeError::EphemeralKeyAgreement(CurveError::NoKeyTypeIdentifier).into(); + assert_eq!(pe.to_string(), "ephemeral key agreement failed"); + let src = std::error::Error::source(&pe).expect("source preserved"); + let curve = src + .downcast_ref::() + .expect("downcasts to CurveError through transparent wrapper"); + assert!(matches!(curve, CurveError::NoKeyTypeIdentifier)); + } +} diff --git a/src/request.rs b/src/request.rs index 0dacf5a50..0edf44687 100644 --- a/src/request.rs +++ b/src/request.rs @@ -1,5 +1,6 @@ use crate::client::Client; -use crate::socket::error::SocketError; +use crate::client::ClientError; +use crate::socket::error::{EncryptSendError, SocketError}; use futures::FutureExt; use std::sync::Arc; use std::sync::atomic::Ordering; @@ -14,19 +15,23 @@ pub use wacore::request::{InfoQuery, InfoQueryType, RequestUtils}; pub enum IqError { #[error("IQ request timed out")] Timeout, - #[error("Client is not connected")] + #[error("client is not connected")] NotConnected, - #[error("Socket error: {0}")] + #[error("socket error")] Socket(#[from] SocketError), - #[error("Received disconnect node during IQ wait: {0:?}")] + #[error("encrypted send pipeline failed")] + EncryptSend(#[from] EncryptSendError), + #[error("client state prevented send")] + ClientState(#[source] ClientError), + #[error("received disconnect node during IQ wait: {0:?}")] Disconnected(Node), - #[error("Received a server error response: code={code}, text='{text}'")] + #[error("received a server error response: code={code}, text='{text}'")] ServerError { code: u16, text: String }, - #[error("Internal channel closed unexpectedly")] + #[error("internal channel closed unexpectedly")] InternalChannelClosed, - #[error("Failed to encode IQ request: {0}")] - EncodeError(anyhow::Error), - #[error("Failed to parse IQ response: {0}")] + #[error("failed to encode IQ request")] + EncodeError(#[source] anyhow::Error), + #[error("failed to parse IQ response")] ParseError(#[from] anyhow::Error), } @@ -40,7 +45,6 @@ impl From for IqError { Self::ServerError { code, text } } wacore::request::IqError::InternalChannelClosed => Self::InternalChannelClosed, - wacore::request::IqError::Network(msg) => Self::Socket(SocketError::Crypto(msg)), } } } @@ -225,9 +229,12 @@ impl Client { if let Err(e) = send_fn.await { self.response_waiters.lock().await.remove(&req_id); return match e { - crate::client::ClientError::Socket(s_err) => Err(IqError::Socket(s_err)), - crate::client::ClientError::NotConnected => Err(IqError::NotConnected), - _ => Err(IqError::Socket(SocketError::Crypto(e.to_string()))), + ClientError::Socket(s_err) => Err(IqError::Socket(s_err)), + ClientError::EncryptSend(es_err) => Err(IqError::EncryptSend(es_err)), + ClientError::NotConnected => Err(IqError::NotConnected), + other @ (ClientError::AlreadyConnected | ClientError::NotLoggedIn) => { + Err(IqError::ClientState(other)) + } }; } diff --git a/src/socket/error.rs b/src/socket/error.rs index 9b6910607..8e3d41f35 100644 --- a/src/socket/error.rs +++ b/src/socket/error.rs @@ -1,15 +1,17 @@ use thiserror::Error; +use wacore::handshake::NoiseError; +use wacore_binary::error::BinaryError; #[derive(Debug, Error)] pub enum SocketError { - #[error("Socket is closed")] + #[error("socket is closed")] SocketClosed, - #[error("Noise handshake failed: {0}")] - NoiseHandshake(String), - #[error("I/O error: {0}")] + #[error("I/O error")] Io(#[from] std::io::Error), - #[error("Crypto error: {0}")] - Crypto(String), + #[error("noise cipher operation failed")] + Cipher(#[from] NoiseError), + #[error("binary protocol marshalling failed")] + Marshal(#[source] BinaryError), } pub type Result = std::result::Result; @@ -80,3 +82,38 @@ impl EncryptSendError { ) } } + +#[cfg(test)] +mod tests { + use super::*; + use wacore::libsignal::crypto::CryptoProviderError; + + #[test] + fn cipher_preserves_noise_source_through_socket_error() { + let noise = NoiseError::Decrypt(CryptoProviderError::AuthFailed); + let se: SocketError = noise.into(); + // First hop: SocketError → NoiseError + let src = std::error::Error::source(&se).expect("source preserved"); + let ne = src + .downcast_ref::() + .expect("downcasts to NoiseError"); + assert!(matches!(ne, NoiseError::Decrypt(_))); + // Second hop: NoiseError → CryptoProviderError + let inner = std::error::Error::source(ne).expect("inner source preserved"); + let cpe = inner + .downcast_ref::() + .expect("downcasts to CryptoProviderError"); + assert!(matches!(cpe, CryptoProviderError::AuthFailed)); + } + + #[test] + fn marshal_preserves_binary_error_source() { + let be = BinaryError::InvalidNode; + let se = SocketError::Marshal(be); + let src = std::error::Error::source(&se).expect("source preserved"); + let inner = src + .downcast_ref::() + .expect("downcasts to BinaryError"); + assert!(matches!(inner, BinaryError::InvalidNode)); + } +} diff --git a/src/socket/noise_socket.rs b/src/socket/noise_socket.rs index b4b42fc78..baddb5efb 100644 --- a/src/socket/noise_socket.rs +++ b/src/socket/noise_socket.rs @@ -180,7 +180,7 @@ impl NoiseSocket { let counter = self.read_counter.fetch_add(1, Ordering::SeqCst); self.read_key .decrypt_in_place_with_counter(counter, &mut ciphertext) - .map_err(|e| SocketError::Crypto(e.to_string()))?; + .map_err(SocketError::Cipher)?; Ok(ciphertext) } } diff --git a/src/store/error.rs b/src/store/error.rs index f0944f10f..5e10b4934 100644 --- a/src/store/error.rs +++ b/src/store/error.rs @@ -1,2 +1,2 @@ // Re-export error types from wacore for compatibility -pub use wacore::store::error::{Result, StoreError, db_err}; +pub use wacore::store::error::{Result, StoreError}; diff --git a/src/store/persistence_manager.rs b/src/store/persistence_manager.rs index fcb62f63e..9fec958a4 100644 --- a/src/store/persistence_manager.rs +++ b/src/store/persistence_manager.rs @@ -1,4 +1,4 @@ -use super::error::{StoreError, db_err}; +use super::error::StoreError; use crate::store::Device; use crate::store::traits::Backend; use async_lock::RwLock; @@ -27,15 +27,15 @@ impl PersistenceManager { pub async fn new(backend: Arc) -> Result { debug!("PersistenceManager: Ensuring device row exists."); // Ensure a device row exists for this backend's device_id; create it if not. - let exists = backend.exists().await.map_err(db_err)?; + let exists = backend.exists().await?; if !exists { debug!("PersistenceManager: No device row found. Creating new device row."); - let id = backend.create().await.map_err(db_err)?; + let id = backend.create().await?; debug!("PersistenceManager: Created device row with id={id}."); } debug!("PersistenceManager: Attempting to load device data via Backend."); - let device_data_opt = backend.load().await.map_err(db_err)?; + let device_data_opt = backend.load().await?; let device = if let Some(serializable_device) = device_data_opt { debug!( @@ -104,7 +104,7 @@ impl PersistenceManager { if let Err(e) = self.backend.save(&serializable_device).await { // Restore dirty flag so the next tick retries the save self.dirty.store(true, Ordering::Release); - return Err(db_err(e)); + return Err(e); } debug!("Device state saved successfully."); } @@ -122,10 +122,7 @@ impl PersistenceManager { { // Ensure pending changes are saved first self.save_to_disk().await?; - self.backend - .snapshot_db(name, extra_content) - .await - .map_err(db_err) + self.backend.snapshot_db(name, extra_content).await } #[cfg(not(feature = "debug-snapshots"))] { @@ -242,10 +239,7 @@ impl PersistenceManager { &self, group_jid: &str, ) -> Result, StoreError> { - self.backend - .get_sender_key_devices(group_jid) - .await - .map_err(db_err) + self.backend.get_sender_key_devices(group_jid).await } pub async fn set_sender_key_status( @@ -253,17 +247,11 @@ impl PersistenceManager { group_jid: &str, entries: &[(&str, bool)], ) -> Result<(), StoreError> { - self.backend - .set_sender_key_status(group_jid, entries) - .await - .map_err(db_err) + self.backend.set_sender_key_status(group_jid, entries).await } pub async fn clear_sender_key_devices(&self, group_jid: &str) -> Result<(), StoreError> { - self.backend - .clear_sender_key_devices(group_jid) - .await - .map_err(db_err) + self.backend.clear_sender_key_devices(group_jid).await } } diff --git a/src/store/signal.rs b/src/store/signal.rs index 6ac558e85..be0582db4 100644 --- a/src/store/signal.rs +++ b/src/store/signal.rs @@ -220,9 +220,7 @@ impl IdentityKeyStore for Device { })?, ) .await - .map_err(|e| { - SignalProtocolError::InvalidState("backend put_identity", e.to_string()) - })?; + .map_err(|e| SignalProtocolError::BackendError("backend put_identity", Box::new(e)))?; match existing_identity_opt { None => Ok(IdentityChange::NewOrUnchanged), @@ -248,9 +246,7 @@ impl IdentityKeyStore for Device { .backend .load_identity(address.as_str()) .await - .map_err(|e| { - SignalProtocolError::InvalidState("backend get_identity", e.to_string()) - })?; + .map_err(|e| SignalProtocolError::BackendError("backend get_identity", Box::new(e)))?; match identity_bytes { Some(bytes) if !bytes.is_empty() => { @@ -479,7 +475,7 @@ impl SenderKeyStore for Device { self.backend .put_sender_key(sender_key_name.cache_key(), &serialized_record) .await - .map_err(|e| SignalProtocolError::InvalidState("store_sender_key", e.to_string())) + .map_err(|e| SignalProtocolError::BackendError("store_sender_key", Box::new(e))) } async fn load_sender_key( @@ -490,7 +486,7 @@ impl SenderKeyStore for Device { .backend .get_sender_key(sender_key_name.cache_key()) .await - .map_err(|e| SignalProtocolError::InvalidState("load_sender_key", e.to_string()))? + .map_err(|e| SignalProtocolError::BackendError("load_sender_key", Box::new(e)))? { Some(data) => { let record = SenderKeyRecord::deserialize(&data)?; diff --git a/src/store/signal_adapter.rs b/src/store/signal_adapter.rs index 0c058bdad..0098b4118 100644 --- a/src/store/signal_adapter.rs +++ b/src/store/signal_adapter.rs @@ -15,10 +15,11 @@ use wacore::libsignal::store::{ PreKeyStore as WacorePreKeyStore, SignedPreKeyStore as WacoreSignedPreKeyStore, }; -fn signal_err( - context: &'static str, -) -> impl FnOnce(E) -> SignalProtocolError { - move |e| SignalProtocolError::InvalidState(context, e.to_string()) +fn signal_err(context: &'static str) -> impl FnOnce(E) -> SignalProtocolError +where + E: Into>, +{ + move |e| SignalProtocolError::BackendError(context, e.into()) } #[derive(Clone)] diff --git a/storages/sqlite-storage/src/sqlite_store.rs b/storages/sqlite-storage/src/sqlite_store.rs index 9577c9363..a9921e8a3 100644 --- a/storages/sqlite-storage/src/sqlite_store.rs +++ b/storages/sqlite-storage/src/sqlite_store.rs @@ -27,7 +27,7 @@ enum DieselOrStore { impl From for StoreError { fn from(e: DieselOrStore) -> Self { match e { - DieselOrStore::Diesel(e) => StoreError::Database(e.to_string()), + DieselOrStore::Diesel(e) => StoreError::Database(Box::new(e)), DieselOrStore::Store(e) => e, } } @@ -124,7 +124,7 @@ impl diesel::r2d2::CustomizeConnection fn parse_database_path(database_url: &str) -> Result { // Reject in-memory databases if database_url == ":memory:" { - return Err(StoreError::Database( + return Err(StoreError::InvalidConfig( "Snapshot not supported for in-memory databases".to_string(), )); } @@ -140,7 +140,7 @@ fn parse_database_path(database_url: &str) -> Result { // Check if the resulting path looks like an in-memory marker if path == ":memory:" || path.starts_with(":memory:?") { - return Err(StoreError::Database( + return Err(StoreError::InvalidConfig( "Snapshot not supported for in-memory databases".to_string(), )); } @@ -158,25 +158,25 @@ impl SqliteStore { .max_size(pool_size) .connection_customizer(Box::new(ConnectionOptions)) .build(manager) - .map_err(|e| StoreError::Connection(e.to_string()))?; + .map_err(|e| StoreError::Connection(Box::new(e)))?; let pool_clone = pool.clone(); tokio::task::spawn_blocking(move || -> std::result::Result<(), StoreError> { let mut conn = pool_clone .get() - .map_err(|e| StoreError::Connection(e.to_string()))?; + .map_err(|e| StoreError::Connection(Box::new(e)))?; diesel::sql_query("PRAGMA journal_mode = WAL;") .execute(&mut conn) - .map_err(|e| StoreError::Database(e.to_string()))?; + .map_err(|e| StoreError::Database(Box::new(e)))?; conn.run_pending_migrations(MIGRATIONS) - .map_err(|e| StoreError::Migration(e.to_string()))?; + .map_err(StoreError::Migration)?; Ok(()) }) .await - .map_err(|e| StoreError::Database(e.to_string()))??; + .map_err(|e| StoreError::Database(Box::new(e)))??; let database_path = parse_database_path(database_url)?; @@ -211,14 +211,14 @@ impl SqliteStore { .clone() .acquire_owned() .await - .map_err(|e| StoreError::Database(format!("Semaphore error: {}", e)))?; + .map_err(|e| StoreError::Database(Box::new(e)))?; let result = tokio::task::spawn_blocking(move || { let res = f(); drop(permit); res }) .await - .map_err(|e| StoreError::Database(e.to_string()))??; + .map_err(|e| StoreError::Database(Box::new(e)))??; Ok(result) } @@ -240,7 +240,7 @@ impl SqliteStore { .clone() .acquire_owned() .await - .map_err(|e| StoreError::Database(format!("Semaphore error: {}", e)))?; + .map_err(|e| StoreError::Database(Box::new(e)))?; let pool = self.pool.clone(); let op = make_op(); @@ -250,7 +250,7 @@ impl SqliteStore { let _permit = permit; let mut conn = pool .get() - .map_err(|e| DieselOrStore::Store(StoreError::Connection(e.to_string())))?; + .map_err(|e| DieselOrStore::Store(StoreError::Connection(Box::new(e))))?; op(&mut conn).map_err(DieselOrStore::Diesel) }) .await; @@ -264,14 +264,13 @@ impl SqliteStore { tokio::time::sleep(tokio::time::Duration::from_millis(delay_ms)).await; } Ok(Err(e)) => return Err(e.into()), - Err(e) => return Err(StoreError::Database(e.to_string())), + Err(e) => return Err(StoreError::Database(Box::new(e))), } } - Err(StoreError::Database(format!( - "{} exhausted retries", - op_name - ))) + Err(StoreError::RetriesExhausted { + op: op_name.to_string(), + }) } fn serialize_keypair(&self, key_pair: &KeyPair) -> Result> { @@ -283,16 +282,16 @@ impl SqliteStore { fn deserialize_keypair(&self, bytes: &[u8]) -> Result { if bytes.len() != 64 { - return Err(StoreError::Serialization(format!( + return Err(StoreError::Validation(format!( "Invalid KeyPair length: {}", bytes.len() ))); } let private_key = PrivateKey::deserialize(&bytes[0..32]) - .map_err(|e| StoreError::Serialization(e.to_string()))?; + .map_err(|e| StoreError::Serialization(Box::new(e)))?; let public_key = PublicKey::from_djb_public_key_bytes(&bytes[32..64]) - .map_err(|e| StoreError::Serialization(e.to_string()))?; + .map_err(|e| StoreError::Serialization(Box::new(e)))?; Ok(KeyPair::new(public_key, private_key)) } @@ -485,18 +484,18 @@ impl SqliteStore { tokio::task::spawn_blocking(move || -> Result { let mut conn = pool .get() - .map_err(|e| StoreError::Connection(e.to_string()))?; + .map_err(|e| StoreError::Connection(Box::new(e)))?; let count: i64 = device::table .filter(device::id.eq(device_id)) .count() .get_result(&mut conn) - .map_err(|e| StoreError::Database(e.to_string()))?; + .map_err(|e| StoreError::Database(Box::new(e)))?; Ok(count > 0) }) .await - .map_err(|e| StoreError::Database(e.to_string()))? + .map_err(|e| StoreError::Database(Box::new(e)))? } pub async fn load_device_data_for_device(&self, device_id: i32) -> Result> { @@ -506,16 +505,16 @@ impl SqliteStore { let row = tokio::task::spawn_blocking(move || -> Result> { let mut conn = pool .get() - .map_err(|e| StoreError::Connection(e.to_string()))?; + .map_err(|e| StoreError::Connection(Box::new(e)))?; let result = device::table .filter(device::id.eq(device_id)) .first::(&mut conn) .optional() - .map_err(|e| StoreError::Database(e.to_string()))?; + .map_err(|e| StoreError::Database(Box::new(e)))?; Ok(result) }) .await - .map_err(|e| StoreError::Database(e.to_string()))??; + .map_err(|e| StoreError::Database(Box::new(e)))??; if let Some(row) = row { let pn = if !row.pn.is_empty() { @@ -535,18 +534,19 @@ impl SqliteStore { let signed_pre_key_signature: [u8; 64] = row.signed_pre_key_signature.try_into().map_err(|_| { - StoreError::Serialization("Invalid signed_pre_key_signature length".to_string()) + StoreError::Validation("Invalid signed_pre_key_signature length".to_string()) })?; - let adv_secret_key: [u8; 32] = row.adv_secret_key.try_into().map_err(|_| { - StoreError::Serialization("Invalid adv_secret_key length".to_string()) - })?; + let adv_secret_key: [u8; 32] = row + .adv_secret_key + .try_into() + .map_err(|_| StoreError::Validation("Invalid adv_secret_key length".to_string()))?; let account = row .account .map(|data| { wacore::store::device::account_serde::from_bytes(&data) - .map_err(|e| StoreError::Serialization(e.to_string())) + .map_err(|e| StoreError::Serialization(Box::new(e))) }) .transpose()?; @@ -594,10 +594,11 @@ impl SqliteStore { const MAX_RETRIES: u32 = 5; for attempt in 0..=MAX_RETRIES { - let permit = - db_semaphore.clone().acquire_owned().await.map_err(|e| { - StoreError::Database(format!("Failed to acquire semaphore: {}", e)) - })?; + let permit = db_semaphore + .clone() + .acquire_owned() + .await + .map_err(|e| StoreError::Database(Box::new(e)))?; let pool_clone = pool.clone(); let address_clone = address_owned.clone(); @@ -607,7 +608,7 @@ impl SqliteStore { tokio::task::spawn_blocking(move || -> std::result::Result<(), DieselOrStore> { let mut conn = pool_clone .get() - .map_err(|e| DieselOrStore::Store(StoreError::Connection(e.to_string())))?; + .map_err(|e| DieselOrStore::Store(StoreError::Connection(Box::new(e))))?; diesel::insert_into(identities::table) .values(( identities::address.eq(address_clone), @@ -640,14 +641,13 @@ impl SqliteStore { continue; } Ok(Err(e)) => return Err(e.into()), - Err(e) => return Err(StoreError::Database(format!("Task join error: {}", e))), + Err(e) => return Err(StoreError::Database(Box::new(e))), } } - Err(StoreError::Database(format!( - "Identity write failed after {} attempts", - MAX_RETRIES + 1 - ))) + Err(StoreError::RetriesExhausted { + op: format!("identity_write (after {} attempts)", MAX_RETRIES + 1), + }) } pub async fn delete_identity_for_device(&self, address: &str, device_id: i32) -> Result<()> { @@ -657,18 +657,18 @@ impl SqliteStore { tokio::task::spawn_blocking(move || -> Result<()> { let mut conn = pool .get() - .map_err(|e| StoreError::Connection(e.to_string()))?; + .map_err(|e| StoreError::Connection(Box::new(e)))?; diesel::delete( identities::table .filter(identities::address.eq(address_owned)) .filter(identities::device_id.eq(device_id)), ) .execute(&mut conn) - .map_err(|e| StoreError::Database(e.to_string()))?; + .map_err(|e| StoreError::Database(Box::new(e)))?; Ok(()) }) .await - .map_err(|e| StoreError::Database(e.to_string()))??; + .map_err(|e| StoreError::Database(Box::new(e)))??; Ok(()) } @@ -684,14 +684,14 @@ impl SqliteStore { .with_semaphore(move || -> Result>> { let mut conn = pool .get() - .map_err(|e| StoreError::Connection(e.to_string()))?; + .map_err(|e| StoreError::Connection(Box::new(e)))?; let res: Option> = identities::table .select(identities::key) .filter(identities::address.eq(address)) .filter(identities::device_id.eq(device_id)) .first(&mut conn) .optional() - .map_err(|e| StoreError::Database(e.to_string()))?; + .map_err(|e| StoreError::Database(Box::new(e)))?; Ok(res) }) .await?; @@ -710,14 +710,14 @@ impl SqliteStore { .with_semaphore(move || -> Result>> { let mut conn = pool .get() - .map_err(|e| StoreError::Connection(e.to_string()))?; + .map_err(|e| StoreError::Connection(Box::new(e)))?; let res: Option> = sessions::table .select(sessions::record) .filter(sessions::address.eq(address_for_query.clone())) .filter(sessions::device_id.eq(device_id)) .first(&mut conn) .optional() - .map_err(|e| StoreError::Database(e.to_string()))?; + .map_err(|e| StoreError::Database(Box::new(e)))?; Ok(res) }) @@ -740,10 +740,11 @@ impl SqliteStore { const MAX_RETRIES: u32 = 5; for attempt in 0..=MAX_RETRIES { - let permit = - db_semaphore.clone().acquire_owned().await.map_err(|e| { - StoreError::Database(format!("Failed to acquire semaphore: {}", e)) - })?; + let permit = db_semaphore + .clone() + .acquire_owned() + .await + .map_err(|e| StoreError::Database(Box::new(e)))?; let pool_clone = pool.clone(); let address_clone = address_owned.clone(); @@ -753,7 +754,7 @@ impl SqliteStore { tokio::task::spawn_blocking(move || -> std::result::Result<(), DieselOrStore> { let mut conn = pool_clone .get() - .map_err(|e| DieselOrStore::Store(StoreError::Connection(e.to_string())))?; + .map_err(|e| DieselOrStore::Store(StoreError::Connection(Box::new(e))))?; diesel::insert_into(sessions::table) .values(( sessions::address.eq(address_clone), @@ -786,14 +787,13 @@ impl SqliteStore { continue; } Ok(Err(e)) => return Err(e.into()), - Err(e) => return Err(StoreError::Database(format!("Task join error: {}", e))), + Err(e) => return Err(StoreError::Database(Box::new(e))), } } - Err(StoreError::Database(format!( - "Session write failed after {} attempts", - MAX_RETRIES + 1 - ))) + Err(StoreError::RetriesExhausted { + op: format!("session_write (after {} attempts)", MAX_RETRIES + 1), + }) } pub async fn delete_session_for_device(&self, address: &str, device_id: i32) -> Result<()> { @@ -803,18 +803,18 @@ impl SqliteStore { tokio::task::spawn_blocking(move || -> Result<()> { let mut conn = pool .get() - .map_err(|e| StoreError::Connection(e.to_string()))?; + .map_err(|e| StoreError::Connection(Box::new(e)))?; diesel::delete( sessions::table .filter(sessions::address.eq(address_owned)) .filter(sessions::device_id.eq(device_id)), ) .execute(&mut conn) - .map_err(|e| StoreError::Database(e.to_string()))?; + .map_err(|e| StoreError::Database(Box::new(e)))?; Ok(()) }) .await - .map_err(|e| StoreError::Database(e.to_string()))??; + .map_err(|e| StoreError::Database(Box::new(e)))??; Ok(()) } @@ -831,7 +831,7 @@ impl SqliteStore { tokio::task::spawn_blocking(move || -> Result<()> { let mut conn = pool .get() - .map_err(|e| StoreError::Connection(e.to_string()))?; + .map_err(|e| StoreError::Connection(Box::new(e)))?; diesel::insert_into(sender_keys::table) .values(( sender_keys::address.eq(address), @@ -842,11 +842,11 @@ impl SqliteStore { .do_update() .set(sender_keys::record.eq(&record_vec)) .execute(&mut conn) - .map_err(|e| StoreError::Database(e.to_string()))?; + .map_err(|e| StoreError::Database(Box::new(e)))?; Ok(()) }) .await - .map_err(|e| StoreError::Database(e.to_string()))??; + .map_err(|e| StoreError::Database(Box::new(e)))??; Ok(()) } @@ -860,18 +860,18 @@ impl SqliteStore { tokio::task::spawn_blocking(move || -> Result>> { let mut conn = pool .get() - .map_err(|e| StoreError::Connection(e.to_string()))?; + .map_err(|e| StoreError::Connection(Box::new(e)))?; let res: Option> = sender_keys::table .select(sender_keys::record) .filter(sender_keys::address.eq(address)) .filter(sender_keys::device_id.eq(device_id)) .first(&mut conn) .optional() - .map_err(|e| StoreError::Database(e.to_string()))?; + .map_err(|e| StoreError::Database(Box::new(e)))?; Ok(res) }) .await - .map_err(|e| StoreError::Database(e.to_string()))? + .map_err(|e| StoreError::Database(Box::new(e)))? } pub async fn delete_sender_key_for_device(&self, address: &str, device_id: i32) -> Result<()> { @@ -880,18 +880,18 @@ impl SqliteStore { tokio::task::spawn_blocking(move || -> Result<()> { let mut conn = pool .get() - .map_err(|e| StoreError::Connection(e.to_string()))?; + .map_err(|e| StoreError::Connection(Box::new(e)))?; diesel::delete( sender_keys::table .filter(sender_keys::address.eq(address)) .filter(sender_keys::device_id.eq(device_id)), ) .execute(&mut conn) - .map_err(|e| StoreError::Database(e.to_string()))?; + .map_err(|e| StoreError::Database(Box::new(e)))?; Ok(()) }) .await - .map_err(|e| StoreError::Database(e.to_string()))??; + .map_err(|e| StoreError::Database(Box::new(e)))??; Ok(()) } @@ -906,22 +906,22 @@ impl SqliteStore { tokio::task::spawn_blocking(move || -> Result>> { let mut conn = pool .get() - .map_err(|e| StoreError::Connection(e.to_string()))?; + .map_err(|e| StoreError::Connection(Box::new(e)))?; let res: Option> = app_state_keys::table .select(app_state_keys::key_data) .filter(app_state_keys::key_id.eq(&key_id)) .filter(app_state_keys::device_id.eq(device_id)) .first(&mut conn) .optional() - .map_err(|e| StoreError::Database(e.to_string()))?; + .map_err(|e| StoreError::Database(Box::new(e)))?; Ok(res) }) .await - .map_err(|e| StoreError::Database(e.to_string()))??; + .map_err(|e| StoreError::Database(Box::new(e)))??; if let Some(data) = res { let (key, _) = bincode::serde::decode_from_slice(&data, bincode::config::standard()) - .map_err(|e| StoreError::Serialization(e.to_string()))?; + .map_err(|e| StoreError::Serialization(Box::new(e)))?; Ok(Some(key)) } else { Ok(None) @@ -937,11 +937,11 @@ impl SqliteStore { let pool = self.pool.clone(); let key_id = key_id.to_vec(); let data = bincode::serde::encode_to_vec(&key, bincode::config::standard()) - .map_err(|e| StoreError::Serialization(e.to_string()))?; + .map_err(|e| StoreError::Serialization(Box::new(e)))?; tokio::task::spawn_blocking(move || -> Result<()> { let mut conn = pool .get() - .map_err(|e| StoreError::Connection(e.to_string()))?; + .map_err(|e| StoreError::Connection(Box::new(e)))?; diesel::insert_into(app_state_keys::table) .values(( app_state_keys::key_id.eq(&key_id), @@ -952,11 +952,11 @@ impl SqliteStore { .do_update() .set(app_state_keys::key_data.eq(&data)) .execute(&mut conn) - .map_err(|e| StoreError::Database(e.to_string()))?; + .map_err(|e| StoreError::Database(Box::new(e)))?; Ok(()) }) .await - .map_err(|e| StoreError::Database(e.to_string()))??; + .map_err(|e| StoreError::Database(Box::new(e)))??; Ok(()) } @@ -969,18 +969,18 @@ impl SqliteStore { tokio::task::spawn_blocking(move || -> Result>> { let mut conn = pool .get() - .map_err(|e| StoreError::Connection(e.to_string()))?; + .map_err(|e| StoreError::Connection(Box::new(e)))?; let res: Option> = app_state_keys::table .select(app_state_keys::key_id) .filter(app_state_keys::device_id.eq(device_id)) .order(app_state_keys::key_id.desc()) .first(&mut conn) .optional() - .map_err(|e| StoreError::Database(e.to_string()))?; + .map_err(|e| StoreError::Database(Box::new(e)))?; Ok(res) }) .await - .map_err(|e| StoreError::Database(e.to_string()))??; + .map_err(|e| StoreError::Database(Box::new(e)))??; Ok(res) } @@ -995,22 +995,22 @@ impl SqliteStore { tokio::task::spawn_blocking(move || -> Result>> { let mut conn = pool .get() - .map_err(|e| StoreError::Connection(e.to_string()))?; + .map_err(|e| StoreError::Connection(Box::new(e)))?; let res: Option> = app_state_versions::table .select(app_state_versions::state_data) .filter(app_state_versions::name.eq(name)) .filter(app_state_versions::device_id.eq(device_id)) .first(&mut conn) .optional() - .map_err(|e| StoreError::Database(e.to_string()))?; + .map_err(|e| StoreError::Database(Box::new(e)))?; Ok(res) }) .await - .map_err(|e| StoreError::Database(e.to_string()))??; + .map_err(|e| StoreError::Database(Box::new(e)))??; if let Some(data) = res { let (state, _) = bincode::serde::decode_from_slice(&data, bincode::config::standard()) - .map_err(|e| StoreError::Serialization(e.to_string()))?; + .map_err(|e| StoreError::Serialization(Box::new(e)))?; Ok(state) } else { Ok(HashState::default()) @@ -1025,7 +1025,7 @@ impl SqliteStore { ) -> Result<()> { let name = name.to_string(); let data = bincode::serde::encode_to_vec(&state, bincode::config::standard()) - .map_err(|e| StoreError::Serialization(e.to_string()))?; + .map_err(|e| StoreError::Serialization(Box::new(e)))?; self.with_retry("set_app_state_version", || { let name = name.clone(); let data = data.clone(); @@ -1150,7 +1150,7 @@ impl SqliteStore { tokio::task::spawn_blocking(move || -> Result>> { let mut conn = pool .get() - .map_err(|e| StoreError::Connection(e.to_string()))?; + .map_err(|e| StoreError::Connection(Box::new(e)))?; let res: Option> = app_state_mutation_macs::table .select(app_state_mutation_macs::value_mac) .filter(app_state_mutation_macs::name.eq(&name)) @@ -1158,11 +1158,11 @@ impl SqliteStore { .filter(app_state_mutation_macs::device_id.eq(device_id)) .first(&mut conn) .optional() - .map_err(|e| StoreError::Database(e.to_string()))?; + .map_err(|e| StoreError::Database(Box::new(e)))?; Ok(res) }) .await - .map_err(|e| StoreError::Database(e.to_string()))? + .map_err(|e| StoreError::Database(Box::new(e)))? } } @@ -1181,7 +1181,7 @@ impl SignalStore for SqliteStore { match blob { None => Ok(None), Some(v) => Ok(Some(v.try_into().map_err(|v: Vec| { - StoreError::Serialization(format!( + StoreError::Validation(format!( "identity key for '{}' has invalid length {} (expected 32)", address, v.len() @@ -1209,14 +1209,14 @@ impl SignalStore for SqliteStore { self.with_semaphore(move || -> Result { let mut conn = pool .get() - .map_err(|e| StoreError::Connection(e.to_string()))?; + .map_err(|e| StoreError::Connection(Box::new(e)))?; let exists = diesel::select(diesel::dsl::exists( sessions::table .filter(sessions::address.eq(&address_owned)) .filter(sessions::device_id.eq(device_id)), )) .get_result(&mut conn) - .map_err(|e| StoreError::Database(e.to_string()))?; + .map_err(|e| StoreError::Database(Box::new(e)))?; Ok(exists) }) .await @@ -1241,10 +1241,11 @@ impl SignalStore for SqliteStore { const MAX_RETRIES: u32 = 5; for attempt in 0..=MAX_RETRIES { - let permit = - db_semaphore.clone().acquire_owned().await.map_err(|e| { - StoreError::Database(format!("Failed to acquire semaphore: {}", e)) - })?; + let permit = db_semaphore + .clone() + .acquire_owned() + .await + .map_err(|e| StoreError::Database(Box::new(e)))?; let pool_clone = pool.clone(); let record_clone = record.clone(); @@ -1253,7 +1254,7 @@ impl SignalStore for SqliteStore { tokio::task::spawn_blocking(move || -> std::result::Result<(), DieselOrStore> { let mut conn = pool_clone .get() - .map_err(|e| DieselOrStore::Store(StoreError::Connection(e.to_string())))?; + .map_err(|e| DieselOrStore::Store(StoreError::Connection(Box::new(e))))?; diesel::insert_into(prekeys::table) .values(( prekeys::id.eq(id as i32), @@ -1284,13 +1285,13 @@ impl SignalStore for SqliteStore { tokio::time::sleep(tokio::time::Duration::from_millis(delay_ms)).await; } Ok(Err(e)) => return Err(e.into()), - Err(e) => return Err(StoreError::Database(e.to_string())), + Err(e) => return Err(StoreError::Database(Box::new(e))), } } - Err(StoreError::Database( - "store_prekey exhausted retries".to_string(), - )) + Err(StoreError::RetriesExhausted { + op: "store_prekey".to_string(), + }) } async fn store_prekeys_batch(&self, keys: &[(u32, Bytes)], uploaded: bool) -> Result<()> { @@ -1306,10 +1307,11 @@ impl SignalStore for SqliteStore { const MAX_RETRIES: u32 = 5; for attempt in 0..=MAX_RETRIES { - let permit = - db_semaphore.clone().acquire_owned().await.map_err(|e| { - StoreError::Database(format!("Failed to acquire semaphore: {}", e)) - })?; + let permit = db_semaphore + .clone() + .acquire_owned() + .await + .map_err(|e| StoreError::Database(Box::new(e)))?; let pool_clone = pool.clone(); let keys_clone = keys.clone(); @@ -1318,7 +1320,7 @@ impl SignalStore for SqliteStore { tokio::task::spawn_blocking(move || -> std::result::Result<(), DieselOrStore> { let mut conn = pool_clone .get() - .map_err(|e| DieselOrStore::Store(StoreError::Connection(e.to_string())))?; + .map_err(|e| DieselOrStore::Store(StoreError::Connection(Box::new(e))))?; conn.transaction(|conn| { for (id, record) in &keys_clone { @@ -1354,13 +1356,13 @@ impl SignalStore for SqliteStore { tokio::time::sleep(tokio::time::Duration::from_millis(delay_ms)).await; } Ok(Err(e)) => return Err(e.into()), - Err(e) => return Err(StoreError::Database(e.to_string())), + Err(e) => return Err(StoreError::Database(Box::new(e))), } } - Err(StoreError::Database( - "store_prekeys_batch exhausted retries".to_string(), - )) + Err(StoreError::RetriesExhausted { + op: "store_prekeys_batch".to_string(), + }) } async fn load_prekey(&self, id: u32) -> Result> { @@ -1369,18 +1371,18 @@ impl SignalStore for SqliteStore { tokio::task::spawn_blocking(move || -> Result> { let mut conn = pool .get() - .map_err(|e| StoreError::Connection(e.to_string()))?; + .map_err(|e| StoreError::Connection(Box::new(e)))?; let res: Option> = prekeys::table .select(prekeys::key) .filter(prekeys::id.eq(id as i32)) .filter(prekeys::device_id.eq(device_id)) .first(&mut conn) .optional() - .map_err(|e| StoreError::Database(e.to_string()))?; + .map_err(|e| StoreError::Database(Box::new(e)))?; Ok(res.map(Bytes::from)) }) .await - .map_err(|e| StoreError::Database(e.to_string()))? + .map_err(|e| StoreError::Database(Box::new(e)))? } async fn load_prekeys_batch(&self, ids: &[u32]) -> Result> { @@ -1393,13 +1395,13 @@ impl SignalStore for SqliteStore { self.with_semaphore(move || -> Result> { let mut conn = pool .get() - .map_err(|e| StoreError::Connection(e.to_string()))?; + .map_err(|e| StoreError::Connection(Box::new(e)))?; let rows: Vec<(i32, Vec)> = prekeys::table .select((prekeys::id, prekeys::key)) .filter(prekeys::id.eq_any(&ids)) .filter(prekeys::device_id.eq(device_id)) .load(&mut conn) - .map_err(|e| StoreError::Database(e.to_string()))?; + .map_err(|e| StoreError::Database(Box::new(e)))?; Ok(rows .into_iter() .map(|(id, key)| (id as u32, Bytes::from(key))) @@ -1416,10 +1418,11 @@ impl SignalStore for SqliteStore { const MAX_RETRIES: u32 = 5; for attempt in 0..=MAX_RETRIES { - let permit = - db_semaphore.clone().acquire_owned().await.map_err(|e| { - StoreError::Database(format!("Failed to acquire semaphore: {}", e)) - })?; + let permit = db_semaphore + .clone() + .acquire_owned() + .await + .map_err(|e| StoreError::Database(Box::new(e)))?; let pool_clone = pool.clone(); @@ -1427,7 +1430,7 @@ impl SignalStore for SqliteStore { tokio::task::spawn_blocking(move || -> std::result::Result<(), DieselOrStore> { let mut conn = pool_clone .get() - .map_err(|e| DieselOrStore::Store(StoreError::Connection(e.to_string())))?; + .map_err(|e| DieselOrStore::Store(StoreError::Connection(Box::new(e))))?; diesel::delete( prekeys::table .filter(prekeys::id.eq(id as i32)) @@ -1450,13 +1453,13 @@ impl SignalStore for SqliteStore { tokio::time::sleep(tokio::time::Duration::from_millis(delay_ms)).await; } Ok(Err(e)) => return Err(e.into()), - Err(e) => return Err(StoreError::Database(e.to_string())), + Err(e) => return Err(StoreError::Database(Box::new(e))), } } - Err(StoreError::Database( - "remove_prekey exhausted retries".to_string(), - )) + Err(StoreError::RetriesExhausted { + op: "remove_prekey".to_string(), + }) } async fn get_max_prekey_id(&self) -> Result { @@ -1466,22 +1469,22 @@ impl SignalStore for SqliteStore { let _permit = db_semaphore .acquire() .await - .map_err(|e| StoreError::Database(format!("Failed to acquire semaphore: {}", e)))?; + .map_err(|e| StoreError::Database(Box::new(e)))?; tokio::task::spawn_blocking(move || -> Result { let mut conn = pool .get() - .map_err(|e| StoreError::Connection(e.to_string()))?; + .map_err(|e| StoreError::Connection(Box::new(e)))?; use diesel::dsl::max; let result: Option = prekeys::table .filter(prekeys::device_id.eq(device_id)) .select(max(prekeys::id)) .first(&mut conn) - .map_err(|e| StoreError::Database(e.to_string()))?; + .map_err(|e| StoreError::Database(Box::new(e)))?; Ok(result.unwrap_or(0) as u32) }) .await - .map_err(|e| StoreError::Database(e.to_string()))? + .map_err(|e| StoreError::Database(Box::new(e)))? } async fn store_signed_prekey(&self, id: u32, record: &[u8]) -> Result<()> { @@ -1493,10 +1496,11 @@ impl SignalStore for SqliteStore { const MAX_RETRIES: u32 = 5; for attempt in 0..=MAX_RETRIES { - let permit = - db_semaphore.clone().acquire_owned().await.map_err(|e| { - StoreError::Database(format!("Failed to acquire semaphore: {}", e)) - })?; + let permit = db_semaphore + .clone() + .acquire_owned() + .await + .map_err(|e| StoreError::Database(Box::new(e)))?; let pool_clone = pool.clone(); let record_clone = record.clone(); @@ -1505,7 +1509,7 @@ impl SignalStore for SqliteStore { tokio::task::spawn_blocking(move || -> std::result::Result<(), DieselOrStore> { let mut conn = pool_clone .get() - .map_err(|e| DieselOrStore::Store(StoreError::Connection(e.to_string())))?; + .map_err(|e| DieselOrStore::Store(StoreError::Connection(Box::new(e))))?; diesel::insert_into(signed_prekeys::table) .values(( signed_prekeys::id.eq(id as i32), @@ -1532,13 +1536,13 @@ impl SignalStore for SqliteStore { tokio::time::sleep(tokio::time::Duration::from_millis(delay_ms)).await; } Ok(Err(e)) => return Err(e.into()), - Err(e) => return Err(StoreError::Database(e.to_string())), + Err(e) => return Err(StoreError::Database(Box::new(e))), } } - Err(StoreError::Database( - "store_signed_prekey exhausted retries".to_string(), - )) + Err(StoreError::RetriesExhausted { + op: "store_signed_prekey".to_string(), + }) } async fn load_signed_prekey(&self, id: u32) -> Result>> { @@ -1547,18 +1551,18 @@ impl SignalStore for SqliteStore { tokio::task::spawn_blocking(move || -> Result>> { let mut conn = pool .get() - .map_err(|e| StoreError::Connection(e.to_string()))?; + .map_err(|e| StoreError::Connection(Box::new(e)))?; let res: Option> = signed_prekeys::table .select(signed_prekeys::record) .filter(signed_prekeys::id.eq(id as i32)) .filter(signed_prekeys::device_id.eq(device_id)) .first(&mut conn) .optional() - .map_err(|e| StoreError::Database(e.to_string()))?; + .map_err(|e| StoreError::Database(Box::new(e)))?; Ok(res) }) .await - .map_err(|e| StoreError::Database(e.to_string()))? + .map_err(|e| StoreError::Database(Box::new(e)))? } async fn load_all_signed_prekeys(&self) -> Result)>> { @@ -1567,19 +1571,19 @@ impl SignalStore for SqliteStore { tokio::task::spawn_blocking(move || -> Result)>> { let mut conn = pool .get() - .map_err(|e| StoreError::Connection(e.to_string()))?; + .map_err(|e| StoreError::Connection(Box::new(e)))?; let results: Vec<(i32, Vec)> = signed_prekeys::table .select((signed_prekeys::id, signed_prekeys::record)) .filter(signed_prekeys::device_id.eq(device_id)) .load(&mut conn) - .map_err(|e| StoreError::Database(e.to_string()))?; + .map_err(|e| StoreError::Database(Box::new(e)))?; Ok(results .into_iter() .map(|(id, record)| (id as u32, record)) .collect()) }) .await - .map_err(|e| StoreError::Database(e.to_string()))? + .map_err(|e| StoreError::Database(Box::new(e)))? } async fn remove_signed_prekey(&self, id: u32) -> Result<()> { @@ -1590,10 +1594,11 @@ impl SignalStore for SqliteStore { const MAX_RETRIES: u32 = 5; for attempt in 0..=MAX_RETRIES { - let permit = - db_semaphore.clone().acquire_owned().await.map_err(|e| { - StoreError::Database(format!("Failed to acquire semaphore: {}", e)) - })?; + let permit = db_semaphore + .clone() + .acquire_owned() + .await + .map_err(|e| StoreError::Database(Box::new(e)))?; let pool_clone = pool.clone(); @@ -1601,7 +1606,7 @@ impl SignalStore for SqliteStore { tokio::task::spawn_blocking(move || -> std::result::Result<(), DieselOrStore> { let mut conn = pool_clone .get() - .map_err(|e| DieselOrStore::Store(StoreError::Connection(e.to_string())))?; + .map_err(|e| DieselOrStore::Store(StoreError::Connection(Box::new(e))))?; diesel::delete( signed_prekeys::table .filter(signed_prekeys::id.eq(id as i32)) @@ -1624,13 +1629,13 @@ impl SignalStore for SqliteStore { tokio::time::sleep(tokio::time::Duration::from_millis(delay_ms)).await; } Ok(Err(e)) => return Err(e.into()), - Err(e) => return Err(StoreError::Database(e.to_string())), + Err(e) => return Err(StoreError::Database(Box::new(e))), } } - Err(StoreError::Database( - "remove_signed_prekey exhausted retries".to_string(), - )) + Err(StoreError::RetriesExhausted { + op: "remove_signed_prekey".to_string(), + }) } async fn put_sender_key(&self, address: &str, record: &[u8]) -> Result<()> { @@ -1708,20 +1713,20 @@ impl ProtocolStore for SqliteStore { tokio::task::spawn_blocking(move || -> Result> { let mut conn = pool .get() - .map_err(|e| StoreError::Connection(e.to_string()))?; + .map_err(|e| StoreError::Connection(Box::new(e)))?; let rows: Vec<(String, i32)> = sender_key_devices::table .select((sender_key_devices::device_jid, sender_key_devices::has_key)) .filter(sender_key_devices::group_jid.eq(&group_jid)) .filter(sender_key_devices::device_id.eq(device_id)) .load(&mut conn) - .map_err(|e| StoreError::Database(e.to_string()))?; + .map_err(|e| StoreError::Database(Box::new(e)))?; Ok(rows .into_iter() .map(|(jid, has_key)| (jid, has_key != 0)) .collect()) }) .await - .map_err(|e| StoreError::Database(e.to_string()))? + .map_err(|e| StoreError::Database(Box::new(e)))? } async fn set_sender_key_status(&self, group_jid: &str, entries: &[(&str, bool)]) -> Result<()> { @@ -1817,7 +1822,7 @@ impl ProtocolStore for SqliteStore { tokio::task::spawn_blocking(move || -> Result> { let mut conn = pool .get() - .map_err(|e| StoreError::Connection(e.to_string()))?; + .map_err(|e| StoreError::Connection(Box::new(e)))?; let row: Option<(String, String, i64, String, i64)> = lid_pn_mapping::table .select(( lid_pn_mapping::lid, @@ -1830,7 +1835,7 @@ impl ProtocolStore for SqliteStore { .filter(lid_pn_mapping::device_id.eq(device_id)) .first(&mut conn) .optional() - .map_err(|e| StoreError::Database(e.to_string()))?; + .map_err(|e| StoreError::Database(Box::new(e)))?; Ok(row.map( |(lid, phone_number, created_at, learning_source, updated_at)| LidPnMappingEntry { lid, @@ -1842,7 +1847,7 @@ impl ProtocolStore for SqliteStore { )) }) .await - .map_err(|e| StoreError::Database(e.to_string()))? + .map_err(|e| StoreError::Database(Box::new(e)))? } async fn get_pn_mapping(&self, phone: &str) -> Result> { @@ -1852,7 +1857,7 @@ impl ProtocolStore for SqliteStore { tokio::task::spawn_blocking(move || -> Result> { let mut conn = pool .get() - .map_err(|e| StoreError::Connection(e.to_string()))?; + .map_err(|e| StoreError::Connection(Box::new(e)))?; let row: Option<(String, String, i64, String, i64)> = lid_pn_mapping::table .select(( lid_pn_mapping::lid, @@ -1866,7 +1871,7 @@ impl ProtocolStore for SqliteStore { .order(lid_pn_mapping::updated_at.desc()) .first(&mut conn) .optional() - .map_err(|e| StoreError::Database(e.to_string()))?; + .map_err(|e| StoreError::Database(Box::new(e)))?; Ok(row.map( |(lid, phone_number, created_at, learning_source, updated_at)| LidPnMappingEntry { lid, @@ -1878,7 +1883,7 @@ impl ProtocolStore for SqliteStore { )) }) .await - .map_err(|e| StoreError::Database(e.to_string()))? + .map_err(|e| StoreError::Database(Box::new(e)))? } async fn put_lid_mapping(&self, entry: &LidPnMappingEntry) -> Result<()> { @@ -1930,7 +1935,7 @@ impl ProtocolStore for SqliteStore { tokio::task::spawn_blocking(move || -> Result> { let mut conn = pool .get() - .map_err(|e| StoreError::Connection(e.to_string()))?; + .map_err(|e| StoreError::Connection(Box::new(e)))?; let rows: Vec<(String, String, i64, String, i64)> = lid_pn_mapping::table .select(( lid_pn_mapping::lid, @@ -1941,7 +1946,7 @@ impl ProtocolStore for SqliteStore { )) .filter(lid_pn_mapping::device_id.eq(device_id)) .load(&mut conn) - .map_err(|e| StoreError::Database(e.to_string()))?; + .map_err(|e| StoreError::Database(Box::new(e)))?; Ok(rows .into_iter() .map( @@ -1958,7 +1963,7 @@ impl ProtocolStore for SqliteStore { .collect()) }) .await - .map_err(|e| StoreError::Database(e.to_string()))? + .map_err(|e| StoreError::Database(Box::new(e)))? } async fn save_base_key(&self, address: &str, message_id: &str, base_key: &[u8]) -> Result<()> { @@ -1971,7 +1976,7 @@ impl ProtocolStore for SqliteStore { tokio::task::spawn_blocking(move || -> Result<()> { let mut conn = pool .get() - .map_err(|e| StoreError::Connection(e.to_string()))?; + .map_err(|e| StoreError::Connection(Box::new(e)))?; diesel::insert_into(base_keys::table) .values(( base_keys::address.eq(&address), @@ -1988,11 +1993,11 @@ impl ProtocolStore for SqliteStore { .do_update() .set(base_keys::base_key.eq(&base_key)) .execute(&mut conn) - .map_err(|e| StoreError::Database(e.to_string()))?; + .map_err(|e| StoreError::Database(Box::new(e)))?; Ok(()) }) .await - .map_err(|e| StoreError::Database(e.to_string()))??; + .map_err(|e| StoreError::Database(Box::new(e)))??; Ok(()) } @@ -2010,7 +2015,7 @@ impl ProtocolStore for SqliteStore { tokio::task::spawn_blocking(move || -> Result { let mut conn = pool .get() - .map_err(|e| StoreError::Connection(e.to_string()))?; + .map_err(|e| StoreError::Connection(Box::new(e)))?; let stored_key: Option> = base_keys::table .select(base_keys::base_key) .filter(base_keys::address.eq(&address)) @@ -2018,11 +2023,11 @@ impl ProtocolStore for SqliteStore { .filter(base_keys::device_id.eq(device_id)) .first(&mut conn) .optional() - .map_err(|e| StoreError::Database(e.to_string()))?; + .map_err(|e| StoreError::Database(Box::new(e)))?; Ok(stored_key.as_ref() == Some(¤t_base_key)) }) .await - .map_err(|e| StoreError::Database(e.to_string()))? + .map_err(|e| StoreError::Database(Box::new(e)))? } async fn delete_base_key(&self, address: &str, message_id: &str) -> Result<()> { @@ -2033,7 +2038,7 @@ impl ProtocolStore for SqliteStore { tokio::task::spawn_blocking(move || -> Result<()> { let mut conn = pool .get() - .map_err(|e| StoreError::Connection(e.to_string()))?; + .map_err(|e| StoreError::Connection(Box::new(e)))?; diesel::delete( base_keys::table .filter(base_keys::address.eq(&address)) @@ -2041,11 +2046,11 @@ impl ProtocolStore for SqliteStore { .filter(base_keys::device_id.eq(device_id)), ) .execute(&mut conn) - .map_err(|e| StoreError::Database(e.to_string()))?; + .map_err(|e| StoreError::Database(Box::new(e)))?; Ok(()) }) .await - .map_err(|e| StoreError::Database(e.to_string()))??; + .map_err(|e| StoreError::Database(Box::new(e)))??; Ok(()) } @@ -2053,12 +2058,12 @@ impl ProtocolStore for SqliteStore { let pool = self.pool.clone(); let device_id = self.device_id; let devices_json = serde_json::to_string(&record.devices) - .map_err(|e| StoreError::Serialization(e.to_string()))?; + .map_err(|e| StoreError::Serialization(Box::new(e)))?; let now = wacore::time::now_secs() as i32; tokio::task::spawn_blocking(move || -> Result<()> { let mut conn = pool .get() - .map_err(|e| StoreError::Connection(e.to_string()))?; + .map_err(|e| StoreError::Connection(Box::new(e)))?; let raw_id_i32 = record.raw_id.map(|r| r as i32); diesel::insert_into(device_registry::table) .values(( @@ -2080,11 +2085,11 @@ impl ProtocolStore for SqliteStore { device_registry::raw_id.eq(raw_id_i32), )) .execute(&mut conn) - .map_err(|e| StoreError::Database(e.to_string()))?; + .map_err(|e| StoreError::Database(Box::new(e)))?; Ok(()) }) .await - .map_err(|e| StoreError::Database(e.to_string()))??; + .map_err(|e| StoreError::Database(Box::new(e)))??; Ok(()) } @@ -2095,7 +2100,7 @@ impl ProtocolStore for SqliteStore { tokio::task::spawn_blocking(move || -> Result> { let mut conn = pool .get() - .map_err(|e| StoreError::Connection(e.to_string()))?; + .map_err(|e| StoreError::Connection(Box::new(e)))?; let row: Option<(String, String, i32, Option, Option)> = device_registry::table .select(( @@ -2109,11 +2114,11 @@ impl ProtocolStore for SqliteStore { .filter(device_registry::device_id.eq(device_id)) .first(&mut conn) .optional() - .map_err(|e| StoreError::Database(e.to_string()))?; + .map_err(|e| StoreError::Database(Box::new(e)))?; match row { Some((user, devices_json, timestamp, phash, raw_id)) => { let devices: Vec = serde_json::from_str(&devices_json) - .map_err(|e| StoreError::Serialization(e.to_string()))?; + .map_err(|e| StoreError::Serialization(Box::new(e)))?; Ok(Some(DeviceListRecord { user, devices, @@ -2126,7 +2131,7 @@ impl ProtocolStore for SqliteStore { } }) .await - .map_err(|e| StoreError::Database(e.to_string()))? + .map_err(|e| StoreError::Database(Box::new(e)))? } async fn delete_devices(&self, user: &str) -> Result<()> { @@ -2136,18 +2141,18 @@ impl ProtocolStore for SqliteStore { tokio::task::spawn_blocking(move || -> Result<()> { let mut conn = pool .get() - .map_err(|e| StoreError::Connection(e.to_string()))?; + .map_err(|e| StoreError::Connection(Box::new(e)))?; diesel::delete( device_registry::table .filter(device_registry::user_id.eq(&user)) .filter(device_registry::device_id.eq(device_id)), ) .execute(&mut conn) - .map_err(|e| StoreError::Database(e.to_string()))?; + .map_err(|e| StoreError::Database(Box::new(e)))?; Ok(()) }) .await - .map_err(|e| StoreError::Database(e.to_string()))??; + .map_err(|e| StoreError::Database(Box::new(e)))??; Ok(()) } @@ -2158,7 +2163,7 @@ impl ProtocolStore for SqliteStore { tokio::task::spawn_blocking(move || -> Result> { let mut conn = pool .get() - .map_err(|e| StoreError::Connection(e.to_string()))?; + .map_err(|e| StoreError::Connection(Box::new(e)))?; let row: Option<(Vec, i64, Option)> = tc_tokens::table .select(( tc_tokens::token, @@ -2169,7 +2174,7 @@ impl ProtocolStore for SqliteStore { .filter(tc_tokens::device_id.eq(device_id)) .first(&mut conn) .optional() - .map_err(|e| StoreError::Database(e.to_string()))?; + .map_err(|e| StoreError::Database(Box::new(e)))?; Ok( row.map(|(token, token_timestamp, sender_timestamp)| TcTokenEntry { token, @@ -2179,7 +2184,7 @@ impl ProtocolStore for SqliteStore { ) }) .await - .map_err(|e| StoreError::Database(e.to_string()))? + .map_err(|e| StoreError::Database(Box::new(e)))? } async fn put_tc_token(&self, jid: &str, entry: &TcTokenEntry) -> Result<()> { @@ -2191,7 +2196,7 @@ impl ProtocolStore for SqliteStore { tokio::task::spawn_blocking(move || -> Result<()> { let mut conn = pool .get() - .map_err(|e| StoreError::Connection(e.to_string()))?; + .map_err(|e| StoreError::Connection(Box::new(e)))?; diesel::insert_into(tc_tokens::table) .values(( tc_tokens::jid.eq(&jid), @@ -2210,11 +2215,11 @@ impl ProtocolStore for SqliteStore { tc_tokens::updated_at.eq(now), )) .execute(&mut conn) - .map_err(|e| StoreError::Database(e.to_string()))?; + .map_err(|e| StoreError::Database(Box::new(e)))?; Ok(()) }) .await - .map_err(|e| StoreError::Database(e.to_string()))??; + .map_err(|e| StoreError::Database(Box::new(e)))??; Ok(()) } @@ -2225,18 +2230,18 @@ impl ProtocolStore for SqliteStore { tokio::task::spawn_blocking(move || -> Result<()> { let mut conn = pool .get() - .map_err(|e| StoreError::Connection(e.to_string()))?; + .map_err(|e| StoreError::Connection(Box::new(e)))?; diesel::delete( tc_tokens::table .filter(tc_tokens::jid.eq(&jid)) .filter(tc_tokens::device_id.eq(device_id)), ) .execute(&mut conn) - .map_err(|e| StoreError::Database(e.to_string()))?; + .map_err(|e| StoreError::Database(Box::new(e)))?; Ok(()) }) .await - .map_err(|e| StoreError::Database(e.to_string()))??; + .map_err(|e| StoreError::Database(Box::new(e)))??; Ok(()) } @@ -2246,16 +2251,16 @@ impl ProtocolStore for SqliteStore { tokio::task::spawn_blocking(move || -> Result> { let mut conn = pool .get() - .map_err(|e| StoreError::Connection(e.to_string()))?; + .map_err(|e| StoreError::Connection(Box::new(e)))?; let jids: Vec = tc_tokens::table .select(tc_tokens::jid) .filter(tc_tokens::device_id.eq(device_id)) .load(&mut conn) - .map_err(|e| StoreError::Database(e.to_string()))?; + .map_err(|e| StoreError::Database(Box::new(e)))?; Ok(jids) }) .await - .map_err(|e| StoreError::Database(e.to_string()))? + .map_err(|e| StoreError::Database(Box::new(e)))? } async fn delete_expired_tc_tokens(&self, cutoff_timestamp: i64) -> Result { @@ -2264,18 +2269,18 @@ impl ProtocolStore for SqliteStore { tokio::task::spawn_blocking(move || -> Result { let mut conn = pool .get() - .map_err(|e| StoreError::Connection(e.to_string()))?; + .map_err(|e| StoreError::Connection(Box::new(e)))?; let deleted = diesel::delete( tc_tokens::table .filter(tc_tokens::token_timestamp.lt(cutoff_timestamp)) .filter(tc_tokens::device_id.eq(device_id)), ) .execute(&mut conn) - .map_err(|e| StoreError::Database(e.to_string()))?; + .map_err(|e| StoreError::Database(Box::new(e)))?; Ok(deleted as u32) }) .await - .map_err(|e| StoreError::Database(e.to_string()))? + .map_err(|e| StoreError::Database(Box::new(e)))? } async fn store_sent_message( @@ -2347,18 +2352,18 @@ impl ProtocolStore for SqliteStore { tokio::task::spawn_blocking(move || -> Result { let mut conn = pool .get() - .map_err(|e| StoreError::Connection(e.to_string()))?; + .map_err(|e| StoreError::Connection(Box::new(e)))?; let deleted = diesel::delete( sent_messages::table .filter(sent_messages::created_at.lt(cutoff_timestamp)) .filter(sent_messages::device_id.eq(device_id)), ) .execute(&mut conn) - .map_err(|e| StoreError::Database(e.to_string()))?; + .map_err(|e| StoreError::Database(Box::new(e)))?; Ok(deleted as u32) }) .await - .map_err(|e| StoreError::Database(e.to_string()))? + .map_err(|e| StoreError::Database(Box::new(e)))? } } @@ -2405,13 +2410,13 @@ impl DeviceStore for SqliteStore { let sanitized = sanitized.trim_matches(['/', '\\', '.']); if sanitized.is_empty() { - return Err(StoreError::Database( + return Err(StoreError::InvalidConfig( "Snapshot name cannot be empty after sanitization".to_string(), )); } if sanitized.len() > MAX_LENGTH { - return Err(StoreError::Database(format!( + return Err(StoreError::InvalidConfig(format!( "Snapshot name exceeds maximum length of {} characters", MAX_LENGTH ))); @@ -2429,7 +2434,7 @@ impl DeviceStore for SqliteStore { tokio::task::spawn_blocking(move || -> Result<()> { let mut conn = pool .get() - .map_err(|e| StoreError::Connection(e.to_string()))?; + .map_err(|e| StoreError::Connection(Box::new(e)))?; let timestamp = wacore::time::now_secs(); @@ -2442,20 +2447,18 @@ impl DeviceStore for SqliteStore { diesel::sql_query(query) .execute(&mut conn) - .map_err(|e| StoreError::Database(e.to_string()))?; + .map_err(|e| StoreError::Database(Box::new(e)))?; // Save extra content if provided if let Some(data) = extra_data { let extra_path = format!("{}.json", target_path); - std::fs::write(&extra_path, data).map_err(|e| { - StoreError::Database(format!("Failed to write snapshot extra content: {}", e)) - })?; + std::fs::write(&extra_path, data)?; } Ok(()) }) .await - .map_err(|e| StoreError::Database(e.to_string()))??; + .map_err(|e| StoreError::Database(Box::new(e)))??; Ok(()) } diff --git a/wacore/libsignal/src/protocol/error.rs b/wacore/libsignal/src/protocol/error.rs index 931b76eff..d44b70b20 100644 --- a/wacore/libsignal/src/protocol/error.rs +++ b/wacore/libsignal/src/protocol/error.rs @@ -24,6 +24,12 @@ pub enum SignalProtocolError { /// invalid state for call to {0} to succeed: {1} InvalidState(&'static str, String), + /// backend store error in {0} + BackendError( + &'static str, + #[source] Box, + ), + /// protobuf encoding was invalid InvalidProtobufEncoding, @@ -116,3 +122,25 @@ impl From for SignalProtocolError { } } } + +#[cfg(test)] +mod tests { + use super::*; + + #[derive(Debug, thiserror::Error)] + #[error("synthetic backend failure: {code}")] + struct DummyBackendError { + code: u32, + } + + #[test] + fn backend_error_preserves_typed_source_via_downcast() { + let dummy = DummyBackendError { code: 42 }; + let spe = SignalProtocolError::BackendError("test_context", Box::new(dummy)); + let src = std::error::Error::source(&spe).expect("source preserved"); + let inner = src + .downcast_ref::() + .expect("downcasts to DummyBackendError"); + assert_eq!(inner.code, 42); + } +} diff --git a/wacore/noise/src/error.rs b/wacore/noise/src/error.rs index b6d499456..92c7726ad 100644 --- a/wacore/noise/src/error.rs +++ b/wacore/noise/src/error.rs @@ -1,25 +1,32 @@ use thiserror::Error; +use wacore_libsignal::crypto::CryptoProviderError; /// Errors that can occur during Noise protocol operations. -#[derive(Debug, Clone, Error)] +#[derive(Debug, Error)] pub enum NoiseError { - #[error("Invalid pattern length: expected {expected}, got {got}")] + #[error("invalid pattern length: expected {expected}, got {got}")] InvalidPatternLength { expected: usize, got: usize }, - #[error("Cryptographic operation failed: {0}")] - CryptoError(String), + #[error("AES-GCM encryption failed")] + Encrypt(#[source] CryptoProviderError), + + #[error("AES-GCM decryption failed")] + Decrypt(#[source] CryptoProviderError), + + #[error("ciphertext too short to contain authentication tag")] + CiphertextTooShort, #[error("HKDF expansion failed")] HkdfExpandFailed, - #[error("Invalid key length for {name}: expected {expected}, got {got}")] + #[error("invalid key length for {name}: expected {expected}, got {got}")] InvalidKeyLength { name: &'static str, expected: usize, got: usize, }, - #[error("Counter exhausted: nonce would be reused after 2^32 messages")] + #[error("counter exhausted: nonce would be reused after 2^32 messages")] CounterExhausted, } diff --git a/wacore/noise/src/state.rs b/wacore/noise/src/state.rs index 308624bcb..f59528552 100644 --- a/wacore/noise/src/state.rs +++ b/wacore/noise/src/state.rs @@ -40,7 +40,7 @@ impl NoiseCipher { let iv = generate_iv(counter); let mut out = Vec::with_capacity(plaintext.len() + TAG_LEN); aes_256_gcm_encrypt(&self.key, &iv, b"", plaintext, &mut out) - .map_err(|e| NoiseError::CryptoError(format!("{e}")))?; + .map_err(NoiseError::Encrypt)?; Ok(out) } @@ -54,8 +54,7 @@ impl NoiseCipher { buffer: &mut B, ) -> Result<()> { let iv = generate_iv(counter); - aes_256_gcm_encrypt_in_place(&self.key, &iv, b"", buffer) - .map_err(|e| NoiseError::CryptoError(format!("{e}"))) + aes_256_gcm_encrypt_in_place(&self.key, &iv, b"", buffer).map_err(NoiseError::Encrypt) } /// Decrypts ciphertext (with 16-byte tag appended) in-place within the @@ -68,8 +67,7 @@ impl NoiseCipher { buffer: &mut B, ) -> Result<()> { let iv = generate_iv(counter); - aes_256_gcm_decrypt_in_place(&self.key, &iv, b"", buffer) - .map_err(|e| NoiseError::CryptoError(format!("Decrypt failed: {e}"))) + aes_256_gcm_decrypt_in_place(&self.key, &iv, b"", buffer).map_err(NoiseError::Decrypt) } } @@ -154,7 +152,7 @@ impl NoiseState { let iv = generate_iv(self.post_increment_counter()?); let mut out = Vec::with_capacity(plaintext.len() + TAG_LEN); aes_256_gcm_encrypt(&self.key, &iv, &self.hash, plaintext, &mut out) - .map_err(|e| NoiseError::CryptoError(format!("{e}")))?; + .map_err(NoiseError::Encrypt)?; self.authenticate(&out); Ok(out) } @@ -164,8 +162,7 @@ impl NoiseState { let iv = generate_iv(self.post_increment_counter()?); let aad = self.hash; let start = out.len(); - aes_256_gcm_encrypt(&self.key, &iv, &aad, plaintext, out) - .map_err(|e| NoiseError::CryptoError(format!("{e}")))?; + aes_256_gcm_encrypt(&self.key, &iv, &aad, plaintext, out).map_err(NoiseError::Encrypt)?; self.authenticate(&out[start..]); Ok(()) } @@ -176,7 +173,7 @@ impl NoiseState { let iv = generate_iv(self.post_increment_counter()?); let mut out = Vec::with_capacity(ciphertext.len().saturating_sub(TAG_LEN)); aes_256_gcm_decrypt(&self.key, &iv, &aad, ciphertext, &mut out) - .map_err(|e| NoiseError::CryptoError(format!("Noise decrypt failed: {e}")))?; + .map_err(NoiseError::Decrypt)?; self.authenticate(ciphertext); Ok(out) } @@ -184,14 +181,11 @@ impl NoiseState { /// Zero-allocation decryption that appends the plaintext to the provided buffer. pub fn decrypt_into(&mut self, ciphertext: &[u8], out: &mut Vec) -> Result<()> { if ciphertext.len() < TAG_LEN { - return Err(NoiseError::CryptoError( - "Ciphertext too short (missing tag)".into(), - )); + return Err(NoiseError::CiphertextTooShort); } let aad = self.hash; let iv = generate_iv(self.post_increment_counter()?); - aes_256_gcm_decrypt(&self.key, &iv, &aad, ciphertext, out) - .map_err(|e| NoiseError::CryptoError(format!("Noise decrypt failed: {e}")))?; + aes_256_gcm_decrypt(&self.key, &iv, &aad, ciphertext, out).map_err(NoiseError::Decrypt)?; self.authenticate(ciphertext); Ok(()) } diff --git a/wacore/src/appstate_sync.rs b/wacore/src/appstate_sync.rs index 8c0d87528..f8a73bb93 100644 --- a/wacore/src/appstate_sync.rs +++ b/wacore/src/appstate_sync.rs @@ -40,7 +40,7 @@ fn lookup_app_state_key( pub enum AppStateSyncError { #[error("app state key not found: {0}")] KeyNotFound(String), - #[error("store error: {0}")] + #[error("store error")] Store(#[from] crate::store::error::StoreError), #[error(transparent)] Other(#[from] anyhow::Error), diff --git a/wacore/src/download.rs b/wacore/src/download.rs index 621781c3d..cd94550e5 100644 --- a/wacore/src/download.rs +++ b/wacore/src/download.rs @@ -1,4 +1,7 @@ -use crate::libsignal::crypto::{CryptographicMac, aes_256_cbc_decrypt_into}; +use crate::libsignal::crypto::{ + CryptographicMac, DecryptionError as AesCbcDecryptionError, Error as CryptoError, + aes_256_cbc_decrypt_into, +}; use anyhow::{Result, anyhow}; use base64::Engine as _; use base64::prelude::*; @@ -17,8 +20,10 @@ pub enum MediaDecryptionError { PayloadTooShort, #[error("invalid MAC signature")] InvalidMac, - #[error("decryption error: {0}")] - Decryption(String), + #[error("AES-CBC decryption failed")] + Decryption(#[source] AesCbcDecryptionError), + #[error("HMAC initialization failed")] + Mac(#[source] CryptoError), #[error(transparent)] Other(#[from] anyhow::Error), } @@ -497,7 +502,7 @@ impl DownloadUtils { pub fn decrypt_cbc(cipher_key: &[u8], iv: &[u8], ciphertext: &[u8]) -> Result> { let mut output = Vec::new(); aes_256_cbc_decrypt_into(ciphertext, cipher_key, iv, &mut output) - .map_err(|e| anyhow!(e.to_string()))?; + .map_err(anyhow::Error::new)?; Ok(output) } @@ -517,8 +522,8 @@ impl DownloadUtils { let (iv, cipher_key, mac_key) = Self::get_media_keys(media_key, media_type)?; let computed_mac_full = { - let mut mac = CryptographicMac::new("HmacSha256", &mac_key) - .map_err(|e| MediaDecryptionError::Decryption(e.to_string()))?; + let mut mac = + CryptographicMac::new("HmacSha256", &mac_key).map_err(MediaDecryptionError::Mac)?; mac.update(&iv); mac.update(ciphertext); mac.finalize() @@ -529,7 +534,7 @@ impl DownloadUtils { let mut output = Vec::new(); aes_256_cbc_decrypt_into(ciphertext, &cipher_key, &iv, &mut output) - .map_err(|e| MediaDecryptionError::Decryption(e.to_string()))?; + .map_err(MediaDecryptionError::Decryption)?; Ok(output) } } @@ -708,4 +713,26 @@ mod tests { .unwrap_err(); assert!(err.to_string().contains("SHA-256 mismatch")); } + + #[test] + fn media_decryption_decryption_preserves_aes_cbc_source() { + let inner = AesCbcDecryptionError::BadKeyOrIv; + let mde = MediaDecryptionError::Decryption(inner); + let src = std::error::Error::source(&mde).expect("source preserved"); + let cbc = src + .downcast_ref::() + .expect("downcasts to AesCbcDecryptionError"); + assert!(matches!(cbc, AesCbcDecryptionError::BadKeyOrIv)); + } + + #[test] + fn media_decryption_mac_preserves_crypto_error_source() { + let inner = CryptoError::UnknownAlgorithm("MAC", "BogusAlg".into()); + let mde = MediaDecryptionError::Mac(inner); + let src = std::error::Error::source(&mde).expect("source preserved"); + let ce = src + .downcast_ref::() + .expect("downcasts to CryptoError"); + assert!(matches!(ce, CryptoError::UnknownAlgorithm("MAC", _))); + } } diff --git a/wacore/src/handshake/mod.rs b/wacore/src/handshake/mod.rs index 673bd9e28..9e555bdf4 100644 --- a/wacore/src/handshake/mod.rs +++ b/wacore/src/handshake/mod.rs @@ -1,6 +1,6 @@ // Re-export everything from wacore-noise pub use wacore_noise::{ EdgeRoutingError, HandshakeError, HandshakeResult as Result, HandshakeState, HandshakeUtils, - MAX_EDGE_ROUTING_LEN, NoiseCipher, NoiseHandshake, WA_CERT_PUB_KEY, + MAX_EDGE_ROUTING_LEN, NoiseCipher, NoiseError, NoiseHandshake, WA_CERT_PUB_KEY, build_edge_routing_preintro, build_handshake_header, generate_iv, }; diff --git a/wacore/src/pair_code.rs b/wacore/src/pair_code.rs index 0d290a37c..cf1f0e169 100644 --- a/wacore/src/pair_code.rs +++ b/wacore/src/pair_code.rs @@ -22,8 +22,8 @@ use crate::companion_reg::{ CompanionWebClientType, companion_platform_display, companion_web_client_type_for_props, }; -use crate::libsignal::crypto::aes_256_gcm_encrypt; -use crate::libsignal::protocol::{KeyPair, PublicKey}; +use crate::libsignal::crypto::{CryptoProviderError, aes_256_gcm_encrypt}; +use crate::libsignal::protocol::{CurveError, KeyPair, PublicKey}; use aes::cipher::{KeyIvInit, StreamCipher}; use ctr::Ctr128BE; use hkdf::Hkdf; @@ -412,29 +412,21 @@ impl PairCodeUtils { primary_identity_pub: &[u8; 32], identity_key: &KeyPair, ) -> Result<(Vec, [u8; 32]), PairCodeError> { - // Parse primary's ephemeral public key - let primary_eph_pub = - PublicKey::from_djb_public_key_bytes(primary_ephemeral_pub).map_err(|e| { - PairCodeError::CryptoError(format!("Invalid primary ephemeral key: {e}")) - })?; - - // Parse primary's identity public key - let primary_id_pub = - PublicKey::from_djb_public_key_bytes(primary_identity_pub).map_err(|e| { - PairCodeError::CryptoError(format!("Invalid primary identity key: {e}")) - })?; - - // DH 1: Ephemeral key exchange + let primary_eph_pub = PublicKey::from_djb_public_key_bytes(primary_ephemeral_pub) + .map_err(PairCodeError::InvalidPrimaryEphemeralKey)?; + + let primary_id_pub = PublicKey::from_djb_public_key_bytes(primary_identity_pub) + .map_err(PairCodeError::InvalidPrimaryIdentityKey)?; + let ephemeral_shared = ephemeral_keypair .private_key .calculate_agreement(&primary_eph_pub) - .map_err(|e| PairCodeError::CryptoError(format!("Ephemeral DH failed: {e}")))?; + .map_err(PairCodeError::EphemeralKeyAgreement)?; - // DH 2: Identity key exchange (for ADV secret derivation) let identity_shared = identity_key .private_key .calculate_agreement(&primary_id_pub) - .map_err(|e| PairCodeError::CryptoError(format!("Identity DH failed: {e}")))?; + .map_err(PairCodeError::IdentityKeyAgreement)?; // Generate random bytes for ADV secret derivation let mut random_bytes = [0u8; 32]; @@ -451,7 +443,7 @@ impl PairCodeUtils { let mut new_adv_secret = [0u8; 32]; hk_adv .expand(b"adv_secret", &mut new_adv_secret) - .map_err(|_| PairCodeError::CryptoError("HKDF expand for adv_secret failed".into()))?; + .map_err(|_| PairCodeError::AdvSecretKeyDerivation)?; // Prepare bundle: companion_identity_pub (32) + primary_identity_pub (32) + random_bytes (32) = 96 bytes let mut bundle = Vec::with_capacity(96); @@ -469,9 +461,7 @@ impl PairCodeUtils { let mut enc_key = [0u8; 32]; hk_bundle .expand(b"link_code_pairing_key_bundle_encryption_key", &mut enc_key) - .map_err(|_| { - PairCodeError::CryptoError("HKDF expand for bundle encryption key failed".into()) - })?; + .map_err(|_| PairCodeError::BundleKeyDerivation)?; // Generate random IV for AES-GCM (12 bytes) let mut iv = [0u8; 12]; @@ -482,7 +472,7 @@ impl PairCodeUtils { wrapped_bundle.extend_from_slice(&key_bundle_salt); wrapped_bundle.extend_from_slice(&iv); aes_256_gcm_encrypt(&enc_key, &iv, b"", &bundle, &mut wrapped_bundle) - .map_err(|e| PairCodeError::CryptoError(format!("AES-GCM encryption failed: {e}")))?; + .map_err(PairCodeError::BundleAead)?; Ok((wrapped_bundle, new_adv_secret)) } @@ -493,35 +483,53 @@ impl PairCodeUtils { } } -/// Errors that can occur during pair code operations. +/// Errors raised by wacore-side pair-code validation, key derivation, and +/// protocol-bundle building. The high-level crate wraps this in +/// `whatsapp_rust::pair_code::PairError` and adds an IQ-failure variant for the +/// transport layer. #[derive(Debug, thiserror::Error)] pub enum PairCodeError { - #[error("Phone number is required")] + #[error("phone number is required")] PhoneNumberRequired, - #[error("Phone number is too short (must be at least 7 digits)")] + #[error("phone number is too short (must be at least 7 digits)")] PhoneNumberTooShort, - #[error("Phone number must not start with 0 (use international format)")] + #[error("phone number must not start with 0 (use international format)")] PhoneNumberNotInternational, - #[error("Invalid custom code: must be 8 characters from Crockford Base32 alphabet")] + #[error("invalid custom code: must be 8 characters from Crockford Base32 alphabet")] InvalidCustomCode, - #[error("Invalid wrapped data: expected {expected} bytes, got {got}")] + #[error("invalid wrapped data: expected {expected} bytes, got {got}")] InvalidWrappedData { expected: usize, got: usize }, - #[error("Cryptographic operation failed: {0}")] - CryptoError(String), + #[error("primary device sent an invalid ephemeral public key")] + InvalidPrimaryEphemeralKey(#[source] CurveError), + + #[error("primary device sent an invalid identity public key")] + InvalidPrimaryIdentityKey(#[source] CurveError), + + #[error("ephemeral key agreement failed")] + EphemeralKeyAgreement(#[source] CurveError), + + #[error("identity key agreement failed")] + IdentityKeyAgreement(#[source] CurveError), + + #[error("HKDF expand failed for adv_secret")] + AdvSecretKeyDerivation, + + #[error("HKDF expand failed for bundle encryption key")] + BundleKeyDerivation, + + #[error("AES-GCM encryption of key bundle failed")] + BundleAead(#[source] CryptoProviderError), - #[error("Not in waiting state for pair code notification")] + #[error("not in waiting state for pair code notification")] NotWaiting, - #[error("Server response missing pairing ref")] + #[error("server response missing pairing ref")] MissingPairingRef, - - #[error("Request failed: {0}")] - RequestFailed(String), } #[cfg(test)] @@ -931,18 +939,18 @@ mod tests { #[test] fn test_pair_code_error_display() { let err = PairCodeError::PhoneNumberRequired; - assert_eq!(err.to_string(), "Phone number is required"); + assert_eq!(err.to_string(), "phone number is required"); let err = PairCodeError::PhoneNumberTooShort; assert_eq!( err.to_string(), - "Phone number is too short (must be at least 7 digits)" + "phone number is too short (must be at least 7 digits)" ); let err = PairCodeError::InvalidCustomCode; assert_eq!( err.to_string(), - "Invalid custom code: must be 8 characters from Crockford Base32 alphabet" + "invalid custom code: must be 8 characters from Crockford Base32 alphabet" ); let err = PairCodeError::InvalidWrappedData { @@ -951,10 +959,30 @@ mod tests { }; assert_eq!( err.to_string(), - "Invalid wrapped data: expected 80 bytes, got 50" + "invalid wrapped data: expected 80 bytes, got 50" ); } + #[test] + fn invalid_primary_ephemeral_key_preserves_curve_source() { + let err = PairCodeError::InvalidPrimaryEphemeralKey(CurveError::NoKeyTypeIdentifier); + let src = std::error::Error::source(&err).expect("source preserved"); + let curve = src + .downcast_ref::() + .expect("downcasts to CurveError"); + assert!(matches!(curve, CurveError::NoKeyTypeIdentifier)); + } + + #[test] + fn bundle_aead_preserves_crypto_provider_source() { + let err = PairCodeError::BundleAead(CryptoProviderError::BadInput); + let src = std::error::Error::source(&err).expect("source preserved"); + let cpe = src + .downcast_ref::() + .expect("downcasts to CryptoProviderError"); + assert!(matches!(cpe, CryptoProviderError::BadInput)); + } + #[test] fn test_crockford_encoding_boundary_values() { // Test specific byte patterns diff --git a/wacore/src/request.rs b/wacore/src/request.rs index 5fad1463e..251c2bd54 100644 --- a/wacore/src/request.rs +++ b/wacore/src/request.rs @@ -82,16 +82,14 @@ impl<'a> InfoQuery<'a> { pub enum IqError { #[error("IQ request timed out")] Timeout, - #[error("Client is not connected")] + #[error("client is not connected")] NotConnected, - #[error("Received disconnect node during IQ wait: {0:?}")] + #[error("received disconnect node during IQ wait: {0:?}")] Disconnected(Node), - #[error("Received a server error response: code={code}, text='{text}'")] + #[error("received a server error response: code={code}, text='{text}'")] ServerError { code: u16, text: String }, - #[error("Internal channel closed unexpectedly")] + #[error("internal channel closed unexpectedly")] InternalChannelClosed, - #[error("Network error: {0}")] - Network(String), } /// Lightweight server error that can be embedded in `anyhow::Error` and diff --git a/wacore/src/store/device.rs b/wacore/src/store/device.rs index 7c11943f2..357145955 100644 --- a/wacore/src/store/device.rs +++ b/wacore/src/store/device.rs @@ -68,6 +68,8 @@ pub mod key_pair_serde { if bytes.len() != 64 { return Err(serde::de::Error::invalid_length(bytes.len(), &"64")); } + // reason: serde::de::Error::custom flattens to a String at the boundary — + // serde's error model has no source-chain preservation. let private_key = PrivateKey::deserialize(&bytes[0..32]) .map_err(|e| serde::de::Error::custom(e.to_string()))?; let public_key = PublicKey::from_djb_public_key_bytes(&bytes[32..64]) diff --git a/wacore/src/store/error.rs b/wacore/src/store/error.rs index 766821ea0..6bda7a0b3 100644 --- a/wacore/src/store/error.rs +++ b/wacore/src/store/error.rs @@ -2,36 +2,112 @@ use thiserror::Error; #[derive(Debug, Error)] pub enum StoreError { - #[error("I/O error: {0}")] + #[error("I/O error")] Io(#[from] std::io::Error), - #[error("Serialization/deserialization error: {0}")] - Serialization(String), + #[error("serialization/deserialization error")] + Serialization(#[source] Box), - #[error("Item not found: {0}")] - NotFound(String), + /// Validation failure with a descriptive message and no underlying typed + /// source — e.g. "Invalid foo length: 17". Prefer `Serialization` or a + /// dedicated typed variant if a real source exists. + #[error("data validation failed: {0}")] + Validation(String), - #[error("Database backend error: {0}")] - Backend(#[from] Box), + #[error("database connection error")] + Connection(#[source] Box), - #[error("Database connection error: {0}")] - Connection(String), + #[error("database operation error")] + Database(#[source] Box), - #[error("Database operation error: {0}")] - Database(String), + #[error("database operation '{op}' exhausted retries")] + RetriesExhausted { op: String }, - #[error("Migration error: {0}")] - Migration(String), + #[error("migration error")] + Migration(#[source] Box), - #[error("Device with ID {0} not found")] + #[error("store configuration is invalid: {0}")] + InvalidConfig(String), + + #[error("device with ID {0} not found")] DeviceNotFound(i32), } +impl StoreError { + /// Walks the error source chain and returns true if any layer's `Display` + /// indicates a SQLite busy/locked condition. Used by retry layers that + /// can't depend on a specific backend (Diesel, libsql, etc.) directly. + /// + /// Substring matching is necessary because SQLite reports BUSY/LOCKED + /// through `sqlite3_errmsg()` strings; the error code itself is mapped + /// to `Diesel::DatabaseError(Unknown, _)` (or similar) without further + /// discrimination. + pub fn is_database_busy_or_locked(&self) -> bool { + let mut layer: &dyn std::error::Error = self; + loop { + let s = layer.to_string().to_lowercase(); + if s.contains("locked") || s.contains("busy") { + return true; + } + match layer.source() { + Some(inner) => layer = inner, + None => return false, + } + } + } +} + pub type Result = std::result::Result; -/// Helper to convert any Display error into StoreError::Database. -/// Use with `.map_err(db_err)?` instead of `.map_err(|e| StoreError::Database(e.to_string()))?` -#[inline] -pub fn db_err(e: E) -> StoreError { - StoreError::Database(e.to_string()) +#[cfg(test)] +mod tests { + use super::*; + + #[derive(Debug, thiserror::Error)] + #[error("synthetic backend error: {0}")] + struct DummyBackendError(&'static str); + + #[test] + fn database_preserves_typed_source_via_downcast() { + let inner = DummyBackendError("bang"); + let se = StoreError::Database(Box::new(inner)); + let src = std::error::Error::source(&se).expect("source preserved"); + let downcast = src + .downcast_ref::() + .expect("downcasts to DummyBackendError"); + assert_eq!(downcast.0, "bang"); + } + + #[test] + fn is_busy_or_locked_walks_chain() { + let inner = DummyBackendError("database is locked"); + let se = StoreError::Database(Box::new(inner)); + assert!(se.is_database_busy_or_locked()); + } + + #[test] + fn is_busy_or_locked_negative() { + let inner = DummyBackendError("permission denied"); + let se = StoreError::Database(Box::new(inner)); + assert!(!se.is_database_busy_or_locked()); + } + + #[test] + fn is_busy_or_locked_is_case_insensitive() { + // SQLite drivers in different ecosystems vary on casing for these + // diagnostic strings ("database is LOCKED", "Busy", etc.). The check + // must not depend on the exact casing the underlying driver chose. + for msg in [ + "database is LOCKED", + "SQLITE_BUSY: write contention", + "Busy", + "Locked", + ] { + let se = StoreError::Database(Box::new(DummyBackendError(msg))); + assert!( + se.is_database_busy_or_locked(), + "expected {msg:?} to be detected" + ); + } + } } diff --git a/wacore/src/store/persistence.rs b/wacore/src/store/persistence.rs index 4965f3ddd..361eda789 100644 --- a/wacore/src/store/persistence.rs +++ b/wacore/src/store/persistence.rs @@ -9,7 +9,7 @@ use crate::runtime::{AbortHandle, Runtime, ShutdownSignal, wait_for_shutdown}; use crate::store::commands::{DeviceCommand, apply_command_to_device}; use crate::store::device::Device; -use crate::store::error::{StoreError, db_err}; +use crate::store::error::StoreError; use crate::store::traits::Backend; use async_lock::RwLock; use event_listener::Event; @@ -34,15 +34,15 @@ impl PersistenceManager { /// Create a PersistenceManager with a backend implementation. pub async fn new(backend: Arc) -> Result { debug!("PersistenceManager: Ensuring device row exists."); - let exists = backend.exists().await.map_err(db_err)?; + let exists = backend.exists().await?; if !exists { debug!("PersistenceManager: No device row found. Creating new device row."); - let id = backend.create().await.map_err(db_err)?; + let id = backend.create().await?; debug!("PersistenceManager: Created device row with id={id}."); } debug!("PersistenceManager: Attempting to load device data via Backend."); - let device_data_opt = backend.load().await.map_err(db_err)?; + let device_data_opt = backend.load().await?; let device = if let Some(serializable_device) = device_data_opt { debug!( @@ -103,7 +103,7 @@ impl PersistenceManager { if let Err(e) = self.backend.save(&serializable_device).await { // Restore dirty flag so the next tick retries the save self.dirty.store(true, Ordering::Release); - return Err(db_err(e)); + return Err(e); } debug!("Device state saved successfully."); } @@ -119,10 +119,7 @@ impl PersistenceManager { #[cfg(feature = "debug-snapshots")] { self.save_to_disk().await?; - self.backend - .snapshot_db(name, extra_content) - .await - .map_err(db_err) + self.backend.snapshot_db(name, extra_content).await } #[cfg(not(feature = "debug-snapshots"))] { @@ -199,10 +196,7 @@ impl PersistenceManager { &self, group_jid: &str, ) -> Result, StoreError> { - self.backend - .get_sender_key_devices(group_jid) - .await - .map_err(db_err) + self.backend.get_sender_key_devices(group_jid).await } pub async fn set_sender_key_status( @@ -210,16 +204,10 @@ impl PersistenceManager { group_jid: &str, entries: &[(&str, bool)], ) -> Result<(), StoreError> { - self.backend - .set_sender_key_status(group_jid, entries) - .await - .map_err(db_err) + self.backend.set_sender_key_status(group_jid, entries).await } pub async fn clear_sender_key_devices(&self, group_jid: &str) -> Result<(), StoreError> { - self.backend - .clear_sender_key_devices(group_jid) - .await - .map_err(db_err) + self.backend.clear_sender_key_devices(group_jid).await } }