diff --git a/rust/src/api/orders.rs b/rust/src/api/orders.rs index cbee5399..10c7f085 100644 --- a/rust/src/api/orders.rs +++ b/rust/src/api/orders.rs @@ -3557,23 +3557,33 @@ pub async fn get_trade_role(order_id: String) -> Result= u32::MAX` is not a +/// real trade index: negatives are nonsense, and `u32::MAX` is the reserved +/// terminal index — storing it as the counter would make the next +/// `derive_trade_key` compute `u32::MAX + 1` and overflow (panic in debug, wrap +/// to 0 in release, reissuing index 0 — the exact key-reuse the resync +/// prevents). Such a value is dropped rather than truncated into the counter. +fn sanitize_trade_index(i: i64) -> Option { + u32::try_from(i).ok().filter(|&v| v < u32::MAX) +} + /// Highest trade-key index across all recovered orders and disputes (#217). /// /// The counter must be raised to this so the next `derive_trade_key()` cannot /// hand out an index a recovered trade already owns. Returns `None` when the -/// restore carried no trades (nothing to resync to). Indexes are `i64` on the -/// wire; a value that is negative or beyond `u32::MAX` is not a real trade -/// index, so it is dropped rather than truncated into the counter. +/// restore carried no trades (nothing to resync to). +/// +/// NOTE (#328): this is only a *lower bound* of the daemon's real counter — +/// the restore payload lists only non-finalized orders, so a finalized trade +/// holding a higher index is invisible here. `restore_session` sources its +/// resync floor from `last_trade_index()` (authoritative) and falls back to +/// this only when the daemon does not answer. fn recovered_max_trade_index( info: &mostro_core::message::RestoreSessionInfo, ) -> Option { - // A single adapter drops both negatives and any value >= u32::MAX — neither - // is a real trade index, and truncating one into a small u32 could corrupt - // the counter this exists to protect. u32::MAX itself is dropped: it is the - // reserved terminal index, and storing it as the counter would make the next - // derive_trade_key compute u32::MAX + 1 and overflow (panic in debug, wrap to - // 0 in release — reissuing index 0, the exact key-reuse this resync prevents). - // Collapsed into one filter_map so the two conditions can't drift apart. let all: Vec = info .restore_orders .iter() @@ -3581,10 +3591,7 @@ fn recovered_max_trade_index( .chain(info.restore_disputes.iter().map(|d| d.trade_index)) .collect(); let total = all.len(); - let valid: Vec = all - .into_iter() - .filter_map(|i| u32::try_from(i).ok().filter(|&v| v < u32::MAX)) - .collect(); + let valid: Vec = all.iter().copied().filter_map(sanitize_trade_index).collect(); // A dropped index is not just an odd value: it means the daemon sent // something this client's model does not cover, and a silently-lowered // floor produces a later CantDo(InvalidTradeIndex) with no breadcrumb. Warn @@ -3604,6 +3611,196 @@ fn recovered_max_trade_index( valid.into_iter().max() } +/// Pick the resync floor as the highest of the daemon counter and the +/// restore-payload maximum (#328). +/// +/// The daemon's `LastTradeIndex` answer (`daemon_counter`) is authoritative and +/// is the real high-water mark, including finalized trades. The payload maximum +/// is a proven lower bound — every recovered order carries its own index. +/// +/// Against a consistent daemon the counter is always `>=` the payload maximum: +/// the daemon raises `last_trade_index` to every index it accepts +/// (`update_user_trade_index`) and rejects any index it has already seen, so an +/// order it still returns in the restore payload was necessarily seen at or +/// below the counter. Taking the max is therefore a no-op in practice — kept as +/// cheap defense-in-depth so the floor stays correct independent of that +/// invariant: a stale/partial reply or a daemon bug can never make us resync +/// below a recovered trade's own index and reuse its key. Returns `None` only +/// when neither source has a usable index. +fn resync_floor( + daemon_counter: Option, + info: &mostro_core::message::RestoreSessionInfo, +) -> Option { + let payload_max = recovered_max_trade_index(info); + match (daemon_counter, payload_max) { + (Some(daemon), Some(payload)) => Some(daemon.max(payload)), + (daemon, payload) => daemon.or(payload), + } +} + +/// True only for the reply to THIS `LastTradeIndex` request: the action +/// matches and the daemon echoed our correlation nonce +/// (`mostro/src/app/last_trade_index.rs` copies `request_id` into the reply). +/// +/// A replayed reply from an earlier request carries a different nonce — or +/// none, since this client's own pre-hardening requests sent no id — so +/// accepting `None` would readmit exactly the replays this guards against. +/// Strict matching means a daemon that does not echo falls back to the +/// restore-payload maximum, the same designed path as a silent daemon. +fn is_matching_last_trade_index_reply( + kind: &mostro_core::message::MessageKind, + request_id: u64, +) -> bool { + kind.action == mostro_core::message::Action::LastTradeIndex + && kind.request_id == Some(request_id) +} + +/// Ask the daemon for its authoritative last-known trade index (#328). +/// +/// `Action::LastTradeIndex` is account-scoped: the request is signed with the +/// identity keys for BOTH the Seal and the rumor (see +/// `actions::last_trade_index`), so it derives NO trade key and the daemon +/// resolves the account by sender pubkey. The reply is a kind-14 addressed to +/// the identity pubkey, carrying the counter in `MessageKind::trade_index`. +/// +/// Returns `Ok(Some(idx))` with the sanitized counter, or `Ok(None)` when the +/// daemon does not answer within the timeout or the reply carries no usable +/// index — the caller then falls back to `recovered_max_trade_index`. +/// +/// This is a self-contained request/reply (own subscription + inline wait, like +/// mostro-cli's `wait_for_dm`) rather than a reuse of the per-trade +/// subscription/dispatch path: that path derives its recipient keys from a +/// trade index and is coupled to trade keys, but this reply is encrypted to the +/// identity key. +async fn last_trade_index() -> Result> { + use nostr_sdk::RelayPoolNotification; + use crate::rt::time::{timeout, Duration}; + + let identity_keys = crate::api::identity::get_active_keys().await?; + let identity_pk = identity_keys.public_key(); + let identity_pk_hex = identity_pk.to_hex(); + let mostro_pubkey = nostr_sdk::PublicKey::from_hex(&active_mostro_pubkey())?; + + let pool = crate::api::nostr::get_pool()?; + let client = pool.client(); + + // Grab the notifications receiver BEFORE subscribing so the reply can't + // arrive in the gap between subscribe and the first recv. + let mut rx = client.notifications(); + + // This query, unlike the restore itself, has a fallback (the payload + // maximum), so a shorter wait halves the worst-case restore latency + // against a silent daemon. Shared by the relay-side auto-close and the + // outer wait loop so the two can't drift. + const REPLY_TIMEOUT: Duration = Duration::from_secs(5); + + // limit(0): live-only, same rationale as subscribe_daemon_messages — the + // reply is published after we subscribe, and we never want a replayed + // historical LastTradeIndex to resolve this request. + let filter = nostr_sdk::Filter::new() + .kind(nostr_sdk::Kind::PrivateDirectMessage) + .author(mostro_pubkey) + .pubkey(identity_pk) + .limit(0); + // Auto-close the relay-side subscription — this is a one-shot request/reply + // (mostro-cli's wait_for_dm shape), not a long-lived watcher, so the library + // issues the CLOSE on every path: after the reply (WaitForEventsAfterEOSE(1)) + // and, if the daemon never answers, on the timeout. Leaving it to manual + // bookkeeping is the leak class #182/#255 address. Auto-close subs are + // deliberately excluded from reconnect re-subscription (correct here: if the + // socket drops mid-request we fall back to the payload maximum by design). + let close_opts = nostr_sdk::prelude::SubscribeAutoCloseOptions::default() + .exit_policy(nostr_sdk::prelude::ReqExitPolicy::WaitForEventsAfterEOSE(1)) + .timeout(Some(REPLY_TIMEOUT)); + if let Err(e) = client.subscribe(filter, Some(close_opts)).await { + log::warn!("[orders] last_trade_index subscribe failed: {e}"); + return Ok(None); + } + + // Correlation nonce, echoed by the daemon in its reply. Without it any + // authenticated LastTradeIndex reply resolves this request, so a malicious + // relay could replay an old one. Monotonicity caps the damage (the max in + // resync_floor means a stale counter degrades to the payload fallback, the + // same as a silent daemon) — but the daemon echoes request_id, so binding + // the reply to this request costs nothing. + let request_id: u64 = { + use rand::RngCore; + rand::rngs::OsRng.next_u64().max(1) // 0 is indistinguishable from "unset" + }; + let event_json = + actions::last_trade_index(&identity_keys, &mostro_pubkey, request_id).await?; + publish_event_json(&event_json).await?; + crate::api::logging::blog_info( + "restore", + "LastTradeIndex published — waiting for daemon".to_string(), + ); + + // Wait for the reply. Bounded by REPLY_TIMEOUT — the same budget the + // relay-side auto-close uses, so both give up together. + let deadline = REPLY_TIMEOUT; + let start = crate::rt::time::Instant::now(); + loop { + let remaining = deadline.saturating_sub(start.elapsed()); + if remaining.is_zero() { + break; + } + match timeout(remaining, rx.recv()).await { + Ok(Ok(RelayPoolNotification::Event { event, .. })) => { + if event.kind != nostr_sdk::Kind::PrivateDirectMessage + || event.pubkey != mostro_pubkey + { + continue; + } + let is_for_us = event.tags.iter().any(|t| { + let s = t.as_slice(); + s.first().map(|v| v.as_str()) == Some("p") + && s.get(1).map(|v| v.as_str()) == Some(identity_pk_hex.as_str()) + }); + if !is_for_us { + continue; + } + match crate::nostr::transport::unwrap_mostro_message(&identity_keys, &event).await { + Ok(Some(unwrapped)) => { + // Authenticate: the kind-14 author must be the node. + if unwrapped.sender != mostro_pubkey { + continue; + } + let kind = unwrapped.message.get_inner_message_kind(); + if !is_matching_last_trade_index_reply(kind, request_id) { + continue; + } + let idx = kind.trade_index.and_then(sanitize_trade_index); + crate::api::logging::blog_info("restore", format!( + "LastTradeIndex reply: trade_index={:?} -> floor={idx:?}", + kind.trade_index + )); + return Ok(idx); + } + Ok(None) => continue, + Err(e) => { + log::warn!("[orders] last_trade_index decrypt failed: {e}"); + continue; + } + } + } + Ok(Ok(RelayPoolNotification::Shutdown)) => break, + Ok(Err(broadcast::error::RecvError::Lagged(n))) => { + log::warn!("[orders] last_trade_index lagged by {n} messages"); + continue; + } + Ok(Err(broadcast::error::RecvError::Closed)) => break, + Err(_) => break, // timeout + Ok(Ok(_)) => continue, + } + } + crate::api::logging::blog_warn( + "restore", + "LastTradeIndex: no usable daemon reply — falling back to restore payload max" + .to_string(), + ); + Ok(None) +} + /// Send a `RestoreSession` to the active daemon and return the user's active /// trades/disputes. Mirrors create_order's send/await, minus the order payload. /// @@ -3677,13 +3874,29 @@ pub async fn restore_session() -> Result { - // #217: raise trade_key_index past every recovered trade before - // returning, so the next derive_trade_key() can't reuse a key a - // recovered trade already owns. Monotonic and idempotent. A persist - // failure fails the restore: an un-resynced counter reopens the - // key-reuse bug this closes, so silent success would be worse than - // a surfaced error the caller can retry. - if let Some(floor) = recovered_max_trade_index(&info) { + // #217: raise trade_key_index before returning, so the next + // derive_trade_key() can't reuse a key a recovered trade already + // owns. Monotonic and idempotent. A persist failure fails the + // restore: an un-resynced counter reopens the key-reuse bug this + // closes, so silent success would be worse than a surfaced error + // the caller can retry. + // + // #328: the authoritative floor is the daemon's LastTradeIndex + // counter. The restore payload lists only non-finalized orders, so + // recovered_max_trade_index is a lower bound (a finalized trade + // holding a higher index is invisible) — kept only as a fallback + // for when the daemon does not answer. + let daemon_counter = match last_trade_index().await { + Ok(idx) => idx, + Err(e) => { + crate::api::logging::blog_warn("restore", format!( + "LastTradeIndex request errored ({e}); \ + falling back to restore payload max" + )); + None + } + }; + if let Some(floor) = resync_floor(daemon_counter, &info) { crate::api::identity::ensure_trade_key_index_at_least(floor).await?; } Ok(info) @@ -3813,6 +4026,63 @@ mod tests { ); } + // ── #328 sanitize_trade_index / resync_floor ───────────────────────────── + #[test] + fn sanitize_trade_index_drops_negative_and_out_of_range() { + assert_eq!(sanitize_trade_index(0), Some(0)); + assert_eq!(sanitize_trade_index(42), Some(42)); + assert_eq!(sanitize_trade_index(-1), None); + // u32::MAX is the reserved terminal index (dropped to avoid +1 overflow). + assert_eq!(sanitize_trade_index(i64::from(u32::MAX)), None); + assert_eq!(sanitize_trade_index(i64::from(u32::MAX) - 1), Some(u32::MAX - 1)); + assert_eq!(sanitize_trade_index(i64::from(u32::MAX) + 1), None); + } + + #[test] + fn resync_floor_takes_the_higher_of_daemon_counter_and_payload_max() { + // The #328 scenario: order X open at index 1 (the only non-finalized + // trade the restore returns), order Y canceled at index 2. The payload + // max is 1, but the daemon's LastTradeIndex counter is 2 — the real + // high-water mark — and must win, or the first new order collides. + assert_eq!(resync_floor(Some(2), &restore_info(vec![1], vec![])), Some(2)); + // Defense-in-depth: never resync below a recovered trade's own index. + // A consistent daemon cannot answer below an index it still tracks (it + // raised last_trade_index when it accepted that order), so this only + // guards a stale/partial reply — the payload lower bound then wins. + assert_eq!(resync_floor(Some(1), &restore_info(vec![5], vec![])), Some(5)); + // Daemon present, payload empty -> the daemon value. + assert_eq!(resync_floor(Some(7), &restore_info(vec![], vec![])), Some(7)); + } + + #[test] + fn a_replayed_last_trade_index_reply_is_rejected() { + use mostro_core::message::{Action, MessageKind}; + + let reply = |request_id: Option| { + MessageKind::new(None, request_id, Some(7), Action::LastTradeIndex, None) + }; + // The genuine reply echoes our nonce. + assert!(is_matching_last_trade_index_reply(&reply(Some(42)), 42)); + // A replay of an earlier request's reply carries a different nonce... + assert!(!is_matching_last_trade_index_reply(&reply(Some(41)), 42)); + // ...or none at all — this client's own earlier requests sent no id, + // so their stored replies are exactly the replay material to reject. + assert!(!is_matching_last_trade_index_reply(&reply(None), 42)); + // A different action never matches, even with the right nonce. + let other = + MessageKind::new(None, Some(42), Some(7), Action::RestoreSession, None); + assert!(!is_matching_last_trade_index_reply(&other, 42)); + } + + #[test] + fn resync_floor_falls_back_to_payload_max_when_daemon_is_silent() { + // No LastTradeIndex answer (timeout / error): the restore-payload + // maximum is the best available lower bound. + assert_eq!(resync_floor(None, &restore_info(vec![3, 9], vec![4])), Some(9)); + // Nothing anywhere -> no resync (None). + assert_eq!(resync_floor(None, &restore_info(vec![], vec![])), None); + } + fn insert_pending_create(key: &str, request_id: u64) -> tokio::sync::oneshot::Receiver { let (tx, rx) = tokio::sync::oneshot::channel::(); pending_requests().lock().unwrap().insert( diff --git a/rust/src/mostro/actions.rs b/rust/src/mostro/actions.rs index ff7e31a5..459af572 100644 --- a/rust/src/mostro/actions.rs +++ b/rust/src/mostro/actions.rs @@ -13,7 +13,7 @@ /// arguments — see `api::identity::get_transport_identity_keys`, which /// applies the runtime privacy toggle. use anyhow::Result; -use mostro_core::message::{Action, Message, Payload}; +use mostro_core::message::{Action, Message, MessageKind, Payload}; use nostr_sdk::prelude::*; use uuid::Uuid; @@ -419,6 +419,32 @@ pub async fn restore_session( wrap_message_first_contact(identity_keys, trade_keys, mostro_pubkey, &msg).await } +/// Build and wrap a `LastTradeIndex` request (#328). +/// +/// Account-scoped: `identity_keys` sign BOTH the Seal and the rumor, so the +/// daemon resolves the account by sender pubkey and NO trade key is derived. +/// The reply carries the counter in `MessageKind::trade_index`. Payload must be +/// `None` (enforced by mostro-core). Mirrors `mostro-cli`'s +/// `execute_last_trade_index`, which signs with `identity_keys` for both. +/// +/// `request_id` is the correlation nonce the daemon echoes in its reply +/// (`mostro/src/app/last_trade_index.rs`) — the caller uses it to reject +/// replayed replies from earlier requests. +pub async fn last_trade_index( + identity_keys: &Keys, + mostro_pubkey: &PublicKey, + request_id: u64, +) -> Result { + let msg = Message::Restore(MessageKind::new( + None, + Some(request_id), + None, + Action::LastTradeIndex, + None, + )); + wrap_message_first_contact(identity_keys, identity_keys, mostro_pubkey, &msg).await +} + #[cfg(test)] mod tests { use super::*;