diff --git a/crates/browser-signer-proxy/proxy.js b/crates/browser-signer-proxy/proxy.js index f04465b2..d15e2724 100644 --- a/crates/browser-signer-proxy/proxy.js +++ b/crates/browser-signer-proxy/proxy.js @@ -5,7 +5,7 @@ async function pollForRequests() { isPolling = true; try { - const response = await fetch('/api/pending'); + const response = await fetch('/api/pending?token=' + encodeURIComponent(window.proxySessionToken)); const data = await response.json(); console.log('Polled for requests, got:', data); @@ -94,7 +94,7 @@ async function handleNip07Request(request) { console.log('Sending response:', responsePayload); - await fetch('/api/response', { + await fetch('/api/response?token=' + encodeURIComponent(window.proxySessionToken), { method: 'POST', headers: { 'Content-Type': 'application/json', @@ -117,7 +117,7 @@ async function handleNip07Request(request) { console.log('Sending error response:', errorPayload); - await fetch('/api/response', { + await fetch('/api/response?token=' + encodeURIComponent(window.proxySessionToken), { method: 'POST', headers: { 'Content-Type': 'application/json', diff --git a/crates/browser-signer-proxy/src/lib.rs b/crates/browser-signer-proxy/src/lib.rs index 6617db1b..fc63f29b 100644 --- a/crates/browser-signer-proxy/src/lib.rs +++ b/crates/browser-signer-proxy/src/lib.rs @@ -265,6 +265,10 @@ impl BrowserSignerProxy { let state: Arc = self.inner.state.clone(); let custom_html: &'static str = self.inner.options.custom_html; let shutdown: Arc = self.inner.shutdown.clone(); + // Regenerated on every startup, never part of the opened URL + let session_token: String = Uuid::new_v4().simple().to_string(); + // Only the exact host:port the server binds may appear in the Host header + let expected_host: String = addr.to_string(); smol::spawn(async move { tracing::info!("Starting proxy server on {addr}"); @@ -291,13 +295,22 @@ impl BrowserSignerProxy { AcceptEvent::Connection(Ok((stream, _))) => { let state: Arc = state.clone(); let shutdown: Arc = shutdown.clone(); + let expected_host: String = expected_host.clone(); + let session_token: String = session_token.clone(); smol::spawn(async move { let shutdown_listener = shutdown.listen(); smol::future::or( async { - handle_connection(stream, state, custom_html).await; + handle_connection( + stream, + state, + custom_html, + expected_host, + session_token, + ) + .await; }, async { shutdown_listener.await; @@ -336,6 +349,18 @@ impl BrowserSignerProxy { outgoing_requests.push(request); } + /// Drop a request from both queues so timed-out request data is not retained. + #[inline] + async fn cancel_request(&self, id: Uuid) { + self.inner.state.pending_responses.lock().await.remove(&id); + self.inner + .state + .outgoing_requests + .lock() + .await + .retain(|request| request.id != id); + } + async fn request(&self, method: RequestMethod, params: Value) -> Result where T: DeserializeOwned, @@ -345,6 +370,7 @@ impl BrowserSignerProxy { // Construct the request let request: RequestData = RequestData::new(method, params); + let request_id = request.id; // Create a bounded channel of size 1 as a oneshot replacement let (tx, rx) = channel::bounded::>(1); @@ -361,7 +387,10 @@ impl BrowserSignerProxy { match response { Ok(Ok(res)) => Ok(serde_json::from_value(res)?), Ok(Err(error)) => Err(Error::generic(error)), - Err(TimeoutError) => Err(Error::timeout()), + Err(TimeoutError) => { + self.cancel_request(request_id).await; + Err(Error::timeout()) + } } } @@ -437,10 +466,14 @@ impl AsyncNip44 for BrowserSignerProxy { } } -// ── Minimal HTTP server ────────────────────────────────────────────────── - /// Handle a single HTTP connection. -async fn handle_connection(stream: TcpStream, state: Arc, custom_html: &'static str) { +async fn handle_connection( + stream: TcpStream, + state: Arc, + custom_html: &'static str, + expected_host: String, + session_token: String, +) { let mut reader = BufReader::new(stream); // Read the request line @@ -462,24 +495,49 @@ async fn handle_connection(stream: TcpStream, state: Arc, custom_htm // Read headers until empty line let mut headers = Vec::new(); let mut content_length: usize = 0; + let mut host: Option = None; + loop { let mut line = String::new(); if reader.read_line(&mut line).await.is_err() { return; } + let line = line.trim_end().to_string(); if line.is_empty() { break; } + if let Some(value) = line.strip_prefix("content-length:") { content_length = value.trim().parse().unwrap_or(0); } else if let Some(value) = line.strip_prefix("Content-Length:") { content_length = value.trim().parse().unwrap_or(0); + } else if let Some(value) = line.strip_prefix("host:") { + host = Some(value.trim().to_string()); + } else if let Some(value) = line.strip_prefix("Host:") { + host = Some(value.trim().to_string()); } + headers.push(line); } - match (method.as_str(), path.as_str()) { + // DNS rebinding defense: only the exact bound host:port is accepted + if host.as_deref() != Some(expected_host.as_str()) { + send_response(&mut reader, 403, "Forbidden", "", "").await; + return; + } + + let (route, query) = match path.split_once('?') { + Some((route, query)) => (route, Some(query)), + None => (path.as_str(), None), + }; + + if route.starts_with("/api/") && query_param(query, "token") != Some(session_token.as_str()) { + send_response(&mut reader, 403, "Forbidden", "", "").await; + return; + } + + match (method.as_str(), route) { // Serve the HTML proxy page ("GET", "/") => { let html = if custom_html.is_empty() { @@ -487,7 +545,8 @@ async fn handle_connection(stream: TcpStream, state: Arc, custom_htm } else { custom_html }; - send_response(&mut reader, 200, "OK", "text/html", html).await; + let html = inject_session_token(html, &session_token); + send_response(&mut reader, 200, "OK", "text/html", &html).await; } // Serve the JS proxy script ("GET", "/proxy.js") => { @@ -516,7 +575,7 @@ async fn handle_connection(stream: TcpStream, state: Arc, custom_htm // Clear the outgoing requests after sending them outgoing.clear(); - send_response_cors_json(&mut reader, 200, "OK", &json).await; + send_response(&mut reader, 200, "OK", "application/json", &json).await; } // Receive response from browser extension ("POST", "/api/response") => { @@ -541,27 +600,16 @@ async fn handle_connection(stream: TcpStream, state: Arc, custom_htm let mut pending = state.pending_responses.lock().await; match pending.remove(&id) { - Some(sender) => { - // Use try_send since we already hold the lock - let _ = sender.try_send(message.into_result()); - tracing::info!("Forwarded response for request {id}"); - } + Some(sender) => match sender.try_send(message.into_result()) { + Ok(()) => tracing::info!("Forwarded response for request {id}"), + Err(e) => { + tracing::warn!("Failed to forward response for request {id}: {e}") + } + }, None => tracing::warn!("No pending request found for {id}"), } - send_response_cors(&mut reader, 200, "OK", "text/plain", "OK").await; - } - // CORS preflight - ("OPTIONS", _) => { - let response = "HTTP/1.1 200 OK\r\n\ - Access-Control-Allow-Origin: *\r\n\ - Access-Control-Allow-Methods: GET, POST, OPTIONS\r\n\ - Access-Control-Allow-Headers: Content-Type\r\n\ - Content-Length: 0\r\n\ - Connection: close\r\n\ - \r\n"; - let _ = reader.get_mut().write_all(response.as_bytes()).await; - let _ = reader.get_mut().flush().await; + send_response(&mut reader, 200, "OK", "text/plain", "OK").await; } // 404 - not found _ => { @@ -585,68 +633,44 @@ async fn send_response( } response.push_str(&format!("Content-Length: {}\r\n", body.len())); - response.push_str("Access-Control-Allow-Origin: *\r\n"); response.push_str("Connection: close\r\n"); response.push_str("\r\n"); response.push_str(body); - let _ = stream.write_all(response.as_bytes()).await; - let _ = stream.flush().await; + write_response(stream, &response).await; } -/// Write a response with CORS headers and JSON content type. -async fn send_response_cors_json( - stream: &mut (impl AsyncWriteExt + Unpin), - status: u16, - status_text: &str, - body: &str, -) { - let mut response = format!("HTTP/1.1 {status} {status_text}\r\n"); - response.push_str("Content-Type: application/json\r\n"); - response.push_str(&format!("Content-Length: {}\r\n", body.len())); - response.push_str("Access-Control-Allow-Origin: *\r\n"); - response.push_str("Connection: close\r\n"); - response.push_str("\r\n"); - response.push_str(body); - - let _ = stream.write_all(response.as_bytes()).await; - let _ = stream.flush().await; -} - -/// Write a response with CORS headers. -async fn send_response_cors( - stream: &mut (impl AsyncWriteExt + Unpin), - status: u16, - status_text: &str, - content_type: &str, - body: &str, -) { - let mut response = format!("HTTP/1.1 {status} {status_text}\r\n"); - - if !content_type.is_empty() { - response.push_str(&format!("Content-Type: {content_type}\r\n")); +async fn write_response(stream: &mut (impl AsyncWriteExt + Unpin), response: &str) { + if let Err(e) = stream.write_all(response.as_bytes()).await { + tracing::debug!("Failed to write response to peer: {e}"); + } + if let Err(e) = stream.flush().await { + tracing::debug!("Failed to flush response to peer: {e}"); } - - response.push_str(&format!("Content-Length: {}\r\n", body.len())); - response.push_str("Access-Control-Allow-Origin: *\r\n"); - response.push_str("Connection: close\r\n"); - response.push_str("\r\n"); - response.push_str(body); - - let _ = stream.write_all(response.as_bytes()).await; - let _ = stream.flush().await; } -// ── Timeout helper ─────────────────────────────────────────────────────── +/// Get a parameter value from an optional query string. +fn query_param<'a>(query: Option<&'a str>, name: &str) -> Option<&'a str> { + let query = query?; + query.split('&').find_map(|pair| { + let (key, value) = pair.split_once('=')?; + (key == name).then_some(value) + }) +} + +fn inject_session_token(html: &str, token: &str) -> String { + let script = format!(""); + match html.rfind("") { + Some(index) => format!("{}{script}{}", &html[..index], &html[index..]), + None => format!("{html}{script}"), + } +} /// An error indicating that an operation timed out. #[derive(Debug)] struct TimeoutError; /// Races a channel receive against a duration. -/// -/// Returns the channel value on success, or [`TimeoutError`] if the duration -/// elapses first or the channel is closed. async fn race_timeout( duration: Duration, recv: impl Future>, @@ -677,10 +701,6 @@ async fn race_timeout( } } -// ── Utility ────────────────────────────────────────────────────────────── - -/// Gets the current time in seconds since the Unix epoch (1970-01-01). If the -/// time is before the epoch, returns 0. #[inline] fn current_time() -> u64 { SystemTime::now() diff --git a/crates/chat_ui/src/lib.rs b/crates/chat_ui/src/lib.rs index 8f4c0d91..6701b8f3 100644 --- a/crates/chat_ui/src/lib.rs +++ b/crates/chat_ui/src/lib.rs @@ -598,16 +598,23 @@ impl ChatPanel { return; } + let content = SharedString::from(&event.content); let emojis = emojis_from_tags(&event.tags); + let image = Message::reaction_image(&event.content, &emojis) .map(|url| SharedUri::from(url.to_string())); for id in event.tags.event_ids() { - self.reactions.entry(id).or_default().push(( - SharedString::from(&event.content), - image.clone(), - event.pubkey, - )); + let entries = self.reactions.entry(id).or_default(); + // NIP-25: a newer reaction by the same author replaces their previous one + if let Some(existing) = entries + .iter_mut() + .find(|(_, _, author)| *author == event.pubkey) + { + *existing = (content.clone(), image.clone(), event.pubkey); + } else { + entries.push((content.clone(), image.clone(), event.pubkey)); + } } cx.notify(); diff --git a/crates/common/src/media_extractor.rs b/crates/common/src/media_extractor.rs index c688b3f8..34d13f0b 100644 --- a/crates/common/src/media_extractor.rs +++ b/crates/common/src/media_extractor.rs @@ -1,65 +1,161 @@ +use std::sync::LazyLock; + use gpui::SharedUri; use regex::Regex; -/// Extracts media URLs from a string and returns both -/// the extracted URLs and the string with media URLs removed -struct MediaExtractor { - image_regex: Regex, -} +/// Matches image URLs with common image extensions +static IMAGE_RE: LazyLock = LazyLock::new(|| { + Regex::new( + r#"(?i)\bhttps?://[^\s<>"']+\.(?:jpg|jpeg|png|gif|bmp|webp|svg|ico)(?:\?[^\s<>"']*)?\b"#, + ) + .unwrap() +}); + +/// Extracts media URLs from a string and returns the URLs and the cleaned text +struct MediaExtractor; impl MediaExtractor { - /// Creates a new MediaExtractor with compiled regex patterns - fn new() -> Self { - MediaExtractor { - // Match common image extensions - image_regex: Regex::new( - r#"(?i)\bhttps?://[^\s<>"']+\.(?:jpg|jpeg|png|gif|bmp|webp|svg|ico)(?:\?[^\s<>"']*)?\b"#, - ).unwrap(), - } - } - /// Extracts all media URLs from a string - fn extract_media_urls(&self, text: &str) -> Vec { - let mut urls = Vec::new(); - - // Extract image URLs - for capture in self.image_regex.find_iter(text) { - urls.push(capture.as_str().to_string().into()); - } - - urls + fn extract_media_urls(text: &str) -> Vec { + IMAGE_RE + .find_iter(text) + .map(|capture| capture.as_str().to_string().into()) + .collect() } /// Removes all media URLs from a string and returns the cleaned text - fn remove_media_urls(&self, text: &str) -> String { - let mut result = text.to_string(); + fn remove_media_urls(text: &str) -> String { + let mut cleaned = String::with_capacity(text.len()); + let mut last_end = 0; + // Whitespace bordered a removed URL, so one space stands in for it + let mut spaced = false; - // Remove image URLs - result = self.image_regex.replace_all(&result, "").to_string(); + for url in IMAGE_RE.find_iter(text) { + let raw_gap = &text[last_end..url.start()]; + let surrounded = raw_gap.ends_with(char::is_whitespace) + || text[url.end()..].starts_with(char::is_whitespace); - // Clean up extra whitespace that might result from removal - self.cleanup_text(&result) + let gap = if last_end == 0 { + raw_gap.trim_end() + } else { + raw_gap.trim() + }; + + if !gap.is_empty() { + if spaced && !cleaned.is_empty() { + cleaned.push(' '); + } + cleaned.push_str(gap); + } + + spaced |= surrounded; + last_end = url.end(); + } + + let tail = text[last_end..].trim_start(); + if spaced && !cleaned.is_empty() && !tail.is_empty() { + cleaned.push(' '); + } + cleaned.push_str(tail); + + cleaned.trim().to_string() } /// Extracts media URLs and removes them from the string, returning both - fn extract_and_remove(&self, text: &str) -> (Vec, String) { - let urls = self.extract_media_urls(text); - let cleaned_text = self.remove_media_urls(text); - (urls, cleaned_text) - } - - /// Helper function to clean up text after URL removal - fn cleanup_text(&self, text: &str) -> String { - let text = text.trim(); - - // Remove multiple consecutive spaces - let re = Regex::new(r"\s+").unwrap(); - re.replace_all(text, " ").trim().to_string() + fn extract_and_remove(text: &str) -> (Vec, String) { + ( + Self::extract_media_urls(text), + Self::remove_media_urls(text), + ) } } /// Convenience function for one-time extraction and removal pub fn extract_and_remove_media_urls(text: &str) -> (Vec, String) { - let extractor = MediaExtractor::new(); - extractor.extract_and_remove(text) + MediaExtractor::extract_and_remove(text) +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn multi_line_text_without_urls_keeps_line_breaks() { + let text = "first line\n\nsecond line"; + let (urls, cleaned) = extract_and_remove_media_urls(text); + + assert!(urls.is_empty()); + assert_eq!(cleaned, text); + } + + #[test] + fn removes_url_and_tidies_around_it() { + let (urls, cleaned) = + extract_and_remove_media_urls("check https://example.com/pic.png here"); + + assert_eq!(urls, vec![SharedUri::from("https://example.com/pic.png")]); + assert_eq!(cleaned, "check here"); + } + + #[test] + fn removing_url_keeps_other_line_breaks() { + let (urls, cleaned) = extract_and_remove_media_urls( + "first\nsecond https://example.com/pic.png third\nfourth", + ); + + assert_eq!(urls.len(), 1); + assert_eq!(cleaned, "first\nsecond third\nfourth"); + } + + #[test] + fn url_on_its_own_line_joins_neighbors_with_one_space() { + let (urls, cleaned) = + extract_and_remove_media_urls("first\n\nhttps://example.com/pic.png\n\nlast"); + + assert_eq!(urls.len(), 1); + assert_eq!(cleaned, "first last"); + } + + #[test] + fn multiple_urls_collapse_to_single_spaces() { + let (urls, cleaned) = extract_and_remove_media_urls( + "a https://example.com/a.png b https://example.com/b.png c", + ); + + assert_eq!(urls.len(), 2); + assert_eq!(cleaned, "a b c"); + } + + #[test] + fn message_with_only_a_url_becomes_empty() { + let (urls, cleaned) = extract_and_remove_media_urls("https://example.com/pic.png"); + + assert_eq!(urls.len(), 1); + assert_eq!(cleaned, ""); + } + + #[test] + fn digits_glued_to_url_scheme_prevent_match() { + let text = "v1.2https://example.com/pic.png"; + let (urls, cleaned) = extract_and_remove_media_urls(text); + + assert!(urls.is_empty()); + assert_eq!(cleaned, text); + } + + #[test] + fn digits_around_removed_url_survive() { + let (urls, cleaned) = extract_and_remove_media_urls("123 https://example.com/pic.png 456"); + + assert_eq!(urls.len(), 1); + assert_eq!(cleaned, "123 456"); + } + + #[test] + fn emoji_around_removed_url_survives() { + let (urls, cleaned) = extract_and_remove_media_urls("👍 https://example.com/pic.png 👍"); + + assert_eq!(urls.len(), 1); + assert_eq!(cleaned, "👍 👍"); + } } diff --git a/crates/nip29/src/group.rs b/crates/nip29/src/group.rs index 23589d09..e5927852 100644 --- a/crates/nip29/src/group.rs +++ b/crates/nip29/src/group.rs @@ -13,7 +13,7 @@ use crate::LOCAL_KEYS; use crate::protocol::{ ACTIVITY_KINDS, Activity, GroupAdmins, GroupId, GroupKey, GroupMembers, GroupMetadata, GroupPins, GroupRoles, MEMBERSHIP_KINDS, Membership, PIN_LIST, Pin, RENDER_KINDS, STATE_KINDS, - TimelineWindow, tag_value, + TimelineWindow, is_newer_event, tag_value, }; const REPLAY_LIMIT: usize = 100; @@ -91,7 +91,8 @@ pub struct Group { membership: Membership, membership_events: Vec, elsewhere: Option, - state_ids: HashMap, + /// The last accepted (created_at, id) per state kind. + state_ids: HashMap, rows: Vec, index: HashSet, reactions: HashMap>, @@ -694,6 +695,15 @@ impl Group { } fn absorb_state(&mut self, event: &Event, me: PublicKey) -> bool { + // Relays deliver same-kind state events in any order. + if !is_newer_event( + self.state_ids.get(&event.kind).copied(), + event.created_at, + event.id, + ) { + return false; + } + match event.kind { Kind::GroupMetadata => match GroupMetadata::parse(event) { Ok(metadata) => self.metadata = Some(metadata), @@ -724,7 +734,10 @@ impl Group { _ => return false, } - self.state_ids.insert(event.kind, event.id) != Some(event.id) + self.state_ids + .insert(event.kind, (event.created_at, event.id)); + + true } fn adopt_roster_membership(&mut self, me: PublicKey) -> bool { diff --git a/crates/nip29/src/lib.rs b/crates/nip29/src/lib.rs index c1efc3e9..70cc5464 100644 --- a/crates/nip29/src/lib.rs +++ b/crates/nip29/src/lib.rs @@ -76,6 +76,8 @@ pub struct GroupsRegistry { groups: Vec>, previews: HashMap>, list: GroupList, + /// The (created_at, id) of the last accepted group list event. + list_stamp: Option<(Timestamp, EventId)>, browse: HashMap, routes: HashMap, observers: HashMap, @@ -132,6 +134,7 @@ impl GroupsRegistry { groups: Vec::new(), previews: HashMap::new(), list: GroupList::default(), + list_stamp: None, browse: HashMap::new(), routes: HashMap::new(), observers: HashMap::new(), @@ -455,6 +458,8 @@ impl GroupsRegistry { self.fork_authors.clear(); self.syncing.clear(); self.list = GroupList::default(); + // A new signer's list events must not be judged against the old stamp. + self.list_stamp = None; let nostr = NostrRegistry::global(cx); let client = nostr.read(cx).client(); @@ -488,24 +493,26 @@ impl GroupsRegistry { let signer = nostr.read(cx).signer(); // Show the cached groups before the relays answer. - let task: Task> = cx.background_spawn(async move { - let me = signer.get_public_key_async().await?; - let filter = Filter::new().author(me).kind(Kind::SimpleGroups).limit(1); + let task: Task), Error>> = + cx.background_spawn(async move { + let me = signer.get_public_key_async().await?; + let filter = Filter::new().author(me).kind(Kind::SimpleGroups).limit(1); - match client.database().query(filter).await?.into_iter().next() { - Some(event) => { - let list = GroupList::parse(&event)?; - Ok(Self::hydrate_list(list, &signer, &me).await) + match client.database().query(filter).await?.into_iter().next() { + Some(event) => { + let list = + Self::hydrate_list(GroupList::parse(&event)?, &signer, &me).await; + Ok((list, Some(event))) + } + None => Ok((GroupList::default(), None)), } - None => Ok(GroupList::default()), - } - }); + }); self.tasks .push(cx.spawn(async move |this, cx| match task.await { - Ok(list) => { + Ok((list, event)) => { this.update(cx, |this, cx| { - this.merge(list, cx); + this.merge(list, event.as_ref(), cx); }) .ok(); } @@ -538,7 +545,16 @@ impl GroupsRegistry { self.watch_forks(cx); } - fn merge(&mut self, list: GroupList, cx: &mut Context) { + fn merge(&mut self, list: GroupList, event: Option<&Event>, cx: &mut Context) { + if let Some(event) = event { + // Relays replay list events out of order; only strictly newer applies. + if !is_newer_event(self.list_stamp, event.created_at, event.id) { + return; + } + + self.list_stamp = Some((event.created_at, event.id)); + } + // Carry the event's private section so a later publish can preserve it. self.list .set_content(list.content().map(str::to_owned), list.content_readable()); @@ -589,20 +605,21 @@ impl GroupsRegistry { }; if list.content().is_none() { - self.merge(list, cx); + self.merge(list, Some(&event), cx); return; } let nostr = NostrRegistry::global(cx); let signer = nostr.read(cx).signer(); let Some(me) = nostr.read(cx).current_user() else { - self.merge(list, cx); + self.merge(list, Some(&event), cx); return; }; self.tasks.push(cx.spawn(async move |this, cx| { let list = Self::hydrate_list(list, &signer, &me).await; - this.update(cx, |this, cx| this.merge(list, cx)).ok(); + this.update(cx, |this, cx| this.merge(list, Some(&event), cx)) + .ok(); })); } @@ -864,7 +881,7 @@ impl GroupsRegistry { let me = nostr.read(cx).current_user(); let list = self.list.clone(); - self.tasks.push(cx.spawn(async move |_this, _cx| { + self.tasks.push(cx.spawn(async move |this, cx| { // Re-encrypt the private section when it carries local entries; a // section this session could not read is carried verbatim. let content = match (list.has_private(), list.content_readable()) { @@ -893,6 +910,15 @@ impl GroupsRegistry { } }; + // Our own publish is the newest list until a newer event arrives. + if let Err(error) = this.update(cx, |this, _cx| { + if is_newer_event(this.list_stamp, event.created_at, event.id) { + this.list_stamp = Some((event.created_at, event.id)); + } + }) { + log::warn!("recording the published list failed: {error}"); + } + match client.send_event(&event).to_nip65().await { Ok(output) if output.failed.is_empty() => {} Ok(output) => { diff --git a/crates/nip29/src/protocol.rs b/crates/nip29/src/protocol.rs index 0d680229..26502327 100644 --- a/crates/nip29/src/protocol.rs +++ b/crates/nip29/src/protocol.rs @@ -50,6 +50,15 @@ pub(crate) fn tag_value<'a>(tags: &'a Tags, name: &str) -> Option<&'a str> { .and_then(|tag| tag.content()) } +/// Whether `(created_at, id)` is strictly newer than the last accepted stamp. +pub(crate) fn is_newer_event( + accepted: Option<(Timestamp, EventId)>, + created_at: Timestamp, + id: EventId, +) -> bool { + accepted.is_none_or(|last| (created_at, id) > last) +} + #[derive(Debug, Clone, PartialEq, Eq, PartialOrd, Ord, Hash)] pub struct GroupId(String); @@ -1152,4 +1161,19 @@ mod tests { let malformed = Tag::custom("previous", ["nothex"]); assert_eq!(window.references_seen(Some(&malformed)), Some(false)); } + + #[test] + fn is_newer_event_accepts_only_strictly_newer_events() { + let lower = EventId::from_byte_array([0x00; 32]); + let higher = EventId::from_byte_array([0xff; 32]); + let earlier = Timestamp::from(100); + let later = Timestamp::from(200); + + assert!(is_newer_event(None, earlier, lower)); + assert!(!is_newer_event(Some((earlier, lower)), earlier, lower)); + assert!(is_newer_event(Some((earlier, lower)), later, lower)); + assert!(!is_newer_event(Some((later, lower)), earlier, higher)); + assert!(is_newer_event(Some((earlier, lower)), earlier, higher)); + assert!(!is_newer_event(Some((earlier, higher)), earlier, lower)); + } } diff --git a/crates/nip60/src/lib.rs b/crates/nip60/src/lib.rs index 4e2473ce..4e032cf7 100644 --- a/crates/nip60/src/lib.rs +++ b/crates/nip60/src/lib.rs @@ -55,6 +55,9 @@ pub struct HistoryEntry { pub created_at: Timestamp, } +/// How many times a failed nutzap redemption is retried before giving up +const MAX_REDEEM_ATTEMPTS: u32 = 3; + pub struct WalletRegistry { wallet: Option, tokens: HashMap, @@ -69,6 +72,10 @@ pub struct WalletRegistry { pending_nutzaps: Vec, /// Nutzaps with a redemption job in flight redeeming: HashSet, + /// Redemption attempts per nutzap, reset on success + redeem_attempts: HashMap, + /// A job errored after possibly mutating the store + publish_failed: bool, store: Option>, wallets: HashMap, loading: bool, @@ -119,6 +126,8 @@ impl WalletRegistry { redeemed: HashSet::new(), pending_nutzaps: Vec::new(), redeeming: HashSet::new(), + redeem_attempts: HashMap::new(), + publish_failed: false, store: None, wallets: HashMap::new(), loading: false, @@ -378,6 +387,8 @@ impl WalletRegistry { self.redeemed.clear(); self.pending_nutzaps.clear(); self.redeeming.clear(); + self.redeem_attempts.clear(); + self.publish_failed = false; self.store = None; self.wallets.clear(); self.loading = false; @@ -397,6 +408,7 @@ impl WalletRegistry { let signer = nostr.read(cx).signer().clone(); let (tx, rx) = flume::bounded::(100); + let consumer_signer = signer.clone(); self.notification_listener = Some(cx.background_spawn(async move { const MAX_PROCESSED: usize = 1000; @@ -417,7 +429,7 @@ impl WalletRegistry { | Kind::CashuWalletSpendingHistory ); - let wallet_event = ours && is_cashu || is_token_deletion(&event); + let wallet_event = ours && (is_cashu || is_token_deletion(&event)); let nutzap = event.kind == Kind::Custom(9321) && me.is_some_and(|me| event.tags.public_keys().any(|key| key == me)); @@ -443,21 +455,22 @@ impl WalletRegistry { })); self.tasks.push(cx.spawn(async move |this, cx| { + let me = consumer_signer.get_public_key_async().await.ok(); while let Ok(event) = rx.recv_async().await { this.update(cx, |this, cx| { - this.absorb(event, cx); + this.absorb(event, me, cx); }) .ok(); } })); } - fn absorb(&mut self, event: Event, cx: &mut Context) { + fn absorb(&mut self, event: Event, me: Option, cx: &mut Context) { match event.kind { Kind::CashuWallet => self.absorb_wallet(event, cx), Kind::CashuWalletUnspentProof => self.absorb_token(event, cx), Kind::CashuWalletSpendingHistory => self.absorb_history(event, cx), - Kind::EventDeletion => self.absorb_deletion(event, cx), + Kind::EventDeletion => self.absorb_deletion(event, me, cx), Kind::Custom(9321) => self.absorb_nutzap(event, cx), _ => {} } @@ -549,7 +562,15 @@ impl WalletRegistry { })); } - fn absorb_deletion(&mut self, event: Event, cx: &mut Context) { + fn absorb_deletion(&mut self, event: Event, me: Option, cx: &mut Context) { + if me != Some(event.pubkey) { + log::warn!( + "ignoring kind 5 deletion {} from a foreign author", + event.id + ); + return; + } + let mut removed_any = false; for tag in event.tags.iter() { if tag.kind() != "e" { @@ -807,6 +828,9 @@ impl WalletRegistry { self.job_running = true; + let job_handle = job.clone(); + let publish_failed = self.publish_failed; + self.tasks.push(cx.spawn(async move |this, cx| { let result = Tokio::spawn_result(cx, async move { let ctx = ops::JobCtx { @@ -815,6 +839,7 @@ impl WalletRegistry { signer, client, tokens, + publish_failed, }; job.run(ctx).await @@ -823,7 +848,7 @@ impl WalletRegistry { this.update(cx, |this, cx| { this.job_running = false; - this.apply_outcome(result, cx); + this.apply_outcome(job_handle, result, cx); for job in std::mem::take(&mut this.job_queued) { this.schedule_job(job, cx); @@ -833,10 +858,38 @@ impl WalletRegistry { })); } - fn apply_outcome(&mut self, result: Result, cx: &mut Context) { + fn apply_outcome( + &mut self, + job: ops::Job, + result: Result, + cx: &mut Context, + ) { let outcome = match result { Ok(outcome) => outcome, Err(error) => { + if !matches!(job, ops::Job::CreateQuote { .. }) { + self.publish_failed = true; + } + + if !matches!(job, ops::Job::CreateQuote { .. } | ops::Job::Reconcile) { + self.schedule_job(ops::Job::Reconcile, cx); + } + + if let ops::Job::RedeemNutzap { nutzap_id, .. } = &job { + self.redeeming.remove(nutzap_id); + + let attempts = self.redeem_attempts.entry(*nutzap_id).or_default(); + *attempts += 1; + + if *attempts <= MAX_REDEEM_ATTEMPTS { + self.job_queued.push(job.clone()); + } else { + log::warn!( + "giving up redeeming nutzap {nutzap_id} after {MAX_REDEEM_ATTEMPTS} attempts" + ); + } + } + self.emit_error(format!("{error:#}"), cx); return; } @@ -845,6 +898,7 @@ impl WalletRegistry { match outcome { ops::Outcome::Idle => {} ops::Outcome::Reconciled { updates } => { + self.publish_failed = false; let changed = !updates.is_empty(); for update in updates { @@ -905,6 +959,7 @@ impl WalletRegistry { amount, } => { self.redeeming.remove(&nutzap_id); + self.redeem_attempts.remove(&nutzap_id); self.redeemed.insert(nutzap_id); self.apply_token_update(update); self.emit_updated(cx); diff --git a/crates/nip60/src/ops.rs b/crates/nip60/src/ops.rs index 722c37e4..e347c636 100644 --- a/crates/nip60/src/ops.rs +++ b/crates/nip60/src/ops.rs @@ -361,6 +361,8 @@ pub(crate) struct JobCtx { pub(crate) client: Client, /// Relay token events at schedule time. pub(crate) tokens: HashMap, + /// A previous job errored after possibly mutating the store + pub(crate) publish_failed: bool, } impl JobCtx @@ -426,35 +428,14 @@ where log::warn!("saga recovery failed for {mint_url}: {error:#}"); } - let minted = wallet.mint_unissued_quotes().await?; - if minted == Amount::ZERO { - continue; - } + let local_ys = proof_ys(wallet.get_unspent_proofs().await?.iter())?; + let states = vec![State::Pending, State::Reserved, State::PendingSpent]; - let update = self - .sync_token_events( - wallet, - nostr_mint(mint_url)?, - &previous_ids(&tokens, mint_url), - Some(keys::History { - direction: TransactionDirection::In, - amount: u64::from(minted), - redeemed: Vec::new(), - sender: None, - }), - ) - .await?; + let in_flight = wallet.get_proofs_by_states(states).await?; + let in_flight_ys = proof_ys(in_flight.iter())?; - for id in &update.destroyed { - tokens.remove(id); - } - tokens.insert(update.event_id, update.token.clone()); - updates.push(update); - } - - for (mint_url, wallet) in &self.wallets { let mut relay_ys = HashSet::new(); - let mut added = Vec::new(); + let mut unseen = Vec::new(); for token in tokens .values() @@ -468,23 +449,104 @@ where continue; } }; - if relay_ys.insert(info.y) { - added.push(info); + + if !relay_ys.insert(info.y) { + continue; } + + if local_ys.contains(&info.y) || in_flight_ys.contains(&info.y) { + continue; + } + + unseen.push(info); } } - let unspent = wallet.get_unspent_proofs().await?; - let removed = unspent - .iter() - .filter(|proof| match proof.y() { - Ok(y) => !relay_ys.contains(&y), - Err(_) => false, - }) - .map(|proof| proof.y()) - .collect::, _>>()?; + if unseen.is_empty() { + continue; + } - wallet.localstore.update_proofs(added, removed).await?; + let unseen_proofs: Proofs = unseen.iter().map(|info| info.proof.clone()).collect(); + + let spent = match wallet.check_proofs_spent(unseen_proofs).await { + Ok(states) => states + .into_iter() + .filter(|state| state.state == State::Spent) + .map(|state| state.y) + .collect::>(), + Err(error) => { + log::warn!("mint state check failed, accepting unseen proofs: {error:#}"); + HashSet::new() + } + }; + + let added: Vec = unseen + .into_iter() + .filter(|info| !spent.contains(&info.y)) + .collect(); + + if !added.is_empty() { + wallet.localstore.update_proofs(added, Vec::new()).await?; + } + } + + for (mint_url, wallet) in &self.wallets { + let minted = wallet.mint_unissued_quotes().await?; + if minted != Amount::ZERO { + let update = self + .sync_token_events( + wallet, + nostr_mint(mint_url)?, + &previous_ids(&tokens, mint_url), + Some(keys::History { + direction: TransactionDirection::In, + amount: u64::from(minted), + redeemed: Vec::new(), + sender: None, + }), + ) + .await?; + + for id in &update.destroyed { + tokens.remove(id); + } + tokens.insert(update.event_id, update.token.clone()); + updates.push(update); + } + + let relay_ys = relay_proof_ys(&tokens, mint_url); + let local_ys = proof_ys(wallet.get_unspent_proofs().await?.iter())?; + + if local_ys != relay_ys { + if self.publish_failed { + let update = self + .sync_token_events( + wallet, + nostr_mint(mint_url)?, + &previous_ids(&tokens, mint_url), + None, + ) + .await?; + + for id in &update.destroyed { + tokens.remove(id); + } + tokens.insert(update.event_id, update.token.clone()); + updates.push(update); + } else { + let removed = wallet + .get_unspent_proofs() + .await? + .into_iter() + .filter(|proof| match proof.y() { + Ok(y) => !relay_ys.contains(&y), + Err(_) => false, + }) + .map(|proof| proof.y()) + .collect::, _>>()?; + wallet.localstore.update_proofs(Vec::new(), removed).await?; + } + } } Ok(Outcome::Reconciled { updates }) @@ -499,6 +561,34 @@ fn previous_ids(tokens: &HashMap, mint: &MintUrl) -> Vec( + proofs: impl Iterator, +) -> Result, Error> { + proofs.map(|proof| proof.y().map_err(Error::from)).collect() +} + +/// Proof ys referenced by the relay token events of one mint +fn relay_proof_ys( + tokens: &HashMap, + mint: &MintUrl, +) -> HashSet { + let mut ys = HashSet::new(); + for token in tokens + .values() + .filter(|token| token.mint.as_str() == mint.to_string()) + { + for proof in &token.proofs { + match proof_info(proof, mint) { + Ok(info) => { + ys.insert(info.y); + } + Err(error) => log::warn!("skipping proof: {error:#}"), + } + } + } + ys +} + /// Resolve a recipient's 10019 wallet info and pick a mint we share with them. async fn recipient_wallet_info( ctx: &JobCtx, diff --git a/crates/settings/src/lib.rs b/crates/settings/src/lib.rs index a850bc7d..04ade7bb 100644 --- a/crates/settings/src/lib.rs +++ b/crates/settings/src/lib.rs @@ -223,6 +223,11 @@ impl Global for GlobalAppSettings {} pub struct AppSettings { /// Settings inner: Entity, + /// The pending save task, so writes to the settings file complete in order + #[cfg(not(target_arch = "wasm32"))] + save_task: Option>, + /// Set after a failed load so its own save cannot overwrite the file with defaults + suppress_next_save: bool, /// Event subscriptions _subscriptions: SmallVec<[Subscription; 2]>, } @@ -315,11 +320,16 @@ impl AppSettings { // Run at the end of current cycle cx.defer(move |cx| { - entity.update(cx, |this, cx| this.load(cx)).ok(); + if let Err(error) = entity.update(cx, |this, cx| this.load(cx)) { + log::error!("failed to load settings: {error:#}"); + } }); Self { inner, + #[cfg(not(target_arch = "wasm32"))] + save_task: None, + suppress_next_save: false, _subscriptions: subscriptions, } } @@ -381,27 +391,46 @@ impl AppSettings { /// Load settings fn load(&mut self, cx: &mut Context) { - let task: Task> = cx.background_spawn(async move { + let task: Task, Error>> = cx.background_spawn(async move { #[cfg(not(target_arch = "wasm32"))] { let path = config_dir().join(".settings"); - if let Ok(content) = smol::fs::read_to_string(&path).await { - return Ok(serde_json::from_str(&content)?); + match smol::fs::read_to_string(&path).await { + Ok(content) => { + let settings: Settings = + serde_json::from_str(&content).map_err(|error| { + anyhow!("failed to parse {}: {error}", path.display()) + })?; + return Ok(Some(settings)); + } + // No settings file yet on a fresh install, so keep the defaults + Err(error) if error.kind() == std::io::ErrorKind::NotFound => {} + Err(error) => { + return Err(anyhow!("failed to read {}: {error}", path.display())); + } } } #[cfg(target_arch = "wasm32")] { if let Some(content) = common::web_store::get(SETTINGS_KEY)? { - return Ok(serde_json::from_str(&content)?); + let settings: Settings = serde_json::from_str(&content) + .map_err(|error| anyhow!("failed to parse saved settings: {error}"))?; + return Ok(Some(settings)); } } - Err(anyhow!("Not found")) + Ok(None) }); cx.spawn(async move |this, cx| { - let mut settings = task.await.unwrap_or(Settings::default()); + let (mut settings, load_failed) = match task.await { + Ok(settings) => (settings.unwrap_or_default(), false), + Err(error) => { + log::error!("failed to load settings, keeping defaults: {error:#}"); + (Settings::default(), true) + } + }; // Move settings still pointed at the old default file server over to the new one if settings.file_server.host_str() == Some(LEGACY_FILE_SERVER) { @@ -409,31 +438,63 @@ impl AppSettings { } // Update settings - this.update(cx, |this, cx| { + if let Err(error) = this.update(cx, |this, cx| { + // A failed load must not trigger the save that would overwrite the file + this.suppress_next_save = load_failed; this.set_settings(settings, cx); this.apply_theme(cx); cx.refresh_windows(); - }) - .ok(); + }) { + log::error!("failed to apply loaded settings: {error:#}"); + } }) .detach(); } /// Save settings pub fn save(&mut self, cx: &mut Context) { - let settings = self.inner.read(cx); - if let Ok(content) = serde_json::to_string(&settings) { - #[cfg(not(target_arch = "wasm32"))] - cx.background_spawn(async move { - let path = config_dir().join(".settings"); - smol::fs::write(&path, content).await.ok(); - }) - .detach(); + // Consume the failed-load suppression, so the next user change saves again + if self.suppress_next_save { + self.suppress_next_save = false; + return; + } - #[cfg(target_arch = "wasm32")] - if let Err(e) = common::web_store::set(SETTINGS_KEY, &content) { - log::warn!("failed to persist settings: {e}"); + let settings = self.inner.read(cx); + let content = match serde_json::to_string(settings) { + Ok(content) => content, + Err(error) => { + log::error!("failed to serialize settings: {error:#}"); + return; } + }; + + #[cfg(not(target_arch = "wasm32"))] + { + let previous_task = self.save_task.take(); + self.save_task = Some(cx.background_spawn(async move { + // Await the previous write, so saves to the file complete in order + if let Some(previous_task) = previous_task { + previous_task.await; + } + + let path = config_dir().join(".settings"); + let temporary_path = config_dir().join(".settings.tmp"); + + let result: std::io::Result<()> = async { + smol::fs::write(&temporary_path, &content).await?; + smol::fs::rename(&temporary_path, &path).await + } + .await; + + if let Err(error) = result { + log::error!("failed to save settings to {}: {error}", path.display()); + } + })); + } + + #[cfg(target_arch = "wasm32")] + if let Err(error) = common::web_store::set(SETTINGS_KEY, &content) { + log::error!("failed to persist settings: {error:#}"); } } diff --git a/crates/speech/src/parakeet.rs b/crates/speech/src/parakeet.rs index 6c19229b..fe7c38fa 100644 --- a/crates/speech/src/parakeet.rs +++ b/crates/speech/src/parakeet.rs @@ -1,5 +1,5 @@ use std::path::PathBuf; -use std::sync::OnceLock; +use std::sync::{Mutex, OnceLock}; use std::time::Duration; use anyhow::{Context as _, Result, anyhow}; @@ -30,7 +30,7 @@ impl SpeechRecognizer for ParakeetRecognizer { cx: &mut App, ) -> Result, SpeechError> { if super::model::installed() { - worker().send_or_log(Command::Warm(super::model::dir())); + send(Command::Warm(super::model::dir())).map_err(SpeechError::recognizer)?; } else if super::model::status().is_downloading() { return Err(SpeechError::recognizer(anyhow!( "The speech model is still downloading. Try again when it finishes." @@ -79,7 +79,11 @@ impl RecognitionSession for ParakeetSession { } let (sender, receiver) = flume::unbounded(); - worker().send_or_log(Command::Transcribe { samples, sender }); + + if let Err(error) = send(Command::Transcribe { samples, sender }) { + sink.error(SpeechError::recognizer(error), cx); + return; + } self._result = Some(cx.spawn(async move |cx| { let result = receiver.recv_async().await; @@ -109,23 +113,50 @@ enum Command { Unload, } -/// A handle to the single inference worker. -fn worker() -> &'static flume::Sender { - static WORKER: OnceLock> = OnceLock::new(); +fn worker_slot() -> &'static Mutex>> { + static WORKER: OnceLock>>> = OnceLock::new(); + WORKER.get_or_init(|| Mutex::new(None)) +} - WORKER.get_or_init(|| { - let (sender, receiver) = flume::unbounded::(); +/// A handle to the live worker, spawning a replacement after an idle exit. +fn worker() -> Result> { + let mut slot = worker_slot() + .lock() + .unwrap_or_else(|error| error.into_inner()); - let spawned = std::thread::Builder::new() - .name("speech-parakeet".into()) - .spawn(move || run_worker(receiver)); + if slot.as_ref().is_some_and(|sender| sender.is_disconnected()) { + *slot = None; + } - if let Err(error) = spawned { - log::error!("failed to spawn the speech worker: {error}"); + if let Some(sender) = slot.as_ref() { + return Ok(sender.clone()); + } + + let (sender, receiver) = flume::unbounded::(); + + match std::thread::Builder::new() + .name("speech-parakeet".into()) + .spawn(move || run_worker(receiver)) + { + Ok(_) => { + *slot = Some(sender.clone()); + Ok(sender) } + Err(error) => { + log::error!("failed to spawn the speech worker: {error}"); + Err(anyhow!("Could not start the speech worker: {error}")) + } + } +} - sender - }) +/// Deliver a command, retrying once on a fresh worker if the previous one exited. +fn send(command: Command) -> Result<()> { + let command = match worker()?.send(command) { + Ok(()) => return Ok(()), + Err(error) => error.into_inner(), + }; + // The worker can exit between the liveness check and the send, so retry once. + worker()?.send(command).context("the speech worker exited") } fn run_worker(receiver: flume::Receiver) { @@ -192,21 +223,11 @@ fn transcribe(cached: &mut Option<(PathBuf, ParakeetTDT)>, samples: Vec) -> .map_err(|error| anyhow!("Could not transcribe this recording: {error}")) } -trait SendOrLog { - fn send_or_log(&self, command: Command); -} - -impl SendOrLog for flume::Sender { - fn send_or_log(&self, command: Command) { - if self.send(command).is_err() { - log::debug!("speech worker is gone; command dropped"); - } - } -} - /// Retire cached weights, e.g. after the model was deleted. pub(crate) fn unload() { - worker().send_or_log(Command::Unload); + if let Err(error) = send(Command::Unload) { + log::warn!("failed to unload the speech model: {error}"); + } } fn flatten_panic(result: std::thread::Result) -> Result { diff --git a/crates/state/Cargo.toml b/crates/state/Cargo.toml index 116a721d..af27e9d0 100644 --- a/crates/state/Cargo.toml +++ b/crates/state/Cargo.toml @@ -51,6 +51,7 @@ sqlite-wasm-vfs = "0.2" [target.'cfg(not(target_arch = "wasm32"))'.dependencies] browser-signer-proxy = { path = "../browser-signer-proxy" } nostr-lmdb.workspace = true +nostr-memory.workspace = true smol.workspace = true gpui_tokio.workspace = true rustls = "0.23" diff --git a/crates/state/src/file.rs b/crates/state/src/file.rs index 3f7f6e87..2aba636e 100644 --- a/crates/state/src/file.rs +++ b/crates/state/src/file.rs @@ -403,7 +403,7 @@ pub async fn download_and_decrypt_file( .clone() .unwrap_or_else(|| fallback_name(&file.url)); - let local = store_file(&name, &file.mime, &data).await?; + let local = store_file(&file.mime, &data).await?; Ok(DownloadedFile { local, @@ -415,10 +415,11 @@ pub async fn download_and_decrypt_file( } /// Persist decrypted bytes somewhere the platform can render or open them -async fn store_file(name: &str, mime: &str, data: &[u8]) -> Result { +async fn store_file(mime: &str, data: &[u8]) -> Result { #[cfg(not(target_arch = "wasm32"))] { - let digest = sha256_hex(name.as_bytes()); + // Key the cache entry on the content hash, so same-named attachments can't collide + let digest = sha256_hex(data); let extension = mime_guess::get_mime_extensions_str(mime) .and_then(|extensions| extensions.first()) .copied() @@ -428,24 +429,21 @@ async fn store_file(name: &str, mime: &str, data: &[u8]) -> Result (Arc::new(database) as Arc, None), + Err(e) => { + log::error!("Failed to open the local database, falling back to memory: {e}"); + + ( + Arc::new(MemoryDatabase::unbounded()) as Arc, + Some(format!( + "Local database unavailable, using an in-memory database instead: {e}" + )), + ) + } + }; #[cfg(target_arch = "wasm32")] - let database = web_database::take(); + let (database, database_error) = (web_database::take(), None::); // Construct the nostr client let client = ClientBuilder::default() @@ -141,6 +155,10 @@ impl NostrRegistry { // Connect to bootstrap relays once the registry has been returned to the app cx.defer(move |cx| { let _ = entity.update(cx, |this, cx| { + if let Some(error) = database_error { + cx.emit(StateEvent::error(error)); + } + this.connect_bootstrap_relays(cx); if let Some(secret) = cli_key { @@ -445,8 +463,14 @@ impl NostrRegistry { }; let keys = Keys::generate(); - let credential = with_master_key(&uri_string, &keys); - let write = credentials::write(cx, "bunker", credential.as_bytes()); + + // The URI embeds the client key, so never persist it where web stores it unencrypted + #[cfg(not(target_arch = "wasm32"))] + let write = + credentials::write(cx, "bunker", with_master_key(&uri_string, &keys).as_bytes()); + + #[cfg(target_arch = "wasm32")] + let write = Task::ready(Ok::<(), Error>(())); cx.spawn(async move |this, cx| { let signer = NostrConnect::builder(connect_uri, keys) @@ -709,6 +733,7 @@ fn extract_master_key(credential: &str) -> (&str, Keys) { } // Embed the master key in a bunker URI so reconnects reuse the same local key +#[cfg(not(target_arch = "wasm32"))] fn with_master_key(uri: &str, keys: &Keys) -> String { let separator = if uri.contains('?') { '&' } else { '?' }; let nsec = keys.secret_key().to_bech32().expect("infallible"); diff --git a/crates/workspace/src/panels/contact_list.rs b/crates/workspace/src/panels/contact_list.rs index 059eb876..1e6a3c57 100644 --- a/crates/workspace/src/panels/contact_list.rs +++ b/crates/workspace/src/panels/contact_list.rs @@ -29,26 +29,17 @@ pub fn init(window: &mut Window, cx: &mut App) -> Entity { pub struct ContactListPanel { name: SharedString, focus_handle: FocusHandle, - /// Npub input input: Entity, - /// Whether the panel is updating updating: bool, - /// Error message error: Option, - /// All contacts - contacts: HashSet, - + contacts: HashSet, /// Person directory for the contacts directory: Entity, - - /// Event subscriptions _subscriptions: SmallVec<[Subscription; 1]>, - - /// Background tasks tasks: Vec>>, } @@ -92,33 +83,50 @@ impl ContactListPanel { return; }; - let task: Task, Error>> = cx.background_spawn(async move { + let task: Task, Error>> = cx.background_spawn(async move { let filter = Filter::new() .author(public_key) .kind(Kind::ContactList) .limit(1); - let contact_list: HashSet = client + let contact_list: HashSet = client .database() .query(filter) .await? .into_iter() .next() - .map(|event| event.tags.public_keys().collect()) + .map(|event| { + event + .tags + .iter() + .filter_map(|tag| match Nip02Tag::try_from(tag) { + Ok(Nip02Tag::PublicKey { + public_key, + relay_hint, + alias, + }) => Some(Contact { + public_key, + relay_url: relay_hint, + alias, + }), + Err(_) => None, + }) + .collect() + }) .unwrap_or_default(); Ok(contact_list) }); self.tasks.push(cx.spawn_in(window, async move |this, cx| { - let public_keys = task.await?; + let contacts = task.await?; // Update state this.update(cx, |this, cx| { this.directory.update(cx, |directory, cx| { - directory.track(public_keys.iter().copied(), cx) + directory.track(contacts.iter().map(|contact| contact.public_key), cx) }); - this.contacts.extend(public_keys); + this.contacts.extend(contacts); cx.notify(); })?; @@ -130,7 +138,7 @@ impl ContactListPanel { let value = self.input.read(cx).value().to_string(); if let Ok(public_key) = PublicKey::parse(&value) { - if self.contacts.insert(public_key) { + if self.contacts.insert(Contact::new(public_key)) { self.directory .update(cx, |directory, cx| directory.track([public_key], cx)); self.input.update(cx, |this, cx| { @@ -144,7 +152,8 @@ impl ContactListPanel { } fn remove(&mut self, public_key: &PublicKey, cx: &mut Context) { - self.contacts.remove(public_key); + self.contacts + .retain(|contact| contact.public_key != *public_key); cx.notify(); } @@ -184,11 +193,7 @@ impl ContactListPanel { let signer = nostr.read(cx).signer(); // Get contacts - let contacts: Vec = self - .contacts - .iter() - .map(|public_key| Contact::new(*public_key)) - .collect(); + let contacts: Vec = self.contacts.iter().cloned().collect(); // Set updating state self.set_updating(true, cx); @@ -230,8 +235,9 @@ impl ContactListPanel { fn render_list_items(&mut self, cx: &mut Context) -> Vec { let mut items = Vec::new(); - for (ix, public_key) in self.contacts.iter().enumerate() { - let profile = self.directory.read(cx).person(public_key, cx); + for (ix, contact) in self.contacts.iter().enumerate() { + let public_key = contact.public_key; + let profile = self.directory.read(cx).person(&public_key, cx); items.push( h_flex() @@ -262,12 +268,9 @@ impl ContactListPanel { .ghost() .invisible() .group_hover("", |this| this.visible()) - .on_click({ - let public_key = public_key.to_owned(); - cx.listener(move |this, _ev, _window, cx| { - this.remove(&public_key, cx); - }) - }), + .on_click(cx.listener(move |this, _ev, _window, cx| { + this.remove(&public_key, cx); + })), ), ) } diff --git a/docs/audit-stable-release.md b/docs/audit-stable-release.md index 826ca3db..bd59ffec 100644 --- a/docs/audit-stable-release.md +++ b/docs/audit-stable-release.md @@ -2,6 +2,10 @@ Date: 2026-11-12 · Revision audited: `8fa79aef` (master) · Scope: entire workspace (~35k lines, 18 crates + `desktop` + `web`) +> **Remediation status (2026-11-12, later the same day):** all ten P0 items from §8 have been +> implemented and validated (workspace clippy clean, `cargo fmt --check` clean, 35 tests passing). +> See §9 for the per-fix log. + ## 1. Scope and Method Every crate was audited file-by-file against the project conventions (`AGENTS.md`), Rust best @@ -823,3 +827,81 @@ claims were verified against the vendored `gpui-pre 0.3.12`, `gpui-base 0.9.1`, `gpui-component 0.9.1`, and rust-nostr sources; the Critical and top-High findings were independently re-verified by hand against the working tree at `8fa79aef`. No files were modified during the audit.* + +## 9. P0 Remediation Log (2026-11-12) + +All ten P0 items fixed. Validation after the full batch: `cargo clippy --workspace --all-targets` +zero warnings, `cargo fmt --all --check` clean, `cargo test --workspace` 35 passed / 0 failed. +Diff: 15 files, +816/−273. + +1. **nip60 deletion spoofing** — `nip60/src/lib.rs`: precedence fixed to + `ours && (is_cashu || is_token_deletion(..))`; `absorb`/`absorb_deletion` now take `me` and + reject foreign-authored kind 5 events (defense in depth; the live notification gate was the + only protection before, and `absorb_deletion` had no author check). The DB load path was + already author-filtered and is unchanged. +2. **browser-signer-proxy hardening** — per-session UUID token generated at startup, embedded in + the served page (`inject_session_token`), required on `/api/*` (403 otherwise); exact `Host` + match (DNS-rebinding defense); all CORS headers and the OPTIONS preflight removed; + `cancel_request` removes timed-out requests from both maps so plaintext is dropped; + centralized best-effort `write_response`. `proxy.js` authenticates all three fetches. Token + never appears in the URL the app opens. +3. **state startup crash loop** — `NostrLmdb::open` failure now logs and falls back to + `MemoryDatabase::unbounded()` (nostr-memory added to the desktop deps), emitting + `StateEvent::error` via the existing deferred path; the `block_on` stays (safe: the LMDB open + schedules nothing back onto the GPUI executor — now commented). +4. **state web localStorage key** — `import_bunker` on wasm32 no longer persists the bunker + credential (session is memory-only; desktop keyring unchanged); `restore_session` degrades to + `NoSigner` when nothing is stored (verified). **Known behavior change: web users re-enter their + bunker URI after each reload.** +5. **state file-cache collision** — cache filename now derived from the decrypted content's + sha256 instead of the filename; the metadata guard is gone and every store goes through + temp-write + atomic rename. +6. **nip60 reconcile money-safety** — `ops.rs::reconcile` rewritten: relay tokens are converged + into the store BEFORE any publish (closes the `del`-before-converge race); relay proofs already + tracked locally (unspent or in-flight Pending/Reserved/PendingSpent) are never re-added, and + unseen proofs are checked against the mint (NUT-08 `check_proofs_spent`, best-effort) before + being accepted as Unspent; when a prior job errored after mutating the store + (`publish_failed` flag), reconcile republishes the store's unspent set instead of trusting + relay state — healing stranded sends/receives; without the flag the relay-trust removal path + (cross-device spend convergence) is preserved. A failing Reconcile does not reschedule itself + (no infinite loop). +7. **nip60 stuck nutzap redemptions** — `apply_outcome` now receives the failed job; on + `RedeemNutzap` error the id is removed from `redeeming` and requeued up to + `MAX_REDEEM_ATTEMPTS = 3` (cleared on success, reset on signer change), so transient mint + failures no longer wedge redemptions until restart. +8. **Reactions double-count** — `chat_ui::insert_reaction` replaces per (author, target) per + NIP-25, making the backup gift-wrap echo idempotent (verified: optimistic rumor and the + re-delivered seal carry the same event id) and history reloads non-duplicating. +9. **Whitespace collapse** — `common/media_extractor.rs` rewritten: whitespace is tidied only at + URL-removal boundaries; line breaks in untouched regions survive; regex hoisted to + `LazyLock`; 9 unit tests added (multi-line preservation, URL-only messages, digit/emoji edges). +10. **Contact-list data loss** — `contact_list.rs` now parses the stored event into full + `nostr::nips::nip02::Contact` values (relay hint + alias preserved through load → add/remove + → republish); new contacts still get `Contact::new`. +11. **Speech worker lifetime** — `parakeet.rs`: the `OnceLock` became a resettable + `Mutex>`; a dead worker (30 s idle exit) is detected via `is_disconnected`, the + slot cleared, and a replacement thread spawned lazily on the next command; spawn failure no + longer caches a dead sender; failures surface through `sink.error` / returned `Err` instead + of a debug-level swallow (one retry covers the liveness-check/send race). +12. **Settings durability** — a failed load logs and keeps defaults in memory without + persisting (`suppress_next_save`, consumed by the next real change); fresh-install NotFound is + no longer conflated with read/parse failures; saves are serialized through a single + `save_task` slot (chained, FIFO order) with temp-file + atomic rename; serialization and + write failures are logged. +13. **nip29 ordering guards** — shared `is_newer_event` (strict `(created_at, id)` total order): + `Group::absorb_state` keeps a per-kind stamp (written only after a successful apply, so a + malformed newer event can't poison the guard) and rejects stale replays while ascending cache + hydration still applies; the registry's `list_stamp` seeds from the cached list event at + startup, is checked in `merge` before applying, updated by `publish_list` (closing the + leave→resurrect window), and reset on signer change. 3 new tests across protocol/group. +14. **nip60_ui invoice flow** — *not* in this batch: it is P1 (the audit lists it as High but the + P0 list ranks it 16th via `nip60_ui/wallet.rs`); `mint_quotes()`/`claim_mint_quote()` remain + dead until the P1 UI work lands. No other P0 code paths depend on it. + +Behavior changes to be aware of when testing: +- Web: bunker credentials are no longer persisted (re-connect after each reload). +- Desktop: a corrupted/unreadable local database now starts with an empty in-memory history and + shows an error event instead of crash-looping; relays re-sync as before. +- nip60: reconcile now performs one NUT-08 mint round-trip when unseen relay proofs appear. +- Reactions render per NIP-25 replace semantics (one reaction per author per target). +- Messages keep their line breaks; only URL-adjacent whitespace is tidied.