feat: Trojan-Go test function

Basis/Methods.py

@@ -124,3 +124,6 @@ xtlsFlows = {x: x.replace('-', '-rprx-') for x in xtlsFlows}
 # v2ray / Xray Info
 quicMethods = ['none', 'aes-128-gcm', 'chacha20-poly1305']
 udpObfuscations = ['none', 'srtp', 'utp', 'wechat-video', 'dtls', 'wireguard']
+
+# Trojan-Go Info
+trojanGoMethods = ['aes-128-gcm', 'aes-256-gcm', 'chacha20-ietf-poly1305']

Builder/TrojanGo.py

@@ -59,6 +59,6 @@ def load(proxyInfo: dict, socksInfo: dict, configFile: str) -> tuple[list, str,
         'ssl': sslConfig(proxyInfo),
         'websocket': wsConfig(proxyInfo),
         'shadowsocks': ssConfig(proxyInfo),
-        'transport_plugin': pluginConfig(proxyInfo)
+        'transport_plugin': pluginConfig(proxyInfo),
     }
     return ['trojan-go', '-config', configFile], json.dumps(trojanGoConfig), {}

Tester/Plugin.py

@@ -256,7 +256,41 @@ def cloakLoad() -> None:
     ]
 
 
-def ssInject(server: Process, pluginInfo: dict) -> Process:
+def rabbitShadowsocks(server: Process, pluginInfo: dict) -> Process:
+    ssConfig = json.loads(server.file[0]['content'])  # modify origin config
+    ssConfig.pop('plugin')  # remove plugin option
+    ssConfig.pop('plugin_opts')
+    rabbitBind = ipFormat(ssConfig['server'], v6Bracket=True)  # ipv4 / [ipv6]
+    rabbitPort = ssConfig['server_port']
+    ssConfig['server'] = '127.0.0.1'  # SIP003 use ipv4 localhost for communication
+    ssConfig['server_port'] = int(pluginInfo['server']['param'])  # aka ${RABBIT_PORT}
+    server.file[0]['content'] = json.dumps(ssConfig)
+    server.setCmd(['sh', '-c', paramFill(
+        'rabbit -mode s -password ${PASSWD} -rabbit-addr %s:%s' % (rabbitBind, rabbitPort)  # start rabbit-tcp
+    ) + ' &\nexec ' + ' '.join(server.cmd)])  # shadowsocks as main process (rabbit as sub process)
+    return server
+
+
+def rabbitTrojanGo(server: Process, pluginInfo: dict) -> Process:
+    trojanConfig = json.loads(server.file[0]['content'])  # modify origin config
+    rabbitBind = ipFormat(trojanConfig['local_addr'], v6Bracket=True)  # ipv4 / [ipv6]
+    rabbitPort = trojanConfig['local_port']
+    trojanConfig['local_addr'] = '127.0.0.1'  # SIP003 use ipv4 localhost for communication
+    trojanConfig['local_port'] = int(pluginInfo['server']['param'])  # aka ${RABBIT_PORT}
+    trojanConfig['transport_plugin'] = {
+        'enabled': True,
+        'type': 'other',
+        'command': 'rabbit',
+        'arg': [
+            '-mode', 's', '-password', paramFill('${PASSWD}'),
+            '-rabbit-addr', '%s:%s' % (rabbitBind, rabbitPort)
+        ]
+    }
+    server.file[0]['content'] = json.dumps(trojanConfig)
+    return server
+
+
+def inject(server: Process, pluginInfo: dict) -> Process:
     if pluginInfo['type'] == 'cloak':
         ckConfig = paramFill(json.dumps({
             'BypassUID': ['${CK_UID}'],
@@ -272,21 +306,21 @@ def ssInject(server: Process, pluginInfo: dict) -> Process:
             'path': pluginInfo['server']['param'],
             'content': paramFill(json.dumps({'key': '${PASSWD}'}))
         }])
-    elif pluginInfo['type'] == 'rabbit':  # hijack rabbit plugin config
-        ssConfig = json.loads(server.file[0]['content'])  # modify origin config
-        ssConfig.pop('plugin')  # remove plugin option
-        ssConfig.pop('plugin_opts')
-        rabbitBind = ipFormat(ssConfig['server'], v6Bracket = True)  # ipv4 / [ipv6]
-        rabbitPort = ssConfig['server_port']
-        ssConfig['server'] = '127.0.0.1'  # SIP003 use ipv4 localhost for communication
-        ssConfig['server_port'] = int(pluginInfo['server']['param'])  # aka ${RABBIT_PORT}
-        server.file[0]['content'] = json.dumps(ssConfig)
-        server.setCmd(['sh', '-c', paramFill(
-            'rabbit -mode s -password ${PASSWD} -rabbit-addr %s:%s' % (rabbitBind, rabbitPort)  # start rabbit-tcp
-        ) + ' &\nexec ' + ' '.join(server.cmd)])  # shadowsocks as main process (rabbit as sub process)
     return server
 
 
+def ssInject(server: Process, pluginInfo: dict) -> Process:
+    if pluginInfo['type'] == 'rabbit':  # hijack rabbit plugin config
+        return rabbitShadowsocks(server, pluginInfo)
+    return inject(server, pluginInfo)
+
+
+def trojanInject(server: Process, pluginInfo: dict) -> Process:
+    if pluginInfo['type'] == 'rabbit':  # hijack rabbit plugin config
+        return rabbitTrojanGo(server, pluginInfo)
+    return inject(server, pluginInfo)
+
+
 def paramFill(param: str) -> str:
     if '${RANDOM}' in param:  # refresh RANDOM field
         pluginParams['RANDOM'] = genFlag(length = 8)
@@ -295,7 +329,9 @@ def paramFill(param: str) -> str:
     return param
 
 
-def load():
+def load(proxyType: str):
+    if proxyType not in ['ss', 'trojan-go']:
+        raise RuntimeError('Unknown proxy type for sip003 plugin')
     cloakLoad()  # init cloak config
     kcptunLoad()  # init kcptun config
     for pluginType in pluginConfig:
@@ -313,5 +349,5 @@ def load():
                     'type': plugins[pluginType]['client'],
                     'param': paramFill(pluginTestInfo[1]),
                 },
-                'inject': ssInject  # for some special plugins (only server part)
+                'inject': ssInject if proxyType == 'ss' else trojanInject  # for some special plugins
             }

Tester/Shadowsocks.py

@@ -129,7 +129,7 @@ def loadTest(serverType: str, clientType: str, method: str, plugin: dict or None
     pluginServer = {'plugin': None if plugin is None else plugin['server']}
     configName = '%s_%s_%s' % (serverType, clientType, method)  # prefix of config file name
     if plugin is not None:
-        configName += '_%s_%s' % (plugin['type'], md5Sum(plugin['caption'])[:8])
+        configName += '_%s_%s' % (plugin['type'], md5Sum(plugin['type'] + plugin['caption'])[:8])
     pluginText = '' if plugin is None else (' [%s -> %s]' % (plugin['type'], plugin['caption']))
     testInfo = {  # release test info
         'title': 'Shadowsocks test: {%s <- %s -> %s}%s' % (serverType, method, clientType, pluginText),
@@ -149,7 +149,7 @@ def loadTest(serverType: str, clientType: str, method: str, plugin: dict or None
 
 def load(isExtra: bool = False):
     pluginTest = []
-    pluginIter = Plugin.load()
+    pluginIter = Plugin.load('ss')
     while True:
         try:
             pluginTest.append(next(pluginIter))  # export data of plugin generator

Tester/TrojanGo.py

@@ -0,0 +1,154 @@
+#!/usr/bin/env python3
+# -*- coding: utf-8 -*-
+
+import os
+import json
+from Tester import Plugin
+from Builder import TrojanGo
+from Basis.Logger import logging
+from Basis.Process import Process
+from Basis.Functions import md5Sum
+from Basis.Functions import genFlag
+from Basis.Methods import trojanGoMethods
+from Basis.Functions import getAvailablePort
+
+settings = {
+    'serverBind': '127.0.0.1',
+    'clientBind': '127.0.0.1',
+    # 'serverBind': '::1',
+    # 'clientBind': '::1',
+    'workDir': '/tmp/ProxyC',
+    'host': '343.re',
+    'cert': '/etc/ssl/certs/343.re/fullchain.pem',
+    'key': '/etc/ssl/certs/343.re/privkey.pem',
+}
+
+
+# def wsConfig(proxyInfo: dict) -> dict:
+#     if proxyInfo['ws'] is None:
+#         return {'enabled': False}
+#     wsObject = {
+#         'enabled': True,
+#         'path': proxyInfo['ws']['path']
+#     }
+#     if proxyInfo['ws']['host'] != '':
+#         wsObject['host'] = proxyInfo['ws']['host']
+#     return wsObject
+#
+#
+# def ssConfig(proxyInfo: dict) -> dict:
+#     return {**{
+#         'enabled': False if proxyInfo['ss'] is None else True
+#     }, **({} if proxyInfo['ss'] is None else {
+#         'method': proxyInfo['ss']['method'],
+#         'password': proxyInfo['ss']['passwd'],
+#     })}
+#
+#
+# def pluginConfig(proxyInfo: dict) -> dict:
+#     return {**{
+#         'enabled': False if proxyInfo['plugin'] is None else True
+#     }, **({} if proxyInfo['plugin'] is None else {
+#         'type': 'shadowsocks',
+#         'command': proxyInfo['plugin']['type'],
+#         'option': proxyInfo['plugin']['param'],
+#     })}
+
+
+def loadServer(configFile: str, proxyInfo: dict) -> Process:
+    trojanGoConfig = {
+        'run_type': 'server',
+        'local_addr': proxyInfo['server'],
+        'local_port': proxyInfo['port'],
+        'remote_addr': '127.0.0.1',  # remote address are only for shadowsocks fallback (pointless here)
+        'remote_port': getAvailablePort(),  # random port (will not be used)
+        'password': [
+            proxyInfo['passwd']
+        ],
+        'disable_http_check': True,
+        'ssl': {
+            'cert': settings['cert'],
+            'key': settings['key']
+        },
+        'websocket': TrojanGo.wsConfig(proxyInfo),
+        'shadowsocks': TrojanGo.ssConfig(proxyInfo),
+        'transport_plugin': TrojanGo.pluginConfig(proxyInfo),
+    }
+    serverFile = os.path.join(settings['workDir'], configFile)
+    return Process(settings['workDir'], cmd = ['trojan-go', '-config', serverFile], file = {
+        'path': serverFile,
+        'content': json.dumps(trojanGoConfig)
+    }, isStart = False)
+
+
+def loadClient(configFile: str, proxyInfo: dict, socksInfo: dict) -> Process:  # load client process
+    clientFile = os.path.join(settings['workDir'], configFile)
+    trojanGoCommand, trojanGoConfig, _ = TrojanGo.load(proxyInfo, socksInfo, clientFile)
+    return Process(settings['workDir'], cmd = trojanGoCommand, file = {
+        'path': clientFile,
+        'content': trojanGoConfig
+    }, isStart = False)
+
+
+def loadTest(wsObject: dict or None, ssObject: dict or None, plugin: dict or None = None) -> dict:
+    proxyInfo = {  # connection info
+        'server': settings['serverBind'],
+        'port': getAvailablePort(),
+        'passwd': genFlag(length = 8),  # random password
+        'sni': settings['host'],
+        'alpn': None,
+        'verify': True,
+        'ws': wsObject,
+        'ss': ssObject,
+    }
+    socksInfo = {  # socks5 interface for test
+        'addr': settings['clientBind'],
+        'port': getAvailablePort()
+    }
+    configName = 'trojan-go%s%s%s' % (
+        ('' if wsObject is None else '_ws'),
+        ('' if ssObject is None else '_' + ssObject['method']),
+        ('' if plugin is None else '_' + md5Sum(plugin['type'] + plugin['caption'])[:8])
+    )
+    pluginClient = {'plugin': None if plugin is None else plugin['client']}
+    pluginServer = {'plugin': None if plugin is None else plugin['server']}
+    testInfo = {  # release test info
+        'title': 'Trojan-Go test: original' + \
+                 ('' if ssObject is None else ' (with %s encrypt)' % ssObject['method']) + \
+                 ('' if wsObject is None else ' (with websocket)') + \
+                 ('' if plugin is None else ' [%s -> %s]' % (plugin['type'], plugin['caption'])),
+        'client': loadClient(configName + '_client.json', {**proxyInfo, **pluginClient}, socksInfo),
+        'server': loadServer(configName + '_server.json', {**proxyInfo, **pluginServer}),
+        'socks': socksInfo,  # exposed socks5 address
+        'interface': {
+            'addr': proxyInfo['server'],
+            'port': proxyInfo['port'],
+        }
+    }
+    if plugin is not None:
+        testInfo['server'] = plugin['inject'](testInfo['server'], plugin)
+    logging.debug('New trojan-go test -> %s' % testInfo)
+    return testInfo
+
+
+def load():
+    pluginTest = []
+    pluginIter = Plugin.load('trojan-go')
+    while True:
+        try:
+            pluginTest.append(next(pluginIter))  # export data of plugin generator
+        except StopIteration:
+            break
+    wsObject = {
+        'host': settings['host'],
+        'path': '/' + genFlag(length = 6),
+    }
+    yield loadTest(None, None, None)
+    for method in [''] + trojanGoMethods:  # different encryption for trojan-go
+        ssObject = {
+            'method': method,
+            'passwd': genFlag(length = 8)
+        }
+        yield loadTest(wsObject, None if ssObject['method'] == '' else ssObject, None)
+    for plugin in pluginTest:  # different plugin for trojan-go
+        yield loadTest(None, None, plugin)

test.py

@@ -8,6 +8,7 @@ from threading import Thread
 from Tester import VMess
 from Tester import VLESS
 from Tester import Trojan
+from Tester import TrojanGo
 from Tester import Shadowsocks
 from Tester import ShadowsocksR
 
@@ -99,11 +100,13 @@ ssr = ShadowsocksR.load()
 vmess = VMess.load()
 vless = VLESS.load()
 trojan = Trojan.load()
+trojanGo = TrojanGo.load()
 
 logging.critical('test start')
 # runTest(ss, 64)
 # runTest(ssr, 64)
 # runTest(vmess, 64)
 # runTest(vless, 64)
-runTest(trojan, 64)
+# runTest(trojan, 64)
+runTest(trojanGo, 64)
 logging.critical('test complete')