You can not select more than 25 topics Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.
 
 

112 lines
4.2 KiB

#!/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 Tester.Settings import Settings
from Basis.Constant import trojanGoMethods
from Basis.Functions import md5Sum, genFlag, getAvailablePort
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']}
testTitle = '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']))
testInfo = { # release test info
'caption': testTitle,
'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)