Refactor vless (#86)

fix issue #76

apps/deno-vless/src/main.ts

@@ -1,10 +1,11 @@
-import { serve } from 'https://deno.land/std@0.167.0/http/server.ts';
-import { parse, stringify, validate } from 'https://jspm.dev/uuid';
-import { chunk, join } from 'https://jspm.dev/lodash-es';
+import { serve } from 'https://deno.land/std@0.170.0/http/server.ts';
+import * as uuid from 'https://jspm.dev/uuid';
+import * as lodash from 'https://jspm.dev/lodash-es';
 import { serveClient } from './deno/client.ts';
+import { processSocket } from '../../../libs/vless-js/src/lib/vless-js.ts';
 
 const userID = Deno.env.get('UUID') || '';
-let isVaildUser = validate(userID);
+let isVaildUser = uuid.validate(userID);
 if (!isVaildUser) {
   console.log('not set valid UUID');
 }
@@ -26,181 +27,27 @@ const handler = async (req: Request): Promise<Response> => {
     return await serveClient(req, userID);
   }
   const { socket, response } = Deno.upgradeWebSocket(req);
-  let remoteConnection: Deno.TcpConn;
-  let skipHeader = false;
-  let address = '';
-  let port = 0;
-  socket.onopen = () => console.log('socket opened');
-  socket.onmessage = async (e) => {
-    try {
-      if (!(e.data instanceof ArrayBuffer)) {
-        return;
-      }
-      const vlessBuffer: ArrayBuffer = e.data;
+  socket.addEventListener('open', () => {});
 
-      if (remoteConnection) {
-        const number = await remoteConnection.write(
-          new Uint8Array(vlessBuffer)
-        );
-      } else {
-        //https://github.com/v2ray/v2ray-core/issues/2636
-        // 1 字节	  16 字节     1 字节	       M 字节	      1 字节  2 字节   1 字节	 S 字节	X 字节
-        // 协议版本	  等价 UUID	  附加信息长度 M	附加信息 ProtoBuf  指令	    端口	地址类型   地址	请求数据
-
-        // 1 字节	              1 字节	      N 字节	         Y 字节
-        // 协议版本，与请求的一致	附加信息长度 N	附加信息 ProtoBuf	响应数据
-        if (vlessBuffer.byteLength < 24) {
-          console.log('invalid data');
-          return;
-        }
-        const version = new Uint8Array(vlessBuffer.slice(0, 1));
-        let isValidUser = false;
-        if (stringify(new Uint8Array(vlessBuffer.slice(1, 17))) === userID) {
-          isValidUser = true;
-        }
-        if (!isValidUser) {
-          console.log('in valid user');
-          return;
-        }
-
-        const optLength = new Uint8Array(vlessBuffer.slice(17, 18))[0];
-        //skip opt for now
-
-        const command = new Uint8Array(
-          vlessBuffer.slice(18 + optLength, 18 + optLength + 1)
-        )[0];
-        // 0x01 TCP
-        // 0x02 UDP
-        // 0x03 MUX
-        if (command === 1) {
-        } else {
-          console.log(
-            `command ${command} is not support, command 01-tcp,02-udp,03-mux`
-          );
-          socket.close();
-          return;
-        }
-        const portIndex = 18 + optLength + 1;
-        const portBuffer = vlessBuffer.slice(portIndex, portIndex + 2);
-        // port is big-Endian in raw data etc 80 == 0x005d
-        const portRemote = new DataView(portBuffer).getInt16(0);
-        port = portRemote;
-        let addressIndex = portIndex + 2;
-        const addressBuffer = new Uint8Array(
-          vlessBuffer.slice(addressIndex, addressIndex + 1)
-        );
-
-        // 1--> ipv4  addressLength =4
-        // 2--> domain name addressLength=addressBuffer[1]
-        // 3--> ipv6  addressLength =16
-        const addressType = addressBuffer[0];
-        let addressLength = 0;
-        let addressValueIndex = addressIndex + 1;
-        let addressValue = '';
-        switch (addressType) {
-          case 1:
-            addressLength = 4;
-            addressValue = new Uint8Array(
-              vlessBuffer.slice(
-                addressValueIndex,
-                addressValueIndex + addressLength
-              )
-            ).join('.');
-            break;
-          case 2:
-            addressLength = new Uint8Array(
-              vlessBuffer.slice(addressValueIndex, addressValueIndex + 1)
-            )[0];
-            addressValueIndex += 1;
-            addressValue = new TextDecoder().decode(
-              vlessBuffer.slice(
-                addressValueIndex,
-                addressValueIndex + addressLength
-              )
-            );
-            break;
-          case 3:
-            addressLength = 16;
-            const addressChunkBy2: number[][] = chunk(
-              new Uint8Array(
-                vlessBuffer.slice(
-                  addressValueIndex,
-                  addressValueIndex + addressLength
-                )
-              ),
-              2,
-              null
-            );
-            // 2001:0db8:85a3:0000:0000:8a2e:0370:7334
-            addressValue = addressChunkBy2
-              .map((items) =>
-                items.map((item) => item.toString(16).padStart(2, '0')).join('')
-              )
-              .join('.');
-            break;
-          default:
-            console.log(`[${address}:${port}] invild address`);
-        }
-        address = addressValue;
-        if (!addressValue) {
-          console.log(`[${address}:${port}] addressValue is empty`);
-          socket.close();
-          return;
-        }
-        // const addressType = requestAddr >> 4;
-        // const addressLength = requestAddr & 0x0f;
-        console.log(`[${address}:${port}] connecting`);
-        remoteConnection = await Deno.connect({
-          port: port,
-          hostname: addressValue,
-        });
-
-        const rawDataIndex = addressValueIndex + addressLength;
-        const rawClientData = vlessBuffer.slice(rawDataIndex);
-        await remoteConnection.write(new Uint8Array(rawClientData));
-
-        let chunkDatas = [new Uint8Array([version[0], 0])];
-        remoteConnection.readable
-          .pipeTo(
-            new WritableStream({
-              start() {
-                socket.send(new Blob(chunkDatas));
-              },
-              write(chunk, controller) {
-                socket.send(new Blob([chunk]));
-                // if (!skipHeader) {
-                //   console.log(
-                //     `[${address}:${port}] first time write to client socket`
-                //   );
-                //   chunkDatas.push(chunk);
-                //   socket.send(new Blob(chunkDatas));
-                //   chunkDatas = [];
-                // } else {
-                //   socket.send(new Blob([chunk]));
-                // }
-              },
-            })
-          )
-          .catch((error) => {
-            console.log(
-              `[${address}:${port}] remoteConnection pipe to has error`,
-              error
-            );
-          });
-      }
-    } catch (error) {
-      // console.log(
-      //   [...new Uint8Array(vlessBuffer.slice(0, 32))].map((x) =>
-      //     x.toString(16).padStart(2, '0')
-      //   )
-      // );
-      console.log(`[${address}:${port}] request hadler has error`, error);
-    }
-  };
-  socket.onerror = (e) =>
-    console.log(`[${address}:${port}] socket errored:`, e);
-  socket.onclose = () => console.log(`[${address}:${port}] socket closed`);
+  processSocket({
+    userID,
+    socket,
+    rawTCPFactory: (port: number, hostname: string) => {
+      return Deno.connect({
+        port,
+        hostname,
+      });
+    },
+    libs: { uuid, lodash },
+  });
   return response;
 };
 
+globalThis.addEventListener('beforeunload', (e) => {
+  console.log('About to exit...');
+});
+
+globalThis.addEventListener('unload', (e) => {
+  console.log('Exiting');
+});
 serve(handler, { port: 8080, hostname: '0.0.0.0' });

functions/_middleware.ts

@@ -42,13 +42,18 @@ function authentication(context: EventContext<any, any, any>) {
       },
     });
   } else {
-    return new Response(``, {
-      status: 302,
-      headers: {
-        'content-type': 'text/html; charset=utf-8',
-        Location: `./${userID}`,
-      },
-    });
+    const url = new URL(context.request.url);
+    if (url.pathname === '/') {
+      return new Response(``, {
+        status: 302,
+        headers: {
+          'content-type': 'text/html; charset=utf-8',
+          Location: `./${userID}`,
+        },
+      });
+    } else {
+      return context.next();
+    }
   }
 }
 

functions/hello.ts

@@ -1,8 +1,9 @@
+import { vlessJs } from 'vless-js';
 interface Env {
   KV: KVNamespace;
 }
 
 export const onRequest: PagesFunction<Env> = async (context) => {
-  console.log('xxxxx', context.env);
-  return new Response(`Hello, world! ${context.request.url}`);
+  console.log('xxxxx', context.env, vlessJs());
+  return new Response(`Hello, world! ${context.request.url}--${vlessJs()}`);
 };

functions/tsconfig.json

@@ -3,6 +3,9 @@
     "target": "esnext",
     "module": "esnext",
     "lib": ["esnext"],
-    "types": ["@cloudflare/workers-types"]
+    "types": ["@cloudflare/workers-types"],
+    "paths": {
+      "vless-js": ["../libs/vless-js/src/index.ts"]
+    }
   }
 }

functions/ws.ts

@@ -3,16 +3,16 @@ interface Env {
 }
 
 export const onRequest: PagesFunction<Env> = async ({ request }) => {
-  const upgradeHeader = request.headers.get("Upgrade");
-  if (!upgradeHeader || upgradeHeader !== "websocket") {
-    return new Response("Expected Upgrade: websocket", { status: 426 });
+  const upgradeHeader = request.headers.get('Upgrade');
+  if (!upgradeHeader || upgradeHeader !== 'websocket') {
+    return new Response('Expected Upgrade: websocket', { status: 426 });
   }
 
   const webSocketPair = new WebSocketPair();
   const [client, server] = Object.values(webSocketPair);
 
   server.accept();
-  server.addEventListener("message", (event) => {
+  server.addEventListener('message', (event) => {
     console.log(event.data);
     server.send(`server reponse after client sent ${event.data}`);
   });

libs/vless-js/src/index.ts

@@ -1 +1 @@
-export * from './lib/vless-js';
+export { vlessJs, processSocket } from './lib/vless-js';

libs/vless-js/src/lib/vless-js.ts

@@ -1,3 +1,225 @@
 export function vlessJs(): string {
   return 'vless-js';
 }
+
+export async function processSocket({
+  userID,
+  socket,
+  rawTCPFactory,
+  libs: { uuid, lodash },
+}: {
+  userID: string;
+  socket: WebSocket;
+  rawTCPFactory: (port: number, hostname: string) => Promise<any>;
+  libs: { uuid: any; lodash: any };
+}) {
+  let address = '';
+  let port = 0;
+  try {
+    const websocketStream = new ReadableStream({
+      start(controller) {
+        socket.addEventListener('message', async (e) => {
+          const vlessBuffer: ArrayBuffer = e.data;
+          controller.enqueue(vlessBuffer);
+        });
+        socket.addEventListener('error', (e) => {
+          console.log(`[${address}:${port}] socket has error`, e);
+          controller.error(e);
+        });
+        socket.addEventListener('close', () => {
+          try {
+            console.log(`[${address}:${port}] socket is close`);
+            controller.close();
+          } catch (error) {
+            console.log(`[${address}:${port}] websocketStream can't close`);
+          }
+        });
+      },
+      pull(controller) {},
+      cancel(reason) {
+        console.log(`[${address}:${port}] websocketStream is cancel`, reason);
+        socket.close();
+      },
+    });
+    let remoteConnection: {
+      readable: any;
+      write: (arg0: Uint8Array) => any;
+      close: () => void;
+    } | null = null;
+
+    await websocketStream.pipeTo(
+      new WritableStream({
+        async write(chunk, controller) {
+          const vlessBuffer = chunk;
+          if (remoteConnection) {
+            const number = await remoteConnection.write(
+              new Uint8Array(vlessBuffer)
+            );
+            return;
+          }
+          if (vlessBuffer.byteLength < 24) {
+            console.log('invalid data');
+            controller.error('invalid data');
+            return;
+          }
+          const version = new Uint8Array(vlessBuffer.slice(0, 1));
+          let isValidUser = false;
+          if (
+            uuid.stringify(new Uint8Array(vlessBuffer.slice(1, 17))) === userID
+          ) {
+            isValidUser = true;
+          }
+          if (!isValidUser) {
+            console.log('in valid user');
+            controller.error('in valid user');
+            return;
+          }
+
+          const optLength = new Uint8Array(vlessBuffer.slice(17, 18))[0];
+          //skip opt for now
+
+          const command = new Uint8Array(
+            vlessBuffer.slice(18 + optLength, 18 + optLength + 1)
+          )[0];
+          // 0x01 TCP
+          // 0x02 UDP
+          // 0x03 MUX
+          if (command === 1) {
+          } else {
+            controller.error(
+              `command ${command} is not support, command 01-tcp,02-udp,03-mux`
+            );
+            return;
+          }
+          const portIndex = 18 + optLength + 1;
+          const portBuffer = vlessBuffer.slice(portIndex, portIndex + 2);
+          // port is big-Endian in raw data etc 80 == 0x005d
+          const portRemote = new DataView(portBuffer).getInt16(0);
+          port = portRemote;
+          let addressIndex = portIndex + 2;
+          const addressBuffer = new Uint8Array(
+            vlessBuffer.slice(addressIndex, addressIndex + 1)
+          );
+
+          // 1--> ipv4  addressLength =4
+          // 2--> domain name addressLength=addressBuffer[1]
+          // 3--> ipv6  addressLength =16
+          const addressType = addressBuffer[0];
+          let addressLength = 0;
+          let addressValueIndex = addressIndex + 1;
+          let addressValue = '';
+          switch (addressType) {
+            case 1:
+              addressLength = 4;
+              addressValue = new Uint8Array(
+                vlessBuffer.slice(
+                  addressValueIndex,
+                  addressValueIndex + addressLength
+                )
+              ).join('.');
+              break;
+            case 2:
+              addressLength = new Uint8Array(
+                vlessBuffer.slice(addressValueIndex, addressValueIndex + 1)
+              )[0];
+              addressValueIndex += 1;
+              addressValue = new TextDecoder().decode(
+                vlessBuffer.slice(
+                  addressValueIndex,
+                  addressValueIndex + addressLength
+                )
+              );
+              break;
+            case 3:
+              addressLength = 16;
+              const addressChunkBy2: number[][] = lodash.chunk(
+                new Uint8Array(
+                  vlessBuffer.slice(
+                    addressValueIndex,
+                    addressValueIndex + addressLength
+                  )
+                ),
+                2,
+                null
+              );
+              // 2001:0db8:85a3:0000:0000:8a2e:0370:7334
+              addressValue = addressChunkBy2
+                .map((items) =>
+                  items
+                    .map((item) => item.toString(16).padStart(2, '0'))
+                    .join('')
+                )
+                .join(':');
+              break;
+            default:
+              console.log(`[${address}:${port}] invild address`);
+          }
+          address = addressValue;
+          if (!addressValue) {
+            // console.log(`[${address}:${port}] addressValue is empty`);
+            controller.error(`[${address}:${port}] addressValue is empty`);
+            return;
+          }
+          // const addressType = requestAddr >> 4;
+          // const addressLength = requestAddr & 0x0f;
+          console.log(`[${addressValue}:${port}] connecting`);
+          remoteConnection = await rawTCPFactory(port, addressValue);
+
+          const rawDataIndex = addressValueIndex + addressLength;
+          const rawClientData = vlessBuffer.slice(rawDataIndex);
+          await remoteConnection!.write(new Uint8Array(rawClientData));
+          let chunkDatas = [new Uint8Array([version[0], 0])];
+
+          // get response from remoteConnection
+          remoteConnection!.readable
+            .pipeTo(
+              new WritableStream({
+                start() {
+                  socket.send(new Blob(chunkDatas));
+                },
+                write(chunk, controller) {
+                  // ('' as any).toLowerCase1();
+                  socket.send(new Blob([chunk]));
+                },
+                close() {
+                  console.error(
+                    `[${address}:${port}] remoteConnection!.readable is close`
+                  );
+                  socket.close();
+                },
+                abort(reason) {
+                  socket.close();
+                  console.error(
+                    `[${address}:${port}] remoteConnection!.readable abort`,
+                    reason
+                  );
+                },
+              })
+            )
+            .catch((error: any) => {
+              socket.close();
+              console.error(
+                `[${address}:${port}] remoteConnection.readable has error`,
+                error
+              );
+            });
+        },
+        close() {
+          console.log(`[${address}:${port}] websocketStream pipeto is close`);
+        },
+        abort(reason) {
+          console.log(
+            `[${address}:${port}] websocketStream pipeto is abort `,
+            reason
+          );
+          remoteConnection?.close();
+          socket.close();
+        },
+      })
+    );
+  } catch (error: any) {
+    console.error(`[${address}:${port}] processSocket`, error);
+    socket.close();
+  }
+  return;
+}

libs/vless-js/tsconfig.json

@@ -1,7 +1,7 @@
 {
   "extends": "../../tsconfig.base.json",
   "compilerOptions": {
-    "module": "commonjs",
+    "module": "ESNext",
     "forceConsistentCasingInFileNames": true,
     "strict": true,
     "noImplicitOverride": true,

libs/vless-js/tsconfig.lib.json

@@ -1,6 +1,8 @@
 {
   "extends": "./tsconfig.json",
   "compilerOptions": {
+    "target": "esnext",
+    "module": "ESNext",
     "outDir": "../../dist/out-tsc",
     "declaration": true,
     "types": ["node"]

test.mjs

@@ -1,4 +1,44 @@
-import { Readable } from 'stream';
-const readableStream = new Readable();
-readableStream.push('ping!');
-readableStream.push('pong!');
+import { ReadableStream } from 'stream/web';
+
+try {
+  const readableStream = new ReadableStream({
+    start(control) {
+      setInterval(() => {
+        control.enqueue('11');
+      }, 100);
+    },
+    pull(control) {
+      // control.enqueue('11');
+      // undefined.length;
+      // control.close();
+      //   control.error('error');
+      //   undefined.length;
+    },
+    cancel(reason) {
+      console.log('---------', reason);
+    },
+  });
+
+  setTimeout(() => {
+    console.log('cancel');
+  }, 2000);
+
+  await readableStream.pipeTo(
+    new WritableStream({
+      write(chunk, controller) {
+        console.log(chunk);
+      },
+      close() {
+        console.log('close------WritableStream');
+      },
+    })
+  );
+
+  console.log('end--------');
+
+  //   for await (const iterator of readableStream) {
+  //     console.log(iterator);
+  //   }
+} catch (error) {
+  console.log('---end---', error);
+}