From 8adaf6574fed215e452f6402b254ccd82344d743 Mon Sep 17 00:00:00 2001 From: grunch Date: Wed, 29 Jul 2026 19:58:01 -0300 Subject: [PATCH 1/4] feat(chat): migrate P2P chat to the gift-wrap-free envelope MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Implements https://mostro.network/protocol/chat.html (issue #246), replacing the simplified NIP-59 gift wrap whose random ephemeral authors allowed unattributable third-party flooding of any active conversation. - crypto/chat_keys.rs: HKDF-SHA256 split of the trade-key ECDH secret into K_conv (encryption / p tag) and K_sign (outer author), verified against the spec test vector. - nostr/gift_wrap.rs: mostro_wrap / mostro_unwrap — outer kind 14 signed with K_sign carrying a NIP-44 encrypted kind 1 inner event signed by the sender's trade key; full crypto-side validation with a test per rejection path. NIP-59 wrap/unwrap stays for dispute admin chat only. - api/messages.rs: subscription pinned to authors=[pub(K_sign)], bounded by a persisted per-order since cursor (clamped to the local clock) + limit; cheapest-check-first pipeline with outer-id LRU and a 30/min (burst 60) token bucket before any crypto work; flood breaker; inner signature verified and checked against the order's two trade keys; durable replay dedup on the inner event id; attachments ride the same envelope. - Chat history now persists to the messages table (write-through store); message_exists added to the Storage trait (web stub answers false, #233). - No dual-read window: kind 1059 is no longer accepted for peer chat. mostro-core 0.14.1 still ships the superseded envelope; this stays local until the canonical implementation lands upstream. --- CLAUDE.md | 10 +- rust/Cargo.lock | 1 + rust/Cargo.toml | 4 + rust/src/api/messages.rs | 751 +++++++++++++----- rust/src/api/orders.rs | 28 +- rust/src/crypto/chat_keys.rs | 121 +++ rust/src/crypto/mod.rs | 1 + rust/src/db/indexeddb.rs | 6 + rust/src/db/mod.rs | 21 + rust/src/db/sqlite.rs | 78 ++ rust/src/nostr/gift_wrap.rs | 463 ++++++++++- .../contracts/messages.md | 38 +- 12 files changed, 1280 insertions(+), 242 deletions(-) create mode 100644 rust/src/crypto/chat_keys.rs diff --git a/CLAUDE.md b/CLAUDE.md index b07cce23..fa51d112 100644 --- a/CLAUDE.md +++ b/CLAUDE.md @@ -99,8 +99,12 @@ bridged by flutter_rust_bridge. ## Transport (protocol v2) - **Daemon messages** (new-order, take, release, cancel, dispute, rate, invoice, restore): **NIP-44 / signed Kind 14** (transport v2), via `wrap_mostro_message`/`unwrap_mostro_message`. -- **Peer & dispute chat**: **NIP-59 gift wrap / Kind 1059**, via `wrap`/`unwrap`. -- Both live in `rust/src/nostr/gift_wrap.rs` (rename to `transport.rs` pending). +- **Peer chat**: **chat envelope** (kind 14 signed with `K_sign`, NIP-44 inner kind 1 + signed by the trade key — ), via + `mostro_wrap`/`mostro_unwrap` + `crypto/chat_keys.rs`. NIP-59 is gone from this + channel (gift-wrap flood attack, issue #246). +- **Dispute admin chat**: still **NIP-59 gift wrap / Kind 1059**, via `wrap`/`unwrap`. +- All live in `rust/src/nostr/gift_wrap.rs` (rename to `transport.rs` pending). - Wire status strings are **kebab-case** (`waiting-buyer-invoice`, `fiat-sent`). ## Translations @@ -133,7 +137,7 @@ bridged by flutter_rust_bridge. ## Domain gotchas (durable) - **Reputation/ratings come from Kind 38383 event tags, not a DB.** In-memory `RATING_STORE`/`DISPUTE_STORE` are correct by design — don't invent "persist to DB" tasks. - The only real persistence gap is **chat history**. + Chat history persists to the `messages` table since #246 (web still memory-only, #233). - **Order book is sourced only from daemon Kind 38383 events.** `create_order` waits for daemon confirmation; on timeout it returns an error and **persists nothing** (no phantom order). diff --git a/rust/Cargo.lock b/rust/Cargo.lock index f3d6fcdd..b67d8941 100644 --- a/rust/Cargo.lock +++ b/rust/Cargo.lock @@ -2267,6 +2267,7 @@ dependencies = [ "chacha20poly1305", "flutter_rust_bridge", "hex", + "hkdf", "indexed_db_futures", "k256", "log", diff --git a/rust/Cargo.toml b/rust/Cargo.toml index a79c28ea..8d86ebb4 100644 --- a/rust/Cargo.toml +++ b/rust/Cargo.toml @@ -50,6 +50,10 @@ rand = { version = "0.8", features = ["getrandom"] } # Hashing (nym derivation) sha2 = "0.10" +# HKDF-SHA256 split of the chat ECDH secret into K_conv / K_sign +# (P2P chat envelope — https://mostro.network/protocol/chat.html, issue #246) +hkdf = "0.12" + # secp256k1 scalar math (ECDH shared-key derivation for P2P chat) k256 = { version = "0.13", features = ["ecdh", "arithmetic"] } diff --git a/rust/src/api/messages.rs b/rust/src/api/messages.rs index 51b89bec..015083bf 100644 --- a/rust/src/api/messages.rs +++ b/rust/src/api/messages.rs @@ -1,9 +1,23 @@ /// Messages API — encrypted P2P chat during trades. /// -/// P2P chat uses the ECDH-derived shared key (NIP-44 v2). -/// Admin/dispute chat uses the BIP-32 trade key. -/// All outbound messages are NIP-59 Gift Wrapped. -/// Messages persist locally (in-memory until DB layer is wired in Phase 10+). +/// P2P chat rides the chat envelope of the protocol spec +/// (, issue #246): a kind 14 outer +/// event signed with `K_sign` and `p`-tagged to `pub(K_conv)` — both derived +/// from the trade-key ECDH secret via `crate::crypto::chat_keys` — carrying a +/// NIP-44 encrypted kind 1 inner event signed by the sender's trade key. The +/// old NIP-59 gift wrap (kind 1059) is gone from this channel: its random +/// ephemeral authors made third-party flooding unattributable and unfilterable. +/// Admin/dispute chat (api/disputes.rs) still uses gift wrap. +/// +/// Messages persist to the `messages` table (native; web is memory-only until +/// IndexedDB lands, #233); the in-memory store is a write-through cache. The +/// stored inner-event ids double as the durable replay dedup the spec +/// requires, and the per-order `chat_cursor:` setting bounds the subscription +/// backlog. +/// +/// **Isolation invariant**: everything here runs on its own spawned task and +/// bounded channels; a chat failure or flood must never block the order state +/// machine, the daemon transport, or opening a dispute. /// /// Streams: `on_new_message(trade_id)`, `on_unread_count_changed()`, /// `on_attachment_progress(message_id)`. @@ -13,6 +27,7 @@ use std::sync::{Arc, OnceLock}; use tokio::sync::{broadcast, RwLock}; use crate::api::types::{AttachmentInfo, ChatMessage, DownloadStatus, FileType, MessageType}; +use crate::db::Storage; use crate::nostr::blossom; // ── Types ──────────────────────────────────────────────────────────────────── @@ -30,8 +45,13 @@ pub struct FileDownloadResult { // ── Message store ───────────────────────────────────────────────────────────── struct MessageStore { - /// Messages keyed by trade_id. + /// Messages keyed by trade_id. Write-through cache over the `messages` + /// table: adds persist immediately, reads hydrate from the DB once per + /// trade. Where no DB backend exists (web, unit tests) it degrades to + /// memory-only. messages: Arc>>>, + /// Trades whose persisted history has been loaded into `messages`. + hydrated: Arc>>, /// Broadcast channel for new messages (payload = trade_id of new message). new_message_tx: broadcast::Sender, /// Broadcast channel for global unread count changes. @@ -47,13 +67,46 @@ impl MessageStore { let (attachment_tx, _) = broadcast::channel(64); Self { messages: Arc::new(RwLock::new(HashMap::new())), + hydrated: Arc::new(RwLock::new(std::collections::HashSet::new())), new_message_tx, unread_tx, attachment_tx, } } + /// Load the persisted history for `trade_id` into memory, once. + /// + /// Memory wins on id collision: an in-flight message may already sit in + /// the cache with fresher state (e.g. attachment download progress). + async fn ensure_hydrated(&self, trade_id: &str) { + if self.hydrated.read().await.contains(trade_id) { + return; + } + let persisted = match crate::db::app_db::db() { + Some(db) => match db.list_messages(trade_id).await { + Ok(msgs) => msgs, + Err(e) => { + log::warn!("[messages] history load failed trade={trade_id}: {e}"); + Vec::new() + } + }, + None => Vec::new(), + }; + let mut store = self.messages.write().await; + let entry = store.entry(trade_id.to_string()).or_default(); + for msg in persisted { + if !entry.iter().any(|m| m.id == msg.id) { + entry.push(msg); + } + } + drop(store); + self.hydrated.write().await.insert(trade_id.to_string()); + } + async fn add_message(&self, msg: ChatMessage) { + // Hydrate first so the persisted history is not masked by a fresher + // in-memory entry created before the first read. + self.ensure_hydrated(&msg.trade_id).await; { let mut store = self.messages.write().await; store @@ -61,17 +114,48 @@ impl MessageStore { .or_default() .push(msg.clone()); } + // Write-through: chat history and the durable replay dedup both live + // in the `messages` table. Failure is logged, never propagated — a + // full disk must not take the chat (let alone the trade) down. + if let Some(db) = crate::db::app_db::db() { + if let Err(e) = db.save_message(&msg).await { + log::warn!("[messages] persist failed id={}: {e}", msg.id); + } + } let _ = self.new_message_tx.send(msg.clone()); let unread = self.unread_count_inner().await; let _ = self.unread_tx.send(unread); } + /// `true` if this message id was already accepted, in memory or on disk. + /// + /// This is the spec's durable inner-id replay dedup: a re-wrapped inner + /// event keeps the id it had the first time, so a hit here rejects it. + async fn is_known(&self, trade_id: &str, id: &str) -> bool { + { + let store = self.messages.read().await; + if let Some(msgs) = store.get(trade_id) { + if msgs.iter().any(|m| m.id == id) { + return true; + } + } + } + match crate::db::app_db::db() { + Some(db) => db.message_exists(id) + .await + .unwrap_or(false), + None => false, + } + } + async fn get_messages(&self, trade_id: &str) -> Vec { + self.ensure_hydrated(trade_id).await; let store = self.messages.read().await; store.get(trade_id).cloned().unwrap_or_default() } async fn mark_as_read(&self, trade_id: &str) { + self.ensure_hydrated(trade_id).await; let mut store = self.messages.write().await; if let Some(msgs) = store.get_mut(trade_id) { for m in msgs.iter_mut() { @@ -79,6 +163,11 @@ impl MessageStore { } } drop(store); + if let Some(db) = crate::db::app_db::db() { + if let Err(e) = db.mark_messages_read(trade_id).await { + log::warn!("[messages] mark_messages_read failed trade={trade_id}: {e}"); + } + } let unread = self.unread_count_inner().await; let _ = self.unread_tx.send(unread); } @@ -103,10 +192,56 @@ fn message_store() -> &'static MessageStore { // ── Public API ──────────────────────────────────────────────────────────────── +/// The chat-key material for one conversation, derived from the session. +struct ChatContext { + trade_keys: nostr_sdk::Keys, + /// `K_conv` — NIP-44 encryption; `pub(K_conv)` is the `p` tag. + conv: nostr_sdk::Keys, + /// `K_sign` — outer-event author; what relays and clients filter on. + sign: nostr_sdk::Keys, +} + +/// Derive the conversation keys for a session's trade-key index and peer. +/// +/// Cheap enough to derive on demand (one ECDH + two HKDF expands), which +/// keeps the secrets out of long-lived session state. +async fn chat_context(trade_key_index: u32, peer_hex: &str) -> Result { + let trade_keys = crate::api::identity::get_active_trade_keys(trade_key_index) + .await + .map_err(|e| anyhow!("key retrieval failed: {e}"))?; + let peer_pubkey = nostr_sdk::PublicKey::from_hex(peer_hex) + .map_err(|e| anyhow!("invalid peer pubkey: {e}"))?; + let (conv, sign) = crate::crypto::chat_keys::derive_chat_keys(&trade_keys, &peer_pubkey)?; + Ok(ChatContext { + trade_keys, + conv, + sign, + }) +} + +/// Wrap `payload` in the chat envelope and publish it. +/// +/// Returns the signed inner event on success — its id and timestamp are the +/// message's durable identity (shared with the recipient's replay dedup). +async fn publish_chat_payload(ctx: &ChatContext, payload: &str) -> Result { + let (outer, inner) = + crate::nostr::gift_wrap::mostro_wrap(&ctx.trade_keys, &ctx.conv, &ctx.sign, payload) + .await?; + let pool = crate::api::nostr::get_pool().map_err(|_| anyhow!("relay pool not ready"))?; + pool.client() + .send_event(&outer) + .await + .map_err(|e| anyhow!("publish failed: {e}"))?; + Ok(inner) +} + /// Send an encrypted text message to the trade counterparty. /// -/// Validates that `content` is non-empty. Encrypts via NIP-59 and publishes -/// to relays. If offline, the message is queued (queue wired in Phase 10+). +/// Validates that `content` is non-empty, wraps it in the chat envelope +/// (kind 14 signed with `K_sign`, inner kind 1 signed with the trade key) and +/// publishes it. If the session, peer, or relay pool is not available the +/// message is stored locally with a warning — same graceful degradation as +/// before, chat never throws for transport reasons. /// /// Returns the sent `ChatMessage` (with `is_mine: true`). pub async fn send_message(trade_id: String, content: String) -> Result { @@ -117,95 +252,40 @@ pub async fn send_message(trade_id: String, content: String) -> Result Err(anyhow!("key retrieval failed: {e}")), - (Ok(_), None) => { - log::warn!("[messages] session exists but peer not yet known — local-only"); - Ok(()) - } - (Ok(keys), Some(peer_hex)) => match nostr_sdk::PublicKey::from_hex(peer_hex) { - Err(e) => Err(anyhow!("invalid peer pubkey: {e}")), - Ok(peer_pubkey) => { - // Derive the shared-key pubkey per the Mostro P2P chat protocol: - // the p-tag of the gift wrap MUST be the ECDH shared pubkey, - // not the peer's trade pubkey, so that only the two parties - // can find (and decrypt) each other's messages. - let shared_pubkey = match s - .shared_key - .and_then(|sk| nostr_sdk::SecretKey::from_slice(&sk).ok()) - .map(|sk| nostr_sdk::Keys::new(sk).public_key()) - { - Some(pk) => pk, - None => { - // Derive on the fly if not cached. - let raw = - crate::crypto::ecdh::derive_nip04_shared_key(keys, &peer_pubkey) - .map_err(|e| anyhow!("ECDH derive failed: {e}"))?; - nostr_sdk::SecretKey::from_slice(&raw) - .map(|sk| nostr_sdk::Keys::new(sk).public_key()) - .map_err(|e| anyhow!("shared key→pubkey failed: {e}"))? - } - }; - let payload = serde_json::json!({ "text": content }).to_string(); - match crate::nostr::gift_wrap::wrap( - keys, - &shared_pubkey, - &payload, - nostr_sdk::Kind::from(14u16), - ) - .await - { - Err(e) => Err(anyhow!("gift wrap failed: {e}")), - Ok(event_json) => { - if let Ok(pool) = crate::api::nostr::get_pool() { - match serde_json::from_str::(&event_json) { - Ok(event) => pool - .client() - .send_event(&event) - .await - .map(|_| ()) - .map_err(|e| anyhow!("publish failed: {e}")), - Err(e) => Err(anyhow!("event parse failed: {e}")), - } - } else { - log::warn!( - "[messages] relay pool not ready — message stored locally" - ); - Ok(()) - } + // Local-only defaults, replaced on successful publish by the inner + // event's identity so both sides agree on the message id. + let mut id = uuid::Uuid::new_v4().to_string(); + let mut created_at = unix_now(); + let mut sender_pubkey = String::new(); + + match &session { + None => log::warn!("[messages] no session for trade={trade_id} — local-only"), + Some(s) => match &s.peer_pubkey { + None => log::warn!("[messages] session exists but peer not yet known — local-only"), + Some(peer_hex) => match chat_context(s.trade_key_index, peer_hex).await { + Err(e) => log::warn!("[messages] send_message trade={trade_id}: {e}"), + Ok(ctx) => { + sender_pubkey = ctx.trade_keys.public_key().to_hex(); + match publish_chat_payload(&ctx, &content).await { + Err(e) => log::warn!("[messages] send_message trade={trade_id}: {e}"), + Ok(inner) => { + id = inner.id.to_hex(); + created_at = inner.created_at.as_secs() as i64; } } } }, - }; - - (sender_pubkey, result) - } else { - log::warn!("[messages] no session for trade={trade_id} — local-only"); - (String::new(), Ok(())) - }; - - if let Err(e) = publish_result { - log::warn!("[messages] send_message trade={trade_id}: {e}"); + }, } let msg = ChatMessage { - id: uuid::Uuid::new_v4().to_string(), + id, trade_id: trade_id.clone(), sender_pubkey, content, @@ -214,7 +294,7 @@ pub async fn send_message(trade_id: String, content: String) -> Result = None; + if let Some(peer_hex) = &peer_pubkey_hex { - match nostr_sdk::PublicKey::from_hex(peer_hex) { - Err(e) => log::warn!("[messages] send_file invalid peer pubkey: {e}"), - Ok(peer_pubkey) => { - // Use shared-key pubkey as gift-wrap p-tag per protocol. - let shared_pubkey_res = nostr_sdk::SecretKey::from_slice(&shared_key) - .map(|sk| nostr_sdk::Keys::new(sk).public_key()) - .map_err(|_| ()) - .or_else(|_| { - crate::crypto::ecdh::derive_nip04_shared_key(&sender_keys, &peer_pubkey) - .and_then(|raw| { - nostr_sdk::SecretKey::from_slice(&raw) - .map(|sk| nostr_sdk::Keys::new(sk).public_key()) - .map_err(|e| anyhow::anyhow!("{e}")) - }) - }); - match shared_pubkey_res { - Err(e) => log::warn!("[messages] send_file shared key derive failed: {e}"), - Ok(shared_pubkey) => match crate::nostr::gift_wrap::wrap( - &sender_keys, - &shared_pubkey, - &payload, - nostr_sdk::Kind::from(14u16), - ) - .await - { - Err(e) => log::warn!("[messages] send_file gift wrap failed: {e}"), - Ok(event_json) => match crate::api::nostr::get_pool() { - Err(_) => log::warn!("[messages] send_file relay pool not ready"), - Ok(pool) => match serde_json::from_str::(&event_json) - { - Err(e) => { - log::warn!("[messages] send_file event parse failed: {e}") - } - Ok(event) => { - if let Err(e) = pool.client().send_event(&event).await { - log::warn!("[messages] send_file publish failed: {e}"); - } - } - }, - }, - }, + match chat_context(trade_key_index, peer_hex).await { + Err(e) => log::warn!("[messages] send_file trade={trade_id}: {e}"), + Ok(ctx) => match publish_chat_payload(&ctx, &payload).await { + Err(e) => log::warn!("[messages] send_file trade={trade_id}: {e}"), + Ok(inner) => { + published_id = Some(inner.id.to_hex()); + msg_created_at = inner.created_at.as_secs() as i64; } - } + }, } } else { log::warn!("[messages] send_file peer not yet known — local-only"); @@ -378,9 +430,10 @@ pub async fn send_file( local_path: None, }; - let now = unix_now(); + // Prefer the inner event id so the stored message matches the identity + // the recipient (and our own restart catch-up) dedups on. let msg = ChatMessage { - id: msg_id.clone(), + id: published_id.unwrap_or(msg_id), trade_id: trade_id.clone(), sender_pubkey, content: blossom_url, @@ -389,7 +442,7 @@ pub async fn send_file( is_read: true, has_attachment: true, attachment: Some(attachment), - created_at: now, + created_at: msg_created_at, }; message_store().add_message(msg.clone()).await; @@ -645,21 +698,168 @@ fn mime_to_file_type(mime: &str) -> FileType { // ── Incoming-chat subscription ──────────────────────────────────────────────── -/// Spawn a background task that listens for NIP-59 gift-wrap (Kind 1059) events -/// addressed to `shared_pubkey` and delivers decrypted messages into the in-memory -/// message store, firing `on_new_message` so the Dart UI can react. +/// Cap on the backlog requested from relays in one subscription. +const CHAT_BACKLOG_LIMIT: usize = 500; + +/// Token bucket sizing per the spec: ~30 messages/minute sustained with a +/// burst of 60, refused **before** any cryptographic work. +const RATE_CAPACITY: f64 = 60.0; +const RATE_PER_SEC: f64 = 0.5; + +/// Consecutive rejected events before the conversation is marked flooded and +/// processing stops. At the sustained rate this is several minutes of pure +/// garbage from the only author able to produce it — the counterparty. +const FLOOD_TRIP_REJECTIONS: u32 = 180; + +/// Entries kept in the outer-event-id LRU. Pre-decryption filter against +/// duplicate relay deliveries only — the security-bearing dedup is the +/// durable inner-id check in `MessageStore::is_known`. +const OUTER_LRU_CAP: usize = 512; + +/// Bounded insert-only id set with FIFO eviction (outer-id LRU, step 5). +struct BoundedIdSet { + set: std::collections::HashSet, + order: std::collections::VecDeque, + cap: usize, +} + +impl BoundedIdSet { + fn new(cap: usize) -> Self { + Self { + set: std::collections::HashSet::new(), + order: std::collections::VecDeque::new(), + cap, + } + } + + /// Insert `id`; returns `false` if it was already present. + fn insert(&mut self, id: &str) -> bool { + if self.set.contains(id) { + return false; + } + if self.order.len() >= self.cap { + if let Some(evicted) = self.order.pop_front() { + self.set.remove(&evicted); + } + } + self.set.insert(id.to_string()); + self.order.push_back(id.to_string()); + true + } +} + +/// Token bucket refilled continuously, drained one token per event (step 6). +struct TokenBucket { + tokens: f64, + last: crate::rt::time::Instant, +} + +impl TokenBucket { + fn new(now: crate::rt::time::Instant) -> Self { + Self { + tokens: RATE_CAPACITY, + last: now, + } + } + + /// Take one token at time `now`; `false` when the budget is exhausted. + fn try_take(&mut self, now: crate::rt::time::Instant) -> bool { + let elapsed = now.duration_since(self.last).as_secs_f64(); + self.last = now; + self.tokens = (self.tokens + elapsed * RATE_PER_SEC).min(RATE_CAPACITY); + if self.tokens >= 1.0 { + self.tokens -= 1.0; + true + } else { + false + } + } +} + +/// Read the persisted `since` cursor for `order_id`, if any. +async fn load_chat_cursor(order_id: &str) -> Option { + let db = crate::db::app_db::db()?; + db.get_setting(&crate::db::settings_keys::chat_cursor(order_id)) + .await + .ok() + .flatten()? + .parse() + .ok() +} + +/// Persist the `since` cursor. Best-effort: on web this is a no-op until +/// IndexedDB lands (#233), so the backlog bound degrades to per-process. +async fn store_chat_cursor(order_id: &str, ts: i64) { + if let Some(db) = crate::db::app_db::db() { + let key = crate::db::settings_keys::chat_cursor(order_id); + if let Err(e) = db.set_setting(&key, &ts.to_string()).await { + log::warn!("[messages] cursor persist failed order={order_id}: {e}"); + } + } +} + +/// Interpret a validated inner-event payload. /// -/// Called by `orders::on_peer_pubkey_received` as soon as the shared key is known. -/// Runs until the relay pool shuts down or an idle-timeout fires (30 min of silence). +/// Attachments travel as a JSON pointer object (`type: "file"`) — everything +/// else is plaintext. Returns `(content, attachment)` where `content` is the +/// display text (the Blossom URL for attachments, mirroring `send_file`). +fn parse_chat_payload(payload: &str) -> (String, Option) { + if let Ok(v) = serde_json::from_str::(payload) { + if v.get("type").and_then(|t| t.as_str()) == Some("file") { + if let (Some(url), Some(name), Some(mime)) = ( + v.get("url").and_then(|x| x.as_str()), + v.get("name").and_then(|x| x.as_str()), + v.get("mime_type").and_then(|x| x.as_str()), + ) { + let attachment = AttachmentInfo { + file_name: name.to_string(), + mime_type: mime.to_string(), + file_size: v.get("size").and_then(|s| s.as_u64()).unwrap_or(0), + file_type: mime_to_file_type(mime), + download_status: DownloadStatus::Pending, + local_path: None, + }; + return (url.to_string(), Some(attachment)); + } + } + } + (payload.to_string(), None) +} + +/// Spawn-able listener for the P2P chat conversation of one order. +/// +/// Subscribes with **`authors = [pub(K_sign)]`** — the rule that eliminates +/// third-party flooding: relays drop everything not signed by the +/// conversation key, so junk never reaches us — bounded by the persisted +/// `since` cursor plus a `limit`, so a restart never re-downloads an +/// unbounded backlog. +/// +/// Incoming events run the spec's cheapest-check-first pipeline: author → +/// outer-id LRU → rate-limit budget → `mostro_unwrap` (p tag, timestamp +/// bounds, size, both signatures, allowed signers) → durable inner-id dedup. +/// The one deliberate deviation: the LRU and budget run *before* the p-tag / +/// timestamp / size checks rather than after — all five are O(1) compares, +/// and what matters is that no signature or decryption work happens before +/// the budget gate. +/// +/// Isolation: this is its own task over a bounded notification channel. It +/// only ever drops chat events; it cannot touch the order state machine, the +/// daemon transport, or dispute flows. +/// +/// Called by `orders::on_peer_pubkey_received` as soon as the peer (and thus +/// the conversation keys) are known. Runs until the relay pool shuts down, an +/// idle-timeout fires (30 min of silence), or the conversation trips the +/// flood breaker. pub(crate) async fn subscribe_incoming_chat( order_id: String, - trade_pubkey_hex: String, - shared_pubkey: nostr_sdk::PublicKey, - recipient_keys: nostr_sdk::Keys, + my_trade_pubkey: nostr_sdk::PublicKey, + peer_pubkey: nostr_sdk::PublicKey, + conv: nostr_sdk::Keys, + sign: nostr_sdk::Keys, ) { + use crate::rt::time::{timeout, Duration}; use nostr_sdk::RelayPoolNotification; use tokio::sync::broadcast; - use crate::rt::time::{timeout, Duration}; const IDLE_TIMEOUT_SECS: u64 = 30 * 60; @@ -669,9 +869,19 @@ pub(crate) async fn subscribe_incoming_chat( }; let client = pool.client(); - let filter = nostr_sdk::Filter::new() - .kind(nostr_sdk::Kind::from(1059u16)) - .pubkey(shared_pubkey); + let sign_pubkey = sign.public_key(); + let allowed_signers = [my_trade_pubkey, peer_pubkey]; + + // `since` from the persisted cursor: everything older is already stored + // locally (the cursor only advances on accepted messages). + let mut cursor = load_chat_cursor(&order_id).await.unwrap_or(0); + let mut filter = nostr_sdk::Filter::new() + .kind(nostr_sdk::Kind::PrivateDirectMessage) + .author(sign_pubkey) + .limit(CHAT_BACKLOG_LIMIT); + if cursor > 0 { + filter = filter.since(nostr_sdk::Timestamp::from_secs(cursor as u64)); + } // Obtain the receiver BEFORE subscribing — same pattern as subscribe_gift_wraps. // This avoids a race where an event arrives between subscribe() and notifications() @@ -683,12 +893,15 @@ pub(crate) async fn subscribe_incoming_chat( return; } - let shared_pubkey_hex = shared_pubkey.to_hex(); log::info!( - "[messages] incoming-chat subscription active order={order_id} shared_pubkey={shared_pubkey_hex}" + "[messages] incoming-chat subscription active order={order_id} author={} since={cursor}", + sign_pubkey.to_hex() ); let mut last_activity = crate::rt::time::Instant::now(); + let mut outer_seen = BoundedIdSet::new(OUTER_LRU_CAP); + let mut bucket = TokenBucket::new(crate::rt::time::Instant::now()); + let mut consecutive_rejected: u32 = 0; loop { let remaining = @@ -700,97 +913,104 @@ pub(crate) async fn subscribe_incoming_chat( match timeout(remaining, rx.recv()).await { Ok(Ok(RelayPoolNotification::Event { event, .. })) => { - if event.kind != nostr_sdk::Kind::from(1059u16) { + if event.kind != nostr_sdk::Kind::PrivateDirectMessage { continue; } - // Only process events addressed to our shared pubkey. - 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(shared_pubkey_hex.as_str()) - }); - if !is_for_us { + // Step 1 — author. Kind 14 is shared with the daemon + // transport; a different author is somebody else's traffic + // (routed by its own subscription), not a violation. + if event.pubkey != sign_pubkey { continue; } last_activity = crate::rt::time::Instant::now(); - // Decrypt gift wrap. - let event_json = match serde_json::to_string(&*event) { - Ok(j) => j, - Err(e) => { - log::warn!("[messages] incoming-chat event serialize failed: {e}"); - continue; + // Step 5 — outer-id LRU: duplicate relay deliveries cost one + // hash lookup, nothing more. + if !outer_seen.insert(&event.id.to_hex()) { + continue; + } + + // Step 6 — rate-limit budget, before any cryptographic work. + if !bucket.try_take(crate::rt::time::Instant::now()) { + consecutive_rejected += 1; + if consecutive_rejected >= FLOOD_TRIP_REJECTIONS { + log::error!( + "[messages] conversation flooded — halting chat for order={order_id} \ + (author={}); the trade itself stays fully operational", + sign_pubkey.to_hex() + ); + crate::api::logging::blog_info( + "messages", + format!("chat flooded, processing stopped order={order_id}"), + ); + return; } - }; - let rumor_json = - match crate::nostr::gift_wrap::unwrap(&recipient_keys, &event_json).await { - Ok(j) => j, - Err(e) => { - log::warn!("[messages] incoming-chat decrypt failed: {e}"); - continue; - } - }; + continue; + } - // Parse rumor into a kind-1 inner event JSON. - let inner: serde_json::Value = match serde_json::from_str(&rumor_json) { - Ok(v) => v, + // Steps 2,3,4,7–11,13 — the crypto-side validation. + let inner = match crate::nostr::gift_wrap::mostro_unwrap( + &conv, + &sign_pubkey, + &allowed_signers, + &event, + nostr_sdk::Timestamp::now(), + ) { + Ok(inner) => inner, Err(e) => { - log::warn!("[messages] incoming-chat rumor parse failed: {e}"); + // Only the counterparty can author a validly-signed + // outer event, so failures here are attributable. + log::warn!("[messages] incoming-chat rejected order={order_id}: {e}"); + consecutive_rejected += 1; + if consecutive_rejected >= FLOOD_TRIP_REJECTIONS { + log::error!( + "[messages] conversation flooded — halting chat for \ + order={order_id}; the trade itself stays fully operational" + ); + return; + } continue; } }; + consecutive_rejected = 0; + + // Advance the persisted cursor for every event that passed + // validation (echoes included), clamped to our own clock so a + // counterparty dating events at the skew-tolerance edge can + // never push it into the future and silence the conversation. + let accepted_ts = (event.created_at.as_secs() as i64).min(unix_now()); + if accepted_ts > cursor { + cursor = accepted_ts; + store_chat_cursor(&order_id, cursor).await; + } - // Sender pubkey lives in the inner event `pubkey` field. - // - // Protocol guarantee (Mostro P2P chat spec): the inner kind-1 - // event MUST be signed by the sender's trade key. `inner.pubkey` - // is therefore the sender's trade pubkey. Key rotation is not - // supported — a trader always uses the same BIP-32 derived key - // for the entire lifetime of a trade session. - // - // We compare against `trade_pubkey_hex` (our own trade key) to - // filter echo messages: relays reflect our own gift-wraps back - // to us because the subscription filter uses only the shared-key - // p-tag, which both parties share. The outer event pubkey is an - // ephemeral key generated per-message and carries no identity - // information — only the inner `pubkey` is authoritative here. - let sender_pubkey = inner - .get("pubkey") - .and_then(|v| v.as_str()) - .unwrap_or("") - .to_string(); - - // Ignore echoes of our own messages (inner sender == our trade key). - if sender_pubkey == trade_pubkey_hex { - log::debug!("[messages] incoming-chat ignoring own echo"); + // Step 12 — durable replay dedup on the inner id. Also what + // skips echoes of our own already-stored sends. + let inner_id = inner.id.to_hex(); + if message_store().is_known(&order_id, &inner_id).await { + log::debug!("[messages] incoming-chat duplicate inner id={inner_id}"); continue; } - let content = inner - .get("content") - .and_then(|v| v.as_str()) - .unwrap_or("") - .to_string(); - let created_at = inner - .get("created_at") - .and_then(|v| v.as_i64()) - .unwrap_or_else(unix_now); - - let msg = crate::api::types::ChatMessage { - id: inner - .get("id") - .and_then(|v| v.as_str()) - .map(|s| s.to_string()) - .unwrap_or_else(|| uuid::Uuid::new_v4().to_string()), + // An unknown echo of our own message (this device lost its + // local copy, or another device of ours sent it): store it as + // ours so history reconstructs, but never as unread. + let is_echo = inner.pubkey == my_trade_pubkey; + let (content, attachment) = parse_chat_payload(&inner.content); + + let msg = ChatMessage { + id: inner_id, trade_id: order_id.clone(), - sender_pubkey, + sender_pubkey: inner.pubkey.to_hex(), content, - message_type: crate::api::types::MessageType::Peer, - is_mine: false, - is_read: false, - has_attachment: false, - attachment: None, - created_at, + message_type: MessageType::Peer, + is_mine: is_echo, + is_read: is_echo, + has_attachment: attachment.is_some(), + attachment, + // Presentation orders by the inner timestamp, which the + // relative bound has already tied to the outer one. + created_at: inner.created_at.as_secs() as i64, }; log::debug!("[messages] incoming-chat rx order={order_id} id={}", msg.id); @@ -798,6 +1018,8 @@ pub(crate) async fn subscribe_incoming_chat( } Ok(Ok(RelayPoolNotification::Shutdown)) => break, Ok(Err(broadcast::error::RecvError::Lagged(n))) => { + // The bounded notification channel dropped n events under + // pressure — chat data loss, never trade-traffic loss. log::warn!("[messages] incoming-chat lagged by {n} messages"); continue; } @@ -993,6 +1215,101 @@ mod tests { assert_eq!(msgs.len(), 2); } + #[test] + fn token_bucket_sustains_the_spec_rate_and_burst() { + use crate::rt::time::{Duration, Instant}; + + let start = Instant::now(); + let mut bucket = TokenBucket::new(start); + + // Full burst available immediately. + for i in 0..RATE_CAPACITY as u32 { + assert!(bucket.try_take(start), "burst token {i} refused"); + } + // Exhausted: the 61st in the same instant is refused. + assert!(!bucket.try_take(start)); + + // After 2 seconds one token has refilled (0.5/s), not two. + let later = start + Duration::from_secs(2); + assert!(bucket.try_take(later)); + assert!(!bucket.try_take(later)); + + // A long quiet period refills only up to the cap. + let much_later = start + Duration::from_secs(24 * 3600); + for _ in 0..RATE_CAPACITY as u32 { + assert!(bucket.try_take(much_later)); + } + assert!(!bucket.try_take(much_later)); + } + + #[test] + fn outer_id_lru_dedups_and_evicts_fifo() { + let mut set = BoundedIdSet::new(2); + assert!(set.insert("a")); + assert!(!set.insert("a"), "duplicate must be refused"); + assert!(set.insert("b")); + // Capacity 2: inserting c evicts a (FIFO)… + assert!(set.insert("c")); + assert!(set.insert("a"), "evicted id is acceptable again"); + // …which is exactly why this LRU carries no security requirement: + // the durable inner-id dedup does. + } + + #[test] + fn chat_payload_parses_files_and_plaintext() { + // Attachment pointer → content is the URL, attachment populated. + let file = serde_json::json!({ + "url": "https://blossom.example.com/abc", + "name": "receipt.jpg", + "mime_type": "image/jpeg", + "size": 12345, + "type": "file", + }) + .to_string(); + let (content, att) = parse_chat_payload(&file); + assert_eq!(content, "https://blossom.example.com/abc"); + let att = att.expect("attachment expected"); + assert_eq!(att.file_name, "receipt.jpg"); + assert_eq!(att.file_size, 12345); + assert!(matches!(att.file_type, FileType::Image)); + assert!(matches!(att.download_status, DownloadStatus::Pending)); + + // Plaintext stays as-is. + let (content, att) = parse_chat_payload("hola, ¿pagaste?"); + assert_eq!(content, "hola, ¿pagaste?"); + assert!(att.is_none()); + + // JSON that is not a file pointer is displayed verbatim, not + // misinterpreted. + let (content, att) = parse_chat_payload(r#"{"type":"file","url":"x"}"#); + assert_eq!(content, r#"{"type":"file","url":"x"}"#); + assert!(att.is_none(), "incomplete pointer must not become an attachment"); + } + + #[tokio::test] + async fn is_known_finds_messages_already_in_memory() { + let trade_id = uuid::Uuid::new_v4().to_string(); + let store = message_store(); + let id = uuid::Uuid::new_v4().to_string(); + store + .add_message(ChatMessage { + id: id.clone(), + trade_id: trade_id.clone(), + sender_pubkey: "peer".to_string(), + content: "hello".to_string(), + message_type: MessageType::Peer, + is_mine: false, + is_read: false, + has_attachment: false, + attachment: None, + created_at: unix_now(), + }) + .await; + + assert!(store.is_known(&trade_id, &id).await); + assert!(!store.is_known(&trade_id, "unknown-id").await); + } + #[tokio::test] async fn on_new_message_stream_fires_for_correct_trade() { let trade_id = uuid::Uuid::new_v4().to_string(); diff --git a/rust/src/api/orders.rs b/rust/src/api/orders.rs index 00b60f72..fc3d5bfd 100644 --- a/rust/src/api/orders.rs +++ b/rust/src/api/orders.rs @@ -1880,7 +1880,7 @@ async fn dispatch_mostro_message( ); // Derive the ECDH shared key and store in session so the chat API // can encrypt/decrypt P2P messages and subscribe to the right p-tag. - on_peer_pubkey_received(&order_id, trade_pubkey_hex, &peer_pubkey_hex).await; + on_peer_pubkey_received(&order_id, &peer_pubkey_hex).await; // Sync the order status from the payload so the trade doesn't stay // stuck at Pending in the DB and in-memory order book. @@ -2195,7 +2195,7 @@ fn map_core_status(s: mostro_core::order::Status) -> Option { /// stores it in the session, and spawns an incoming-chat subscription on the /// shared-key pubkey so we receive peer messages from the moment the trade /// goes active. -async fn on_peer_pubkey_received(order_id: &str, trade_pubkey_hex: &str, peer_pubkey_hex: &str) { +async fn on_peer_pubkey_received(order_id: &str, peer_pubkey_hex: &str) { // Resolve trade key index from order_id. let trade_index = match get_trade_key_index(order_id).await { Some(idx) => idx, @@ -2256,15 +2256,26 @@ async fn on_peer_pubkey_received(order_id: &str, trade_pubkey_hex: &str, peer_pu "[orders] on_peer_pubkey_received: session not found for order={order_id}, skipping session update — incoming subscription still spawned" ); } - // Spawn incoming-chat subscription on shared-key pubkey. + // Derive the chat conversation keys (K_conv / K_sign — HKDF split of the + // trade-key ECDH secret, protocol chat spec) and spawn the incoming-chat + // subscription pinned to their author key. + let (conv, sign) = match crate::crypto::chat_keys::derive_chat_keys(&trade_keys, &peer_pubkey) + { + Ok(pair) => pair, + Err(e) => { + log::error!("[orders] on_peer_pubkey_received: chat key derivation failed: {e}"); + return; + } + }; let order_id_owned = order_id.to_string(); - let trade_pubkey_hex_owned = trade_pubkey_hex.to_string(); + let my_trade_pubkey = trade_keys.public_key(); crate::rt::spawn(async move { crate::api::messages::subscribe_incoming_chat( order_id_owned, - trade_pubkey_hex_owned, - shared_pubkey, - trade_keys, + my_trade_pubkey, + peer_pubkey, + conv, + sign, ) .await; }); @@ -3505,8 +3516,7 @@ mod tests { // Use a random order_id that has no session — should log a warning only. on_peer_pubkey_received( &uuid::Uuid::new_v4().to_string(), - "aabbccdd", // trade_pubkey_hex (irrelevant, no trade key stored) - "aabbccdd", // peer_pubkey_hex (also irrelevant) + "aabbccdd", // peer_pubkey_hex (irrelevant, no trade key stored) ) .await; // If we reach here without panicking the test passes. diff --git a/rust/src/crypto/chat_keys.rs b/rust/src/crypto/chat_keys.rs new file mode 100644 index 00000000..5e2e4800 --- /dev/null +++ b/rust/src/crypto/chat_keys.rs @@ -0,0 +1,121 @@ +/// Chat key derivation for the P2P chat envelope. +/// +/// Implements the "Shared Key" section of the protocol chat spec +/// (): the ECDH secret shared by +/// the two trade keys is split with HKDF-SHA256 (empty salt, domain-separated +/// `info` strings) into two secp256k1 keypairs: +/// +/// * `K_conv` — NIP-44 encryption of the payload; `pub(K_conv)` is the +/// conversation address carried in the `p` tag. Disclosed to a solver +/// during a dispute (read-only grant). +/// * `K_sign` — signs the outer kind 14 event; `pub(K_sign)` is the author +/// every client filters on. Never disclosed. +/// +/// The ECDH here is `nostr::util::generate_shared_key` (the raw x-coordinate +/// of the shared point) — **not** `ecdh::derive_nip04_shared_key`, which +/// hashes it with SHA-256. The spec's test vector is derived from the raw +/// form; mixing the two silently yields a different conversation. +use anyhow::{anyhow, Result}; +// Leading `::` selects the `hkdf` crate: `nostr_sdk::prelude` also exports a +// module by that name, so a plain `use hkdf::Hkdf` is ambiguous. +use ::hkdf::Hkdf; +use nostr_sdk::prelude::*; +use nostr_sdk::util::generate_shared_key; +use sha2::Sha256; + +/// HKDF `info` strings. Changing either value changes the wire format. +const CONV_INFO: &[u8] = b"mostro:chat:conv:v1"; +const SIGN_INFO: &[u8] = b"mostro:chat:sign:v1"; + +/// Derive the domain-separated conversation and signing keys for one order. +/// +/// Both parties reach the same pair: the ECDH secret is symmetric, and HKDF +/// is deterministic. +/// +/// Returns `(K_conv, K_sign)`. +pub fn derive_chat_keys(own_trade: &Keys, peer_trade: &PublicKey) -> Result<(Keys, Keys)> { + let shared = generate_shared_key(own_trade.secret_key(), peer_trade) + .map_err(|e| anyhow!("chat ECDH failed: {e}"))?; + let hkdf = Hkdf::::new(None, &shared); + + let derive = |info: &[u8]| -> Result { + // Retry with a counter byte on the negligible chance that the output + // is not a valid secp256k1 secret key (zero or >= curve order). + for counter in 0u16..=255 { + let mut labelled = info.to_vec(); + if counter > 0 { + labelled.push(counter as u8); + } + let mut out = [0u8; 32]; + hkdf.expand(&labelled, &mut out) + .map_err(|e| anyhow!("HKDF expand failed: {e}"))?; + if let Ok(sk) = SecretKey::from_slice(&out) { + return Ok(Keys::new(sk)); + } + } + Err(anyhow!("HKDF failed to produce a valid secret key")) + }; + + Ok((derive(CONV_INFO)?, derive(SIGN_INFO)?)) +} + +#[cfg(test)] +mod tests { + use super::*; + + // Trade keys from the spec's test vector. + const ALICE_SK: &str = "548f68890c49fa42f104c60352395e60ff030b0b407e955f1eed1400d6c0347a"; + const BOB_SK: &str = "f258e73f07386d37133718b6127f873dd7c391b8f43b331ff8254034a13d2943"; + + #[test] + fn derivation_matches_the_spec_test_vector() { + let alice = Keys::parse(ALICE_SK).unwrap(); + let bob = Keys::parse(BOB_SK).unwrap(); + + let (conv, sign) = derive_chat_keys(&alice, &bob.public_key()).unwrap(); + + assert_eq!( + conv.public_key().to_hex(), + "bceb1cd2a8e98ee9729122a1693edcc39c3ace04582ff96a26705c5e4078a6f2", + "pub(K_conv) diverges from the spec test vector", + ); + assert_eq!( + sign.public_key().to_hex(), + "1dba04571059183f76b148119cfa6f8004dad30cb4e810180a6df17386a7f0b4", + "pub(K_sign) diverges from the spec test vector", + ); + } + + #[test] + fn both_parties_derive_the_same_pair() { + let alice = Keys::parse(ALICE_SK).unwrap(); + let bob = Keys::parse(BOB_SK).unwrap(); + + let (a_conv, a_sign) = derive_chat_keys(&alice, &bob.public_key()).unwrap(); + let (b_conv, b_sign) = derive_chat_keys(&bob, &alice.public_key()).unwrap(); + + assert_eq!(a_conv.public_key(), b_conv.public_key()); + assert_eq!(a_sign.public_key(), b_sign.public_key()); + } + + #[test] + fn conv_and_sign_keys_differ() { + let alice = Keys::parse(ALICE_SK).unwrap(); + let bob = Keys::parse(BOB_SK).unwrap(); + + let (conv, sign) = derive_chat_keys(&alice, &bob.public_key()).unwrap(); + assert_ne!(conv.public_key(), sign.public_key()); + } + + #[test] + fn different_orders_yield_different_conversations() { + // Fresh trade keys — a different order derives unrelated chat keys. + let alice1 = Keys::generate(); + let alice2 = Keys::generate(); + let bob = Keys::generate(); + + let (conv1, _) = derive_chat_keys(&alice1, &bob.public_key()).unwrap(); + let (conv2, _) = derive_chat_keys(&alice2, &bob.public_key()).unwrap(); + assert_ne!(conv1.public_key(), conv2.public_key()); + } +} diff --git a/rust/src/crypto/mod.rs b/rust/src/crypto/mod.rs index bbd378f2..310b4703 100644 --- a/rust/src/crypto/mod.rs +++ b/rust/src/crypto/mod.rs @@ -1,3 +1,4 @@ +pub mod chat_keys; pub mod ecdh; pub mod file_enc; pub mod keys; diff --git a/rust/src/db/indexeddb.rs b/rust/src/db/indexeddb.rs index f06bb7d7..a39030b7 100644 --- a/rust/src/db/indexeddb.rs +++ b/rust/src/db/indexeddb.rs @@ -49,6 +49,12 @@ impl Storage for IndexedDbStorage { async fn mark_messages_read(&self, _trade_id: &str) -> Result<()> { Err(anyhow!("IndexedDB not yet implemented")) } + async fn message_exists(&self, _id: &str) -> Result { + // Stub contract (#233): answer "not stored" so the caller falls back + // to its in-memory dedup. On web, replay dedup is process-lifetime + // only until IndexedDB lands. + Ok(false) + } async fn save_relay(&self, _relay: &RelayInfo) -> Result<()> { Err(anyhow!("IndexedDB not yet implemented")) } diff --git a/rust/src/db/mod.rs b/rust/src/db/mod.rs index cb5bda57..c04251ce 100644 --- a/rust/src/db/mod.rs +++ b/rust/src/db/mod.rs @@ -25,6 +25,18 @@ pub mod settings_keys { /// Developer mint-URL override, pointing Cashu at a local mint instead of /// the one the node advertises. pub const CASHU_MINT_URL_OVERRIDE: &str = "cashu_mint_url_override"; + + /// Per-order chat `since` cursor — the `created_at` (unix seconds, decimal + /// string) of the newest accepted outer chat event, clamped to the local + /// clock. Full key is `chat_cursor:`; build it with + /// [`chat_cursor`]. Bounds the chat subscription backlog so a flood is + /// never re-downloaded on restart (protocol chat spec, issue #246). + pub const CHAT_CURSOR_PREFIX: &str = "chat_cursor:"; + + /// Build the settings key holding the chat `since` cursor for `order_id`. + pub fn chat_cursor(order_id: &str) -> String { + format!("{CHAT_CURSOR_PREFIX}{order_id}") + } } /// Storage trait — implemented by both SQLite (native) and IndexedDB (WASM). @@ -52,6 +64,15 @@ pub trait Storage: Send + Sync { async fn list_messages(&self, trade_id: &str) -> Result>; async fn mark_messages_read(&self, trade_id: &str) -> Result<()>; + /// `true` if a message with this id was already accepted and stored. + /// + /// This is the **durable inner-event-id dedup** required by the chat spec: + /// both parties hold `K_sign`, so either can re-wrap a previously received + /// inner event inside a fresh outer one ("I sent the fiat", replayed). An + /// in-memory LRU is not enough — an evicted entry makes the message + /// replayable again — so the check must reach persisted history. + async fn message_exists(&self, id: &str) -> Result; + async fn save_relay(&self, relay: &crate::api::types::RelayInfo) -> Result<()>; async fn delete_relay(&self, url: &str) -> Result<()>; async fn list_relays(&self) -> Result>; diff --git a/rust/src/db/sqlite.rs b/rust/src/db/sqlite.rs index 979a582a..819806c9 100644 --- a/rust/src/db/sqlite.rs +++ b/rust/src/db/sqlite.rs @@ -209,6 +209,14 @@ impl Storage for SqliteStorage { .collect() } + async fn message_exists(&self, id: &str) -> Result { + let row: Option<(i64,)> = sqlx::query_as("SELECT 1 FROM messages WHERE id = ?") + .bind(id) + .fetch_optional(&self.pool) + .await?; + Ok(row.is_some()) + } + async fn mark_messages_read(&self, trade_id: &str) -> Result<()> { sqlx::query( "UPDATE messages SET is_read = 1 WHERE trade_id = ? AND is_read = 0", @@ -530,6 +538,76 @@ mod tests { std::env::temp_dir().join(format!("mostro_test_{}_{n}.db", std::process::id())) } + #[tokio::test] + async fn message_exists_is_durable_replay_dedup() { + use crate::api::types::*; + + let path = temp_db_path(); + let storage = SqliteStorage::open(path.to_str().unwrap()).await.unwrap(); + + // messages.trade_id has a FK to trades(id) — store the trade first, + // exactly as the take-order flow does before chat ever runs. + let trade_id = "trade-dedup-1".to_string(); + let trade = TradeInfo { + id: trade_id.clone(), + order: OrderInfo { + id: "order-dedup-1".into(), + kind: OrderKind::Sell, + status: OrderStatus::Active, + amount_sats: Some(1000), + fiat_amount: Some(10.0), + fiat_amount_min: None, + fiat_amount_max: None, + fiat_code: "VES".into(), + payment_method: "bank".into(), + premium: 0.0, + creator_pubkey: "maker".into(), + created_at: 1, + expires_at: None, + is_mine: false, + rating: 0.0, + total_reviews: 0, + days_active: 0, + }, + role: TradeRole::Buyer, + counterparty_pubkey: "peer".into(), + current_step: TradeStep::Buyer(BuyerStep::FiatSent), + hold_invoice: None, + buyer_invoice: None, + trade_key_index: 1, + cooperative_cancel_state: None, + timeout_at: None, + started_at: 1, + completed_at: None, + outcome: None, + }; + storage.save_trade(&trade).await.unwrap(); + + let inner_id = "3f".repeat(32); + assert!(!storage.message_exists(&inner_id).await.unwrap()); + + let msg = ChatMessage { + id: inner_id.clone(), + trade_id, + sender_pubkey: "peer".into(), + content: "I sent the fiat".into(), + message_type: MessageType::Peer, + is_mine: false, + is_read: false, + has_attachment: false, + attachment: None, + created_at: 2, + }; + storage.save_message(&msg).await.unwrap(); + + // A re-wrapped replay carries the same inner id — now known, durably. + assert!(storage.message_exists(&inner_id).await.unwrap()); + assert!(!storage.message_exists("un".repeat(32).as_str()).await.unwrap()); + + drop(storage); + let _ = std::fs::remove_file(&path); + } + #[tokio::test] async fn active_mostro_pubkey_round_trip() { let path = temp_db_path(); diff --git a/rust/src/nostr/gift_wrap.rs b/rust/src/nostr/gift_wrap.rs index b6b8570b..eb82827e 100644 --- a/rust/src/nostr/gift_wrap.rs +++ b/rust/src/nostr/gift_wrap.rs @@ -9,11 +9,17 @@ /// proof, NIP-44 encryption and event signing/verification all live in /// mostro-core. See `specs/005-transport-v2-migration/`. /// +/// * `mostro_wrap` / `mostro_unwrap` — the P2P chat envelope +/// (): a kind 14 event signed +/// with `K_sign` (derived from the trade-key ECDH secret, see +/// `crate::crypto::chat_keys`), carrying a NIP-44 encrypted kind 1 event +/// signed by the sender's trade key. Replaced the simplified NIP-59 gift +/// wrap, which allowed unattributable third-party flooding — issue #246. +/// /// * `wrap` / `unwrap` — raw JSON content gift-wrapped (NIP-59, Kind 1059), -/// used for NIP-17-style text DMs (P2P chat, dispute admin messages). These -/// are **not** part of the v2 migration: they keep using gift wrap. Their -/// payloads are not `mostro_core::Message` values, so they stay on the -/// local helper — see issue #101, "Scope". +/// still used for dispute admin messages only. Their payloads are not +/// `mostro_core::Message` values, so they stay on the local helper — see +/// issue #101, "Scope". use anyhow::{anyhow, Result}; use mostro_core::message::Message; use mostro_core::nip59::{UnwrappedMessage, WrapOptions}; @@ -86,10 +92,188 @@ pub async fn unwrap_mostro_message( .map_err(|e| anyhow!("unwrap_message failed: {e}")) } -// ── NIP-17 text DMs (P2P chat, dispute admin) ──────────────────────────────── +// ── P2P chat envelope (kind 14 signed with K_sign) ─────────────────────────── +// +// Implements the event structure and the crypto-side validation steps of the +// chat spec. The caller (api/messages.rs) owns the stateful steps: outer-id +// LRU, rate-limit budget, durable inner-id dedup, and the `since` cursor. +// +// mostro-core 0.14.1 still ships the superseded gift-wrap chat +// (`wrap_chat_message` / `unwrap_chat_message`); this stays a local +// implementation until the canonical one lands upstream — flagged in #246. + +/// Tolerance for clock skew, applied both between the inner and outer +/// `created_at` and against the recipient's own clock. +pub const MAX_CLOCK_SKEW_SECS: u64 = 60; + +/// Upper bound on the encrypted payload, enforced before decrypting. +pub const MAX_CONTENT_BYTES: usize = 64 * 1024; + +/// Build the outer kind 14 event carrying an encrypted, trade-key-signed +/// kind 1 event, per the P2P chat spec. +/// +/// The inner event authenticates the sender; the outer event authenticates +/// the conversation and is what clients (and relays) filter on. When the +/// connected Mostro requires NIP-13 Proof of Work, the difficulty is applied +/// to the outer event. +/// +/// Returns `(outer, inner)` — the caller publishes `outer` and keeps +/// `inner.id` as the message's durable identity (it is what the recipient's +/// replay dedup keys on, so both sides agree on it). +/// +/// # Arguments +/// - `sender_trade`: the sender's trade keys, used to sign the inner event. +/// - `conv`: `K_conv`, used to encrypt and as the `p` tag. +/// - `sign`: `K_sign`, used to sign the outer event. +/// - `message`: the plaintext payload (text, or attachment-metadata JSON). +pub async fn mostro_wrap( + sender_trade: &Keys, + conv: &Keys, + sign: &Keys, + message: &str, +) -> Result<(Event, Event)> { + // One timestamp for both events: the real moment the message is sent. + // Recipients reject a mismatch, which is what bounds replays. No NIP-59 + // timestamp tweaking — it would break `since`-based sync. + let now = Timestamp::now(); + + let inner = EventBuilder::text_note(message) + .custom_created_at(now) + .build(sender_trade.public_key()) + .sign(sender_trade) + .await + .map_err(|e| anyhow!("inner event sign failed: {e}"))?; + + // NIP-44 self-encryption: K_conv is both sides of the key exchange. + let content = nip44::encrypt( + conv.secret_key(), + &conv.public_key(), + inner.as_json(), + nip44::Version::V2, + ) + .map_err(|e| anyhow!("NIP-44 encrypt failed: {e}"))?; + + // Exactly one `p` tag, ours. Anything else could hide the message from + // the `#p` query a dispute solver uses to rebuild the transcript. + let builder = EventBuilder::new(Kind::PrivateDirectMessage, content) + .tag(Tag::public_key(conv.public_key())) + .custom_created_at(now); + + let pow = crate::mostro::pow::get_pow(); + let builder = if pow > 0 { builder.pow(pow) } else { builder }; + + let outer = builder + .sign_with_keys(sign) + .map_err(|e| anyhow!("outer event sign failed: {e}"))?; + + Ok((outer, inner)) +} + +/// Validate an incoming outer event and return the inner event. +/// +/// Implements the crypto-side steps of the spec's cheapest-check-first +/// validation order (author, `p` tag, absolute timestamp bound, size, outer +/// signature, decrypt, inner signature, allowed signer, inner kind, relative +/// timestamp bound). Three steps are the **caller's**, because they need +/// state this function does not own: the bounded LRU on the outer id, the +/// rate-limit budget, and the durable dedup on the inner id. A caller that +/// skips the durable inner-id check accepts replays. +/// +/// # Arguments +/// - `conv`: `K_conv`, used to decrypt. +/// - `sign_pubkey`: `pub(K_sign)` of this conversation. +/// - `allowed_signers`: the buyer's and the seller's trade pubkeys. +/// - `outer`: the received kind 14 event. +/// - `now`: the recipient's current time, for the absolute timestamp bound. +pub fn mostro_unwrap( + conv: &Keys, + sign_pubkey: &PublicKey, + allowed_signers: &[PublicKey], + outer: &Event, + now: Timestamp, +) -> Result { + // A third party cannot produce a valid signature for this author, so this + // check is what makes flooding impossible. Relays enforce it too, via the + // `authors` filter; we re-check locally. + if outer.pubkey != *sign_pubkey { + return Err(anyhow!( + "outer event is not authored by the conversation signing key" + )); + } + if outer.kind != Kind::PrivateDirectMessage { + return Err(anyhow!("outer event is not kind 14")); + } + + // Exactly one `p` tag, addressing this conversation. Anything else could + // be a message engineered to stay out of a dispute solver's `#p` query. + let mut p_tags = outer + .tags + .iter() + .filter(|t| t.kind() == TagKind::p()); + match (p_tags.next().and_then(|t| t.content()), p_tags.next()) { + (Some(pk), None) if pk == conv.public_key().to_hex() => {} + _ => { + return Err(anyhow!( + "outer event must carry exactly one p tag for this conversation" + )) + } + } + + // Absolute bound against our own clock. Without it a counterparty can + // date both events far in the future — they agree with each other, so the + // relative check below passes — and poison the `since` cursor, silencing + // the conversation until that date. The past is unbounded: catching up + // after being offline is legitimate. + if outer.created_at.as_secs() > now.as_secs().saturating_add(MAX_CLOCK_SKEW_SECS) { + return Err(anyhow!("outer event is dated too far in the future")); + } + + if outer.content.len() > MAX_CONTENT_BYTES { + return Err(anyhow!("encrypted payload exceeds the accepted size")); + } + + outer + .verify() + .map_err(|e| anyhow!("outer signature invalid: {e}"))?; + + let decrypted = nip44::decrypt(conv.secret_key(), &conv.public_key(), &outer.content) + .map_err(|e| anyhow!("NIP-44 decrypt failed: {e}"))?; + let inner = Event::from_json(&decrypted) + .map_err(|e| anyhow!("inner event parse failed: {e}"))?; + + // The only authentication of who wrote the message: both parties can sign + // the outer event, so it cannot tell the two sides apart. Reading the + // inner pubkey without verifying this signature accepts forged senders. + inner + .verify() + .map_err(|e| anyhow!("inner signature invalid: {e}"))?; + if !allowed_signers.contains(&inner.pubkey) { + return Err(anyhow!( + "inner event is signed by a key that is not a party to this order" + )); + } + if inner.kind != Kind::TextNote { + return Err(anyhow!("inner event is not kind 1")); + } + + // Bounds how far back the caller's durable inner-id dedup has to reach: a + // re-wrap older than the tolerance is stale and rejected here, while one + // inside the window is caught by that dedup, never by this check. + let skew = inner + .created_at + .as_secs() + .abs_diff(outer.created_at.as_secs()); + if skew > MAX_CLOCK_SKEW_SECS { + return Err(anyhow!("inner and outer timestamps disagree — stale re-wrap")); + } + + Ok(inner) +} + +// ── NIP-59 gift wrap (dispute admin messages only) ─────────────────────────── // // These wrap arbitrary JSON content in a Kind 14 rumor. Kept as local glue -// until `mostro-core` grows a DM helper or we migrate these off NIP-59. +// for the admin/dispute channel; the P2P chat moved to `mostro_wrap` above. /// Wrap a plaintext JSON payload as a NIP-59 Gift Wrap event addressed to /// `recipient_pubkey`, signed by `sender_keys`. @@ -153,6 +337,273 @@ pub async fn unwrap(recipient_keys: &Keys, gift_wrap_json: &str) -> Result Convo { + let alice_trade = Keys::generate(); + let bob_trade = Keys::generate(); + let (conv, sign) = derive_chat_keys(&alice_trade, &bob_trade.public_key()).unwrap(); + Convo { + alice_trade, + bob_trade, + conv, + sign, + } + } + + fn unwrap_now(c: &Convo, outer: &Event) -> Result { + mostro_unwrap( + &c.conv, + &c.sign.public_key(), + &[c.alice_trade.public_key(), c.bob_trade.public_key()], + outer, + Timestamp::now(), + ) + } + + #[tokio::test] + async fn round_trip_authenticates_the_sender() { + let c = convo(); + let (outer, inner) = mostro_wrap(&c.alice_trade, &c.conv, &c.sign, "hola") + .await + .unwrap(); + + // Wire shape: kind 14 authored by pub(K_sign), one p tag = pub(K_conv), + // and no field anywhere carrying a trade pubkey. + assert_eq!(outer.kind, Kind::PrivateDirectMessage); + assert_eq!(outer.pubkey, c.sign.public_key()); + let outer_json = outer.as_json(); + assert!(!outer_json.contains(&c.alice_trade.public_key().to_hex())); + assert!(!outer_json.contains(&c.bob_trade.public_key().to_hex())); + + let got = unwrap_now(&c, &outer).unwrap(); + assert_eq!(got.id, inner.id); + assert_eq!(got.pubkey, c.alice_trade.public_key()); + assert_eq!(got.content, "hola"); + } + + #[tokio::test] + async fn wrong_outer_author_is_rejected() { + let c = convo(); + let (_, inner) = mostro_wrap(&c.alice_trade, &c.conv, &c.sign, "hola") + .await + .unwrap(); + + // A third party re-encrypts a genuine inner event under K_conv (which + // it could hold after a dispute disclosure) but must sign the outer + // event with its own key — rejected on the author check. + let mallory = Keys::generate(); + let content = nip44::encrypt( + c.conv.secret_key(), + &c.conv.public_key(), + inner.as_json(), + nip44::Version::V2, + ) + .unwrap(); + let forged = EventBuilder::new(Kind::PrivateDirectMessage, content) + .tag(Tag::public_key(c.conv.public_key())) + .custom_created_at(inner.created_at) + .sign_with_keys(&mallory) + .unwrap(); + + let err = unwrap_now(&c, &forged).unwrap_err().to_string(); + assert!(err.contains("not authored"), "got: {err}"); + } + + #[tokio::test] + async fn missing_or_foreign_p_tag_is_rejected() { + let c = convo(); + let (_, inner) = mostro_wrap(&c.alice_trade, &c.conv, &c.sign, "hola") + .await + .unwrap(); + let content = nip44::encrypt( + c.conv.secret_key(), + &c.conv.public_key(), + inner.as_json(), + nip44::Version::V2, + ) + .unwrap(); + + // No p tag: reaches us via the authors filter, but would be invisible + // in the #p transcript a dispute solver retrieves. + let no_p = EventBuilder::new(Kind::PrivateDirectMessage, content.clone()) + .custom_created_at(inner.created_at) + .sign_with_keys(&c.sign) + .unwrap(); + assert!(unwrap_now(&c, &no_p).is_err()); + + // Foreign p tag: same evasion, pointing elsewhere. + let foreign = EventBuilder::new(Kind::PrivateDirectMessage, content.clone()) + .tag(Tag::public_key(Keys::generate().public_key())) + .custom_created_at(inner.created_at) + .sign_with_keys(&c.sign) + .unwrap(); + assert!(unwrap_now(&c, &foreign).is_err()); + + // Two p tags: ours plus a decoy — "exactly one" is the contract. + let two = EventBuilder::new(Kind::PrivateDirectMessage, content) + .tag(Tag::public_key(c.conv.public_key())) + .tag(Tag::public_key(Keys::generate().public_key())) + .custom_created_at(inner.created_at) + .sign_with_keys(&c.sign) + .unwrap(); + assert!(unwrap_now(&c, &two).is_err()); + } + + #[tokio::test] + async fn far_future_timestamp_is_rejected() { + let c = convo(); + // Counterparty dates BOTH events in the future: the relative check + // passes, only the absolute bound against our clock catches it — + // this is the cursor-poisoning defence. + let future = Timestamp::from_secs(Timestamp::now().as_secs() + 7 * 24 * 3600); + let inner = EventBuilder::text_note("poison") + .custom_created_at(future) + .build(c.alice_trade.public_key()) + .sign(&c.alice_trade) + .await + .unwrap(); + let content = nip44::encrypt( + c.conv.secret_key(), + &c.conv.public_key(), + inner.as_json(), + nip44::Version::V2, + ) + .unwrap(); + let outer = EventBuilder::new(Kind::PrivateDirectMessage, content) + .tag(Tag::public_key(c.conv.public_key())) + .custom_created_at(future) + .sign_with_keys(&c.sign) + .unwrap(); + + let err = unwrap_now(&c, &outer).unwrap_err().to_string(); + assert!(err.contains("future"), "got: {err}"); + } + + #[tokio::test] + async fn oversized_payload_is_rejected_before_decrypting() { + let c = convo(); + let big = "x".repeat(MAX_CONTENT_BYTES + 1); + let outer = EventBuilder::new(Kind::PrivateDirectMessage, big) + .tag(Tag::public_key(c.conv.public_key())) + .sign_with_keys(&c.sign) + .unwrap(); + + let err = unwrap_now(&c, &outer).unwrap_err().to_string(); + assert!(err.contains("size"), "got: {err}"); + } + + #[tokio::test] + async fn inner_signed_by_a_stranger_is_rejected() { + let c = convo(); + // Outer is genuine (signed with K_sign) but the inner author is not a + // party to the order — e.g. a solver who obtained K_sign could still + // not impersonate either side. + let stranger = Keys::generate(); + let (outer, _) = mostro_wrap(&stranger, &c.conv, &c.sign, "imposter") + .await + .unwrap(); + + let err = unwrap_now(&c, &outer).unwrap_err().to_string(); + assert!(err.contains("not a party"), "got: {err}"); + } + + #[tokio::test] + async fn tampered_inner_signature_is_rejected() { + let c = convo(); + let (_, inner) = mostro_wrap(&c.alice_trade, &c.conv, &c.sign, "hola") + .await + .unwrap(); + + // Forge the inner: claim Alice's pubkey without her signature. + let mut forged: serde_json::Value = serde_json::from_str(&inner.as_json()).unwrap(); + forged["content"] = serde_json::json!("I sent the fiat"); + let content = nip44::encrypt( + c.conv.secret_key(), + &c.conv.public_key(), + forged.to_string(), + nip44::Version::V2, + ) + .unwrap(); + let outer = EventBuilder::new(Kind::PrivateDirectMessage, content) + .tag(Tag::public_key(c.conv.public_key())) + .custom_created_at(inner.created_at) + .sign_with_keys(&c.sign) + .unwrap(); + + assert!(unwrap_now(&c, &outer).is_err()); + } + + #[tokio::test] + async fn stale_rewrap_outside_the_window_is_rejected() { + let c = convo(); + let (_, inner) = mostro_wrap(&c.alice_trade, &c.conv, &c.sign, "fiat sent") + .await + .unwrap(); + + // Bob re-wraps Alice's genuine old message in a fresh outer event + // dated outside the tolerance window: the relative bound rejects it. + let later = Timestamp::from_secs(inner.created_at.as_secs() + MAX_CLOCK_SKEW_SECS + 10); + let content = nip44::encrypt( + c.conv.secret_key(), + &c.conv.public_key(), + inner.as_json(), + nip44::Version::V2, + ) + .unwrap(); + let rewrap = EventBuilder::new(Kind::PrivateDirectMessage, content) + .tag(Tag::public_key(c.conv.public_key())) + .custom_created_at(later) + .sign_with_keys(&c.sign) + .unwrap(); + + let err = mostro_unwrap( + &c.conv, + &c.sign.public_key(), + &[c.alice_trade.public_key(), c.bob_trade.public_key()], + &rewrap, + later, + ) + .unwrap_err() + .to_string(); + assert!(err.contains("stale"), "got: {err}"); + } + + #[tokio::test] + async fn conv_key_alone_cannot_author_into_the_conversation() { + // Dispute disclosure grant: K_conv decrypts, but an outer event signed + // with K_conv (instead of K_sign) is rejected — read-only access. + let c = convo(); + let (_, inner) = mostro_wrap(&c.alice_trade, &c.conv, &c.sign, "hola") + .await + .unwrap(); + let content = nip44::encrypt( + c.conv.secret_key(), + &c.conv.public_key(), + inner.as_json(), + nip44::Version::V2, + ) + .unwrap(); + let signed_with_conv = EventBuilder::new(Kind::PrivateDirectMessage, content) + .tag(Tag::public_key(c.conv.public_key())) + .custom_created_at(inner.created_at) + .sign_with_keys(&c.conv) + .unwrap(); + + assert!(unwrap_now(&c, &signed_with_conv).is_err()); + } +} + #[cfg(test)] mod tests { use super::*; diff --git a/specs/004-mostro-p2p-client/contracts/messages.md b/specs/004-mostro-p2p-client/contracts/messages.md index c2da86df..655a31d2 100644 --- a/specs/004-mostro-p2p-client/contracts/messages.md +++ b/specs/004-mostro-p2p-client/contracts/messages.md @@ -2,10 +2,30 @@ **Module**: `rust/src/api/messages.rs` -Encrypted peer-to-peer messaging during trades. P2P chat uses sharedKey -(ECDH-derived). Admin/dispute chat uses tradeKey (BIP-32 derived). All -messages are NIP-59 Gift Wrapped. Messages persist locally after decryption. -Supports encrypted file attachments via Blossom servers. +Encrypted peer-to-peer messaging during trades, over the **chat envelope** +of the protocol spec (, issue +#246): a kind 14 outer event signed with `K_sign` and `p`-tagged to +`pub(K_conv)` — both HKDF-SHA256 derivations of the trade-key ECDH secret — +carrying a NIP-44 encrypted kind 1 inner event signed by the sender's trade +key. NIP-59 gift wrap (kind 1059) is no longer used for peer chat (its +random ephemeral authors made third-party flooding unattributable); +admin/dispute chat still uses it. Messages persist locally after +validation. Supports encrypted file attachments via Blossom servers. + +**Security requirements implemented** (see the protocol spec for the +normative list): + +- Subscription pinned to `authors = [pub(K_sign)]`, bounded by a persisted + per-order `since` cursor (clamped to the local clock) plus a `limit`. +- Cheapest-check-first validation; no signature or decryption work before + the outer-id LRU and the rate-limit budget (token bucket, 30 msg/min + sustained, burst 60; sustained violation marks the conversation flooded + and halts chat processing while the trade stays operational). +- Inner signature verified and its author checked against the two trade + keys of the order — the only sender authentication. +- Durable replay dedup on the inner event id (`messages` table). +- Isolation: chat runs on its own task and bounded channels; it can never + block the order state machine, the daemon transport, or a dispute. ## Functions @@ -14,8 +34,11 @@ Send an encrypted message to the trade counterparty. **Validation**: `content` MUST not be empty. Trade MUST be active. -**Side effects**: Encrypts via NIP-59, publishes to relays. If offline, -queues in MessageQueue for delivery on reconnection. +**Side effects**: Wraps in the chat envelope (inner kind 1 signed by the +trade key, outer kind 14 signed with `K_sign`), publishes to relays. The +stored message id is the inner event id, so both sides dedup on the same +identity. If the session, peer, or relay pool is unavailable the message is +stored locally with a warning. **Errors**: `NoActiveTrade`, `TradeNotFound`, `MessageEmpty`. @@ -64,7 +87,8 @@ Encrypt and upload a file attachment, then send as a chat message. 1. Encrypt file with ChaCha20-Poly1305 (random nonce, key derived from sharedKey for P2P messages or tradeKey for admin/dispute messages). 2. Upload encrypted blob to Blossom server. -3. Send Blossom URL + encryption metadata as NIP-59 Gift Wrapped message. +3. Send Blossom URL + metadata as a JSON pointer payload (`type: "file"`) + through the same chat envelope as text messages. **Returns**: ChatMessage with `has_attachment: true` and attachment metadata. From b9f0ff36141e5fe220d54c123b897fb56224332f Mon Sep 17 00:00:00 2001 From: grunch Date: Wed, 29 Jul 2026 20:57:29 -0300 Subject: [PATCH 2/4] =?UTF-8?q?fix(chat):=20review=20round=201=20on=20#247?= =?UTF-8?q?=20=E2=80=94=20CI,=20migration=20window,=20lifecycle,=20durabil?= =?UTF-8?q?ity?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Addresses every review thread (Codex, CodeRabbit, grunch) and the red CI: - CI: mark_as_read_updates_count asserted on the global unread counter and raced with parallel tests — now asserts per-trade read state. - Persistence: messages.trade_id lost its FK to trades(id) (chat keys are per ORDER id; a taker's trades row is a fresh UUID, so every taker save_message failed). One-off table rebuild migrates v2 databases. - mark_messages_read now rewrites the is_read flag inside the JSON blob too — it survives rehydration instead of resurrecting unread badges. - The since cursor advances only after the message is durably stored; add_message reports persistence success. - Replay dedup fails closed: a storage lookup error drops the event. - Catch-up exemption: the token bucket meters only the live stream (post-EOSE); stored backlog is bounded by the filter limit instead, so history above the burst size is never dropped. - Lifecycle: one chat task per order (spawn guard), explicit subscription ids unsubscribed on every exit path, no 30-min idle death, and resubscription of persisted active trades when the pool comes online. - Dual-read migration window (write-new/read-both): inbound kind 1059 from pre-migration peers is accepted until LEGACY_CHAT_DEPRECATION_TS (2026-12-31T00:00:00Z), bounded by the same LRU/budget/size/dedup/quota; decryption uses the shared secret as the recipient key, as v1 sends it. - Per-trade retention quotas (1000 messages / 5 MiB) bound durable growth at a legitimate rate — the isolation invariant now covers storage. - Send-side size validation: stable MessageTooLarge error before publishing anything every receiver must reject. - Raw-event bound: tag-count cap before signature verification defeats junk-tag padding around the ciphertext cap. - Inner events carry a signed uniqueness nonce so two identical same-second sends keep distinct ids (dedup no longer eats double-sends). - Web: IndexedDB now implements chat messages + settings KV (durable replay dedup and cursor on web, fail-closed); the rest stays on #233. 193 rust tests green (17 new), clippy clean on touched files, wasm check green, flutter analyze/test green. --- CLAUDE.md | 5 +- rust/src/api/messages.rs | 964 +++++++++++++++--- rust/src/api/nostr.rs | 5 + rust/src/api/orders.rs | 3 +- rust/src/db/indexeddb.rs | 223 +++- rust/src/db/schema.rs | 31 +- rust/src/db/sqlite.rs | 147 ++- rust/src/nostr/gift_wrap.rs | 116 ++- .../contracts/messages.md | 26 +- 9 files changed, 1266 insertions(+), 254 deletions(-) diff --git a/CLAUDE.md b/CLAUDE.md index fa51d112..d43ce2e5 100644 --- a/CLAUDE.md +++ b/CLAUDE.md @@ -101,8 +101,9 @@ bridged by flutter_rust_bridge. **NIP-44 / signed Kind 14** (transport v2), via `wrap_mostro_message`/`unwrap_mostro_message`. - **Peer chat**: **chat envelope** (kind 14 signed with `K_sign`, NIP-44 inner kind 1 signed by the trade key — ), via - `mostro_wrap`/`mostro_unwrap` + `crypto/chat_keys.rs`. NIP-59 is gone from this - channel (gift-wrap flood attack, issue #246). + `mostro_wrap`/`mostro_unwrap` + `crypto/chat_keys.rs`. Outbound NIP-59 is gone from + this channel (gift-wrap flood attack, issue #246); inbound 1059 is still accepted + from pre-migration peers until the dual-read deadline (`LEGACY_CHAT_DEPRECATION_TS`). - **Dispute admin chat**: still **NIP-59 gift wrap / Kind 1059**, via `wrap`/`unwrap`. - All live in `rust/src/nostr/gift_wrap.rs` (rename to `transport.rs` pending). - Wire status strings are **kebab-case** (`waiting-buyer-invoice`, `fiat-sent`). diff --git a/rust/src/api/messages.rs b/rust/src/api/messages.rs index 015083bf..15806c97 100644 --- a/rust/src/api/messages.rs +++ b/rust/src/api/messages.rs @@ -5,8 +5,10 @@ /// event signed with `K_sign` and `p`-tagged to `pub(K_conv)` — both derived /// from the trade-key ECDH secret via `crate::crypto::chat_keys` — carrying a /// NIP-44 encrypted kind 1 inner event signed by the sender's trade key. The -/// old NIP-59 gift wrap (kind 1059) is gone from this channel: its random -/// ephemeral authors made third-party flooding unattributable and unfilterable. +/// old NIP-59 gift wrap (kind 1059) — whose random ephemeral authors made +/// third-party flooding unattributable and unfilterable — is no longer +/// written; it is still *read* from pre-migration peers until +/// [`LEGACY_CHAT_DEPRECATION_TS`] so mixed-version trades keep chatting. /// Admin/dispute chat (api/disputes.rs) still uses gift wrap. /// /// Messages persist to the `messages` table (native; web is memory-only until @@ -103,7 +105,12 @@ impl MessageStore { self.hydrated.write().await.insert(trade_id.to_string()); } - async fn add_message(&self, msg: ChatMessage) { + /// Store a message; returns `true` when it is **durably** stored (DB + /// write succeeded, or no DB backend exists so memory is the best this + /// platform offers). The chat `since` cursor must only advance past + /// events whose messages returned `true` — otherwise a failed write plus + /// an advanced cursor loses the message permanently. + async fn add_message(&self, msg: ChatMessage) -> bool { // Hydrate first so the persisted history is not masked by a fresher // in-memory entry created before the first read. self.ensure_hydrated(&msg.trade_id).await; @@ -117,34 +124,62 @@ impl MessageStore { // Write-through: chat history and the durable replay dedup both live // in the `messages` table. Failure is logged, never propagated — a // full disk must not take the chat (let alone the trade) down. - if let Some(db) = crate::db::app_db::db() { - if let Err(e) = db.save_message(&msg).await { - log::warn!("[messages] persist failed id={}: {e}", msg.id); - } - } + let stored = match crate::db::app_db::db() { + Some(db) => match db.save_message(&msg).await { + Ok(()) => true, + Err(e) => { + log::warn!("[messages] persist failed id={}: {e}", msg.id); + false + } + }, + None => true, + }; let _ = self.new_message_tx.send(msg.clone()); let unread = self.unread_count_inner().await; let _ = self.unread_tx.send(unread); + stored } /// `true` if this message id was already accepted, in memory or on disk. /// /// This is the spec's durable inner-id replay dedup: a re-wrapped inner /// event keeps the id it had the first time, so a hit here rejects it. - async fn is_known(&self, trade_id: &str, id: &str) -> bool { + /// A storage lookup failure is an `Err` — the caller MUST fail closed + /// (drop the event) rather than treat it as "not seen". + async fn is_known(&self, trade_id: &str, id: &str) -> Result { { let store = self.messages.read().await; if let Some(msgs) = store.get(trade_id) { if msgs.iter().any(|m| m.id == id) { - return true; + return Ok(true); } } } match crate::db::app_db::db() { - Some(db) => db.message_exists(id) + Some(db) => db + .message_exists(id) .await - .unwrap_or(false), + .map_err(|e| anyhow!("dedup lookup failed: {e}")), + None => Ok(false), + } + } + + /// `true` when storing one more incoming message of `incoming_bytes` + /// would exceed the per-trade retention caps. Bounds durable growth from + /// a counterparty writing forever at a legitimate rate — the token + /// bucket limits CPU, this limits memory and disk (isolation invariant). + async fn quota_exceeded(&self, trade_id: &str, incoming_bytes: usize) -> bool { + self.ensure_hydrated(trade_id).await; + let store = self.messages.read().await; + match store.get(trade_id) { None => false, + Some(msgs) => { + if msgs.len() >= MAX_STORED_MESSAGES_PER_TRADE { + return true; + } + let bytes: usize = msgs.iter().map(|m| m.content.len()).sum(); + bytes.saturating_add(incoming_bytes) > MAX_STORED_BYTES_PER_TRADE + } } } @@ -252,6 +287,17 @@ pub async fn send_message(trade_id: String, content: String) -> Result crate::nostr::gift_wrap::MAX_CONTENT_BYTES { + bail!( + "MessageTooLarge: {} bytes exceeds the maximum message size", + content.len() + ); + } + // Look up session to get peer pubkey and trade key index. // If no session exists (e.g. order not yet active), fall back to local-only. let session = crate::mostro::session::session_manager() @@ -273,6 +319,12 @@ pub async fn send_message(trade_id: String, content: String) -> Result { sender_pubkey = ctx.trade_keys.public_key().to_hex(); match publish_chat_payload(&ctx, &content).await { + // A message every receiver must reject is a caller + // error, not a transport hiccup — surface it instead + // of storing a "sent" message the peer never sees. + Err(e) if e.to_string().contains("MessageTooLarge") => { + return Err(e); + } Err(e) => log::warn!("[messages] send_message trade={trade_id}: {e}"), Ok(inner) => { id = inner.id.to_hex(); @@ -297,7 +349,7 @@ pub async fn send_message(trade_id: String, content: String) -> Result nostr_sdk::SubscriptionId { + nostr_sdk::SubscriptionId::new(format!("mostro-chat-{order_id}")) +} + +/// Subscription id for the legacy gift-wrap dual-read of one order. +fn legacy_chat_subscription_id(order_id: &str) -> nostr_sdk::SubscriptionId { + nostr_sdk::SubscriptionId::new(format!("mostro-chat-legacy-{order_id}")) +} + +/// Orders with a live chat task. Single-owner guard: `on_peer_pubkey_received` +/// fires again on daemon replays and reconnect backfills, and a second task +/// for the same order would double-process events and race on the cursor. +static ACTIVE_CHATS: OnceLock>> = + OnceLock::new(); + +fn active_chats() -> &'static tokio::sync::Mutex> { + ACTIVE_CHATS.get_or_init(|| tokio::sync::Mutex::new(std::collections::HashSet::new())) +} + /// Bounded insert-only id set with FIFO eviction (outer-id LRU, step 5). struct BoundedIdSet { set: std::collections::HashSet, @@ -832,37 +920,141 @@ fn parse_chat_payload(payload: &str) -> (String, Option) { /// third-party flooding: relays drop everything not signed by the /// conversation key, so junk never reaches us — bounded by the persisted /// `since` cursor plus a `limit`, so a restart never re-downloads an -/// unbounded backlog. +/// unbounded backlog. Until [`LEGACY_CHAT_DEPRECATION_TS`] it additionally +/// dual-reads the superseded NIP-59 gift wrap (kind 1059) so pre-migration +/// peers keep working; it never *writes* the legacy form. /// /// Incoming events run the spec's cheapest-check-first pipeline: author → /// outer-id LRU → rate-limit budget → `mostro_unwrap` (p tag, timestamp -/// bounds, size, both signatures, allowed signers) → durable inner-id dedup. -/// The one deliberate deviation: the LRU and budget run *before* the p-tag / -/// timestamp / size checks rather than after — all five are O(1) compares, -/// and what matters is that no signature or decryption work happens before -/// the budget gate. +/// bounds, size, both signatures, allowed signers) → durable inner-id dedup +/// (fail-closed) → retention quota. The budget is only metered on the +/// **live** stream (after the relay's EOSE): stored catch-up above the burst +/// size is legitimate history, and dropping it would permanently lose +/// messages the advancing cursor never re-fetches. Two deliberate ordering +/// deviations from the spec text: the LRU and budget run before the p-tag / +/// timestamp / size checks (all are O(1) compares; what matters is that no +/// signature or decryption work happens before the budget gate). +/// +/// Lifecycle: exactly one task per order (`ACTIVE_CHATS` guard — daemon +/// replays re-invoke `on_peer_pubkey_received` and must be no-ops), explicit +/// subscription ids unsubscribed on every exit path, and **no idle timeout**: +/// the listener lives until relay-pool shutdown or a flood trip, because a +/// quiet half hour is normal in a fiat trade and the next peer message must +/// still arrive. After a restart, `resubscribe_active_chats` rebuilds the +/// listeners for persisted active trades. /// /// Isolation: this is its own task over a bounded notification channel. It /// only ever drops chat events; it cannot touch the order state machine, the /// daemon transport, or dispute flows. -/// -/// Called by `orders::on_peer_pubkey_received` as soon as the peer (and thus -/// the conversation keys) are known. Runs until the relay pool shuts down, an -/// idle-timeout fires (30 min of silence), or the conversation trips the -/// flood breaker. pub(crate) async fn subscribe_incoming_chat( order_id: String, - my_trade_pubkey: nostr_sdk::PublicKey, + trade_keys: nostr_sdk::Keys, peer_pubkey: nostr_sdk::PublicKey, conv: nostr_sdk::Keys, sign: nostr_sdk::Keys, ) { - use crate::rt::time::{timeout, Duration}; + // Single-owner guard: a second spawn for the same order is a no-op. + { + let mut active = active_chats().lock().await; + if !active.insert(order_id.clone()) { + log::debug!("[messages] chat task already active order={order_id}"); + return; + } + } + + run_chat_subscription(&order_id, &trade_keys, &peer_pubkey, &conv, &sign).await; + + // Cleanup on every exit path: release ownership and drop the relay + // subscriptions so they never outlive the task. + active_chats().lock().await.remove(&order_id); + if let Ok(pool) = crate::api::nostr::get_pool() { + let client = pool.client(); + client.unsubscribe(&chat_subscription_id(&order_id)).await; + client + .unsubscribe(&legacy_chat_subscription_id(&order_id)) + .await; + } + log::debug!("[messages] incoming-chat subscription exiting order={order_id}"); +} + +/// Mutable per-conversation receive state (see `subscribe_incoming_chat`). +struct ChatRxState { + outer_seen: BoundedIdSet, + bucket: TokenBucket, + consecutive_rejected: u32, + /// `true` once a relay reported EOSE for one of our subscriptions — + /// from then on the token bucket meters arrivals; before that, events + /// are stored catch-up already bounded by the filter `limit`. + live: bool, + cursor: i64, + flooded: bool, +} + +impl ChatRxState { + fn new(cursor: i64) -> Self { + Self { + outer_seen: BoundedIdSet::new(OUTER_LRU_CAP), + bucket: TokenBucket::new(crate::rt::time::Instant::now()), + consecutive_rejected: 0, + live: false, + cursor, + flooded: false, + } + } + + /// Count one rejected event; trips the flood breaker on sustained abuse. + fn reject(&mut self, order_id: &str) { + self.consecutive_rejected += 1; + if self.consecutive_rejected >= FLOOD_TRIP_REJECTIONS { + self.flooded = true; + log::error!( + "[messages] conversation flooded — halting chat for order={order_id}; \ + the trade itself stays fully operational" + ); + crate::api::logging::blog_info( + "messages", + format!("chat flooded, processing stopped order={order_id}"), + ); + } + } + + /// Live-stream budget check (no-op during stored catch-up). + fn budget_ok(&mut self, order_id: &str) -> bool { + if !self.live { + return true; + } + if self.bucket.try_take(crate::rt::time::Instant::now()) { + true + } else { + self.reject(order_id); + false + } + } + + /// Advance the persisted cursor to `event_ts` clamped to our own clock, + /// so a counterparty dating events at the skew-tolerance edge can never + /// push it into the future and silence the conversation. Callers only + /// invoke this once the corresponding message is durably stored (or was + /// already known/durable). + async fn advance_cursor(&mut self, order_id: &str, event_ts: i64) { + let accepted = event_ts.min(unix_now()); + if accepted > self.cursor { + self.cursor = accepted; + store_chat_cursor(order_id, accepted).await; + } + } +} + +async fn run_chat_subscription( + order_id: &str, + trade_keys: &nostr_sdk::Keys, + peer_pubkey: &nostr_sdk::PublicKey, + conv: &nostr_sdk::Keys, + sign: &nostr_sdk::Keys, +) { use nostr_sdk::RelayPoolNotification; use tokio::sync::broadcast; - const IDLE_TIMEOUT_SECS: u64 = 30 * 60; - let Ok(pool) = crate::api::nostr::get_pool() else { log::warn!("[messages] subscribe_incoming_chat: relay pool not initialized"); return; @@ -870,11 +1062,15 @@ pub(crate) async fn subscribe_incoming_chat( let client = pool.client(); let sign_pubkey = sign.public_key(); - let allowed_signers = [my_trade_pubkey, peer_pubkey]; + let my_trade_pubkey = trade_keys.public_key(); + let allowed_signers = [my_trade_pubkey, *peer_pubkey]; // `since` from the persisted cursor: everything older is already stored - // locally (the cursor only advances on accepted messages). - let mut cursor = load_chat_cursor(&order_id).await.unwrap_or(0); + // locally (the cursor only advances on durably stored messages). + let cursor = load_chat_cursor(order_id).await.unwrap_or(0); + let sub_id = chat_subscription_id(order_id); + let legacy_sub_id = legacy_chat_subscription_id(order_id); + let mut filter = nostr_sdk::Filter::new() .kind(nostr_sdk::Kind::PrivateDirectMessage) .author(sign_pubkey) @@ -888,148 +1084,387 @@ pub(crate) async fn subscribe_incoming_chat( // and would otherwise be missed. let mut rx = client.notifications(); - if let Err(e) = client.subscribe(filter, None).await { + if let Err(e) = client.subscribe_with_id(sub_id.clone(), filter, None).await { log::warn!("[messages] subscribe_incoming_chat subscribe failed: {e}"); return; } + // Dual-read window: also accept the superseded gift-wrap envelope from + // pre-migration peers, read-only, until the deprecation date. The legacy + // address is the old NIP-04-style shared pubkey. + let legacy_shared: Option = if unix_now() < LEGACY_CHAT_DEPRECATION_TS { + // The legacy address doubles as the decryption key: v1 gift-wraps to + // the NIP-04-style shared pubkey, so the receiver must unwrap with + // the shared SECRET as its keypair. + let keys = crate::crypto::ecdh::derive_nip04_shared_key(trade_keys, peer_pubkey) + .ok() + .and_then(|raw| nostr_sdk::SecretKey::from_slice(&raw).ok()) + .map(nostr_sdk::Keys::new); + match keys { + None => { + log::warn!("[messages] legacy shared key derivation failed order={order_id}"); + None + } + Some(keys) => { + let legacy_filter = nostr_sdk::Filter::new() + .kind(nostr_sdk::Kind::GiftWrap) + .pubkey(keys.public_key()) + .limit(CHAT_BACKLOG_LIMIT); + match client + .subscribe_with_id(legacy_sub_id.clone(), legacy_filter, None) + .await + { + Ok(_) => Some(keys), + Err(e) => { + log::warn!("[messages] legacy chat subscribe failed: {e}"); + None + } + } + } + } + } else { + None + }; + log::info!( - "[messages] incoming-chat subscription active order={order_id} author={} since={cursor}", - sign_pubkey.to_hex() + "[messages] incoming-chat subscription active order={order_id} author={} since={cursor} legacy={}", + sign_pubkey.to_hex(), + legacy_shared.is_some(), ); - let mut last_activity = crate::rt::time::Instant::now(); - let mut outer_seen = BoundedIdSet::new(OUTER_LRU_CAP); - let mut bucket = TokenBucket::new(crate::rt::time::Instant::now()); - let mut consecutive_rejected: u32 = 0; + let mut state = ChatRxState::new(cursor); loop { - let remaining = - Duration::from_secs(IDLE_TIMEOUT_SECS).saturating_sub(last_activity.elapsed()); - if remaining.is_zero() { - log::debug!("[messages] incoming-chat idle timeout order={order_id}"); - break; - } - - match timeout(remaining, rx.recv()).await { - Ok(Ok(RelayPoolNotification::Event { event, .. })) => { - if event.kind != nostr_sdk::Kind::PrivateDirectMessage { - continue; - } - // Step 1 — author. Kind 14 is shared with the daemon - // transport; a different author is somebody else's traffic - // (routed by its own subscription), not a violation. - if event.pubkey != sign_pubkey { - continue; - } - last_activity = crate::rt::time::Instant::now(); - - // Step 5 — outer-id LRU: duplicate relay deliveries cost one - // hash lookup, nothing more. - if !outer_seen.insert(&event.id.to_hex()) { - continue; - } - - // Step 6 — rate-limit budget, before any cryptographic work. - if !bucket.try_take(crate::rt::time::Instant::now()) { - consecutive_rejected += 1; - if consecutive_rejected >= FLOOD_TRIP_REJECTIONS { - log::error!( - "[messages] conversation flooded — halting chat for order={order_id} \ - (author={}); the trade itself stays fully operational", - sign_pubkey.to_hex() - ); - crate::api::logging::blog_info( - "messages", - format!("chat flooded, processing stopped order={order_id}"), - ); - return; + match rx.recv().await { + Ok(RelayPoolNotification::Event { + subscription_id, + event, + .. + }) => { + if subscription_id == sub_id { + handle_chat_event(order_id, &allowed_signers, conv, &sign_pubkey, &my_trade_pubkey, &event, &mut state) + .await; + } else if subscription_id == legacy_sub_id { + if let Some(shared) = &legacy_shared { + handle_legacy_chat_event( + order_id, + shared, + &my_trade_pubkey, + &allowed_signers, + &event, + &mut state, + ) + .await; } - continue; } - - // Steps 2,3,4,7–11,13 — the crypto-side validation. - let inner = match crate::nostr::gift_wrap::mostro_unwrap( - &conv, - &sign_pubkey, - &allowed_signers, - &event, - nostr_sdk::Timestamp::now(), - ) { - Ok(inner) => inner, - Err(e) => { - // Only the counterparty can author a validly-signed - // outer event, so failures here are attributable. - log::warn!("[messages] incoming-chat rejected order={order_id}: {e}"); - consecutive_rejected += 1; - if consecutive_rejected >= FLOOD_TRIP_REJECTIONS { - log::error!( - "[messages] conversation flooded — halting chat for \ - order={order_id}; the trade itself stays fully operational" - ); - return; - } - continue; - } - }; - consecutive_rejected = 0; - - // Advance the persisted cursor for every event that passed - // validation (echoes included), clamped to our own clock so a - // counterparty dating events at the skew-tolerance edge can - // never push it into the future and silence the conversation. - let accepted_ts = (event.created_at.as_secs() as i64).min(unix_now()); - if accepted_ts > cursor { - cursor = accepted_ts; - store_chat_cursor(&order_id, cursor).await; + if state.flooded { + return; } - - // Step 12 — durable replay dedup on the inner id. Also what - // skips echoes of our own already-stored sends. - let inner_id = inner.id.to_hex(); - if message_store().is_known(&order_id, &inner_id).await { - log::debug!("[messages] incoming-chat duplicate inner id={inner_id}"); - continue; + } + Ok(RelayPoolNotification::Message { message, .. }) => { + // EOSE for one of our subscriptions: stored catch-up is over, + // the token bucket meters everything from here on. + if let nostr_sdk::RelayMessage::EndOfStoredEvents(sid) = message { + if *sid == sub_id || *sid == legacy_sub_id { + state.live = true; + } } - - // An unknown echo of our own message (this device lost its - // local copy, or another device of ours sent it): store it as - // ours so history reconstructs, but never as unread. - let is_echo = inner.pubkey == my_trade_pubkey; - let (content, attachment) = parse_chat_payload(&inner.content); - - let msg = ChatMessage { - id: inner_id, - trade_id: order_id.clone(), - sender_pubkey: inner.pubkey.to_hex(), - content, - message_type: MessageType::Peer, - is_mine: is_echo, - is_read: is_echo, - has_attachment: attachment.is_some(), - attachment, - // Presentation orders by the inner timestamp, which the - // relative bound has already tied to the outer one. - created_at: inner.created_at.as_secs() as i64, - }; - - log::debug!("[messages] incoming-chat rx order={order_id} id={}", msg.id); - message_store().add_message(msg).await; } - Ok(Ok(RelayPoolNotification::Shutdown)) => break, - Ok(Err(broadcast::error::RecvError::Lagged(n))) => { + Ok(RelayPoolNotification::Shutdown) => break, + Err(broadcast::error::RecvError::Lagged(n)) => { // The bounded notification channel dropped n events under // pressure — chat data loss, never trade-traffic loss. log::warn!("[messages] incoming-chat lagged by {n} messages"); continue; } - Ok(Err(broadcast::error::RecvError::Closed)) => break, - Err(_) => break, // idle timeout - Ok(Ok(_)) => continue, + Err(broadcast::error::RecvError::Closed) => break, } } +} - log::debug!("[messages] incoming-chat subscription exiting order={order_id}"); +/// Validate and store one incoming chat-envelope event (see +/// `subscribe_incoming_chat` for the pipeline description). +async fn handle_chat_event( + order_id: &str, + allowed_signers: &[nostr_sdk::PublicKey], + conv: &nostr_sdk::Keys, + sign_pubkey: &nostr_sdk::PublicKey, + my_trade_pubkey: &nostr_sdk::PublicKey, + event: &nostr_sdk::Event, + state: &mut ChatRxState, +) { + if event.kind != nostr_sdk::Kind::PrivateDirectMessage { + return; + } + // Step 1 — author. Kind 14 is shared with the daemon transport; a + // different author is somebody else's traffic, not a violation. + if event.pubkey != *sign_pubkey { + return; + } + // Step 5 — outer-id LRU: duplicate relay deliveries cost one hash lookup. + if !state.outer_seen.insert(&event.id.to_hex()) { + return; + } + // Step 6 — rate-limit budget, before any cryptographic work. + if !state.budget_ok(order_id) { + return; + } + + // Steps 2,3,4,7–11,13 — the crypto-side validation. + let inner = match crate::nostr::gift_wrap::mostro_unwrap( + conv, + sign_pubkey, + allowed_signers, + event, + nostr_sdk::Timestamp::now(), + ) { + Ok(inner) => inner, + Err(e) => { + // Only the counterparty can author a validly-signed outer event, + // so failures here are attributable. + log::warn!("[messages] incoming-chat rejected order={order_id}: {e}"); + state.reject(order_id); + return; + } + }; + state.consecutive_rejected = 0; + + // Step 12 — durable replay dedup on the inner id, fail-closed: a lookup + // error drops the event (and leaves the cursor put, so it is re-fetched + // once storage recovers) instead of accepting a possible replay. + let inner_id = inner.id.to_hex(); + match message_store().is_known(order_id, &inner_id).await { + Err(e) => { + log::warn!("[messages] {e} — dropping event order={order_id}"); + return; + } + Ok(true) => { + // Already durably stored (an echo of our own send, or a replay). + log::debug!("[messages] incoming-chat duplicate inner id={inner_id}"); + state + .advance_cursor(order_id, event.created_at.as_secs() as i64) + .await; + return; + } + Ok(false) => {} + } + + // Retention quota — bounds durable growth at a legitimate send rate. + if message_store() + .quota_exceeded(order_id, inner.content.len()) + .await + { + log::warn!("[messages] retention quota reached order={order_id} — dropping message"); + return; + } + + // An unknown echo of our own message (this device lost its local copy, + // or another device of ours sent it): store it as ours so history + // reconstructs, but never as unread. + let is_echo = inner.pubkey == *my_trade_pubkey; + let (content, attachment) = parse_chat_payload(&inner.content); + + let msg = ChatMessage { + id: inner_id, + trade_id: order_id.to_string(), + sender_pubkey: inner.pubkey.to_hex(), + content, + message_type: MessageType::Peer, + is_mine: is_echo, + is_read: is_echo, + has_attachment: attachment.is_some(), + attachment, + // Presentation orders by the inner timestamp, which the relative + // bound has already tied to the outer one. + created_at: inner.created_at.as_secs() as i64, + }; + + log::debug!("[messages] incoming-chat rx order={order_id} id={}", msg.id); + // Cursor moves only past durably stored messages: a failed write with an + // advanced cursor would lose the message permanently. + if message_store().add_message(msg).await { + state + .advance_cursor(order_id, event.created_at.as_secs() as i64) + .await; + } +} + +/// Validate and store one legacy gift-wrap chat event (dual-read window). +/// +/// The legacy envelope has no author to pin, so this path keeps the old +/// exposure — but bounded: same LRU, same live-stream budget, same size cap +/// before decryption, same durable dedup and quota. It exists only so a +/// pre-migration counterparty is not cut off mid-trade, and disappears at +/// [`LEGACY_CHAT_DEPRECATION_TS`]. +async fn handle_legacy_chat_event( + order_id: &str, + legacy_shared: &nostr_sdk::Keys, + my_trade_pubkey: &nostr_sdk::PublicKey, + allowed_signers: &[nostr_sdk::PublicKey], + event: &nostr_sdk::Event, + state: &mut ChatRxState, +) { + let legacy_shared_hex = legacy_shared.public_key().to_hex(); + if event.kind != nostr_sdk::Kind::GiftWrap { + return; + } + 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(legacy_shared_hex.as_str()) + }); + if !is_for_us { + return; + } + if !state.outer_seen.insert(&event.id.to_hex()) { + return; + } + if !state.budget_ok(order_id) { + return; + } + if event.content.len() > crate::nostr::gift_wrap::MAX_CONTENT_BYTES { + state.reject(order_id); + return; + } + + let event_json = match serde_json::to_string(event) { + Ok(j) => j, + Err(_) => return, + }; + let rumor_json = match crate::nostr::gift_wrap::unwrap(legacy_shared, &event_json).await { + Ok(j) => j, + Err(e) => { + log::debug!("[messages] legacy chat decrypt failed order={order_id}: {e}"); + state.reject(order_id); + return; + } + }; + let rumor: serde_json::Value = match serde_json::from_str(&rumor_json) { + Ok(v) => v, + Err(_) => { + state.reject(order_id); + return; + } + }; + state.consecutive_rejected = 0; + + // The rumor is unsigned (NIP-59), so the claimed sender is only checked + // for membership: this is exactly the weakness the new envelope fixes. + let sender_hex = rumor.get("pubkey").and_then(|v| v.as_str()).unwrap_or(""); + let Ok(sender) = nostr_sdk::PublicKey::from_hex(sender_hex) else { + return; + }; + if !allowed_signers.contains(&sender) { + return; + } + + let rumor_id = rumor + .get("id") + .and_then(|v| v.as_str()) + .map(|s| s.to_string()) + .unwrap_or_else(|| uuid::Uuid::new_v4().to_string()); + match message_store().is_known(order_id, &rumor_id).await { + Err(e) => { + log::warn!("[messages] {e} — dropping legacy event order={order_id}"); + return; + } + Ok(true) => return, + Ok(false) => {} + } + + let raw_content = rumor.get("content").and_then(|v| v.as_str()).unwrap_or(""); + if message_store() + .quota_exceeded(order_id, raw_content.len()) + .await + { + log::warn!("[messages] retention quota reached order={order_id} — dropping message"); + return; + } + + // Legacy senders wrapped text as {"text": ...}; also accept the file + // pointer shape and plain text. + let (content, attachment) = match serde_json::from_str::(raw_content) + .ok() + .and_then(|v| v.get("text").and_then(|t| t.as_str()).map(String::from)) + { + Some(text) => (text, None), + None => parse_chat_payload(raw_content), + }; + + let is_echo = sender == *my_trade_pubkey; + let msg = ChatMessage { + id: rumor_id, + trade_id: order_id.to_string(), + sender_pubkey: sender_hex.to_string(), + content, + message_type: MessageType::Peer, + is_mine: is_echo, + is_read: is_echo, + has_attachment: attachment.is_some(), + attachment, + created_at: rumor + .get("created_at") + .and_then(|v| v.as_i64()) + .unwrap_or_else(unix_now), + }; + let _ = message_store().add_message(msg).await; +} + +/// Rebuild the chat listeners for every persisted trade that can still chat. +/// +/// Called once the relay pool is up (`api::nostr::initialize`): sessions are +/// in-memory, so after a process restart nothing else would resubscribe and +/// the next peer message would be lost until the daemon happened to resend a +/// peer-pubkey notification. +pub(crate) async fn resubscribe_active_chats() { + let Some(db) = crate::db::app_db::db() else { + return; + }; + let trades = match db.list_trades().await { + Ok(t) => t, + Err(e) => { + log::warn!("[messages] resubscribe: list_trades failed: {e}"); + return; + } + }; + for trade in trades.into_iter().filter(chat_still_relevant) { + let order_id = trade.order.id.clone(); + let Ok(trade_keys) = + crate::api::identity::get_active_trade_keys(trade.trade_key_index).await + else { + continue; + }; + let Ok(peer) = nostr_sdk::PublicKey::from_hex(&trade.counterparty_pubkey) else { + continue; + }; + let Ok((conv, sign)) = crate::crypto::chat_keys::derive_chat_keys(&trade_keys, &peer) + else { + continue; + }; + log::info!("[messages] resubscribing chat order={order_id}"); + crate::rt::spawn(subscribe_incoming_chat( + order_id, trade_keys, peer, conv, sign, + )); + } +} + +/// A persisted trade still needs a live chat listener: it has a known peer +/// and has not reached a terminal outcome. +fn chat_still_relevant(trade: &crate::api::types::TradeInfo) -> bool { + use crate::api::types::OrderStatus::*; + trade.outcome.is_none() + && !trade.counterparty_pubkey.is_empty() + && matches!( + trade.order.status, + Pending + | WaitingBuyerInvoice + | WaitingPayment + | Active + | FiatSent + | SettledHoldInvoice + | Dispute + | InProgress + ) } #[cfg(test)] @@ -1104,10 +1539,27 @@ mod tests { }; store.add_message(incoming).await; - let count_before = get_unread_count().await.unwrap(); + // Assert on THIS trade's messages, not the global unread counter: + // the store is a process-wide singleton and other tests add unread + // messages concurrently, so global comparisons are racy (this + // exact flake took CI down on PR #247). + let unread_before = get_messages(trade_id.clone()) + .await + .unwrap() + .iter() + .filter(|m| !m.is_read) + .count(); + assert_eq!(unread_before, 1); + mark_as_read(trade_id.clone()).await.unwrap(); - let count_after = get_unread_count().await.unwrap(); - assert!(count_after < count_before || count_after == 0); + + let unread_after = get_messages(trade_id) + .await + .unwrap() + .iter() + .filter(|m| !m.is_read) + .count(); + assert_eq!(unread_after, 0); } #[test] @@ -1286,6 +1738,204 @@ mod tests { assert!(att.is_none(), "incomplete pointer must not become an attachment"); } + #[test] + fn bucket_is_bypassed_during_stored_catchup() { + use crate::rt::time::Instant; + + // Pre-EOSE (catch-up): a backlog far above the burst size is all + // accepted — dropping stored history would lose it permanently + // because the cursor advances past it. + let mut state = ChatRxState::new(0); + assert!(!state.live); + for _ in 0..(RATE_CAPACITY as u32 * 5) { + assert!(state.budget_ok("order-x")); + } + assert_eq!(state.consecutive_rejected, 0); + + // Post-EOSE (live): the bucket meters normally. + state.live = true; + let now = Instant::now(); + state.bucket = TokenBucket::new(now); + let mut accepted = 0; + for _ in 0..(RATE_CAPACITY as u32 + 10) { + if state.budget_ok("order-x") { + accepted += 1; + } + } + assert_eq!(accepted, RATE_CAPACITY as u32); + assert!(state.consecutive_rejected > 0); + } + + #[tokio::test] + async fn quota_bounds_messages_and_bytes_per_trade() { + let trade_id = uuid::Uuid::new_v4().to_string(); + let store = message_store(); + + // Byte cap: one huge stored message + an incoming one that would + // cross the byte quota. + let _ = store + .add_message(ChatMessage { + id: uuid::Uuid::new_v4().to_string(), + trade_id: trade_id.clone(), + sender_pubkey: "peer".into(), + content: "x".repeat(MAX_STORED_BYTES_PER_TRADE - 10), + message_type: MessageType::Peer, + is_mine: false, + is_read: true, + has_attachment: false, + attachment: None, + created_at: unix_now(), + }) + .await; + assert!(!store.quota_exceeded(&trade_id, 5).await); + assert!(store.quota_exceeded(&trade_id, 50).await); + + // An untouched trade has room. + let other = uuid::Uuid::new_v4().to_string(); + assert!(!store.quota_exceeded(&other, 1024).await); + } + + #[tokio::test] + async fn legacy_gift_wrap_is_accepted_during_the_window() { + use nostr_sdk::prelude::*; + + // Mixed-version pair: the peer still runs the pre-migration client + // and gift-wraps {"text": ...} to the NIP-04-style shared pubkey. + let my_keys = Keys::generate(); + let peer_keys = Keys::generate(); + let shared_keys = { + let raw = + crate::crypto::ecdh::derive_nip04_shared_key(&peer_keys, &my_keys.public_key()) + .unwrap(); + Keys::new(SecretKey::from_slice(&raw).unwrap()) + }; + let shared_pk = shared_keys.public_key(); + let payload = serde_json::json!({ "text": "hola desde v1" }).to_string(); + let event_json = crate::nostr::gift_wrap::wrap( + &peer_keys, + &shared_pk, + &payload, + Kind::PrivateDirectMessage, + ) + .await + .unwrap(); + let event: Event = serde_json::from_str(&event_json).unwrap(); + + let order_id = uuid::Uuid::new_v4().to_string(); + let mut state = ChatRxState::new(0); + handle_legacy_chat_event( + &order_id, + &shared_keys, + &my_keys.public_key(), + &[my_keys.public_key(), peer_keys.public_key()], + &event, + &mut state, + ) + .await; + + let msgs = get_messages(order_id.clone()).await.unwrap(); + assert_eq!(msgs.len(), 1, "legacy message must be accepted"); + assert_eq!(msgs[0].content, "hola desde v1"); + assert!(!msgs[0].is_mine); + + // Replay of the same rumor in a fresh wrap is deduped durably. + let event_json2 = crate::nostr::gift_wrap::wrap( + &peer_keys, + &shared_pk, + &payload, + Kind::PrivateDirectMessage, + ) + .await + .unwrap(); + let event2: Event = serde_json::from_str(&event_json2).unwrap(); + handle_legacy_chat_event( + &order_id, + &shared_keys, + &my_keys.public_key(), + &[my_keys.public_key(), peer_keys.public_key()], + &event2, + &mut state, + ) + .await; + // Different rumor id (new timestamp/id) → may store; a stranger's + // wrap must never store. + let stranger = Keys::generate(); + let stranger_json = crate::nostr::gift_wrap::wrap( + &stranger, + &shared_pk, + &serde_json::json!({ "text": "spoof" }).to_string(), + Kind::PrivateDirectMessage, + ) + .await + .unwrap(); + let stranger_event: Event = serde_json::from_str(&stranger_json).unwrap(); + handle_legacy_chat_event( + &order_id, + &shared_keys, + &my_keys.public_key(), + &[my_keys.public_key(), peer_keys.public_key()], + &stranger_event, + &mut state, + ) + .await; + let msgs = get_messages(order_id).await.unwrap(); + assert!( + msgs.iter().all(|m| m.content != "spoof"), + "stranger-authored rumor must be rejected" + ); + } + + #[test] + fn chat_still_relevant_selects_only_live_trades() { + use crate::api::types::*; + let base = TradeInfo { + id: "t".into(), + order: OrderInfo { + id: "o".into(), + kind: OrderKind::Sell, + status: OrderStatus::Active, + amount_sats: None, + fiat_amount: None, + fiat_amount_min: None, + fiat_amount_max: None, + fiat_code: "VES".into(), + payment_method: "bank".into(), + premium: 0.0, + creator_pubkey: "maker".into(), + created_at: 1, + expires_at: None, + is_mine: false, + rating: 0.0, + total_reviews: 0, + days_active: 0, + }, + role: TradeRole::Buyer, + counterparty_pubkey: "peer".into(), + current_step: TradeStep::Buyer(BuyerStep::FiatSent), + hold_invoice: None, + buyer_invoice: None, + trade_key_index: 1, + cooperative_cancel_state: None, + timeout_at: None, + started_at: 1, + completed_at: None, + outcome: None, + }; + assert!(chat_still_relevant(&base)); + + let mut done = base.clone(); + done.outcome = Some(TradeOutcome::Success); + assert!(!chat_still_relevant(&done)); + + let mut no_peer = base.clone(); + no_peer.counterparty_pubkey = String::new(); + assert!(!chat_still_relevant(&no_peer)); + + let mut canceled = base; + canceled.order.status = OrderStatus::Canceled; + assert!(!chat_still_relevant(&canceled)); + } + #[tokio::test] async fn is_known_finds_messages_already_in_memory() { let trade_id = uuid::Uuid::new_v4().to_string(); @@ -1306,8 +1956,8 @@ mod tests { }) .await; - assert!(store.is_known(&trade_id, &id).await); - assert!(!store.is_known(&trade_id, "unknown-id").await); + assert!(store.is_known(&trade_id, &id).await.unwrap()); + assert!(!store.is_known(&trade_id, "unknown-id").await.unwrap()); } #[tokio::test] diff --git a/rust/src/api/nostr.rs b/rust/src/api/nostr.rs index 8bf560fe..0f27ab50 100644 --- a/rust/src/api/nostr.rs +++ b/rust/src/api/nostr.rs @@ -56,6 +56,11 @@ pub async fn initialize(relays: Option>) -> Result<()> { let _ = flush_message_queue().await; // Start (or re-start) Kind 38383 order book subscription. crate::api::orders::subscribe_orders().await; + // Rebuild chat listeners for persisted active trades — + // sessions are in-memory, so after a restart nothing else + // would resubscribe. Idempotent: orders with a live chat + // task are skipped by the single-owner guard. + crate::api::messages::resubscribe_active_chats().await; } Ok(state) => { log::info!("[nostr] connection state changed: {state:?}"); diff --git a/rust/src/api/orders.rs b/rust/src/api/orders.rs index fc3d5bfd..c187c196 100644 --- a/rust/src/api/orders.rs +++ b/rust/src/api/orders.rs @@ -2268,11 +2268,10 @@ async fn on_peer_pubkey_received(order_id: &str, peer_pubkey_hex: &str) { } }; let order_id_owned = order_id.to_string(); - let my_trade_pubkey = trade_keys.public_key(); crate::rt::spawn(async move { crate::api::messages::subscribe_incoming_chat( order_id_owned, - my_trade_pubkey, + trade_keys, peer_pubkey, conv, sign, diff --git a/rust/src/db/indexeddb.rs b/rust/src/db/indexeddb.rs index a39030b7..221bbec3 100644 --- a/rust/src/db/indexeddb.rs +++ b/rust/src/db/indexeddb.rs @@ -1,8 +1,19 @@ /// IndexedDB storage backend — WASM target only. /// -/// Full implementation is deferred to Phase 3. This stub satisfies the -/// Storage trait so that the WASM build compiles during Phase 2. +/// **Chat messages and the settings KV are fully implemented** (issue #246): +/// the durable inner-event-id replay dedup and the chat `since` cursor are +/// MUST-level security requirements of the P2P chat protocol, so on web they +/// cannot be left to the in-memory fallback — a browser reload would make +/// every already-accepted message replayable again. Lookups fail **closed**: +/// a storage error is returned as `Err`, and the chat pipeline drops the +/// event rather than treating it as unseen. +/// +/// Everything else remains stubbed until the full IndexedDB backend lands +/// (#233): reads answer "nothing stored" and writes are dropped, so callers +/// fall back to their defaults instead of failing. use anyhow::{anyhow, Result}; +use indexed_db_futures::prelude::*; +use web_sys::wasm_bindgen::JsValue; use crate::api::types::{ ChatMessage, IdentityInfo, OrderInfo, QueuedMessageStatus, RelayInfo, TradeInfo, @@ -10,11 +21,103 @@ use crate::api::types::{ use crate::db::Storage; use crate::queue::outbox::QueuedMessage; -pub struct IndexedDbStorage; +const DB_VERSION: u32 = 1; +const MESSAGES_STORE: &str = "messages"; +const SETTINGS_STORE: &str = "settings"; + +/// Map an opaque JS-side error into an `anyhow` error the trait can carry. +fn js_err(context: &str, e: impl std::fmt::Debug) -> anyhow::Error { + anyhow!("{context}: {e:?}") +} + +pub struct IndexedDbStorage { + /// IndexedDB database name, from `init_db`'s path argument. + db_name: String, +} impl IndexedDbStorage { - pub async fn open(_db_name: &str) -> Result { - Ok(Self) + pub async fn open(db_name: &str) -> Result { + let storage = Self { + db_name: db_name.to_string(), + }; + // Open once eagerly so schema creation (and any quota/permission + // failure) surfaces at init time, not on the first message. + storage.open_db().await?; + Ok(storage) + } + + /// Open the database, creating the object stores on first use. + /// + /// `IdbDatabase` wraps JS values and is not `Send`, so it cannot be + /// cached in this struct (the storage singleton must be `Send + Sync`). + /// Opening per operation is cheap: the browser keeps the underlying + /// connection warm. + async fn open_db(&self) -> Result { + let mut req = IdbDatabase::open_u32(&self.db_name, DB_VERSION) + .map_err(|e| js_err("indexeddb open", e))?; + req.set_on_upgrade_needed(Some( + |evt: &IdbVersionChangeEvent| -> Result<(), JsValue> { + let db = evt.db(); + if !db.object_store_names().any(|n| n == MESSAGES_STORE) { + db.create_object_store(MESSAGES_STORE)?; + } + if !db.object_store_names().any(|n| n == SETTINGS_STORE) { + db.create_object_store(SETTINGS_STORE)?; + } + Ok(()) + }, + )); + req.await.map_err(|e| js_err("indexeddb open await", e)) + } + + /// Write one string value under a string key in `store_name`. + async fn put_string(&self, store_name: &str, key: &str, value: &str) -> Result<()> { + let db = self.open_db().await?; + let tx = db + .transaction_on_one_with_mode(store_name, IdbTransactionMode::Readwrite) + .map_err(|e| js_err("tx open", e))?; + let store = tx + .object_store(store_name) + .map_err(|e| js_err("store open", e))?; + store + .put_key_val_owned(key, &JsValue::from_str(value)) + .map_err(|e| js_err("put", e))?; + tx.await.into_result().map_err(|e| js_err("tx commit", e))?; + Ok(()) + } + + /// Read the string value under `key` in `store_name`, if present. + async fn get_string(&self, store_name: &str, key: &str) -> Result> { + let db = self.open_db().await?; + let tx = db + .transaction_on_one_with_mode(store_name, IdbTransactionMode::Readonly) + .map_err(|e| js_err("tx open", e))?; + let store = tx + .object_store(store_name) + .map_err(|e| js_err("store open", e))?; + let value = store + .get_owned(key) + .map_err(|e| js_err("get", e))? + .await + .map_err(|e| js_err("get await", e))?; + Ok(value.and_then(|v| v.as_string())) + } + + /// All string values stored in `store_name`. + async fn get_all_strings(&self, store_name: &str) -> Result> { + let db = self.open_db().await?; + let tx = db + .transaction_on_one_with_mode(store_name, IdbTransactionMode::Readonly) + .map_err(|e| js_err("tx open", e))?; + let store = tx + .object_store(store_name) + .map_err(|e| js_err("store open", e))?; + let array = store + .get_all() + .map_err(|e| js_err("get_all", e))? + .await + .map_err(|e| js_err("get_all await", e))?; + Ok(array.iter().filter_map(|v| v.as_string()).collect()) } } @@ -38,23 +141,50 @@ impl Storage for IndexedDbStorage { Err(anyhow!("IndexedDB not yet implemented")) } async fn list_trades(&self) -> Result> { - Err(anyhow!("IndexedDB not yet implemented")) + // Consumed by `resubscribe_active_chats` at startup: no persisted + // trades yet on web (tracked in #233), so nothing to resubscribe — + // an empty list, not an error. + Ok(Vec::new()) } - async fn save_message(&self, _msg: &ChatMessage) -> Result<()> { - Err(anyhow!("IndexedDB not yet implemented")) + + // ── Chat messages — fully implemented (durable replay dedup, #246) ────── + + async fn save_message(&self, msg: &ChatMessage) -> Result<()> { + let json = serde_json::to_string(msg)?; + self.put_string(MESSAGES_STORE, &msg.id, &json).await } - async fn list_messages(&self, _trade_id: &str) -> Result> { - Err(anyhow!("IndexedDB not yet implemented")) + + async fn list_messages(&self, trade_id: &str) -> Result> { + let mut msgs: Vec = self + .get_all_strings(MESSAGES_STORE) + .await? + .iter() + .filter_map(|json| serde_json::from_str::(json).ok()) + .filter(|m| m.trade_id == trade_id) + .collect(); + msgs.sort_by_key(|m| m.created_at); + Ok(msgs) } - async fn mark_messages_read(&self, _trade_id: &str) -> Result<()> { - Err(anyhow!("IndexedDB not yet implemented")) + + async fn mark_messages_read(&self, trade_id: &str) -> Result<()> { + let unread: Vec = self + .list_messages(trade_id) + .await? + .into_iter() + .filter(|m| !m.is_read) + .collect(); + for mut msg in unread { + msg.is_read = true; + self.save_message(&msg).await?; + } + Ok(()) } - async fn message_exists(&self, _id: &str) -> Result { - // Stub contract (#233): answer "not stored" so the caller falls back - // to its in-memory dedup. On web, replay dedup is process-lifetime - // only until IndexedDB lands. - Ok(false) + + async fn message_exists(&self, id: &str) -> Result { + // Fail closed: an `Err` here makes the chat pipeline DROP the event. + Ok(self.get_string(MESSAGES_STORE, id).await?.is_some()) } + async fn save_relay(&self, _relay: &RelayInfo) -> Result<()> { Err(anyhow!("IndexedDB not yet implemented")) } @@ -91,7 +221,7 @@ impl Storage for IndexedDbStorage { } async fn save_trade_key(&self, _order_id: &str, _key_index: u32) -> Result<()> { - Ok(()) // no-op: IndexedDB persistence not yet implemented + Ok(()) // no-op: IndexedDB persistence not yet implemented (#233) } async fn get_trade_key(&self, _order_id: &str) -> Result> { @@ -99,56 +229,50 @@ impl Storage for IndexedDbStorage { } async fn get_order_id_by_trade_index(&self, _key_index: u32) -> Result> { - Ok(None) // IndexedDB not yet implemented + Ok(None) // IndexedDB not yet implemented (#233) } async fn delete_trade_key(&self, _order_id: &str) -> Result<()> { - Ok(()) // IndexedDB not yet implemented + Ok(()) // IndexedDB not yet implemented (#233) } async fn clear_trade_keys(&self) -> Result<()> { - Ok(()) // IndexedDB not yet implemented + Ok(()) // IndexedDB not yet implemented (#233) } - // Settings k/v: same stub contract as the node pubkey below — writes are - // dropped and reads answer "nothing stored", so callers fall back to their - // defaults instead of failing. Tracked in #233. + // ── Settings KV — fully implemented (chat cursor + preferences, #246) ─── - async fn get_setting(&self, _key: &str) -> Result> { - log::warn!("get_setting: IndexedDB backend not implemented — falling back to default"); - Ok(None) + async fn get_setting(&self, key: &str) -> Result> { + self.get_string(SETTINGS_STORE, key).await } - async fn set_setting(&self, _key: &str, _value: &str) -> Result<()> { - log::warn!("set_setting: IndexedDB backend not implemented — preference will not survive reload"); - Ok(()) + async fn set_setting(&self, key: &str, value: &str) -> Result<()> { + self.put_string(SETTINGS_STORE, key, value).await } - async fn delete_setting(&self, _key: &str) -> Result<()> { - // Nothing was ever stored, so removal is already satisfied — but say so - // for the same reason the writes do: a silent no-op in a storage layer - // is indistinguishable from working storage when reading a log. - log::warn!("delete_setting: IndexedDB backend not implemented — nothing was stored to remove"); + async fn delete_setting(&self, key: &str) -> Result<()> { + let db = self.open_db().await?; + let tx = db + .transaction_on_one_with_mode(SETTINGS_STORE, IdbTransactionMode::Readwrite) + .map_err(|e| js_err("tx open", e))?; + let store = tx + .object_store(SETTINGS_STORE) + .map_err(|e| js_err("store open", e))?; + store.delete_owned(key).map_err(|e| js_err("delete", e))?; + tx.await.into_result().map_err(|e| js_err("tx commit", e))?; Ok(()) } - async fn save_active_mostro_pubkey(&self, _pubkey: &str) -> Result<()> { - // IndexedDB not yet implemented — node selection will not survive reload. - log::warn!("save_active_mostro_pubkey: IndexedDB backend not implemented — node selection will not survive reload"); - Ok(()) + async fn save_active_mostro_pubkey(&self, pubkey: &str) -> Result<()> { + self.set_setting(settings_keys_active_pubkey(), pubkey).await } async fn get_active_mostro_pubkey(&self) -> Result> { - // IndexedDB not yet implemented — callers fall back to the default. - log::warn!("get_active_mostro_pubkey: IndexedDB backend not implemented — falling back to default"); - Ok(None) + self.get_setting(settings_keys_active_pubkey()).await } - async fn get_trade_by_order_id( - &self, - _order_id: &str, - ) -> Result> { - Ok(None) // no persisted trade: role lookup returns None + async fn get_trade_by_order_id(&self, _order_id: &str) -> Result> { + Ok(None) // no persisted trade: role lookup returns None (#233) } async fn update_trade_order_id( @@ -171,3 +295,8 @@ impl Storage for IndexedDbStorage { Ok(()) } } + +/// The active-node settings key (shared with the SQLite backend). +fn settings_keys_active_pubkey() -> &'static str { + crate::db::settings_keys::ACTIVE_MOSTRO_PUBKEY +} diff --git a/rust/src/db/schema.rs b/rust/src/db/schema.rs index 53a2ee77..7eb5a442 100644 --- a/rust/src/db/schema.rs +++ b/rust/src/db/schema.rs @@ -1,6 +1,27 @@ /// Database schema version. Currently unused at runtime — kept as a reference /// for future migration logic (e.g. ALTER TABLE guards or schema-diff checks). -pub const SCHEMA_VERSION: u32 = 2; +pub const SCHEMA_VERSION: u32 = 3; + +/// One-off rebuild for databases created while `messages` still carried a +/// foreign key to `trades(id)` (schema v2). SQLite cannot drop a FK in place, +/// so the table is recreated and the rows copied. Runs after the main DDL; +/// `SqliteStorage::open` executes it only when the old FK is detected. +#[cfg(not(target_arch = "wasm32"))] +pub const SQLITE_DROP_MESSAGES_FK_SQL: &str = r#" +PRAGMA foreign_keys = OFF; +CREATE TABLE messages_v3 ( + id TEXT PRIMARY KEY, + trade_id TEXT NOT NULL, + data TEXT NOT NULL, + is_read INTEGER NOT NULL DEFAULT 0, + created_at INTEGER NOT NULL +); +INSERT INTO messages_v3 SELECT id, trade_id, data, is_read, created_at FROM messages; +DROP TABLE messages; +ALTER TABLE messages_v3 RENAME TO messages; +CREATE INDEX IF NOT EXISTS idx_messages_trade ON messages(trade_id); +PRAGMA foreign_keys = ON; +"#; /// SQLite DDL executed unconditionally on every `SqliteStorage::open()` call. /// Safe to run repeatedly because every statement uses `CREATE TABLE IF NOT @@ -27,13 +48,17 @@ CREATE TABLE IF NOT EXISTS trades ( completed_at INTEGER ); +-- Chat history + durable replay dedup (issue #246). `trade_id` here is the +-- **order id** — the identity chat keys are derived from — which for taken +-- orders differs from the `trades.id` UUID, so deliberately NO foreign key +-- to trades(id): with one, every taker's save_message failed its FK check +-- and history/dedup silently vanished on restart. CREATE TABLE IF NOT EXISTS messages ( id TEXT PRIMARY KEY, trade_id TEXT NOT NULL, data TEXT NOT NULL, -- JSON-serialised ChatMessage is_read INTEGER NOT NULL DEFAULT 0, - created_at INTEGER NOT NULL, - FOREIGN KEY (trade_id) REFERENCES trades(id) + created_at INTEGER NOT NULL ); CREATE INDEX IF NOT EXISTS idx_messages_trade ON messages(trade_id); diff --git a/rust/src/db/sqlite.rs b/rust/src/db/sqlite.rs index 819806c9..218aa1be 100644 --- a/rust/src/db/sqlite.rs +++ b/rust/src/db/sqlite.rs @@ -86,6 +86,24 @@ impl SqliteStorage { } } + // Migration 2 → 3: drop the messages → trades foreign key. Chat keys + // (and therefore `messages.trade_id`) are per **order id**, while a + // taker's trades row uses a fresh UUID — with the FK in place every + // taker `save_message` failed and chat history/replay-dedup was lost + // on restart (PR #247 review). Rows are preserved. + let messages_has_fk: bool = sqlx::query_scalar( + "SELECT COUNT(*) > 0 FROM pragma_foreign_key_list('messages')", + ) + .fetch_one(pool) + .await + .unwrap_or(false); + if messages_has_fk { + log::warn!("[db] migrating messages table from schema v2 to v3 (dropping FK)"); + sqlx::query(crate::db::schema::SQLITE_DROP_MESSAGES_FK_SQL) + .execute(pool) + .await?; + } + Ok(()) } } @@ -218,8 +236,16 @@ impl Storage for SqliteStorage { } async fn mark_messages_read(&self, trade_id: &str) -> Result<()> { + // `list_messages` reconstructs ChatMessage from the JSON `data` blob, + // so the flag must be rewritten there too — updating only the + // denormalized column resurrects unread badges after a restart. + // `json('true')` keeps the field a JSON boolean (json_set with a bare + // 1 would turn it into a number and break deserialization). sqlx::query( - "UPDATE messages SET is_read = 1 WHERE trade_id = ? AND is_read = 0", + "UPDATE messages + SET is_read = 1, + data = json_set(data, '$.is_read', json('true')) + WHERE trade_id = ? AND is_read = 0", ) .bind(trade_id) .execute(&self.pool) @@ -545,43 +571,10 @@ mod tests { let path = temp_db_path(); let storage = SqliteStorage::open(path.to_str().unwrap()).await.unwrap(); - // messages.trade_id has a FK to trades(id) — store the trade first, - // exactly as the take-order flow does before chat ever runs. - let trade_id = "trade-dedup-1".to_string(); - let trade = TradeInfo { - id: trade_id.clone(), - order: OrderInfo { - id: "order-dedup-1".into(), - kind: OrderKind::Sell, - status: OrderStatus::Active, - amount_sats: Some(1000), - fiat_amount: Some(10.0), - fiat_amount_min: None, - fiat_amount_max: None, - fiat_code: "VES".into(), - payment_method: "bank".into(), - premium: 0.0, - creator_pubkey: "maker".into(), - created_at: 1, - expires_at: None, - is_mine: false, - rating: 0.0, - total_reviews: 0, - days_active: 0, - }, - role: TradeRole::Buyer, - counterparty_pubkey: "peer".into(), - current_step: TradeStep::Buyer(BuyerStep::FiatSent), - hold_invoice: None, - buyer_invoice: None, - trade_key_index: 1, - cooperative_cancel_state: None, - timeout_at: None, - started_at: 1, - completed_at: None, - outcome: None, - }; - storage.save_trade(&trade).await.unwrap(); + // Chat persists under the ORDER id — for takers there is no trades + // row with that id (trades.id is a fresh UUID), so this must succeed + // without any trades row at all (the old FK broke exactly this). + let trade_id = "order-dedup-1".to_string(); let inner_id = "3f".repeat(32); assert!(!storage.message_exists(&inner_id).await.unwrap()); @@ -608,6 +601,84 @@ mod tests { let _ = std::fs::remove_file(&path); } + #[tokio::test] + async fn mark_messages_read_survives_rehydration() { + use crate::api::types::*; + + let path = temp_db_path(); + let storage = SqliteStorage::open(path.to_str().unwrap()).await.unwrap(); + + let trade_id = "order-read-1".to_string(); + storage + .save_message(&ChatMessage { + id: "read-msg-1".into(), + trade_id: trade_id.clone(), + sender_pubkey: "peer".into(), + content: "hola".into(), + message_type: MessageType::Peer, + is_mine: false, + is_read: false, + has_attachment: false, + attachment: None, + created_at: 1, + }) + .await + .unwrap(); + + storage.mark_messages_read(&trade_id).await.unwrap(); + + // Reopen: list_messages deserializes the JSON blob — the read flag + // must have been rewritten there, not only in the column. + drop(storage); + let storage = SqliteStorage::open(path.to_str().unwrap()).await.unwrap(); + let msgs = storage.list_messages(&trade_id).await.unwrap(); + assert_eq!(msgs.len(), 1); + assert!(msgs[0].is_read, "is_read lost on rehydration"); + + drop(storage); + let _ = std::fs::remove_file(&path); + } + + #[tokio::test] + async fn v2_messages_fk_is_dropped_and_rows_survive() { + let path = temp_db_path(); + let url = format!("sqlite://{}?mode=rwc", path.to_str().unwrap()); + + // Build a v2-era database by hand: messages with the old FK and one + // row referencing a trades row (the maker case, which used to work). + { + let pool = SqlitePoolOptions::new().connect(&url).await.unwrap(); + sqlx::query( + "CREATE TABLE trades ( + id TEXT PRIMARY KEY, data TEXT NOT NULL, status TEXT NOT NULL, + started_at INTEGER NOT NULL, completed_at INTEGER); + CREATE TABLE messages ( + id TEXT PRIMARY KEY, trade_id TEXT NOT NULL, data TEXT NOT NULL, + is_read INTEGER NOT NULL DEFAULT 0, created_at INTEGER NOT NULL, + FOREIGN KEY (trade_id) REFERENCES trades(id)); + INSERT INTO trades VALUES ('t1', '{}', 'Active', 1, NULL); + INSERT INTO messages VALUES ('m1', 't1', + '{\"id\":\"m1\",\"trade_id\":\"t1\",\"sender_pubkey\":\"p\",\"content\":\"x\",\"message_type\":\"Peer\",\"is_mine\":false,\"is_read\":false,\"has_attachment\":false,\"attachment\":null,\"created_at\":1}', + 0, 1);", + ) + .execute(&pool) + .await + .unwrap(); + pool.close().await; + } + + // open() must detect the FK, rebuild the table, and keep the row. + let storage = SqliteStorage::open(path.to_str().unwrap()).await.unwrap(); + assert!(storage.message_exists("m1").await.unwrap()); + + // And an order-id message with no trades row now persists fine. + let msgs = storage.list_messages("t1").await.unwrap(); + assert_eq!(msgs.len(), 1); + + drop(storage); + let _ = std::fs::remove_file(&path); + } + #[tokio::test] async fn active_mostro_pubkey_round_trip() { let path = temp_db_path(); diff --git a/rust/src/nostr/gift_wrap.rs b/rust/src/nostr/gift_wrap.rs index eb82827e..51b1da74 100644 --- a/rust/src/nostr/gift_wrap.rs +++ b/rust/src/nostr/gift_wrap.rs @@ -106,9 +106,17 @@ pub async fn unwrap_mostro_message( /// `created_at` and against the recipient's own clock. pub const MAX_CLOCK_SKEW_SECS: u64 = 60; -/// Upper bound on the encrypted payload, enforced before decrypting. +/// Upper bound on the encrypted payload, enforced on receive before +/// decrypting and on send before publishing (`MessageTooLarge`). pub const MAX_CONTENT_BYTES: usize = 64 * 1024; +/// Upper bound on the outer event's tag count, enforced before signature +/// verification. The envelope defines exactly one `p` tag (plus optional +/// NIP-13 nonce); anything past this is a peer padding the event with junk +/// tags to inflate pre-decryption work — the ciphertext cap alone does not +/// bound the raw event a relay may deliver. +pub const MAX_OUTER_TAGS: usize = 8; + /// Build the outer kind 14 event carrying an encrypted, trade-key-signed /// kind 1 event, per the P2P chat spec. /// @@ -137,7 +145,18 @@ pub async fn mostro_wrap( // timestamp tweaking — it would break `since`-based sync. let now = Timestamp::now(); + // Signed uniqueness nonce: the inner id is a hash over pubkey, kind, + // created_at, tags and content — with second-resolution timestamps two + // intentional identical sends ("yes", "yes") in the same second would + // otherwise collapse to one id and the receiver's replay dedup would + // silently drop the second one. + let nonce: [u8; 8] = rand::random(); + let inner = EventBuilder::text_note(message) + .tag(Tag::custom( + TagKind::custom("u"), + [hex::encode(nonce)], + )) .custom_created_at(now) .build(sender_trade.public_key()) .sign(sender_trade) @@ -153,6 +172,18 @@ pub async fn mostro_wrap( ) .map_err(|e| anyhow!("NIP-44 encrypt failed: {e}"))?; + // Reject before publishing what every receiver running this protocol + // must discard before decrypting — otherwise the sender stores the + // message as "sent" while the counterparty never sees it. Stable marker: + // Dart maps `MessageTooLarge` to a localized error. + if content.len() > MAX_CONTENT_BYTES { + return Err(anyhow!( + "MessageTooLarge: encrypted payload is {} bytes, limit {}", + content.len(), + MAX_CONTENT_BYTES + )); + } + // Exactly one `p` tag, ours. Anything else could hide the message from // the `#p` query a dispute solver uses to rebuild the transcript. let builder = EventBuilder::new(Kind::PrivateDirectMessage, content) @@ -204,6 +235,18 @@ pub fn mostro_unwrap( return Err(anyhow!("outer event is not kind 14")); } + // Bound the raw event before any signature work: the ciphertext cap + // below does not stop a peer from shipping a small ciphertext inside a + // multi-megabyte event padded with thousands of junk tags, forcing + // hashing and verification cost per event. + if outer.tags.len() > MAX_OUTER_TAGS { + return Err(anyhow!( + "outer event carries {} tags, limit {}", + outer.tags.len(), + MAX_OUTER_TAGS + )); + } + // Exactly one `p` tag, addressing this conversation. Anything else could // be a message engineered to stay out of a dispute solver's `#p` query. let mut p_tags = outer @@ -579,6 +622,77 @@ mod chat_envelope_tests { assert!(err.contains("stale"), "got: {err}"); } + #[tokio::test] + async fn identical_same_second_sends_keep_distinct_identities() { + let c = convo(); + // Two rapid "yes" messages: without the signed nonce they would share + // one inner id and the receiver's dedup would drop the second. + let (o1, i1) = mostro_wrap(&c.alice_trade, &c.conv, &c.sign, "yes") + .await + .unwrap(); + let (o2, i2) = mostro_wrap(&c.alice_trade, &c.conv, &c.sign, "yes") + .await + .unwrap(); + + assert_ne!(i1.id, i2.id, "identical sends collapsed to one inner id"); + assert!(unwrap_now(&c, &o1).is_ok()); + assert!(unwrap_now(&c, &o2).is_ok()); + } + + #[tokio::test] + async fn oversized_message_is_refused_at_send_time() { + let c = convo(); + // Large enough that the NIP-44 ciphertext exceeds what any receiver + // accepts — must fail with the stable marker instead of publishing. + let big = "x".repeat(60 * 1024); + let err = mostro_wrap(&c.alice_trade, &c.conv, &c.sign, &big) + .await + .unwrap_err() + .to_string(); + assert!(err.contains("MessageTooLarge"), "got: {err}"); + + // A comfortably large message still round-trips (largest-accepted + // boundary is fuzzy by design: NIP-44 padding + JSON escaping). + let ok = "y".repeat(30 * 1024); + let (outer, _) = mostro_wrap(&c.alice_trade, &c.conv, &c.sign, &ok) + .await + .unwrap(); + let inner = unwrap_now(&c, &outer).unwrap(); + assert_eq!(inner.content, ok); + } + + #[tokio::test] + async fn junk_tag_padding_is_rejected_before_verification() { + let c = convo(); + let (_, inner) = mostro_wrap(&c.alice_trade, &c.conv, &c.sign, "hola") + .await + .unwrap(); + let content = nip44::encrypt( + c.conv.secret_key(), + &c.conv.public_key(), + inner.as_json(), + nip44::Version::V2, + ) + .unwrap(); + + // Small ciphertext, huge tag list — the raw-event bound must trip. + let mut builder = EventBuilder::new(Kind::PrivateDirectMessage, content) + .tag(Tag::public_key(c.conv.public_key())); + for i in 0..2000 { + builder = builder.tag(Tag::custom( + TagKind::custom("x"), + [format!("junk-{i}")], + )); + } + let padded = builder + .custom_created_at(inner.created_at) + .sign_with_keys(&c.sign) + .unwrap(); + + let err = unwrap_now(&c, &padded).unwrap_err().to_string(); + assert!(err.contains("tags"), "got: {err}"); + } + #[tokio::test] async fn conv_key_alone_cannot_author_into_the_conversation() { // Dispute disclosure grant: K_conv decrypts, but an outer event signed diff --git a/specs/004-mostro-p2p-client/contracts/messages.md b/specs/004-mostro-p2p-client/contracts/messages.md index 655a31d2..5afacf7a 100644 --- a/specs/004-mostro-p2p-client/contracts/messages.md +++ b/specs/004-mostro-p2p-client/contracts/messages.md @@ -7,9 +7,12 @@ of the protocol spec (, issue #246): a kind 14 outer event signed with `K_sign` and `p`-tagged to `pub(K_conv)` — both HKDF-SHA256 derivations of the trade-key ECDH secret — carrying a NIP-44 encrypted kind 1 inner event signed by the sender's trade -key. NIP-59 gift wrap (kind 1059) is no longer used for peer chat (its -random ephemeral authors made third-party flooding unattributable); -admin/dispute chat still uses it. Messages persist locally after +key. NIP-59 gift wrap (kind 1059) is no longer *written* for peer chat (its +random ephemeral authors made third-party flooding unattributable); it is +still *read* from pre-migration peers until the dual-read deadline +(`LEGACY_CHAT_DEPRECATION_TS`, 2026-12-31T00:00:00Z), bounded by the same +LRU / rate budget / size cap / durable dedup / quota as the new envelope. +Admin/dispute chat still uses gift wrap. Messages persist locally after validation. Supports encrypted file attachments via Blossom servers. **Security requirements implemented** (see the protocol spec for the @@ -23,7 +26,22 @@ normative list): and halts chat processing while the trade stays operational). - Inner signature verified and its author checked against the two trade keys of the order — the only sender authentication. -- Durable replay dedup on the inner event id (`messages` table). +- Durable replay dedup on the inner event id (`messages` table on native, + IndexedDB on web), **fail-closed**: a dedup lookup error drops the event. +- The rate budget meters only the live stream (post-EOSE); stored catch-up + is bounded by the filter `limit` instead, so history above the burst size + is never dropped. +- The cursor advances only past durably stored messages. +- Per-trade retention quotas (message count and total bytes) bound durable + growth even at a legitimate send rate. +- Send-side size validation: a message whose encrypted envelope no receiver + would accept fails with a stable `MessageTooLarge` error; each inner event + carries a signed uniqueness nonce so identical same-second sends keep + distinct ids. +- Subscription lifecycle: one task per order (spawn guard), explicit + subscription ids unsubscribed on every exit, no idle timeout, and + automatic resubscription of persisted active trades when the relay pool + comes online. - Isolation: chat runs on its own task and bounded channels; it can never block the order state machine, the daemon transport, or a dispute. From 67f03fe618cde95d78b0cc946c982c319268afce Mon Sep 17 00:00:00 2001 From: grunch Date: Wed, 29 Jul 2026 21:08:33 -0300 Subject: [PATCH 3/4] =?UTF-8?q?fix(chat):=20review=20round=202=20on=20#247?= =?UTF-8?q?=20=E2=80=94=20legacy=20flood=20accounting,=20crash-safe=20migr?= =?UTF-8?q?ation?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - handle_legacy_chat_event: the rejected-streak reset moved after the allowed-signers check — anyone can wrap to the public legacy address, so unauthorized/malformed senders now count toward the flood breaker instead of resetting it. A rumor without an id is rejected outright: a fabricated fallback id would make the same rumor accepted again on every replay. - messages FK migration: rebuild wrapped in a transaction, stray messages_v3 from an interrupted attempt dropped first, foreign_keys pragma kept outside the transaction (SQLite ignores it inside one). Migration test now seeds a leftover messages_v3. - FRB bindings verified unchanged (subscribe_incoming_chat is pub(crate), not bridge surface): ./scripts/frb-generate.sh --check clean. --- rust/src/api/messages.rs | 16 +++++++++++++--- rust/src/db/schema.rs | 8 ++++++++ rust/src/db/sqlite.rs | 3 +++ 3 files changed, 24 insertions(+), 3 deletions(-) diff --git a/rust/src/api/messages.rs b/rust/src/api/messages.rs index 15806c97..34f73f8c 100644 --- a/rust/src/api/messages.rs +++ b/rust/src/api/messages.rs @@ -1346,23 +1346,33 @@ async fn handle_legacy_chat_event( return; } }; - state.consecutive_rejected = 0; // The rumor is unsigned (NIP-59), so the claimed sender is only checked // for membership: this is exactly the weakness the new envelope fixes. + // Anyone can wrap to the public shared address, so an unauthorized or + // malformed sender still counts toward the flood breaker — resetting the + // streak before this check would let a flooder keep it at zero forever. let sender_hex = rumor.get("pubkey").and_then(|v| v.as_str()).unwrap_or(""); let Ok(sender) = nostr_sdk::PublicKey::from_hex(sender_hex) else { + state.reject(order_id); return; }; if !allowed_signers.contains(&sender) { + state.reject(order_id); return; } + state.consecutive_rejected = 0; - let rumor_id = rumor + // No id, no dedup identity: a fabricated fallback id would make the same + // rumor accepted again on every replay, so it is rejected outright. + let Some(rumor_id) = rumor .get("id") .and_then(|v| v.as_str()) .map(|s| s.to_string()) - .unwrap_or_else(|| uuid::Uuid::new_v4().to_string()); + else { + log::debug!("[messages] legacy rumor without id order={order_id} — dropped"); + return; + }; match message_store().is_known(order_id, &rumor_id).await { Err(e) => { log::warn!("[messages] {e} — dropping legacy event order={order_id}"); diff --git a/rust/src/db/schema.rs b/rust/src/db/schema.rs index 7eb5a442..87973acd 100644 --- a/rust/src/db/schema.rs +++ b/rust/src/db/schema.rs @@ -6,9 +6,16 @@ pub const SCHEMA_VERSION: u32 = 3; /// foreign key to `trades(id)` (schema v2). SQLite cannot drop a FK in place, /// so the table is recreated and the rows copied. Runs after the main DDL; /// `SqliteStorage::open` executes it only when the old FK is detected. +/// Crash-safe: the rebuild runs inside one transaction (an interruption +/// rolls back to the untouched v2 table), the stray `messages_v3` a previous +/// interrupted attempt may have left is dropped first, and the +/// `foreign_keys` pragma toggles sit OUTSIDE the transaction — SQLite +/// silently ignores that pragma inside one. #[cfg(not(target_arch = "wasm32"))] pub const SQLITE_DROP_MESSAGES_FK_SQL: &str = r#" PRAGMA foreign_keys = OFF; +BEGIN; +DROP TABLE IF EXISTS messages_v3; CREATE TABLE messages_v3 ( id TEXT PRIMARY KEY, trade_id TEXT NOT NULL, @@ -20,6 +27,7 @@ INSERT INTO messages_v3 SELECT id, trade_id, data, is_read, created_at FROM mess DROP TABLE messages; ALTER TABLE messages_v3 RENAME TO messages; CREATE INDEX IF NOT EXISTS idx_messages_trade ON messages(trade_id); +COMMIT; PRAGMA foreign_keys = ON; "#; diff --git a/rust/src/db/sqlite.rs b/rust/src/db/sqlite.rs index 218aa1be..02fa6056 100644 --- a/rust/src/db/sqlite.rs +++ b/rust/src/db/sqlite.rs @@ -656,6 +656,9 @@ mod tests { id TEXT PRIMARY KEY, trade_id TEXT NOT NULL, data TEXT NOT NULL, is_read INTEGER NOT NULL DEFAULT 0, created_at INTEGER NOT NULL, FOREIGN KEY (trade_id) REFERENCES trades(id)); + -- Leftover from a previous interrupted migration attempt: + -- the rebuild must drop and recreate it, not fail. + CREATE TABLE messages_v3 (leftover INTEGER); INSERT INTO trades VALUES ('t1', '{}', 'Active', 1, NULL); INSERT INTO messages VALUES ('m1', 't1', '{\"id\":\"m1\",\"trade_id\":\"t1\",\"sender_pubkey\":\"p\",\"content\":\"x\",\"message_type\":\"Peer\",\"is_mine\":false,\"is_read\":false,\"has_attachment\":false,\"attachment\":null,\"created_at\":1}', From 612d36b2b15a116331382207fde5bf190137324d Mon Sep 17 00:00:00 2001 From: grunch Date: Wed, 29 Jul 2026 21:42:30 -0300 Subject: [PATCH 4/4] =?UTF-8?q?fix(db):=20don't=20run=20the=20v2=E2=86=92v?= =?UTF-8?q?3=20messages=20rebuild=20on=20a=20v1=20schema?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit A database created before `messages` moved to a JSON `data` blob stores one column per field (`sender_pubkey`, `content_encrypted`, …) and also carries the FK to `trades(id)`. The FK is the only thing `migrate()` checked, so the v2→v3 rebuild fired and its `INSERT INTO messages_v3 SELECT id, trade_id, data, … FROM messages` failed with "no such column: data". That error propagates out of `SqliteStorage::open()`, so `initDb` failed entirely and the app fell back to memory-only mode — losing orders, trades, identity and the outbox on every launch, not just chat history. Detect the v1 table (present, no `data` column) and drop it: the `content_encrypted` rows are ciphertext the current chat code cannot read, so there is nothing to convert. The v2→v3 rebuild is now additionally gated on `data` existing, so it can never run against a schema without it. Regression test builds the v1 table by hand and asserts `open()` succeeds, the rebuilt table is v3 (JSON `data`, no FK), and `save_message` works. --- rust/src/db/schema.rs | 6 ++- rust/src/db/sqlite.rs | 118 +++++++++++++++++++++++++++++++++++++++++- 2 files changed, 122 insertions(+), 2 deletions(-) diff --git a/rust/src/db/schema.rs b/rust/src/db/schema.rs index 87973acd..daa0ce72 100644 --- a/rust/src/db/schema.rs +++ b/rust/src/db/schema.rs @@ -5,7 +5,11 @@ pub const SCHEMA_VERSION: u32 = 3; /// One-off rebuild for databases created while `messages` still carried a /// foreign key to `trades(id)` (schema v2). SQLite cannot drop a FK in place, /// so the table is recreated and the rows copied. Runs after the main DDL; -/// `SqliteStorage::open` executes it only when the old FK is detected. +/// `SqliteStorage::open` executes it only when the old FK is detected **and** +/// the table already has the JSON `data` column this SQL copies — a v1 +/// database (one column per field) also carries that FK, and running this +/// against it aborts `open()` with "no such column: data", taking the whole +/// database down. `migrate()` drops that older table instead. /// Crash-safe: the rebuild runs inside one transaction (an interruption /// rolls back to the untouched v2 table), the stray `messages_v3` a previous /// interrupted attempt may have left is dropped first, and the diff --git a/rust/src/db/sqlite.rs b/rust/src/db/sqlite.rs index 02fa6056..7ae59ad1 100644 --- a/rust/src/db/sqlite.rs +++ b/rust/src/db/sqlite.rs @@ -86,18 +86,52 @@ impl SqliteStorage { } } + // Migration 1 → 3: the original `messages` table stored one column per + // field (`sender_pubkey`, `content_encrypted`, …) instead of the JSON + // `data` blob. It also carries the FK, so without this the v2 → v3 + // rebuild below fires and its `SELECT … data … FROM messages` aborts + // `open()` with "no such column: data" — killing the ENTIRE database + // (orders, trades, identity, outbox), not just chat. + // + // The rows are dropped rather than converted: `content_encrypted` holds + // ciphertext the current chat code cannot read back, so there is nothing + // to recover. Dropping the table takes its legacy indexes with it. + let messages_exists: bool = sqlx::query_scalar( + "SELECT COUNT(*) > 0 FROM sqlite_master WHERE type = 'table' AND name = 'messages'", + ) + .fetch_one(pool) + .await + .unwrap_or(false); + let messages_has_data: bool = sqlx::query_scalar( + "SELECT COUNT(*) > 0 FROM pragma_table_info('messages') WHERE name = 'data'", + ) + .fetch_one(pool) + .await + .unwrap_or(false); + if messages_exists && !messages_has_data { + log::warn!( + "[db] migrating messages table from schema v1 (dropping unreadable rows)" + ); + sqlx::query("DROP TABLE IF EXISTS messages") + .execute(pool) + .await?; + } + // Migration 2 → 3: drop the messages → trades foreign key. Chat keys // (and therefore `messages.trade_id`) are per **order id**, while a // taker's trades row uses a fresh UUID — with the FK in place every // taker `save_message` failed and chat history/replay-dedup was lost // on restart (PR #247 review). Rows are preserved. + // + // Gated on `data` as well: the rebuild copies that column, so it must + // never run against a schema that lacks it (the v1 case handled above). let messages_has_fk: bool = sqlx::query_scalar( "SELECT COUNT(*) > 0 FROM pragma_foreign_key_list('messages')", ) .fetch_one(pool) .await .unwrap_or(false); - if messages_has_fk { + if messages_has_fk && messages_has_data { log::warn!("[db] migrating messages table from schema v2 to v3 (dropping FK)"); sqlx::query(crate::db::schema::SQLITE_DROP_MESSAGES_FK_SQL) .execute(pool) @@ -682,6 +716,88 @@ mod tests { let _ = std::fs::remove_file(&path); } + #[tokio::test] + async fn pre_v2_messages_table_is_rebuilt_not_copied() { + let path = temp_db_path(); + let url = format!("sqlite://{}?mode=rwc", path.to_str().unwrap()); + + // Build a v1-era database by hand: `messages` still stores one column + // per field (no JSON `data` blob) and carries the FK to trades. The + // v2 → v3 rebuild copies `data`, so triggering it here used to abort + // `open()` with "no such column: data" — taking the WHOLE database + // down, not just chat (orders, trades, identity, outbox). + { + let pool = SqlitePoolOptions::new().connect(&url).await.unwrap(); + sqlx::query( + "CREATE TABLE trades ( + id TEXT PRIMARY KEY, data TEXT NOT NULL, status TEXT NOT NULL, + started_at INTEGER NOT NULL, completed_at INTEGER); + CREATE TABLE messages ( + id TEXT NOT NULL PRIMARY KEY, + trade_id TEXT NOT NULL REFERENCES trades(id), + sender_pubkey TEXT NOT NULL, + content_encrypted BLOB NOT NULL, + message_type TEXT NOT NULL, + is_mine INTEGER NOT NULL DEFAULT 0, + is_read INTEGER NOT NULL DEFAULT 0, + attachment_id TEXT, + created_at INTEGER NOT NULL); + CREATE INDEX idx_messages_trade_id ON messages(trade_id); + CREATE INDEX idx_messages_is_read ON messages(is_read); + INSERT INTO trades VALUES ('t1', '{}', 'Active', 1, NULL); + INSERT INTO messages VALUES + ('m0', 't1', 'p', x'00', 'Peer', 0, 0, NULL, 1);", + ) + .execute(&pool) + .await + .unwrap(); + pool.close().await; + } + + // open() must succeed — the legacy table is dropped, not copied. + let storage = SqliteStorage::open(path.to_str().unwrap()).await.unwrap(); + + // The rebuilt table is v3: JSON `data`, no foreign key. + let cols: Vec<(String,)> = + sqlx::query_as("SELECT name FROM pragma_table_info('messages')") + .fetch_all(&storage.pool) + .await + .unwrap(); + let cols: Vec = cols.into_iter().map(|(c,)| c).collect(); + assert!(cols.contains(&"data".to_string()), "columns: {cols:?}"); + assert!( + !cols.contains(&"content_encrypted".to_string()), + "legacy column survived: {cols:?}" + ); + let fks: i64 = + sqlx::query_scalar("SELECT COUNT(*) FROM pragma_foreign_key_list('messages')") + .fetch_one(&storage.pool) + .await + .unwrap(); + assert_eq!(fks, 0, "messages still has a foreign key"); + + // And it is usable: an order-id message with no matching trades row. + storage + .save_message(&crate::api::types::ChatMessage { + id: "m1".into(), + trade_id: "order-1".into(), + sender_pubkey: "peer".into(), + content: "hola".into(), + message_type: crate::api::types::MessageType::Peer, + is_mine: false, + is_read: false, + has_attachment: false, + attachment: None, + created_at: 1, + }) + .await + .unwrap(); + assert_eq!(storage.list_messages("order-1").await.unwrap().len(), 1); + + drop(storage); + let _ = std::fs::remove_file(&path); + } + #[tokio::test] async fn active_mostro_pubkey_round_trip() { let path = temp_db_path();