feat(local-online-config): refactored with local-http client

- removed dependency to reqwest (very large)
- apply outbound configurations on sockets
- fixed #1508

Cargo.toml

@@ -121,7 +121,7 @@ service = ["local", "server", "manager"]
 winservice = ["service", "windows-service"]
 
 # Enables Hickory-DNS for replacing tokio's builtin DNS resolver
-hickory-dns = ["shadowsocks-service/hickory-dns", "reqwest/hickory-dns"]
+hickory-dns = ["shadowsocks-service/hickory-dns"]
 # Hickory-DNS was renamed from Trust-DNS, keep compatibility.
 trust-dns = ["hickory-dns"]
 dns-over-tls = ["shadowsocks-service/dns-over-tls"]
@@ -166,7 +166,6 @@ local-fake-dns = ["local", "shadowsocks-service/local-fake-dns", "ipnet"]
 # https://shadowsocks.org/doc/sip008.html
 local-online-config = [
     "local",
-    "reqwest",
     "mime",
     "shadowsocks-service/local-online-config",
 ]

crates/shadowsocks-service/Cargo.toml

@@ -98,7 +98,7 @@ local-tun = ["local", "etherparse", "tun2", "smoltcp"]
 local-fake-dns = ["local", "trust-dns", "sled", "bson"]
 # sslocal support online URL (SIP008 Online Configuration Delivery)
 # https://shadowsocks.org/doc/sip008.html
-local-online-config = ["local"]
+local-online-config = ["local", "local-http", "mime", "http"]
 
 # Enable Stream Cipher Protocol
 # WARN: Stream Cipher Protocol is proved to be insecure
@@ -157,6 +157,7 @@ libc = "0.2.141"
 
 hyper = { version = "1.3", optional = true, features = ["full"] }
 http-body-util = { version = "0.1", optional = true }
+http = { version = "1.1", optional = true }
 
 hickory-resolver = { version = "0.24", optional = true, features = [
     "serde-config",
@@ -166,6 +167,7 @@ idna = "1.0"
 ipnet = "2.9"
 iprange = "0.6"
 regex = "1.4"
+mime = { version = "0.3", optional = true }
 
 tun2 = { version = "1", optional = true, features = ["async"] }
 etherparse = { version = "0.15", optional = true }

crates/shadowsocks-service/src/config.rs

@@ -234,6 +234,10 @@ struct SSConfig {
     #[cfg(feature = "local-online-config")]
     #[serde(skip_serializing_if = "Option::is_none")]
     version: Option<u32>,
+
+    #[cfg(feature = "local-online-config")]
+    #[serde(skip_serializing_if = "Option::is_none")]
+    online_config: Option<SSOnlineConfig>,
 }
 
 #[derive(Serialize, Deserialize, Debug, Default)]
@@ -406,6 +410,13 @@ struct SSServerExtConfig {
     outbound_bind_interface: Option<String>,
 }
 
+#[cfg(feature = "local-online-config")]
+#[derive(Serialize, Deserialize, Debug, Default)]
+struct SSOnlineConfig {
+    config_url: String,
+    update_interval: Option<u64>,
+}
+
 /// Server config type
 #[derive(Clone, Copy, Debug, Eq, PartialEq)]
 pub enum ConfigType {
@@ -1237,6 +1248,17 @@ impl LocalInstanceConfig {
     }
 }
 
+/// OnlineConfiguration (SIP008)
+/// https://shadowsocks.org/doc/sip008.html
+#[cfg(feature = "local-online-config")]
+#[derive(Debug, Clone)]
+pub struct OnlineConfig {
+    /// SIP008 URL
+    pub config_url: String,
+    /// Update interval, 3600s by default
+    pub update_interval: Option<Duration>,
+}
+
 /// Configuration
 #[derive(Clone, Debug)]
 pub struct Config {
@@ -1341,6 +1363,10 @@ pub struct Config {
     /// Workers in runtime
     /// It should be replaced with metrics APIs: https://github.com/tokio-rs/tokio/issues/4073
     pub worker_count: usize,
+
+    /// OnlineConfiguration (SIP008)
+    /// https://shadowsocks.org/doc/sip008.html
+    pub online_config: Option<OnlineConfig>,
 }
 
 /// Configuration parsing error kind
@@ -1462,6 +1488,8 @@ impl Config {
             config_path: None,
 
             worker_count: 1,
+
+            online_config: None,
         }
     }
 
@@ -2352,6 +2380,13 @@ impl Config {
             nconfig.acl = Some(acl);
         }
 
+        if let Some(online_config) = config.online_config {
+            nconfig.online_config = Some(OnlineConfig {
+                config_url: online_config.config_url,
+                update_interval: online_config.update_interval.map(Duration::from_secs),
+            });
+        }
+
         Ok(nconfig)
     }
 
@@ -3090,6 +3125,14 @@ impl fmt::Display for Config {
             jconf.acl = Some(acl.file_path().to_str().unwrap().to_owned());
         }
 
+        // OnlineConfig
+        if let Some(ref online_config) = self.online_config {
+            jconf.online_config = Some(SSOnlineConfig {
+                config_url: online_config.config_url.clone(),
+                update_interval: online_config.update_interval.as_ref().map(Duration::as_secs),
+            });
+        }
+
         write!(f, "{}", json5::to_string(&jconf).unwrap())
     }
 }

crates/shadowsocks-service/src/local/http/http_client.rs

@@ -2,20 +2,25 @@
 
 use std::{
     collections::VecDeque,
+    fmt::Debug,
+    future::Future,
     io::{self, ErrorKind},
+    pin::Pin,
     sync::Arc,
+    task::{Context, Poll},
     time::{Duration, Instant},
 };
 
+use hyper::rt::{Sleep, Timer};
 use hyper::{
-    body,
+    body::{self, Body},
     client::conn::{http1, http2},
     http::uri::Scheme,
-    Request,
-    Response,
+    Request, Response,
 };
 use log::{error, trace};
 use lru_time_cache::LruCache;
+use pin_project::pin_project;
 use shadowsocks::relay::Address;
 use tokio::sync::Mutex;
 
@@ -29,33 +34,96 @@ use super::{
 
 const CONNECTION_EXPIRE_DURATION: Duration = Duration::from_secs(20);
 
+/// HTTPClient API request errors
 #[derive(thiserror::Error, Debug)]
 pub enum HttpClientError {
+    /// Errors from hyper
     #[error("{0}")]
     Hyper(#[from] hyper::Error),
+    /// std::io::Error
     #[error("{0}")]
     Io(#[from] io::Error),
 }
 
-#[derive(Clone)]
-pub struct HttpClient {
+#[derive(Clone, Debug)]
+pub struct TokioTimer;
+
+impl Timer for TokioTimer {
+    fn sleep(&self, duration: Duration) -> Pin<Box<dyn Sleep>> {
+        Box::pin(TokioSleep {
+            inner: tokio::time::sleep(duration),
+        })
+    }
+
+    fn sleep_until(&self, deadline: Instant) -> Pin<Box<dyn Sleep>> {
+        Box::pin(TokioSleep {
+            inner: tokio::time::sleep_until(deadline.into()),
+        })
+    }
+
+    fn reset(&self, sleep: &mut Pin<Box<dyn Sleep>>, new_deadline: Instant) {
+        if let Some(sleep) = sleep.as_mut().downcast_mut_pin::<TokioSleep>() {
+            sleep.reset(new_deadline.into())
+        }
+    }
+}
+
+#[pin_project]
+pub(crate) struct TokioSleep {
+    #[pin]
+    pub(crate) inner: tokio::time::Sleep,
+}
+
+impl Future for TokioSleep {
+    type Output = ();
+
+    fn poll(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
+        self.project().inner.poll(cx)
+    }
+}
+
+impl Sleep for TokioSleep {}
+
+impl TokioSleep {
+    pub fn reset(self: Pin<&mut Self>, deadline: Instant) {
+        self.project().inner.as_mut().reset(deadline.into());
+    }
+}
+
+/// HTTPClient, supporting HTTP/1.1 and H2, HTTPS.
+pub struct HttpClient<B> {
     #[allow(clippy::type_complexity)]
-    cache_conn: Arc<Mutex<LruCache<Address, VecDeque<(HttpConnection, Instant)>>>>,
+    cache_conn: Arc<Mutex<LruCache<Address, VecDeque<(HttpConnection<B>, Instant)>>>>,
+}
+
+impl<B> Clone for HttpClient<B> {
+    fn clone(&self) -> Self {
+        HttpClient {
+            cache_conn: self.cache_conn.clone(),
+        }
+    }
 }
 
-impl HttpClient {
-    pub fn new() -> HttpClient {
+impl<B> HttpClient<B>
+where
+    B: Body + Send + Unpin + Debug + 'static,
+    B::Data: Send,
+    B::Error: Into<Box<dyn ::std::error::Error + Send + Sync>>,
+{
+    /// Create a new HttpClient
+    pub fn new() -> HttpClient<B> {
         HttpClient {
             cache_conn: Arc::new(Mutex::new(LruCache::with_expiry_duration(CONNECTION_EXPIRE_DURATION))),
         }
     }
 
+    /// Make HTTP requests
     #[inline]
     pub async fn send_request(
         &self,
         context: Arc<ServiceContext>,
-        req: Request<body::Incoming>,
-        balancer: &PingBalancer,
+        req: Request<B>,
+        balancer: Option<&PingBalancer>,
     ) -> Result<Response<body::Incoming>, HttpClientError> {
         let host = match host_addr(req.uri()) {
             Some(h) => h,
@@ -96,7 +164,7 @@ impl HttpClient {
         self.send_request_conn(host, c, req).await.map_err(Into::into)
     }
 
-    async fn get_cached_connection(&self, host: &Address) -> Option<HttpConnection> {
+    async fn get_cached_connection(&self, host: &Address) -> Option<HttpConnection<B>> {
         if let Some(q) = self.cache_conn.lock().await.get_mut(host) {
             while let Some((c, inst)) = q.pop_front() {
                 let now = Instant::now();
@@ -115,8 +183,8 @@ impl HttpClient {
     async fn send_request_conn(
         &self,
         host: Address,
-        mut c: HttpConnection,
-        req: Request<body::Incoming>,
+        mut c: HttpConnection<B>,
+        req: Request<B>,
     ) -> hyper::Result<Response<body::Incoming>> {
         trace!("HTTP making request to host: {}, request: {:?}", host, req);
         let response = c.send_request(req).await?;
@@ -141,19 +209,24 @@ impl HttpClient {
     }
 }
 
-enum HttpConnection {
-    Http1(http1::SendRequest<body::Incoming>),
-    Http2(http2::SendRequest<body::Incoming>),
+enum HttpConnection<B> {
+    Http1(http1::SendRequest<B>),
+    Http2(http2::SendRequest<B>),
 }
 
-impl HttpConnection {
+impl<B> HttpConnection<B>
+where
+    B: Body + Send + Unpin + 'static,
+    B::Data: Send,
+    B::Error: Into<Box<dyn ::std::error::Error + Send + Sync>>,
+{
     async fn connect(
         context: Arc<ServiceContext>,
         scheme: &Scheme,
         host: Address,
         domain: &str,
-        balancer: &PingBalancer,
-    ) -> io::Result<HttpConnection> {
+        balancer: Option<&PingBalancer>,
+    ) -> io::Result<HttpConnection<B>> {
         if *scheme != Scheme::HTTP && *scheme != Scheme::HTTPS {
             return Err(io::Error::new(ErrorKind::InvalidInput, "invalid scheme"));
         }
@@ -173,7 +246,7 @@ impl HttpConnection {
         scheme: &Scheme,
         host: Address,
         stream: AutoProxyClientStream,
-    ) -> io::Result<HttpConnection> {
+    ) -> io::Result<HttpConnection<B>> {
         trace!(
             "HTTP making new HTTP/1.1 connection to host: {}, scheme: {}",
             host,
@@ -207,7 +280,7 @@ impl HttpConnection {
         host: Address,
         domain: &str,
         stream: AutoProxyClientStream,
-    ) -> io::Result<HttpConnection> {
+    ) -> io::Result<HttpConnection<B>> {
         trace!("HTTP making new TLS connection to host: {}, scheme: {}", host, scheme);
 
         // TLS handshake, check alpn for h2 support.
@@ -216,6 +289,7 @@ impl HttpConnection {
         if stream.negotiated_http2() {
             // H2 connnection
             let (send_request, connection) = match http2::Builder::new(TokioExecutor)
+                .timer(TokioTimer)
                 .keep_alive_interval(Duration::from_secs(15))
                 .handshake(TokioIo::new(stream))
                 .await
@@ -254,7 +328,7 @@ impl HttpConnection {
     }
 
     #[inline]
-    pub async fn send_request(&mut self, req: Request<body::Incoming>) -> hyper::Result<Response<body::Incoming>> {
+    pub async fn send_request(&mut self, req: Request<B>) -> hyper::Result<Response<body::Incoming>> {
         match self {
             HttpConnection::Http1(r) => r.send_request(req).await,
             HttpConnection::Http2(r) => r.send_request(req).await,