build: bump version to 0.5.1

Cargo.lock

@@ -723,7 +723,7 @@ checksum = "38bf9645c8b145698bb0b18a4637dcacbc421ea49bef2317e4fd8065a387cf21"
 
 [[package]]
 name = "nostr-rs-relay"
-version = "0.5.0"
+version = "0.5.1"
 dependencies = [
  "bitcoin_hashes 0.9.7",
  "config",

Cargo.toml

@@ -1,6 +1,6 @@
 [package]
 name = "nostr-rs-relay"
-version = "0.5.0"
+version = "0.5.1"
 edition = "2021"
 
 [dependencies]

README.md

@@ -17,7 +17,7 @@ NIPs with a relay-specific implementation are listed here.
 - [x] NIP-01: Id/Author prefix search (_experimental_)
 - [x] NIP-02: Hide old contact list events
 - [ ] NIP-03: OpenTimestamps
-- [ ] NIP-05: Mapping Nostr keys to DNS identifiers
+- [x] NIP-05: Mapping Nostr keys to DNS identifiers
 - [ ] NIP-09: Event deletion
 - [x] NIP-11: Relay information document
 - [x] NIP-12: Generic tag search (_experimental_)

src/db.rs

@@ -27,6 +27,13 @@ use tokio::task;
 
 pub type SqlitePool = r2d2::Pool<r2d2_sqlite::SqliteConnectionManager>;
 pub type PooledConnection = r2d2::PooledConnection<r2d2_sqlite::SqliteConnectionManager>;
+
+/// Events submitted from a client, with a return channel for notices
+pub struct SubmittedEvent {
+    pub event: Event,
+    pub notice_tx: tokio::sync::mpsc::Sender<String>,
+}
+
 /// Database file
 pub const DB_FILE: &str = "nostr.db";
 
@@ -76,7 +83,7 @@ pub fn build_conn(flags: OpenFlags) -> Result<Connection> {
 
 /// Spawn a database writer that persists events to the SQLite store.
 pub async fn db_writer(
-    mut event_rx: tokio::sync::mpsc::Receiver<Event>,
+    mut event_rx: tokio::sync::mpsc::Receiver<SubmittedEvent>,
     bcast_tx: tokio::sync::broadcast::Sender<Event>,
     metadata_tx: tokio::sync::broadcast::Sender<Event>,
     mut shutdown: tokio::sync::broadcast::Receiver<()>,
@@ -131,18 +138,20 @@ pub async fn db_writer(
                 break;
             }
             let mut event_write = false;
-            let event = next_event.unwrap();
-
+            let subm_event = next_event.unwrap();
+            let event = subm_event.event;
+            let notice_tx = subm_event.notice_tx;
             // check if this event is authorized.
             if let Some(allowed_addrs) = whitelist {
-                debug!("Checking against pubkey whitelist");
                 // if the event address is not in allowed_addrs.
                 if !allowed_addrs.contains(&event.pubkey) {
                     info!(
                         "Rejecting event {}, unauthorized author",
                         event.get_event_id_prefix()
                     );
-                    // TODO: define a channel that can send NOTICEs back to the client.
+                    notice_tx
+                        .try_send("pubkey is not allowed to publish to this relay".to_owned())
+                        .ok();
                     continue;
                 }
             }
@@ -171,6 +180,12 @@ pub async fn db_writer(
                                   uv.name.to_string(),
                                   event.get_author_prefix()
                             );
+                            notice_tx
+                                .try_send(
+                                    "NIP-05 verification is no longer valid (expired/wrong domain)"
+                                        .to_owned(),
+                                )
+                                .ok();
                             continue;
                         }
                     }
@@ -179,6 +194,9 @@ pub async fn db_writer(
                             "no verification records found for pubkey: {:?}",
                             event.get_author_prefix()
                         );
+                        notice_tx
+                            .try_send("NIP-05 verification needed to publish events".to_owned())
+                            .ok();
                         continue;
                     }
                     Err(e) => {
@@ -207,6 +225,12 @@ pub async fn db_writer(
                 }
                 Err(err) => {
                     warn!("event insert failed: {:?}", err);
+                    notice_tx
+                        .try_send(
+                            "relay experienced an error trying to publish the latest event"
+                                .to_owned(),
+                        )
+                        .ok();
                 }
             }
 

src/lib.rs

@@ -2,12 +2,11 @@ pub mod close;
 pub mod config;
 pub mod conn;
 pub mod db;
-pub mod schema;
 pub mod error;
 pub mod event;
 pub mod hexrange;
 pub mod info;
 pub mod nip05;
-pub mod protostream;
+pub mod schema;
 pub mod subscription;
 pub mod utils;

src/main.rs

@@ -9,27 +9,33 @@ use hyper::{
 };
 use log::*;
 use nostr_rs_relay::close::Close;
+use nostr_rs_relay::close::CloseCmd;
 use nostr_rs_relay::config;
 use nostr_rs_relay::conn;
 use nostr_rs_relay::db;
+use nostr_rs_relay::db::SubmittedEvent;
 use nostr_rs_relay::error::{Error, Result};
 use nostr_rs_relay::event::Event;
+use nostr_rs_relay::event::EventCmd;
 use nostr_rs_relay::info::RelayInfo;
 use nostr_rs_relay::nip05;
-use nostr_rs_relay::protostream;
-use nostr_rs_relay::protostream::NostrMessage::*;
-use nostr_rs_relay::protostream::NostrResponse::*;
+use nostr_rs_relay::subscription::Subscription;
+use serde::{Deserialize, Serialize};
 use std::collections::HashMap;
 use std::convert::Infallible;
 use std::env;
 use std::net::SocketAddr;
 use std::path::Path;
+use std::time::Duration;
+use std::time::Instant;
 use tokio::runtime::Builder;
 use tokio::sync::broadcast::{self, Receiver, Sender};
 use tokio::sync::mpsc;
 use tokio::sync::oneshot;
 use tokio_tungstenite::WebSocketStream;
+use tungstenite::error::Error as WsError;
 use tungstenite::handshake;
+use tungstenite::protocol::Message;
 use tungstenite::protocol::WebSocketConfig;
 
 /// Return a requested DB name from command line arguments.
@@ -46,7 +52,7 @@ async fn handle_web_request(
     pool: db::SqlitePool,
     remote_addr: SocketAddr,
     broadcast: Sender<Event>,
-    event_tx: tokio::sync::mpsc::Sender<Event>,
+    event_tx: tokio::sync::mpsc::Sender<SubmittedEvent>,
     shutdown: Receiver<()>,
 ) -> Result<Response<Body>, Infallible> {
     match (
@@ -227,7 +233,8 @@ fn main() -> Result<(), Error> {
         let (bcast_tx, _) = broadcast::channel::<Event>(settings.limits.broadcast_buffer);
         // validated events that need to be persisted are sent to the
         // database on via this channel.
-        let (event_tx, event_rx) = mpsc::channel::<Event>(settings.limits.event_persist_buffer);
+        let (event_tx, event_rx) =
+            mpsc::channel::<SubmittedEvent>(settings.limits.event_persist_buffer);
         // establish a channel for letting all threads now about a
         // requested server shutdown.
         let (invoke_shutdown, shutdown_listen) = broadcast::channel::<()>(1);
@@ -312,31 +319,78 @@ fn main() -> Result<(), Error> {
     Ok(())
 }
 
+/// Nostr protocol messages from a client
+#[derive(Deserialize, Serialize, Clone, PartialEq, Debug)]
+#[serde(untagged)]
+pub enum NostrMessage {
+    /// An `EVENT` message
+    EventMsg(EventCmd),
+    /// A `REQ` message
+    SubMsg(Subscription),
+    /// A `CLOSE` message
+    CloseMsg(CloseCmd),
+}
+
+/// Convert Message to NostrMessage
+fn convert_to_msg(msg: String) -> Result<NostrMessage> {
+    let config = config::SETTINGS.read().unwrap();
+    let parsed_res: Result<NostrMessage> = serde_json::from_str(&msg).map_err(|e| e.into());
+    match parsed_res {
+        Ok(m) => {
+            if let NostrMessage::EventMsg(_) = m {
+                if let Some(max_size) = config.limits.max_event_bytes {
+                    // check length, ensure that some max size is set.
+                    if msg.len() > max_size && max_size > 0 {
+                        return Err(Error::EventMaxLengthError(msg.len()));
+                    }
+                }
+            }
+            Ok(m)
+        }
+        Err(e) => {
+            debug!("proto parse error: {:?}", e);
+            debug!("parse error on message: {}", msg.trim());
+            Err(Error::ProtoParseError)
+        }
+    }
+}
+
 /// Handle new client connections.  This runs through an event loop
 /// for all client communication.
 async fn nostr_server(
     pool: db::SqlitePool,
-    ws_stream: WebSocketStream<Upgraded>,
+    mut ws_stream: WebSocketStream<Upgraded>,
     broadcast: Sender<Event>,
-    event_tx: tokio::sync::mpsc::Sender<Event>,
+    event_tx: mpsc::Sender<SubmittedEvent>,
     mut shutdown: Receiver<()>,
 ) {
     // get a broadcast channel for clients to communicate on
     let mut bcast_rx = broadcast.subscribe();
-    // upgrade the TCP connection to WebSocket
-    //let conn = tokio_tungstenite::accept_async_with_config(stream, Some(config)).await;
-    //let ws_stream = conn.expect("websocket handshake error");
-    // wrap websocket into a stream & sink of Nostr protocol messages
-    let mut nostr_stream = protostream::wrap_ws_in_nostr(ws_stream);
     // Track internal client state
     let mut conn = conn::ClientConn::new();
     let cid = conn.get_client_prefix();
     // Create a channel for receiving query results from the database.
     // we will send out the tx handle to any query we generate.
     let (query_tx, mut query_rx) = mpsc::channel::<db::QueryResult>(256);
+    // Create channel for receiving NOTICEs
+    let (notice_tx, mut notice_rx) = mpsc::channel::<String>(32);
+
     // maintain a hashmap of a oneshot channel for active subscriptions.
     // when these subscriptions are cancelled, make a message
     // available to the executing query so it knows to stop.
+
+    // last time this client sent data
+    let mut last_message_time = Instant::now();
+
+    // ping interval (every 5 minutes)
+    let default_ping_dur = Duration::from_secs(300);
+
+    // disconnect after 20 minutes without a ping response or event.
+    let max_quiet_time = Duration::from_secs(60 * 20);
+
+    let start = tokio::time::Instant::now() + default_ping_dur;
+    let mut ping_interval = tokio::time::interval_at(start, default_ping_dur);
+
     let mut running_queries: HashMap<String, oneshot::Sender<()>> = HashMap::new();
     // for stats, keep track of how many events the client published,
     // and how many it received from queries.
@@ -349,11 +403,27 @@ async fn nostr_server(
                 // server shutting down, exit loop
                 break;
             },
+            _ = ping_interval.tick() => {
+                // check how long since we talked to client
+                // if it has been too long, disconnect
+                if last_message_time.elapsed() > max_quiet_time {
+                    debug!("ending connection due to lack of client ping response");
+                    break;
+                }
+                // Send a ping
+                ws_stream.send(Message::Ping(Vec::new())).await.ok();
+            },
+            Some(notice_msg) = notice_rx.recv() => {
+                let n = notice_msg.to_string().replace("\"", "");
+                ws_stream.send(Message::Text(format!("[\"NOTICE\",\"{}\"]", n))).await.ok();
+            },
             Some(query_result) = query_rx.recv() => {
                 // database informed us of a query result we asked for
-                let res = EventRes(query_result.sub_id,query_result.event);
                 client_received_event_count += 1;
-                nostr_stream.send(res).await.ok();
+                // send a result
+                let subesc = query_result.sub_id.replace("\"", "");
+                let send_str = format!("[\"EVENT\",\"{}\",{}]", subesc, &query_result.event);
+                ws_stream.send(Message::Text(send_str)).await.ok();
             },
             // TODO: consider logging the LaggedRecv error
             Ok(global_event) = bcast_rx.recv() => {
@@ -368,17 +438,43 @@ async fn nostr_server(
                                cid, s,
                                global_event.get_event_id_prefix());
                         // create an event response and send it
-                        let res = EventRes(s.to_owned(),event_str);
-                        nostr_stream.send(res).await.ok();
+                        let subesc = s.replace("\"", "");
+                        ws_stream.send(Message::Text(format!("[\"EVENT\",\"{}\",{}]", subesc, event_str))).await.ok();
+                        //nostr_stream.send(res).await.ok();
                     } else {
                         warn!("could not serialize event {:?}", global_event.get_event_id_prefix());
                     }
                 }
             },
-            // check if this client has a subscription
-            proto_next = nostr_stream.next() => {
-                match proto_next {
-                    Some(Ok(EventMsg(ec))) => {
+            ws_next = ws_stream.next() => {
+                // update most recent message time for client
+                last_message_time = Instant::now();
+                // Consume text messages from the client, parse into Nostr messages.
+                let nostr_msg = match ws_next {
+                    Some(Ok(Message::Text(m))) => {
+                        convert_to_msg(m)
+                    },
+                    Some(Ok(Message::Binary(_))) => {
+                        ws_stream.send(Message::Text(format!("[\"NOTICE\",\"{}\"]", "binary messages are not accepted"))).await.ok();
+                        continue;
+                    },
+                    Some(Ok(Message::Ping(_))) | Some(Ok(Message::Pong(_))) => {
+                        // get a ping/pong, ignore
+                        continue;
+                    },
+                    None | Some(Ok(Message::Close(_))) | Some(Err(WsError::AlreadyClosed)) | Some(Err(WsError::ConnectionClosed))  => {
+                        debug!("normal websocket close from client: {:?}",cid);
+                        break;
+                    },
+                    x => {
+                        info!("message was: {:?} (ignoring)", x);
+                        continue;
+                    }
+                };
+
+                // convert ws_next into proto_next
+                match nostr_msg {
+                    Ok(NostrMessage::EventMsg(ec)) => {
                         // An EventCmd needs to be validated to be converted into an Event
                         // handle each type of message
                         let parsed : Result<Event> = Result::<Event>::from(ec);
@@ -386,23 +482,18 @@ async fn nostr_server(
                             Ok(e) => {
                                 let id_prefix:String = e.id.chars().take(8).collect();
                                 debug!("successfully parsed/validated event: {:?} from client: {:?}", id_prefix, cid);
-                                // TODO: consider moving some/all
-                                // authorization checks here, instead
-                                // of the DB module, so we can send a
-                                // proper NOTICE back to the client if
-                                // they are unable to write.
-
-                                // Write this to the database
-                                event_tx.send(e.clone()).await.ok();
+                                // Write this to the database.
+                                let submit_event = SubmittedEvent { event: e.clone(), notice_tx: notice_tx.clone() };
+                                event_tx.send(submit_event).await.ok();
                                 client_published_event_count += 1;
                             },
                             Err(_) => {
                                 info!("client {:?} sent an invalid event", cid);
-                                nostr_stream.send(NoticeRes("event was invalid".to_owned())).await.ok();
+                                ws_stream.send(Message::Text(format!("[\"NOTICE\",\"{}\"]", "event was invalid"))).await.ok();
                             }
                         }
                     },
-                    Some(Ok(SubMsg(s))) => {
+                    Ok(NostrMessage::SubMsg(s)) => {
                         debug!("client {} requesting a subscription", cid);
                         // subscription handling consists of:
                         // * registering the subscription so future events can be matched
@@ -422,12 +513,12 @@ async fn nostr_server(
                             },
                             Err(e) => {
                                 info!("Subscription error: {}", e);
-                                nostr_stream.send(NoticeRes(format!("{}",e))).await.ok();
-
+                                let s = e.to_string().replace("\"", "");
+                                ws_stream.send(Message::Text(format!("[\"NOTICE\",\"{}\"]", s))).await.ok();
                             }
                         }
                     },
-                    Some(Ok(CloseMsg(cc))) => {
+                    Ok(NostrMessage::CloseMsg(cc)) => {
                         // closing a request simply removes the subscription.
                         let parsed : Result<Close> = Result::<Close>::from(cc);
                         match parsed {
@@ -444,23 +535,23 @@ async fn nostr_server(
                             },
                             Err(_) => {
                                 info!("invalid command ignored");
-
+                                ws_stream.send(Message::Text(format!("[\"NOTICE\",\"{}\"]", "could not parse command"))).await.ok();
                             }
                         }
                     },
-                    None => {
-                        debug!("normal websocket close from client: {:?}",cid);
-                        break;
-                    },
-                    Some(Err(Error::ConnError)) => {
+                    Err(Error::ConnError) => {
                         debug!("got connection close/error, disconnecting client: {:?}",cid);
                         break;
                     }
-                    Some(Err(Error::EventMaxLengthError(s))) => {
+                    Err(Error::EventMaxLengthError(s)) => {
                         info!("client {:?} sent event larger ({} bytes) than max size", cid, s);
-                        nostr_stream.send(NoticeRes("event exceeded max size".to_owned())).await.ok();
+                        ws_stream.send(Message::Text(format!("[\"NOTICE\",\"{}\"]", "event exceeded max size"))).await.ok();
                     },
-                    Some(Err(e)) => {
+                    Err(Error::ProtoParseError) => {
+                        info!("client {:?} sent event that could not be parsed", cid);
+                        ws_stream.send(Message::Text(format!("[\"NOTICE\",\"{}\"]", "could not parse command"))).await.ok();
+                    },
+                    Err(e) => {
                         info!("got non-fatal error from client: {:?}, error: {:?}", cid, e);
                     },
                 }

src/protostream.rs

@@ -1,141 +0,0 @@
-//! Nostr protocol layered over WebSocket
-use crate::close::CloseCmd;
-use crate::config;
-use crate::error::{Error, Result};
-use crate::event::EventCmd;
-use crate::subscription::Subscription;
-use core::pin::Pin;
-use futures::sink::Sink;
-use futures::stream::Stream;
-use futures::task::Context;
-use futures::task::Poll;
-use hyper::upgrade::Upgraded;
-use log::*;
-use serde::{Deserialize, Serialize};
-use tokio_tungstenite::WebSocketStream;
-use tungstenite::error::Error as WsError;
-use tungstenite::protocol::Message;
-
-/// Nostr protocol messages from a client
-#[derive(Deserialize, Serialize, Clone, PartialEq, Debug)]
-#[serde(untagged)]
-pub enum NostrMessage {
-    /// An `EVENT` message
-    EventMsg(EventCmd),
-    /// A `REQ` message
-    SubMsg(Subscription),
-    /// A `CLOSE` message
-    CloseMsg(CloseCmd),
-}
-
-/// Nostr protocol messages from a relay/server
-#[derive(Deserialize, Serialize, Clone, PartialEq, Debug)]
-pub enum NostrResponse {
-    /// A `NOTICE` response
-    NoticeRes(String),
-    /// An `EVENT` response, composed of the subscription identifier,
-    /// and serialized event JSON
-    EventRes(String, String),
-}
-
-/// A Nostr protocol stream is layered on top of a Websocket stream.
-pub struct NostrStream {
-    ws_stream: WebSocketStream<Upgraded>,
-}
-
-/// Given a websocket, return a protocol stream wrapper.
-pub fn wrap_ws_in_nostr(ws: WebSocketStream<Upgraded>) -> NostrStream {
-    NostrStream { ws_stream: ws }
-}
-
-/// Implement the [`Stream`] interface to produce Nostr messages.
-impl Stream for NostrStream {
-    type Item = Result<NostrMessage>;
-    fn poll_next(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<Self::Item>> {
-        // get the configuration
-        /// Convert Message to NostrMessage
-        fn convert(msg: String) -> Result<NostrMessage> {
-            let config = config::SETTINGS.read().unwrap();
-            let parsed_res: Result<NostrMessage> = serde_json::from_str(&msg).map_err(|e| e.into());
-            match parsed_res {
-                Ok(m) => {
-                    if let NostrMessage::EventMsg(_) = m {
-                        if let Some(max_size) = config.limits.max_event_bytes {
-                            // check length, ensure that some max size is set.
-                            if msg.len() > max_size && max_size > 0 {
-                                return Err(Error::EventMaxLengthError(msg.len()));
-                            }
-                        }
-                    }
-                    Ok(m)
-                }
-                Err(e) => {
-                    debug!("proto parse error: {:?}", e);
-                    debug!("parse error on message: {}", msg.trim());
-                    Err(Error::ProtoParseError)
-                }
-            }
-        }
-        match Pin::new(&mut self.ws_stream).poll_next(cx) {
-            Poll::Pending => Poll::Pending,
-            Poll::Ready(None) => Poll::Ready(None),
-            Poll::Ready(Some(v)) => match v {
-                Ok(Message::Text(vs)) => Poll::Ready(Some(convert(vs))),
-                Ok(Message::Ping(x)) => {
-                    debug!("client ping ({:?})", x);
-                    //Pin::new(&mut self.ws_stream).start_send(Message::Pong(x));
-                    // TODO: restructure this so that Pongs work
-                    //Pin::new(&mut self.ws_stream).write_pending();
-                    //info!("sent pong");
-                    Poll::Pending
-                }
-                Ok(Message::Binary(_)) => Poll::Ready(Some(Err(Error::ProtoParseError))),
-                Ok(Message::Pong(_)) => Poll::Pending,
-                Ok(Message::Close(_)) => Poll::Ready(None),
-                Err(WsError::AlreadyClosed) | Err(WsError::ConnectionClosed) => Poll::Ready(None),
-                Err(_) => Poll::Ready(Some(Err(Error::ConnError))),
-            },
-        }
-    }
-}
-
-/// Implement the [`Sink`] interface to produce Nostr responses.
-impl Sink<NostrResponse> for NostrStream {
-    type Error = Error;
-
-    fn poll_ready(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Result<(), Self::Error>> {
-        // map the error type
-        match Pin::new(&mut self.ws_stream).poll_ready(cx) {
-            Poll::Ready(Ok(())) => Poll::Ready(Ok(())),
-            Poll::Ready(Err(_)) => Poll::Ready(Err(Error::ConnWriteError)),
-            Poll::Pending => Poll::Pending,
-        }
-    }
-
-    fn start_send(mut self: Pin<&mut Self>, item: NostrResponse) -> Result<(), Self::Error> {
-        // TODO: do real escaping for these - at least on NOTICE,
-        // which surely has some problems if arbitrary text is sent.
-        let send_str = match item {
-            NostrResponse::NoticeRes(msg) => {
-                let s = msg.replace("\"", "");
-                format!("[\"NOTICE\",\"{}\"]", s)
-            }
-            NostrResponse::EventRes(sub, eventstr) => {
-                let subesc = sub.replace("\"", "");
-                format!("[\"EVENT\",\"{}\",{}]", subesc, eventstr)
-            }
-        };
-        match Pin::new(&mut self.ws_stream).start_send(Message::Text(send_str)) {
-            Ok(()) => Ok(()),
-            Err(_) => Err(Error::ConnWriteError),
-        }
-    }
-
-    fn poll_flush(self: Pin<&mut Self>, _: &mut Context<'_>) -> Poll<Result<(), Self::Error>> {
-        Poll::Ready(Ok(()))
-    }
-
-    fn poll_close(self: Pin<&mut Self>, _: &mut Context<'_>) -> Poll<Result<(), Self::Error>> {
-        Poll::Ready(Ok(()))
-    }
-}