feat: basic test of VMess

Check.py

@@ -75,24 +75,23 @@ def proxyTest(
         }
     try:
         status, client = Builder.build(proxyInfo, workDir)
-    except: # 构建发生未知错误
+    except Exception as reason: # 构建发生错误
+        print(str(reason))
         Builder.destroy(client)
         return None
-    if status is None: # 构建错误
+    if not status: # 节点信息有误
-        Builder.destroy(client)
-        return None
-    elif not status: # 节点信息有误
         return {
             'success': False,
             'info': proxyInfo
         }
 
     time.sleep(startDelay) # 延迟等待客户端启动
-    status = Builder.check(client) # 检查客户端状态
+    try:
-    if status is None: # 检测失败
+        status = Builder.check(client) # 检查客户端状态
+    except: # 检测失败
         Builder.destroy(client)
         return None
-    elif not status: # 客户端异常退出
+    if not status: # 客户端异常退出
         Builder.destroy(client)
         return {
             'success': False,

ProxyBuilder/builder.py

@@ -76,13 +76,10 @@ def __getAvailablePort(rangeStart: int = 41952, rangeEnd: int = 65535) -> int or
         time.sleep(0.1) # wait for 100ms
 
 def build(proxyInfo: dict, configDir: str,
-          portRangeStart: int = 1024, portRangeEnd: int = 65535) -> tuple[bool or None, str or dict]:
+          portRangeStart: int = 1024, portRangeEnd: int = 65535) -> tuple[bool, str or dict]:
     """
     创建代理节点客户端
 
-        程序内部错误:
-            return None, {reason}
-
         代理节点无效:
             return False, {reason}
 
@@ -94,83 +91,72 @@ def build(proxyInfo: dict, configDir: str,
                 'process': process
             }
     """
+    taskFlag = __genTaskFlag() # 生成测试标志
+    socksPort = __getAvailablePort(portRangeStart, portRangeEnd) # 获取Socks5测试端口
+
+    if 'type' not in proxyInfo: # 未指定节点类型
+        return False, 'Proxy type not specified'
+    if proxyInfo['type'] == 'ss': # Shadowsocks节点
+        clientObj = Shadowsocks
+    elif proxyInfo['type'] == 'ssr': # ShadowsocksR节点
+        clientObj = ShadowsocksR
+    elif proxyInfo['type'] == 'vmess': # VMess节点
+        clientObj = VMess
+    elif proxyInfo['type'] == 'vless': # VLESS节点
+        clientObj = VLESS
+    else: # 未知类型
+        return False, 'Unknown proxy type'
+
+    configFile = configDir + '/' + taskFlag + '.json' # 配置文件路径
     try:
-        taskFlag = __genTaskFlag() # 生成测试标志
+        startCommand, fileContent, envVar = clientObj.load(proxyInfo, socksPort, configFile) # 载入配置
-        socksPort = __getAvailablePort(portRangeStart, portRangeEnd) # 获取Socks5测试端口
+    except: # 格式出错
-
+        return False, 'Format error with ' + str(proxyInfo['type'])
-        if 'type' not in proxyInfo: # 未指定节点类型
-            return False, 'Proxy type not specified'
-        if proxyInfo['type'] == 'ss': # Shadowsocks节点
-            clientObj = Shadowsocks
-        elif proxyInfo['type'] == 'ssr': # ShadowsocksR节点
-            clientObj = ShadowsocksR
-        elif proxyInfo['type'] == 'vmess': # VMess节点
-            clientObj = VMess
-        elif proxyInfo['type'] == 'vless': # VLESS节点
-            clientObj = VLESS
-        else: # 未知类型
-            return False, 'Unknown proxy type'
-
-        configFile = configDir + '/' + taskFlag + '.json' # 配置文件路径
-        try:
-            startCommand, fileContent, envVar = clientObj.load(proxyInfo, socksPort, configFile) # 载入配置
-        except: # 格式出错
-            return False, 'Format error with ' + str(proxyInfo['type'])
 
-        try:
+    try:
-            with open(configFile, 'w') as fileObject: # 保存配置文件
+        with open(configFile, 'w') as fileObject: # 保存配置文件
-                fileObject.write(fileContent)
+            fileObject.write(fileContent)
-        except: # 配置文件写入失败
+    except: # 配置文件写入失败
-            return None, "Unable write to file " + str(configFile)
+        raise Exception('Unable write to file ' + str(configFile))
-
+
-        process = None
+    process = None
-        try: # 子进程形式启动
+    try: # 子进程形式启动
-            for libcPath in libcPaths:
+        for libcPath in libcPaths:
-                if os.path.exists(libcPath): # 定位libc.so文件
+            if os.path.exists(libcPath): # 定位libc.so文件
-                    break
+                break
-            process = subprocess.Popen( # 启动子进程
+        process = subprocess.Popen( # 启动子进程
-                startCommand,
+            startCommand,
-                env = envVar,
+            env = envVar,
-                stdout = subprocess.DEVNULL,
+            stdout = subprocess.DEVNULL,
-                stderr = subprocess.DEVNULL,
+            stderr = subprocess.DEVNULL,
-                preexec_fn = lambda: ctypes.CDLL(libcPath).prctl(1, 15) # 子进程跟随退出
+            preexec_fn = lambda: ctypes.CDLL(libcPath).prctl(1, 15) # 子进程跟随退出
-            )
+        )
-        except:
-            try:
-                process = subprocess.Popen( # prctl失败 回退正常启动
-                    startCommand,
-                    env = envVar,
-                    stdout = subprocess.DEVNULL,
-                    stderr = subprocess.DEVNULL
-                )
-            except:
-                pass
-        if process is None: # 启动失败
-            return None, 'Subprocess start failed by `' + ' '.join(startCommand) + '`'
-
-        return True, { # 返回连接参数
-            'flag': taskFlag,
-            'port': socksPort,
-            'file': configFile,
-            'process': process
-        }
     except:
-        return None, 'Unknown error'
+        process = subprocess.Popen( # prctl失败 回退正常启动
+            startCommand,
+            env = envVar,
+            stdout = subprocess.DEVNULL,
+            stderr = subprocess.DEVNULL
+        )
+    if process is None: # 启动失败
+        raise Exception('Subprocess start failed by `' + ' '.join(startCommand) + '`')
+
+    return True, { # 返回连接参数
+        'flag': taskFlag,
+        'port': socksPort,
+        'file': configFile,
+        'process': process
+    }
 
 def check(client: dict) -> bool or None:
     """
     检查客户端是否正常运行
 
-        检测出错: return None
-
         工作异常: return False
 
         工作正常: return True
     """
-    try:
+    return client['process'].poll() is None
-        return client['process'].poll() is None
-    except:
-        return None # 异常
 
 def destroy(client: dict) -> bool:
     """

ProxyTester/VMess.py

@@ -0,0 +1,84 @@
+#!/usr/bin/python
+# -*- coding:utf-8 -*-
+
+import copy
+import json
+
+config = {}
+
+vmessMethodList = [
+    'aes-128-gcm',
+    'chacha20-poly1305',
+    'auto',
+    'none',
+    'zero',
+]
+
+def loadServerConfig(inboundObject: dict) -> str:
+    return json.dumps({
+        'log': {
+            'loglevel': 'warning'
+        },
+        'inbounds': [inboundObject],
+        'outbounds': [
+            {
+                'protocol': 'freedom'
+            }
+        ]
+    })
+
+def basicConfig(method: str, alterId: int):
+    filePath = '/tmp/v2ray.json'
+
+    inboundObject = {
+        'protocol': 'vmess',
+        'listen': '127.0.0.1',
+        'port': config['port'],
+        'settings': {
+            'clients': [
+                {
+                    'id': config['id'],
+                    'alterId': alterId
+                }
+            ]
+        }
+    }
+
+    caption = 'VMess method ' + method
+    if alterId == 0:
+        envVar = {}
+        caption += ' (AEAD)'
+    else:
+        envVar = {
+            'v2ray.vmess.aead.forced': 'false'
+        }
+        caption += ' (alterId ' + str(alterId) + ')'
+
+    return {
+        'caption': caption,
+        'proxy': {
+            'type': 'vmess',
+            'server': '127.0.0.1',
+            'port': config['port'],
+            'method': method,
+            'id': config['id'],
+            'aid': alterId
+        },
+        'server': {
+            'startCommand': ['v2ray', '-c', filePath],
+            'fileContent': loadServerConfig(inboundObject),
+            'filePath': filePath,
+            'envVar': envVar
+        },
+        'aider': None
+    }
+
+def vmessTest(vmessConfig: dict) -> list:
+    result = []
+    for key, value in vmessConfig.items(): # vmessConfig -> config
+        config[key] = value
+    for method in vmessMethodList: # methods and AEAD/MD5+AES test
+        result.append(basicConfig(method, 0))
+        result.append(basicConfig(method, 64))
+
+    return result

ProxyTester/tester.py

@@ -3,6 +3,7 @@
 
 from ProxyTester import Shadowsocks
 from ProxyTester import ShadowsocksR
+from ProxyTester import VMess
 
 def test(key: str, config: dict) -> list:
     if key == 'ss':
@@ -10,6 +11,6 @@ def test(key: str, config: dict) -> list:
     elif key == 'ssr':
         return ShadowsocksR.ssrTest(config)
     elif key == 'vmess':
-        pass
+        return VMess.vmessTest(config)
     else:
         return []

Test.py

@@ -14,7 +14,8 @@ testConfig = {
     'passwd': 'dnomd343',
     'host': 'dns.343.re',
     'cert': '/etc/ssl/certs/dns.343.re/certificate.crt',
-    'key': '/etc/ssl/certs/dns.343.re/private.key'
+    'key': '/etc/ssl/certs/dns.343.re/private.key',
+    'id': '1f7aa040-94d8-4b53-ae85-af6946d550bb',
 }
 
 def testBuild(config: dict): # load file and start process