feat: support ss v2ray plugin(without mux)

Author: VendettaReborn

clash_lib/src/proxy/converters/shadowsocks.rs

@@ -126,6 +126,11 @@ impl TryFrom<HashMap<String, serde_yaml::Value>> for V2RayOBFSOption {
             .get("mode")
             .and_then(|x| x.as_str())
             .ok_or(Error::InvalidConfig("obfs mode is required".to_owned()))?;
+        let port = value
+            .get("port")
+            .and_then(|x| x.as_u64())
+            .ok_or(Error::InvalidConfig("obfs port is required".to_owned()))?
+            as u16;
 
         if mode != "websocket" {
             return Err(Error::InvalidConfig(format!(
@@ -134,10 +139,7 @@ impl TryFrom<HashMap<String, serde_yaml::Value>> for V2RayOBFSOption {
             )));
         }
 
-        let path = value
-            .get("path")
-            .and_then(|x| x.as_str())
-            .ok_or(Error::InvalidConfig("obfs path is required".to_owned()))?;
+        let path = value.get("path").and_then(|x| x.as_str()).unwrap_or("");
         let mux = value.get("mux").and_then(|x| x.as_bool()).unwrap_or(false);
         let tls = value.get("tls").and_then(|x| x.as_bool()).unwrap_or(false);
         let skip_cert_verify = value
@@ -159,6 +161,7 @@ impl TryFrom<HashMap<String, serde_yaml::Value>> for V2RayOBFSOption {
         Ok(V2RayOBFSOption {
             mode: mode.to_owned(),
             host: host.to_owned(),
+            port,
             path: path.to_owned(),
             tls,
             headers,

clash_lib/src/proxy/shadowsocks/mod.rs

@@ -31,8 +31,9 @@ use shadowsocks::{
 };
 use std::{collections::HashMap, fmt::Debug, io, sync::Arc};
 use tracing::debug;
+use v2ray::new_websocket_stream;
 
-#[derive(Clone, Copy)]
+#[derive(Clone, Copy, Debug)]
 pub enum SimpleOBFSMode {
     Http,
     Tls,
@@ -47,14 +48,14 @@ pub struct SimpleOBFSOption {
 pub struct V2RayOBFSOption {
     pub mode: String,
     pub host: String,
+    pub port: u16,
     pub path: String,
-    pub tls: bool,
     pub headers: HashMap<String, String>,
+    pub tls: bool,
     pub skip_cert_verify: bool,
     pub mux: bool,
 }
 
-#[derive(Debug)]
 pub struct ShadowTlsOption {
     pub host: String,
     pub password: String,
@@ -106,7 +107,7 @@ impl Handler {
         &self,
         s: AnyStream,
         sess: &Session,
-        #[allow(unused_variables)] _resolver: ThreadSafeDNSResolver,
+        _resolver: ThreadSafeDNSResolver,
     ) -> std::io::Result<AnyStream> {
         let stream: AnyStream = match &self.opts.plugin_opts {
             Some(plugin) => match plugin {
@@ -121,12 +122,16 @@ impl Handler {
                         simple_obfs::SimpleObfsTLS::new(s, opts.host.clone()).into()
                     }
                 },
-                OBFSOption::V2Ray(_opt) => {
-                    todo!("v2ray-plugin is not implemented yet")
+                OBFSOption::V2Ray(opt) => {
+                    new_websocket_stream(
+                        s,
+                        self.opts.server.clone(),
+                        self.opts.port,
+                        opt,
+                    )
+                    .await?
                 }
                 OBFSOption::ShadowTls(opts) => {
-                    tracing::trace!("using shadow-tls");
-
                     (shadow_tls::Connector::wrap(opts, s).await?) as _
                 }
             },

clash_lib/src/proxy/shadowsocks/v2ray/mod.rs

@@ -1 +1,65 @@
+use std::io;
+
+use crate::proxy::{
+    transport::{self, TLSOptions},
+    AnyStream,
+};
+
+use super::V2RayOBFSOption;
+
 pub(crate) mod mux;
+
+pub async fn new_websocket_stream(
+    mut stream: AnyStream,
+    server: String,
+    port: u16,
+    opt: &V2RayOBFSOption,
+) -> std::io::Result<AnyStream> {
+    // this shall already be checked in the config parser
+    if opt.mode != "websocket" {
+        return Err(io::Error::new(
+            io::ErrorKind::Other,
+            format!("invalid obfs mode: {}", opt.mode),
+        ));
+    }
+
+    if opt.mux {
+        return Err(io::Error::new(
+            io::ErrorKind::Other,
+            "v2ray plugin does not support mux",
+        ));
+    }
+
+    let mut headers = opt.headers.clone();
+    if !headers.contains_key("Host") {
+        headers.insert("Host".to_owned(), opt.host.clone());
+    }
+    let ws_builder = transport::WebsocketStreamBuilder::new(
+        server,
+        port,
+        if opt.path.is_empty() {
+            "/".to_owned()
+        } else {
+            opt.path.clone()
+        },
+        headers,
+        None,
+        0,
+        "".to_owned(),
+    );
+
+    if opt.tls {
+        stream = transport::tls::wrap_stream(
+            stream,
+            TLSOptions {
+                sni: opt.host.clone(),
+                skip_cert_verify: opt.skip_cert_verify,
+                alpn: Some(vec!["http/1.1".to_owned()]),
+            },
+            None,
+        )
+        .await?;
+    }
+
+    ws_builder.proxy_stream(stream).await
+}

clash_lib/src/proxy/vmess/mod.rs

@@ -72,6 +72,28 @@ impl Handler {
         }
     }
 
+    /// v2ray may serve as plugin for other outbound handlers
+    /// so we expose this method
+    pub async fn proxy_stream(
+        &self,
+        sess: &Session,
+        resolver: ThreadSafeDNSResolver,
+        connector: &dyn RemoteConnector,
+    ) -> io::Result<AnyStream> {
+        let stream = connector
+            .connect_stream(
+                resolver,
+                self.opts.server.as_str(),
+                self.opts.port,
+                sess.iface.as_ref(),
+                #[cfg(any(target_os = "linux", target_os = "android"))]
+                sess.so_mark,
+            )
+            .await?;
+
+        self.inner_proxy_stream(stream, sess, false).await
+    }
+
     async fn inner_proxy_stream<'a>(
         &'a self,
         s: AnyStream,
@@ -242,18 +264,7 @@ impl OutboundHandler for Handler {
         resolver: ThreadSafeDNSResolver,
         connector: &dyn RemoteConnector,
     ) -> io::Result<BoxedChainedStream> {
-        let stream = connector
-            .connect_stream(
-                resolver,
-                self.opts.server.as_str(),
-                self.opts.port,
-                sess.iface.as_ref(),
-                #[cfg(any(target_os = "linux", target_os = "android"))]
-                sess.so_mark,
-            )
-            .await?;
-
-        let s = self.inner_proxy_stream(stream, sess, false).await?;
+        let s = self.proxy_stream(sess, resolver, connector).await?;
         let chained = ChainedStreamWrapper::new(s);
         chained.append_to_chain(self.name()).await;
         Ok(Box::new(chained))
@@ -265,18 +276,7 @@ impl OutboundHandler for Handler {
         resolver: ThreadSafeDNSResolver,
         connector: &dyn RemoteConnector,
     ) -> io::Result<BoxedChainedDatagram> {
-        let stream = connector
-            .connect_stream(
-                resolver,
-                self.opts.server.as_str(),
-                self.opts.port,
-                sess.iface.as_ref(),
-                #[cfg(any(target_os = "linux", target_os = "android"))]
-                sess.so_mark,
-            )
-            .await?;
-
-        let stream = self.inner_proxy_stream(stream, sess, true).await?;
+        let stream = self.proxy_stream(sess, resolver, connector).await?;
 
         let d = OutboundDatagramVmess::new(stream, sess.destination.clone());