Revert "completely remove the behavior of replay attack detection"

This reverts commit 90f05a5.

We still need replay attack protection in the future, so the Context
is still required in the whole API design.

crates/shadowsocks-service/src/local/context.rs

@@ -7,6 +7,7 @@ use std::{net::IpAddr, time::Duration};
 #[cfg(feature = "local-dns")]
 use lru_time_cache::LruCache;
 use shadowsocks::{
+    config::ServerType,
     context::{Context, SharedContext},
     dns_resolver::DnsResolver,
     net::{AcceptOpts, ConnectOpts},
@@ -44,7 +45,7 @@ impl ServiceContext {
     /// Create a new `ServiceContext`
     pub fn new() -> ServiceContext {
         ServiceContext {
-            context: Context::new_shared(),
+            context: Context::new_shared(ServerType::Local),
             connect_opts: ConnectOpts::default(),
             accept_opts: AcceptOpts::default(),
             acl: None,

crates/shadowsocks-service/src/manager/server.rs

@@ -4,7 +4,7 @@ use std::{collections::HashMap, io, net::SocketAddr, sync::Arc, time::Duration};
 
 use log::{error, info, trace};
 use shadowsocks::{
-    config::{Mode, ServerConfig},
+    config::{Mode, ServerConfig, ServerType},
     context::{Context, SharedContext},
     crypto::v1::CipherKind,
     dns_resolver::DnsResolver,
@@ -62,7 +62,7 @@ pub struct Manager {
 impl Manager {
     /// Create a new manager server from configuration
     pub fn new(svr_cfg: ManagerConfig) -> Manager {
-        Manager::with_context(svr_cfg, Context::new_shared())
+        Manager::with_context(svr_cfg, Context::new_shared(ServerType::Server))
     }
 
     /// Create a new manager server with context and configuration

crates/shadowsocks-service/src/server/context.rs

@@ -3,6 +3,7 @@
 use std::sync::Arc;
 
 use shadowsocks::{
+    config::ServerType,
     context::{Context, SharedContext},
     dns_resolver::DnsResolver,
     net::ConnectOpts,
@@ -26,7 +27,7 @@ pub struct ServiceContext {
 impl Default for ServiceContext {
     fn default() -> Self {
         ServiceContext {
-            context: Context::new_shared(),
+            context: Context::new_shared(ServerType::Server),
             connect_opts: ConnectOpts::default(),
             acl: None,
             flow_stat: Arc::new(FlowStat::new()),

crates/shadowsocks-service/src/server/udprelay.rs

@@ -85,7 +85,7 @@ impl UdpServer {
     }
 
     pub async fn run(mut self, svr_cfg: &ServerConfig) -> io::Result<()> {
-        let socket = ProxySocket::bind(svr_cfg).await?;
+        let socket = ProxySocket::bind(self.context.context(), svr_cfg).await?;
 
         info!(
             "shadowsocks udp server listening on {}",

crates/shadowsocks/Cargo.toml

@@ -41,7 +41,9 @@ byte_string = "1.0"
 base64 = "0.13"
 url = "2.2"
 once_cell = "1.8"
+spin = { version = "0.9", features = ["std"] }
 pin-project = "1.0"
+bloomfilter = "1.0.8"
 thiserror = "1.0"
 
 serde = { version = "1.0", features = ["derive"] }

crates/shadowsocks/src/context.rs

@@ -2,10 +2,100 @@
 
 use std::{io, net::SocketAddr, sync::Arc};
 
-use crate::dns_resolver::DnsResolver;
+use bloomfilter::Bloom;
+use spin::Mutex as SpinMutex;
+
+use crate::{config::ServerType, dns_resolver::DnsResolver};
+
+// Entries for server's bloom filter
+//
+// Borrowed from shadowsocks-libev's default value
+const BF_NUM_ENTRIES_FOR_SERVER: usize = 1_000_000;
+
+// Entries for client's bloom filter
+//
+// Borrowed from shadowsocks-libev's default value
+const BF_NUM_ENTRIES_FOR_CLIENT: usize = 10_000;
+
+// Error rate for server's bloom filter
+//
+// Borrowed from shadowsocks-libev's default value
+const BF_ERROR_RATE_FOR_SERVER: f64 = 1e-6;
+
+// Error rate for client's bloom filter
+//
+// Borrowed from shadowsocks-libev's default value
+const BF_ERROR_RATE_FOR_CLIENT: f64 = 1e-15;
+
+// A bloom filter borrowed from shadowsocks-libev's `ppbloom`
+//
+// It contains 2 bloom filters and each one holds 1/2 entries.
+// Use them as a ring buffer.
+struct PingPongBloom {
+    blooms: [Bloom<[u8]>; 2],
+    bloom_count: [usize; 2],
+    item_count: usize,
+    current: usize,
+}
+
+impl PingPongBloom {
+    fn new(ty: ServerType) -> PingPongBloom {
+        let (mut item_count, fp_p) = if ty.is_local() {
+            (BF_NUM_ENTRIES_FOR_CLIENT, BF_ERROR_RATE_FOR_CLIENT)
+        } else {
+            (BF_NUM_ENTRIES_FOR_SERVER, BF_ERROR_RATE_FOR_SERVER)
+        };
+
+        item_count /= 2;
+
+        PingPongBloom {
+            blooms: [
+                Bloom::new_for_fp_rate(item_count, fp_p),
+                Bloom::new_for_fp_rate(item_count, fp_p),
+            ],
+            bloom_count: [0, 0],
+            item_count,
+            current: 0,
+        }
+    }
+
+    // Check if data in `buf` exist.
+    //
+    // Set into the current bloom filter if not exist.
+    //
+    // Return `true` if data exist in bloom filter.
+    fn check_and_set(&mut self, buf: &[u8]) -> bool {
+        for bloom in &self.blooms {
+            if bloom.check(buf) {
+                return true;
+            }
+        }
+
+        if self.bloom_count[self.current] >= self.item_count {
+            // Current bloom filter is full,
+            // Create a new one and use that one as current.
+
+            self.current = (self.current + 1) % 2;
+
+            self.bloom_count[self.current] = 0;
+            self.blooms[self.current].clear();
+        }
+
+        // Cannot be optimized by `check_and_set`
+        // Because we have to check every filters in `blooms` before `set`
+        self.blooms[self.current].set(buf);
+        self.bloom_count[self.current] += 1;
+
+        false
+    }
+}
 
 /// Service context
 pub struct Context {
+    // Check for duplicated IV/Nonce, for prevent replay attack
+    // https://github.com/shadowsocks/shadowsocks-org/issues/44
+    nonce_ppbloom: SpinMutex<PingPongBloom>,
+
     // trust-dns resolver, which supports REAL asynchronous resolving, and also customizable
     dns_resolver: Arc<DnsResolver>,
 
@@ -18,16 +108,32 @@ pub type SharedContext = Arc<Context>;
 
 impl Context {
     /// Create a new `Context` for `Client` or `Server`
-    pub fn new() -> Context {
+    pub fn new(config_type: ServerType) -> Context {
+        let nonce_ppbloom = SpinMutex::new(PingPongBloom::new(config_type));
         Context {
+            nonce_ppbloom,
             dns_resolver: Arc::new(DnsResolver::system_resolver()),
             ipv6_first: false,
         }
     }
 
     /// Create a new `Context` shared
-    pub fn new_shared() -> SharedContext {
-        SharedContext::new(Context::new())
+    pub fn new_shared(config_type: ServerType) -> SharedContext {
+        SharedContext::new(Context::new(config_type))
+    }
+
+    /// Check if nonce exist or not
+    ///
+    /// If not, set into the current bloom filter
+    pub fn check_nonce_and_set(&self, nonce: &[u8]) -> bool {
+        // Plain cipher doesn't have a nonce
+        // Always treated as non-duplicated
+        if nonce.is_empty() {
+            return false;
+        }
+
+        let mut ppbloom = self.nonce_ppbloom.lock();
+        ppbloom.check_and_set(nonce)
     }
 
     /// Set a DNS resolver

crates/shadowsocks/src/relay/tcprelay/aead.rs

@@ -43,10 +43,13 @@ use std::{
 use byte_string::ByteStr;
 use bytes::{BufMut, Bytes, BytesMut};
 use futures::ready;
-use log::trace;
+use log::{trace, warn};
 use tokio::io::{AsyncRead, AsyncWrite, ReadBuf};
 
-use crate::crypto::v1::{Cipher, CipherKind};
+use crate::{
+    context::Context,
+    crypto::v1::{Cipher, CipherKind},
+};
 
 /// AEAD packet payload must be smaller than 0x3FFF
 pub const MAX_PACKET_SIZE: usize = 0x3FFF;
@@ -64,6 +67,7 @@ pub struct DecryptedReader {
     cipher: Option<Cipher>,
     buffer: BytesMut,
     method: CipherKind,
+    salt: Option<Bytes>,
 }
 
 impl DecryptedReader {
@@ -76,13 +80,15 @@ impl DecryptedReader {
                 cipher: None,
                 buffer: BytesMut::with_capacity(method.salt_len()),
                 method,
+                salt: None,
             }
         } else {
             DecryptedReader {
                 state: DecryptReadState::ReadLength,
                 cipher: Some(Cipher::new(method, key, &[])),
                 buffer: BytesMut::with_capacity(2 + method.tag_len()),
                 method,
+                salt: None,
             }
         }
     }
@@ -91,6 +97,7 @@ impl DecryptedReader {
     pub fn poll_read_decrypted<S>(
         &mut self,
         cx: &mut task::Context<'_>,
+        context: &Context,
         stream: &mut S,
         buf: &mut ReadBuf<'_>,
     ) -> Poll<io::Result<()>>
@@ -118,7 +125,7 @@ impl DecryptedReader {
                     }
                 },
                 DecryptReadState::ReadData { length } => {
-                    ready!(self.poll_read_data(cx, stream, length))?;
+                    ready!(self.poll_read_data(cx, context, stream, length))?;
 
                     self.state = DecryptReadState::BufferedData { pos: 0 };
                 }
@@ -154,6 +161,11 @@ impl DecryptedReader {
         }
 
         let salt = &self.buffer[..salt_len];
+        // #442 Remember salt in filter after first successful decryption.
+        //
+        // If we check salt right here will allow attacker to flood our filter and eventually block all of our legitimate clients' requests.
+        self.salt = Some(Bytes::copy_from_slice(salt));
+
         trace!("got AEAD salt {:?}", ByteStr::new(salt));
 
         let cipher = Cipher::new(self.method, key, salt);
@@ -182,7 +194,13 @@ impl DecryptedReader {
         Ok(Some(length)).into()
     }
 
-    fn poll_read_data<S>(&mut self, cx: &mut task::Context<'_>, stream: &mut S, size: usize) -> Poll<io::Result<()>>
+    fn poll_read_data<S>(
+        &mut self,
+        cx: &mut task::Context<'_>,
+        context: &Context,
+        stream: &mut S,
+        size: usize,
+    ) -> Poll<io::Result<()>>
     where
         S: AsyncRead + Unpin + ?Sized,
     {
@@ -200,6 +218,15 @@ impl DecryptedReader {
             return Err(io::Error::new(ErrorKind::Other, "invalid tag-in")).into();
         }
 
+        // Check repeated salt after first successful decryption #442
+        if self.salt.is_some() {
+            let salt = self.salt.take().unwrap();
+
+            if context.check_nonce_and_set(&salt) {
+                warn!("detected repeated AEAD salt {:?}", ByteStr::new(&salt));
+            }
+        }
+
         // Remote TAG
         self.buffer.truncate(size);