build: bump version to 0.7.7

Cargo.lock

@@ -1096,7 +1096,7 @@ checksum = "38bf9645c8b145698bb0b18a4637dcacbc421ea49bef2317e4fd8065a387cf21"
 
 [[package]]
 name = "nostr-rs-relay"
-version = "0.7.5"
+version = "0.7.7"
 dependencies = [
  "anyhow",
  "bitcoin_hashes",

Cargo.toml

@@ -1,6 +1,6 @@
 [package]
 name = "nostr-rs-relay"
-version = "0.7.5"
+version = "0.7.7"
 edition = "2021"
 authors = ["Greg Heartsfield <scsibug@imap.cc>"]
 description = "A relay implementation for the Nostr protocol"

config.toml

@@ -65,6 +65,11 @@ reject_future_seconds = 1800
 # an integer.  If not set (or set to 0), defaults to unlimited.
 #messages_per_sec = 0
 
+# Limit client subscriptions created per second, averaged over one
+# minute.  Must be an integer.  If not set (or set to 0), defaults to
+# unlimited.
+#subscriptions_per_min = 0
+
 # Limit the maximum size of an EVENT message.  Defaults to 128 KB.
 # Set to 0 for unlimited.
 #max_event_bytes = 131072

src/config.rs

@@ -52,6 +52,7 @@ pub struct Retention {
 #[allow(unused)]
 pub struct Limits {
     pub messages_per_sec: Option<u32>, // Artificially slow down event writing to limit disk consumption (averaged over 1 minute)
+    pub subscriptions_per_min: Option<u32>, // Artificially slow down request (db query) creation to prevent abuse (averaged over 1 minute)
     pub max_event_bytes: Option<usize>,     // Maximum size of an EVENT message
     pub max_ws_message_bytes: Option<usize>,
     pub max_ws_frame_bytes: Option<usize>,
@@ -214,6 +215,7 @@ impl Default for Settings {
             },
             limits: Limits {
                 messages_per_sec: None,
+                subscriptions_per_min: None,
                 max_event_bytes: Some(2 << 17),      // 128K
                 max_ws_message_bytes: Some(2 << 17), // 128K
                 max_ws_frame_bytes: Some(2 << 17),   // 128K

src/db.rs

@@ -428,14 +428,13 @@ fn query_from_filter(f: &ReqFilter) -> (String, Vec<Box<dyn ToSql>>) {
 
     // if the filter is malformed, don't return anything.
     if f.force_no_match {
-        let empty_query =
-            "SELECT DISTINCT(e.content), e.created_at FROM event e WHERE 1=0".to_owned();
+        let empty_query = "SELECT e.content, e.created_at FROM event e WHERE 1=0".to_owned();
         // query parameters for SQLite
         let empty_params: Vec<Box<dyn ToSql>> = vec![];
         return (empty_query, empty_params);
     }
 
-    let mut query = "SELECT DISTINCT(e.content), e.created_at FROM event e".to_owned();
+    let mut query = "SELECT e.content, e.created_at FROM event e".to_owned();
     // query parameters for SQLite
     let mut params: Vec<Box<dyn ToSql>> = vec![];
 
@@ -591,7 +590,7 @@ fn query_from_sub(sub: &Subscription) -> (String, Vec<Box<dyn ToSql>>) {
     // encapsulate subqueries into select statements
     let subqueries_selects: Vec<String> = subqueries
         .iter()
-        .map(|s| format!("SELECT content, created_at FROM ({})", s))
+        .map(|s| format!("SELECT distinct content, created_at FROM ({})", s))
         .collect();
     let query: String = subqueries_selects.join(" UNION ");
     (query, params)

src/hexrange.rs

@@ -3,7 +3,7 @@ use crate::utils::is_hex;
 use hex;
 
 /// Types of hexadecimal queries.
-#[derive(PartialEq, Eq, Debug, Clone)]
+#[derive(PartialEq, Eq, PartialOrd, Ord, Debug, Clone)]
 pub enum HexSearch {
     // when no range is needed, exact 32-byte
     Exact(Vec<u8>),

src/server.rs

@@ -14,6 +14,7 @@ use crate::notice::Notice;
 use crate::subscription::Subscription;
 use futures::SinkExt;
 use futures::StreamExt;
+use governor::{Jitter, Quota, RateLimiter};
 use http::header::HeaderMap;
 use hyper::header::ACCEPT;
 use hyper::service::{make_service_fn, service_fn};
@@ -438,6 +439,19 @@ async fn nostr_server(
     let mut bcast_rx = broadcast.subscribe();
     // Track internal client state
     let mut conn = conn::ClientConn::new(client_info.remote_ip);
+    // subscription creation rate limiting
+    let mut sub_lim_opt = None;
+    // 100ms jitter when the rate limiter returns
+    let jitter = Jitter::up_to(Duration::from_millis(100));
+    let sub_per_min_setting = settings.limits.subscriptions_per_min;
+    if let Some(sub_per_min) = sub_per_min_setting {
+        if sub_per_min > 0 {
+            trace!("Rate limits for sub creation ({}/min)", sub_per_min);
+            let quota_time = core::num::NonZeroU32::new(sub_per_min).unwrap();
+            let quota = Quota::per_minute(quota_time);
+            sub_lim_opt = Some(RateLimiter::direct(quota));
+        }
+    }
     // Use the remote IP as the client identifier
     let cid = conn.get_client_prefix();
     // Create a channel for receiving query results from the database.
@@ -606,11 +620,15 @@ async fn nostr_server(
                     Ok(NostrMessage::SubMsg(s)) => {
                         debug!("subscription requested (cid: {}, sub: {:?})", cid, s.id);
                         // subscription handling consists of:
+                        // * check for rate limits
                         // * registering the subscription so future events can be matched
                         // * making a channel to cancel to request later
                         // * sending a request for a SQL query
             // Do nothing if the sub already exists.
             if !current_subs.contains(&s) {
+                if let Some(ref lim) = sub_lim_opt {
+                    lim.until_ready_with_jitter(jitter).await;
+                }
                 current_subs.push(s.clone());
                             let (abandon_query_tx, abandon_query_rx) = oneshot::channel::<()>();
                             match conn.subscribe(s.clone()) {