From 5cc5237c8e0524d94d9ad68e9a2ec557e383dfcb Mon Sep 17 00:00:00 2001 From: dnomd343 Date: Sun, 31 Jul 2022 22:18:57 +0800 Subject: [PATCH] update: global test process --- Tester/Brook.py | 26 +++++---------- Tester/Hysteria.py | 30 ++++++------------ Tester/Plugin.py | 33 +++++++------------ Tester/Shadowsocks.py | 24 ++++++-------- Tester/ShadowsocksR.py | 26 ++++++--------- Tester/Trojan.py | 36 ++++++++------------- Tester/TrojanGo.py | 72 ++++++++++-------------------------------- Tester/V2ray.py | 25 ++++++--------- Tester/VLESS.py | 21 ++++-------- Tester/VMess.py | 26 ++++++--------- Tester/Xray.py | 19 ++++------- Tester/__init__.py | 12 +++++++ test.py | 14 ++++---- 13 files changed, 128 insertions(+), 236 deletions(-) create mode 100644 Tester/__init__.py diff --git a/Tester/Brook.py b/Tester/Brook.py index 30eb8c7..02b4017 100644 --- a/Tester/Brook.py +++ b/Tester/Brook.py @@ -4,23 +4,13 @@ import copy import itertools from Builder import Brook +from Tester import Settings from Basis.Logger import logging from Basis.Process import Process from Basis.Functions import genFlag from Basis.Functions import hostFormat 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 originStream(isUot: bool) -> dict: return { @@ -45,8 +35,8 @@ def loadWsCommand(proxyInfo: dict) -> list: # load start command for brook serv '--path', proxyInfo['stream']['path'], '--password', proxyInfo['passwd'], ] + ([] if proxyInfo['stream']['secure'] is None else [ - '--cert', settings['cert'], - '--certkey', settings['key'], + '--cert', Settings['cert'], + '--certkey', Settings['key'], ]) + (['--withoutBrookProtocol'] if proxyInfo['stream']['raw'] else []) @@ -55,7 +45,7 @@ def wsStream(isRaw: bool, isSecure: bool): 'caption': 'websocket' + (' (with tls)' if isSecure else '') + (' (without brook)' if isRaw else ''), 'info': { 'type': 'ws', - 'host': settings['host'], + 'host': Settings['host'], 'path': '/' + genFlag(length = 6), 'raw': isRaw, 'secure': {'verify': True} if isSecure else None, @@ -66,21 +56,21 @@ def wsStream(isRaw: bool, isSecure: bool): def loadTest(stream: dict) -> dict: proxyInfo = { - 'server': settings['serverBind'], + 'server': Settings['serverBind'], 'port': getAvailablePort(), 'passwd': genFlag(), 'stream': stream['info'] } socksInfo = { # socks5 interface for test - 'addr': settings['clientBind'], + 'addr': Settings['clientBind'], 'port': getAvailablePort() } clientCommand, _, _ = Brook.load(proxyInfo, socksInfo, '') serverCommand = ['brook', '--debug', '--listen', ':'] + stream['command'](proxyInfo) testInfo = { # release test info 'title': 'Brook test: ' + stream['caption'], - 'client': Process(settings['workDir'], cmd = clientCommand, isStart = False), - 'server': Process(settings['workDir'], cmd = serverCommand, isStart = False), + 'client': Process(Settings['workDir'], cmd = clientCommand, isStart = False), + 'server': Process(Settings['workDir'], cmd = serverCommand, isStart = False), 'socks': socksInfo, # exposed socks5 address 'interface': { 'addr': proxyInfo['server'], diff --git a/Tester/Hysteria.py b/Tester/Hysteria.py index 914b6a1..112481d 100644 --- a/Tester/Hysteria.py +++ b/Tester/Hysteria.py @@ -4,6 +4,7 @@ import os import json import itertools +from Tester import Settings from Builder import Hysteria from Basis.Logger import logging from Basis.Process import Process @@ -12,30 +13,19 @@ from Basis.Functions import hostFormat from Basis.Methods import hysteriaProtocols 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 loadServer(configFile: str, hysteriaConfig: dict) -> Process: - serverFile = os.path.join(settings['workDir'], configFile) - return Process(settings['workDir'], cmd = ['hysteria', '-c', serverFile, 'server'], file = { + serverFile = os.path.join(Settings['workDir'], configFile) + return Process(Settings['workDir'], cmd = ['hysteria', '-c', serverFile, 'server'], file = { 'path': serverFile, 'content': json.dumps(hysteriaConfig) }, isStart = False) def loadClient(configFile: str, proxyInfo: dict, socksInfo: dict) -> Process: - clientFile = os.path.join(settings['workDir'], configFile) + clientFile = os.path.join(Settings['workDir'], configFile) hysteriaCommand, hysteriaConfig, _ = Hysteria.load(proxyInfo, socksInfo, clientFile) - return Process(settings['workDir'], cmd = hysteriaCommand, file = { + return Process(Settings['workDir'], cmd = hysteriaCommand, file = { 'path': clientFile, 'content': hysteriaConfig }, isStart = False) @@ -44,26 +34,26 @@ def loadClient(configFile: str, proxyInfo: dict, socksInfo: dict) -> Process: def loadTest(protocol: str, isObfs: bool, isAuth: bool) -> dict: proxyInfo = { 'type': 'hysteria', - 'server': settings['serverBind'], + 'server': Settings['serverBind'], 'port': getAvailablePort(), 'protocol': protocol, 'obfs': None, 'passwd': None, 'up': 10, 'down': 50, - 'sni': settings['host'], + 'sni': Settings['host'], 'alpn': None, 'verify': True, } socksInfo = { # socks5 interface for test - 'addr': settings['clientBind'], + 'addr': Settings['clientBind'], 'port': getAvailablePort() } serverConfig = { 'listen': '%s:%i' % (hostFormat(proxyInfo['server'], v6Bracket = True), proxyInfo['port']), 'protocol': proxyInfo['protocol'], - 'cert': settings['cert'], - 'key': settings['key'], + 'cert': Settings['cert'], + 'key': Settings['key'], } configName = 'hysteria_' + protocol caption = 'Hysteria protocol ' + protocol diff --git a/Tester/Plugin.py b/Tester/Plugin.py index a48fc6d..faaa63e 100644 --- a/Tester/Plugin.py +++ b/Tester/Plugin.py @@ -4,6 +4,7 @@ import os import re import json +from Tester import Settings from Basis.Logger import logging from Basis.Process import Process from Basis.Methods import plugins @@ -11,22 +12,12 @@ from Basis.Functions import genFlag from Basis.Functions import hostFormat from Basis.Functions import getAvailablePort -settings = { - 'serverBind': '127.0.0.1', - 'workDir': '/tmp/ProxyC', - 'site': 'www.bing.com', - 'host': '343.re', - 'cert': '/etc/ssl/certs/343.re/fullchain.pem', - 'key': '/etc/ssl/certs/343.re/privkey.pem', -} pluginParams = { - 'SITE': settings['site'], - 'HOST': settings['host'], - 'CERT': settings['cert'], - 'KEY': settings['key'], - 'PATH': '/' + genFlag(length = 6), - 'PASSWD': genFlag(length = 8), + 'SITE': Settings['site'], + 'HOST': Settings['host'], + 'CERT': Settings['cert'], + 'KEY': Settings['key'], } pluginConfig = { @@ -169,11 +160,11 @@ pluginConfig = { 'cloak': {}, 'go-quiet': { 'chrome fingerprint': [ - os.path.join(settings['workDir'], 'go-quiet_config_${RANDOM}.json'), + os.path.join(Settings['workDir'], 'go-quiet_config_${RANDOM}.json'), 'ServerName=${SITE};key=${PASSWD};TicketTimeHint=300;Browser=chrome', ], 'firefox fingerprint': [ - os.path.join(settings['workDir'], 'go-quiet_config_${RANDOM}.json'), + os.path.join(Settings['workDir'], 'go-quiet_config_${RANDOM}.json'), 'ServerName=${SITE};key=${PASSWD};TicketTimeHint=300;Browser=firefox', ], }, @@ -242,7 +233,7 @@ def cloakLoad() -> None: pluginParams['CK_PUBLIC'], pluginParams['CK_PRIVATE'] )) ckPrefix = 'UID=${CK_UID};PublicKey=${CK_PUBLIC};ServerName=${SITE};' # cloak plugin's basic command - ckConfigPath = os.path.join(settings['workDir'], 'cloak_config_${RANDOM}.json') # clock server's config + ckConfigPath = os.path.join(Settings['workDir'], 'cloak_config_${RANDOM}.json') # clock server's config for ckMethod in ['plain', 'aes-128-gcm', 'aes-256-gcm', 'chacha20-poly1305']: # traverse cloak encrypt methods pluginConfig['cloak']['%s method' % ckMethod] = [ ckConfigPath, ckPrefix + 'EncryptionMethod=' + ckMethod @@ -322,8 +313,6 @@ def trojanInject(server: Process, pluginInfo: dict) -> Process: def paramFill(param: str) -> str: - if '${RANDOM}' in param: # refresh RANDOM field - pluginParams['RANDOM'] = genFlag(length = 8) for field in pluginParams: param = param.replace('${%s}' % field, pluginParams[field]) # fill ${XXX} field return param @@ -334,10 +323,12 @@ def load(proxyType: str): raise RuntimeError('Unknown proxy type for sip003 plugin') cloakLoad() # init cloak config kcptunLoad() # init kcptun config + pluginParams['PASSWD'] = genFlag(length = 8) # random password for test + pluginParams['PATH'] = '/' + genFlag(length = 6) # random uri path for test for pluginType in pluginConfig: for pluginTest, pluginTestInfo in pluginConfig[pluginType].items(): # traverse all plugin test item - if pluginType == 'rabbit': - pluginParams['RABBIT_PORT'] = str(getAvailablePort()) # allocate port before rabbit plugin start + pluginParams['RANDOM'] = genFlag(length = 8) # refresh RANDOM field + pluginParams['RABBIT_PORT'] = str(getAvailablePort()) # allocate port before rabbit plugin start yield { 'type': pluginType, 'caption': pluginTest, diff --git a/Tester/Shadowsocks.py b/Tester/Shadowsocks.py index a88908b..0491e89 100644 --- a/Tester/Shadowsocks.py +++ b/Tester/Shadowsocks.py @@ -6,20 +6,14 @@ import json import base64 import itertools from Tester import Plugin +from Tester import Settings from Builder import Shadowsocks from Basis.Logger import logging from Basis.Process import Process from Basis.Functions import md5Sum +from Basis.Functions import genFlag +from Basis.Functions import getAvailablePort from Basis.Methods import ssMethods, ssAllMethods -from Basis.Functions import genFlag, getAvailablePort - -settings = { - 'serverBind': '127.0.0.1', - 'clientBind': '127.0.0.1', - # 'serverBind': '::1', - # 'clientBind': '::1', - 'workDir': '/tmp/ProxyC' -} def loadConfig(proxyInfo: dict) -> dict: # load basic config option @@ -93,8 +87,8 @@ def loadClient(ssType: str, configFile: str, proxyInfo: dict, socksInfo: dict) - 'ss-python': Shadowsocks.ssPython, 'ss-python-legacy': Shadowsocks.ssPythonLegacy }[ssType](proxyInfo, socksInfo, isUdp = False) # disable udp in test mode - clientFile = os.path.join(settings['workDir'], configFile) - return Process(settings['workDir'], cmd = ssClient + ['-c', clientFile], file = { # load client process + clientFile = os.path.join(Settings['workDir'], configFile) + return Process(Settings['workDir'], cmd = ssClient + ['-c', clientFile], file = { # load client process 'path': clientFile, 'content': json.dumps(ssConfig) }, isStart = False) @@ -107,8 +101,8 @@ def loadServer(ssType: str, configFile: str, proxyInfo: dict) -> Process: 'ss-python': ssPython, 'ss-python-legacy': ssPythonLegacy }[ssType](proxyInfo, isUdp = False) # disable udp in test mode - serverFile = os.path.join(settings['workDir'], configFile) - return Process(settings['workDir'], cmd = ssServer + ['-c', serverFile], file = { # load server process + serverFile = os.path.join(Settings['workDir'], configFile) + return Process(Settings['workDir'], cmd = ssServer + ['-c', serverFile], file = { # load server process 'path': serverFile, 'content': json.dumps(ssConfig) }, isStart = False) @@ -116,13 +110,13 @@ def loadServer(ssType: str, configFile: str, proxyInfo: dict) -> Process: def loadTest(serverType: str, clientType: str, method: str, plugin: dict or None = None) -> dict: proxyInfo = { # connection info - 'server': settings['serverBind'], + 'server': Settings['serverBind'], 'port': getAvailablePort(), 'method': method, 'passwd': loadPassword(method), } socksInfo = { # socks5 interface for test - 'addr': settings['clientBind'], + 'addr': Settings['clientBind'], 'port': getAvailablePort(), } pluginClient = {'plugin': None if plugin is None else plugin['client']} diff --git a/Tester/ShadowsocksR.py b/Tester/ShadowsocksR.py index 0210593..50303bd 100644 --- a/Tester/ShadowsocksR.py +++ b/Tester/ShadowsocksR.py @@ -3,20 +3,14 @@ import os import json -from Builder import ShadowsocksR +from Tester import Settings from Basis.Logger import logging +from Builder import ShadowsocksR from Basis.Process import Process -from Basis.Functions import genFlag, getAvailablePort +from Basis.Functions import genFlag +from Basis.Functions import getAvailablePort from Basis.Methods import ssrMethods, ssrProtocols, ssrObfuscations -settings = { - 'serverBind': '127.0.0.1', - 'clientBind': '127.0.0.1', - # 'serverBind': '::1', - # 'clientBind': '::1', - 'workDir': '/tmp/ProxyC' -} - def loadServer(configFile: str, proxyInfo: dict) -> Process: # load server process ssrConfig = { @@ -29,17 +23,17 @@ def loadServer(configFile: str, proxyInfo: dict) -> Process: # load server proc 'obfs': proxyInfo['obfs'], 'obfs_param': proxyInfo['obfsParam'], } - serverFile = os.path.join(settings['workDir'], configFile) - return Process(settings['workDir'], cmd = ['ssr-server', '-vv', '-c', serverFile], file = { + serverFile = os.path.join(Settings['workDir'], configFile) + return Process(Settings['workDir'], cmd = ['ssr-server', '-vv', '-c', serverFile], file = { 'path': serverFile, 'content': json.dumps(ssrConfig) }, isStart = False) def loadClient(configFile: str, proxyInfo: dict, socksInfo: dict) -> Process: # load client process - clientFile = os.path.join(settings['workDir'], configFile) + clientFile = os.path.join(Settings['workDir'], configFile) ssrCommand, ssrConfig, _ = ShadowsocksR.load(proxyInfo, socksInfo, clientFile) - return Process(settings['workDir'], cmd = ssrCommand, file = { + return Process(Settings['workDir'], cmd = ssrCommand, file = { 'path': clientFile, 'content': ssrConfig }, isStart = False) @@ -47,7 +41,7 @@ def loadClient(configFile: str, proxyInfo: dict, socksInfo: dict) -> Process: # def loadTest(method: str, protocol: str, obfs: str) -> dict: proxyInfo = { # connection info - 'server': settings['serverBind'], + 'server': Settings['serverBind'], 'port': getAvailablePort(), 'passwd': genFlag(length = 8), # random password 'method': method, @@ -57,7 +51,7 @@ def loadTest(method: str, protocol: str, obfs: str) -> dict: 'obfsParam': '', } socksInfo = { # socks5 interface for test - 'addr': settings['clientBind'], + 'addr': Settings['clientBind'], 'port': getAvailablePort() } configName = 'ssr_%s_%s_%s' % (method, protocol, obfs) # prefix of config file name diff --git a/Tester/Trojan.py b/Tester/Trojan.py index 1a0da42..19019a2 100644 --- a/Tester/Trojan.py +++ b/Tester/Trojan.py @@ -5,6 +5,7 @@ import os import json from Tester import Xray from Builder import Trojan +from Tester import Settings from Basis.Logger import logging from Basis.Process import Process from Basis.Functions import md5Sum @@ -12,17 +13,6 @@ from Basis.Methods import xtlsFlows from Basis.Functions import genFlag 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 loadServer(configFile: str, proxyInfo: dict, streamConfig: dict, xtlsFlow: str or None) -> Process: trojanConfig = Xray.loadConfig({ @@ -38,17 +28,17 @@ def loadServer(configFile: str, proxyInfo: dict, streamConfig: dict, xtlsFlow: s }, 'streamSettings': streamConfig }) - serverFile = os.path.join(settings['workDir'], configFile) - return Process(settings['workDir'], cmd = ['xray', '-c', serverFile], file = { + serverFile = os.path.join(Settings['workDir'], configFile) + return Process(Settings['workDir'], cmd = ['xray', '-c', serverFile], file = { 'path': serverFile, 'content': json.dumps(trojanConfig) }, isStart = False) def loadClient(configFile: str, proxyInfo: dict, socksInfo: dict) -> Process: # load client process - clientFile = os.path.join(settings['workDir'], configFile) + clientFile = os.path.join(Settings['workDir'], configFile) trojanCommand, trojanConfig, _ = Trojan.load(proxyInfo, socksInfo, clientFile) - return Process(settings['workDir'], cmd = trojanCommand, file = { + return Process(Settings['workDir'], cmd = trojanCommand, file = { 'path': clientFile, 'content': trojanConfig }, isStart = False) @@ -56,13 +46,13 @@ def loadClient(configFile: str, proxyInfo: dict, socksInfo: dict) -> Process: # def loadBasicTest(tcpTlsStream: dict) -> dict: proxyInfo = { # connection info - 'server': settings['serverBind'], + 'server': Settings['serverBind'], 'port': getAvailablePort(), 'passwd': genFlag(length = 8), # random password 'stream': tcpTlsStream['info'] } socksInfo = { # socks5 interface for test - 'addr': settings['clientBind'], + 'addr': Settings['clientBind'], 'port': getAvailablePort() } trojanConfig = { @@ -72,12 +62,12 @@ def loadBasicTest(tcpTlsStream: dict) -> dict: 'password': [proxyInfo['passwd']], 'log_level': 0, # 0 -> ALL / 1 -> INFO / 2 -> WARN / 3 -> ERROR / 4 -> FATAL / 5 -> OFF 'ssl': { - 'cert': settings['cert'], - 'key': settings['key'] + 'cert': Settings['cert'], + 'key': Settings['key'] } } - serverFile = os.path.join(settings['workDir'], 'trojan_basic_server.json') - trojanServer = Process(settings['workDir'], cmd = ['trojan', '-c', serverFile], file = { + serverFile = os.path.join(Settings['workDir'], 'trojan_basic_server.json') + trojanServer = Process(Settings['workDir'], cmd = ['trojan', '-c', serverFile], file = { 'path': serverFile, 'content': json.dumps(trojanConfig) }, isStart = False) @@ -97,13 +87,13 @@ def loadBasicTest(tcpTlsStream: dict) -> dict: def loadTest(stream: dict) -> dict: proxyInfo = { # connection info - 'server': settings['serverBind'], + 'server': Settings['serverBind'], 'port': getAvailablePort(), 'passwd': genFlag(length = 8), # random password 'stream': stream['info'] } socksInfo = { # socks5 interface for test - 'addr': settings['clientBind'], + 'addr': Settings['clientBind'], 'port': getAvailablePort() } xtlsFlow = None diff --git a/Tester/TrojanGo.py b/Tester/TrojanGo.py index fba4cc4..fe28dcb 100644 --- a/Tester/TrojanGo.py +++ b/Tester/TrojanGo.py @@ -4,6 +4,7 @@ import os import json from Tester import Plugin +from Tester import Settings from Builder import TrojanGo from Basis.Logger import logging from Basis.Process import Process @@ -12,48 +13,6 @@ 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 = { @@ -67,24 +26,24 @@ def loadServer(configFile: str, proxyInfo: dict) -> Process: ], 'disable_http_check': True, 'ssl': { - 'cert': settings['cert'], - 'key': settings['key'] + '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 = { + 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) + clientFile = os.path.join(Settings['workDir'], configFile) trojanGoCommand, trojanGoConfig, _ = TrojanGo.load(proxyInfo, socksInfo, clientFile) - return Process(settings['workDir'], cmd = trojanGoCommand, file = { + return Process(Settings['workDir'], cmd = trojanGoCommand, file = { 'path': clientFile, 'content': trojanGoConfig }, isStart = False) @@ -92,17 +51,17 @@ def loadClient(configFile: str, proxyInfo: dict, socksInfo: dict) -> Process: # def loadTest(wsObject: dict or None, ssObject: dict or None, plugin: dict or None = None) -> dict: proxyInfo = { # connection info - 'server': settings['serverBind'], + 'server': Settings['serverBind'], 'port': getAvailablePort(), 'passwd': genFlag(length = 8), # random password - 'sni': settings['host'], + 'sni': Settings['host'], 'alpn': None, 'verify': True, 'ws': wsObject, 'ss': ssObject, } socksInfo = { # socks5 interface for test - 'addr': settings['clientBind'], + 'addr': Settings['clientBind'], 'port': getAvailablePort() } configName = 'trojan-go%s%s%s' % ( @@ -112,11 +71,12 @@ def loadTest(wsObject: dict or None, ssObject: dict or None, plugin: dict or Non ) 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 - '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'])), + 'title': testTitle, 'client': loadClient(configName + '_client.json', {**proxyInfo, **pluginClient}, socksInfo), 'server': loadServer(configName + '_server.json', {**proxyInfo, **pluginServer}), 'socks': socksInfo, # exposed socks5 address @@ -140,7 +100,7 @@ def load(): except StopIteration: break wsObject = { - 'host': settings['host'], + 'host': Settings['host'], 'path': '/' + genFlag(length = 6), } yield loadTest(None, None, None) diff --git a/Tester/V2ray.py b/Tester/V2ray.py index 4448805..4e0d5b3 100644 --- a/Tester/V2ray.py +++ b/Tester/V2ray.py @@ -3,18 +3,11 @@ import copy import itertools +from Tester import Settings from Basis.Functions import genFlag from Basis.Methods import quicMethods from Basis.Methods import udpObfuscations -settings = { - 'site': 'www.bing.com', - 'host': '343.re', - 'cert': '/etc/ssl/certs/343.re/fullchain.pem', - 'key': '/etc/ssl/certs/343.re/privkey.pem', -} - - httpConfig = { 'version': '1.1', 'status': '200', @@ -44,7 +37,7 @@ kcpConfig = { def addSecure(streamConfig: dict) -> dict: streamConfig['caption'] += ' (with tls)' streamConfig['info']['secure'] = { # secure options for client - 'sni': settings['host'], + 'sni': Settings['host'], 'alpn': None, 'verify': True, } @@ -52,8 +45,8 @@ def addSecure(streamConfig: dict) -> dict: streamConfig['server']['tlsSettings'] = { # cert and key for server 'alpn': ['h2', 'http/1.1'], 'certificates': [{ - 'certificateFile': settings['cert'], - 'keyFile': settings['key'], + 'certificateFile': Settings['cert'], + 'keyFile': Settings['key'], }] } return streamConfig @@ -65,7 +58,7 @@ def tcpStream(isObfs: bool) -> dict: 'info': { 'type': 'tcp', 'obfs': None if not isObfs else { - 'host': settings['site'], # obfs website + 'host': Settings['site'], # obfs website 'path': '/', }, 'secure': None, @@ -109,7 +102,7 @@ def wsStream(isEd: bool) -> dict: 'caption': 'WebSocket stream' + (' (Max-Early-Data 2048)' if isEd else ''), 'info': { 'type': 'ws', - 'host': settings['host'], + 'host': Settings['host'], 'path': path, 'ed': 2048 if isEd else None, 'secure': None, @@ -119,7 +112,7 @@ def wsStream(isEd: bool) -> dict: 'wsSettings': {**{ 'path': path, 'headers': { - 'Host': settings['host'] + 'Host': Settings['host'] } }, **({} if not isEd else { 'maxEarlyData': 2048, @@ -135,14 +128,14 @@ def h2Stream() -> dict: 'caption': 'HTTP/2 stream', 'info': { 'type': 'h2', - 'host': settings['host'], + 'host': Settings['host'], 'path': path, 'secure': None, # HTTP/2 stream force enable tls }, 'server': { 'network': 'http', 'httpSettings': { - 'host': [settings['host']], + 'host': [Settings['host']], 'path': path } } diff --git a/Tester/VLESS.py b/Tester/VLESS.py index 53f4745..0f786b4 100644 --- a/Tester/VLESS.py +++ b/Tester/VLESS.py @@ -5,6 +5,7 @@ import os import json from Tester import Xray from Builder import VLESS +from Tester import Settings from Basis.Logger import logging from Basis.Process import Process from Basis.Functions import md5Sum @@ -12,14 +13,6 @@ from Basis.Methods import xtlsFlows from Basis.Functions import genUUID from Basis.Functions import getAvailablePort -settings = { - 'serverBind': '127.0.0.1', - 'clientBind': '127.0.0.1', - # 'serverBind': '::1', - # 'clientBind': '::1', - 'workDir': '/tmp/ProxyC', -} - def loadServer(configFile: str, proxyInfo: dict, streamConfig: dict, xtlsFlow: str or None) -> Process: vlessConfig = Xray.loadConfig({ @@ -36,17 +29,17 @@ def loadServer(configFile: str, proxyInfo: dict, streamConfig: dict, xtlsFlow: s }, 'streamSettings': streamConfig }) - serverFile = os.path.join(settings['workDir'], configFile) - return Process(settings['workDir'], cmd = ['xray', '-c', serverFile], file = { + serverFile = os.path.join(Settings['workDir'], configFile) + return Process(Settings['workDir'], cmd = ['xray', '-c', serverFile], file = { 'path': serverFile, 'content': json.dumps(vlessConfig) }, isStart = False) def loadClient(configFile: str, proxyInfo: dict, socksInfo: dict) -> Process: # load client process - clientFile = os.path.join(settings['workDir'], configFile) + clientFile = os.path.join(Settings['workDir'], configFile) vlessCommand, vlessConfig, _ = VLESS.load(proxyInfo, socksInfo, clientFile) - return Process(settings['workDir'], cmd = vlessCommand, file = { + return Process(Settings['workDir'], cmd = vlessCommand, file = { 'path': clientFile, 'content': vlessConfig }, isStart = False) @@ -54,14 +47,14 @@ def loadClient(configFile: str, proxyInfo: dict, socksInfo: dict) -> Process: # def loadTest(stream: dict) -> dict: proxyInfo = { # connection info - 'server': settings['serverBind'], + 'server': Settings['serverBind'], 'port': getAvailablePort(), 'method': 'none', 'id': genUUID(), # random uuid v5 'stream': stream['info'] } socksInfo = { # socks5 interface for test - 'addr': settings['clientBind'], + 'addr': Settings['clientBind'], 'port': getAvailablePort() } xtlsFlow = None diff --git a/Tester/VMess.py b/Tester/VMess.py index 261b764..2f5e125 100644 --- a/Tester/VMess.py +++ b/Tester/VMess.py @@ -6,22 +6,14 @@ import json import itertools from Tester import V2ray from Builder import VMess +from Tester import Settings +from Builder import pathEnv from Basis.Logger import logging from Basis.Process import Process from Basis.Functions import md5Sum from Basis.Functions import genUUID -from Basis.Functions import getAvailablePort - -from Builder import pathEnv from Basis.Methods import vmessMethods - -settings = { - 'serverBind': '127.0.0.1', - 'clientBind': '127.0.0.1', - # 'serverBind': '::1', - # 'clientBind': '::1', - 'workDir': '/tmp/ProxyC', -} +from Basis.Functions import getAvailablePort def loadServer(configFile: str, proxyInfo: dict, streamConfig: dict) -> Process: # load server process @@ -37,8 +29,8 @@ def loadServer(configFile: str, proxyInfo: dict, streamConfig: dict) -> Process: }, 'streamSettings': streamConfig }) - serverFile = os.path.join(settings['workDir'], configFile) - return Process(settings['workDir'], cmd = ['v2ray', '-c', serverFile], file = { + serverFile = os.path.join(Settings['workDir'], configFile) + return Process(Settings['workDir'], cmd = ['v2ray', '-c', serverFile], file = { 'path': serverFile, 'content': json.dumps(vmessConfig) }, env= { @@ -48,9 +40,9 @@ def loadServer(configFile: str, proxyInfo: dict, streamConfig: dict) -> Process: def loadClient(configFile: str, proxyInfo: dict, socksInfo: dict) -> Process: # load client process - clientFile = os.path.join(settings['workDir'], configFile) + clientFile = os.path.join(Settings['workDir'], configFile) vmessCommand, vmessConfig, _ = VMess.load(proxyInfo, socksInfo, clientFile) - return Process(settings['workDir'], cmd = vmessCommand, file = { + return Process(Settings['workDir'], cmd = vmessCommand, file = { 'path': clientFile, 'content': vmessConfig }, isStart = False) @@ -58,7 +50,7 @@ def loadClient(configFile: str, proxyInfo: dict, socksInfo: dict) -> Process: # def loadTest(method: str, aid: int, stream: dict) -> dict: proxyInfo = { # connection info - 'server': settings['serverBind'], + 'server': Settings['serverBind'], 'port': getAvailablePort(), 'method': method, 'id': genUUID(), # random uuid v5 @@ -66,7 +58,7 @@ def loadTest(method: str, aid: int, stream: dict) -> dict: 'stream': stream['info'] } socksInfo = { # socks5 interface for test - 'addr': settings['clientBind'], + 'addr': Settings['clientBind'], 'port': getAvailablePort() } configName = 'vmess_%s_%i_%s' % (method, aid, md5Sum(stream['caption'])[:8]) diff --git a/Tester/Xray.py b/Tester/Xray.py index ef5b6b9..e75c212 100644 --- a/Tester/Xray.py +++ b/Tester/Xray.py @@ -3,8 +3,8 @@ import copy import itertools - from Tester import V2ray +from Tester import Settings from Basis.Methods import xtlsFlows from Basis.Functions import genFlag from Basis.Methods import quicMethods @@ -17,13 +17,6 @@ h2Stream = V2ray.h2Stream quicStream = V2ray.quicStream grpcStream = V2ray.grpcStream -settings = { - 'site': 'www.bing.com', - 'host': '343.re', - 'cert': '/etc/ssl/certs/343.re/fullchain.pem', - 'key': '/etc/ssl/certs/343.re/privkey.pem', -} - def addSecure(streamConfig: dict, xtlsFlow: str or None = None, isUdp443: bool = False) -> dict: # add TLS or XTLS streamConfig['caption'] += ' (with %s)' % ( @@ -31,7 +24,7 @@ def addSecure(streamConfig: dict, xtlsFlow: str or None = None, isUdp443: bool = ) streamConfig['info']['secure'] = {**{ # secure options for client 'type': 'tls' if xtlsFlow is None else 'xtls', - 'sni': settings['host'], + 'sni': Settings['host'], 'alpn': None, 'verify': True, }, **({} if xtlsFlow is None else { @@ -42,8 +35,8 @@ def addSecure(streamConfig: dict, xtlsFlow: str or None = None, isUdp443: bool = streamConfig['server']['%sSettings' % streamConfig['server']['security']] = { # cert and key for server 'alpn': ['h2', 'http/1.1'], 'certificates': [{ - 'certificateFile': settings['cert'], - 'keyFile': settings['key'], + 'certificateFile': Settings['cert'], + 'keyFile': Settings['key'], }] } return streamConfig @@ -55,7 +48,7 @@ def wsStream(isEd: bool) -> dict: 'caption': 'WebSocket stream' + (' (Max-Early-Data 2048)' if isEd else ''), 'info': { 'type': 'ws', - 'host': settings['host'], + 'host': Settings['host'], 'path': path, 'ed': 2048 if isEd else None, 'secure': None, @@ -65,7 +58,7 @@ def wsStream(isEd: bool) -> dict: 'wsSettings': { 'path': path + ('?ed=2048' if isEd else ''), 'headers': { - 'Host': settings['host'] + 'Host': Settings['host'] } } } diff --git a/Tester/__init__.py b/Tester/__init__.py new file mode 100644 index 0000000..72378d2 --- /dev/null +++ b/Tester/__init__.py @@ -0,0 +1,12 @@ +#!/usr/bin/env python3 +# -*- coding: utf-8 -*- + +Settings = { + 'workDir': '/tmp/ProxyC', + 'serverBind': '127.0.0.1', + 'clientBind': '127.0.0.1', + 'site': 'www.bing.com', + 'host': '343.re', + 'cert': '/etc/ssl/certs/343.re/fullchain.pem', + 'key': '/etc/ssl/certs/343.re/privkey.pem', +} diff --git a/test.py b/test.py index 5942d7b..12190a4 100755 --- a/test.py +++ b/test.py @@ -107,12 +107,12 @@ brook = Brook.load() hysteria = Hysteria.load() logging.critical('test start') -# runTest(ss, 64) -# runTest(ssr, 64) -# runTest(vmess, 64) -# runTest(vless, 64) -# runTest(trojan, 64) -# runTest(trojanGo, 64) -# runTest(brook, 64) +runTest(ss, 64) +runTest(ssr, 64) +runTest(vmess, 64) +runTest(vless, 64) +runTest(trojan, 64) +runTest(trojanGo, 64) +runTest(brook, 64) runTest(hysteria, 64) logging.critical('test complete')