implement vless protocol

Signed-off-by: Uncle Jack <[email protected]>

.gitignore

@@ -1,3 +1,4 @@
 .env
 /target
 /node_modules
+/.wrangler

src/config.rs

@@ -3,4 +3,5 @@ use uuid::Uuid;
 pub struct Config {
     pub uuid: Uuid,
     pub host: String,
+    pub outbound: String,
 }

src/lib.rs

@@ -16,8 +16,16 @@ async fn main(req: Request, env: Env, _: Context) -> Result<Response> {
     let uuid = env
         .var("UUID")
         .map(|x| Uuid::parse_str(&x.to_string()).unwrap_or_default())?;
+    let outbound = env
+        .var("OUTBOUND")
+        .map(|x| x.to_string())
+        .unwrap_or_default();
     let host = req.url()?.host().map(|x| x.to_string()).unwrap_or_default();
-    let config = Config { uuid, host };
+    let config = Config {
+        uuid,
+        host,
+        outbound,
+    };
 
     Router::with_data(config)
         .on_async("/", tunnel)
@@ -31,8 +39,14 @@ async fn tunnel(_: Request, cx: RouteContext<Config>) -> Result<Response> {
 
     server.accept()?;
     wasm_bindgen_futures::spawn_local(async move {
+        let config = cx.data;
         let events = server.events().unwrap();
-        if let Err(e) = VmessStream::new(cx.data, &server, events).process().await {
+
+        if let Err(e) = match config.outbound.as_str() {
+            "VLESS" => VlessStream::new(config, &server, events).process().await,
+            "VMESS" => VmessStream::new(config, &server, events).process().await,
+            _ => Err(Error::RustError("invalid outbound".to_string())),
+        } {
             console_log!("[tunnel]: {}", e);
         }
     });

src/proxy/mod.rs

@@ -1,2 +1,4 @@
+pub mod vless;
 pub mod vmess;
+pub use vless::*;
 pub use vmess::*;

src/proxy/vless.rs

@@ -0,0 +1,130 @@
+use crate::config::Config;
+
+use std::pin::Pin;
+use std::task::{Context, Poll};
+
+use bytes::{BufMut, BytesMut};
+use futures_util::Stream;
+use pin_project_lite::pin_project;
+use tokio::io::{AsyncRead, AsyncReadExt, AsyncWrite, AsyncWriteExt, ReadBuf};
+use worker::*;
+
+pin_project! {
+    pub struct VlessStream<'a> {
+        pub config: Config,
+        pub ws: &'a WebSocket,
+        pub buffer: BytesMut,
+        #[pin]
+        pub events: EventStream<'a>,
+    }
+}
+
+impl<'a> VlessStream<'a> {
+    pub fn new(config: Config, ws: &'a WebSocket, events: EventStream<'a>) -> Self {
+        let buffer = BytesMut::new();
+
+        Self {
+            config,
+            ws,
+            buffer,
+            events,
+        }
+    }
+
+    pub async fn process(&mut self) -> Result<()> {
+        // https://xtls.github.io/Xray-docs-next/en/development/protocols/vless.html
+        // +------------------+-----------------+---------------------------------+---------------------------------+-------------+---------+--------------+---------+--------------+
+        // |      1 byte      |    16 bytes     |             1 byte              |             M bytes             |   1 byte    | 2 bytes |    1 byte    | S bytes |   X bytes    |
+        // +------------------+-----------------+---------------------------------+---------------------------------+-------------+---------+--------------+---------+--------------+
+        // | Protocol Version | Equivalent UUID | Additional Information Length M | Additional Information ProtoBuf | Instruction | Port    | Address Type | Address | Request Data |
+        // +------------------+-----------------+---------------------------------+---------------------------------+-------------+---------+--------------+---------+--------------+
+
+        // ignore protocl version
+        self.read_u8().await?;
+
+        // UUID
+        let mut uuid = [0u8; 16];
+        self.read_exact(&mut uuid).await?;
+        let uuid = uuid::Uuid::from_bytes(uuid);
+        if self.config.uuid != uuid {
+            return Err(Error::RustError("incorrect uuid".to_string()));
+        }
+
+        // additional information
+        let len = self.read_u8().await?;
+        let mut addon = vec![0u8; len as _];
+        self.read_exact(&mut addon).await?;
+
+        // instruction
+        self.read_u8().await?;
+
+        // port
+        let mut port = [0u8; 2];
+        self.read_exact(&mut port).await?;
+        let port = u16::from_be_bytes(port);
+
+        let addr = crate::common::parse_addr(self).await?;
+
+        console_log!("connecting to upstream {}:{}", addr, port);
+        let mut upstream = Socket::builder().connect(addr, port)?;
+
+        // +-----------------------------------------------+------------------------------------+------------------------------------+---------------+
+        // |                    1 Byte                     |               1 Byte               |              N Bytes               |    Y Bytes    |
+        // +-----------------------------------------------+------------------------------------+------------------------------------+---------------+
+        // | Protocol Version, consistent with the request | Length of additional information N | Additional information in ProtoBuf | Response data |
+        // +-----------------------------------------------+------------------------------------+------------------------------------+---------------+
+        self.write(&[0u8; 2]).await?; // no additional information
+        tokio::io::copy_bidirectional(self, &mut upstream).await?;
+
+        Ok(())
+    }
+}
+
+impl<'a> AsyncRead for VlessStream<'a> {
+    fn poll_read(
+        self: Pin<&mut Self>,
+        cx: &mut Context<'_>,
+        buf: &mut ReadBuf<'_>,
+    ) -> Poll<tokio::io::Result<()>> {
+        let mut this = self.project();
+
+        loop {
+            let size = std::cmp::min(this.buffer.len(), buf.remaining());
+            if size > 0 {
+                buf.put_slice(&this.buffer.split_to(size));
+                return Poll::Ready(Ok(()));
+            }
+
+            match this.events.as_mut().poll_next(cx) {
+                Poll::Ready(Some(Ok(WebsocketEvent::Message(msg)))) => {
+                    msg.bytes().iter().for_each(|x| this.buffer.put_slice(&x));
+                }
+                Poll::Pending => return Poll::Pending,
+                _ => return Poll::Ready(Ok(())),
+            }
+        }
+    }
+}
+
+impl<'a> AsyncWrite for VlessStream<'a> {
+    fn poll_write(
+        self: Pin<&mut Self>,
+        _: &mut Context<'_>,
+        buf: &[u8],
+    ) -> Poll<tokio::io::Result<usize>> {
+        return Poll::Ready(
+            self.ws
+                .send_with_bytes(buf)
+                .map(|_| buf.len())
+                .map_err(|e| std::io::Error::new(std::io::ErrorKind::Other, e.to_string())),
+        );
+    }
+
+    fn poll_flush(self: Pin<&mut Self>, _: &mut Context<'_>) -> Poll<tokio::io::Result<()>> {
+        Poll::Ready(Ok(()))
+    }
+
+    fn poll_shutdown(self: Pin<&mut Self>, _: &mut Context<'_>) -> Poll<tokio::io::Result<()>> {
+        unimplemented!()
+    }
+}

wrangler.toml

@@ -11,3 +11,4 @@ build = { command = "cargo install -q worker-build && worker-build --dev" }
 
 [vars]
 UUID = "0fbf4f81-2598-4b6a-a623-0ead4cb9efa8"
+OUTBOUND = "VMESS"