Browse Source

update: Tester framework

master
Dnomd343 3 years ago
parent
commit
e3cf03d444
  1. 39
      ProxyTester/Shadowsocks.py
  2. 10
      ProxyTester/tester.py
  3. 86
      test.py

39
ProxyTester/Shadowsocks.py

@ -4,9 +4,7 @@
import copy import copy
import ProxyTester.Plugin as sip003 import ProxyTester.Plugin as sip003
defaultHost = 'dns.343.re' config = {}
defaultCert = '/etc/ssl/certs/dns.343.re/certificate.crt'
defaultKey = '/etc/ssl/certs/dns.343.re/private.key'
ssMethodList = [ ssMethodList = [
'aes-128-gcm', 'aes-128-gcm',
@ -76,32 +74,28 @@ sip003PluginList = [ # SIP003插件列表
] ]
def __ssServerConfig(method: str, plugin: str or None) -> list: def __ssServerConfig(method: str, plugin: str or None) -> list:
port = 12345
rabbitPort = 20191 rabbitPort = 20191
passwd = 'dnomd343'
proxyInfo = { proxyInfo = {
'type': 'ss', 'type': 'ss',
'server': '127.0.0.1', 'server': '127.0.0.1',
'port': port, 'port': config['port'],
'passwd': passwd, 'passwd': config['passwd'],
'method': method 'method': method
} }
serverCommand = []
caption = 'Shadowsocks method ' + method caption = 'Shadowsocks method ' + method
if method in ['plain', 'none']: # plain / none -> ss-rust if method in ['plain', 'none']: # plain / none -> ss-rust
serverCommand = [ serverCommand = [
'ss-rust-server', 'ss-rust-server',
'-s', '0.0.0.0:' + str(port), '-s', '0.0.0.0:' + str(config['port']),
'-k', passwd, '-k', config['passwd'],
'-m', method '-m', method
] ]
elif method == 'salsa20-ctr': # salsa20-ctr -> ss-python-legacy elif method == 'salsa20-ctr': # salsa20-ctr -> ss-python-legacy
serverCommand = [ serverCommand = [
'ss-bootstrap-server', '--no-udp', 'ss-bootstrap-server', '--no-udp',
'--shadowsocks', 'ss-python-legacy-server', '--shadowsocks', 'ss-python-legacy-server',
'-p', str(port), '-p', str(config['port']),
'-k', passwd, '-k', config['passwd'],
'-m', method '-m', method
] ]
else: # others -> ss-python else: # others -> ss-python
@ -118,8 +112,8 @@ def __ssServerConfig(method: str, plugin: str or None) -> list:
serverCommand = [ serverCommand = [
'ss-bootstrap-server', '--no-udp', 'ss-bootstrap-server', '--no-udp',
'--shadowsocks', 'ss-python-server', '--shadowsocks', 'ss-python-server',
'-p', str(port), '-p', str(config['port']),
'-k', passwd, '-k', config['passwd'],
'-m', method '-m', method
] ]
if method == 'idea-cfb' or method == 'seed-cfb': if method == 'idea-cfb' or method == 'seed-cfb':
@ -142,7 +136,7 @@ def __ssServerConfig(method: str, plugin: str or None) -> list:
proxyInfo['port'] = rabbitPort proxyInfo['port'] = rabbitPort
proxyInfo['plugin'] = { proxyInfo['plugin'] = {
'type': 'rabbit-plugin', 'type': 'rabbit-plugin',
'param': 'serviceAddr=127.0.0.1:' + str(port) + ';password=' + passwd 'param': 'serviceAddr=127.0.0.1:' + str(config['port']) + ';password=' + config['passwd']
} }
return [{ return [{
'proxy': proxyInfo, 'proxy': proxyInfo,
@ -157,7 +151,7 @@ def __ssServerConfig(method: str, plugin: str or None) -> list:
'startCommand': [ 'startCommand': [
'rabbit', 'rabbit',
'-mode', 's', '-mode', 's',
'-password', passwd, '-password', config['passwd'],
'-rabbit-addr', ':' + str(rabbitPort) '-rabbit-addr', ':' + str(rabbitPort)
], ],
'fileContent': None, 'fileContent': None,
@ -168,7 +162,7 @@ def __ssServerConfig(method: str, plugin: str or None) -> list:
# others plugin # others plugin
result = [] result = []
pluginConfig = sip003.loadPluginConfig(plugin, defaultHost, defaultCert, defaultKey) # 载入插件配置 pluginConfig = sip003.loadPluginConfig(plugin, config['host'], config['cert'], config['key']) # 载入插件配置
serverBaseCommand = copy.deepcopy(serverCommand) serverBaseCommand = copy.deepcopy(serverCommand)
for pluginOption in pluginConfig: for pluginOption in pluginConfig:
serverCommand = copy.deepcopy(serverBaseCommand) serverCommand = copy.deepcopy(serverBaseCommand)
@ -190,11 +184,12 @@ def __ssServerConfig(method: str, plugin: str or None) -> list:
})) }))
return result return result
def ssTest() -> list: def ssTest(ssConfig: dict) -> list:
result = [] result = []
for method in ssMethodList: for key, value in ssConfig.items(): # ssConfig -> config
config[key] = value
for method in ssMethodList: # all Shadowsocks methods
result += __ssServerConfig(method, None) result += __ssServerConfig(method, None)
for plugin in sip003PluginList: for plugin in sip003PluginList: # all SIP003 plugin
result += __ssServerConfig('aes-256-ctr', plugin) result += __ssServerConfig('aes-256-ctr', plugin)
# result += __ssServerConfig('aes-256-gcm', None)
return result return result

10
ProxyTester/tester.py

@ -3,15 +3,9 @@
from ProxyTester import Shadowsocks from ProxyTester import Shadowsocks
testInfo = { def test(key: str, config: dict) -> list:
'type': 'ss',
'method': 'aes-256-ctr',
'plugin': None
}
def test(key: str) -> list:
if key == 'ss': if key == 'ss':
return Shadowsocks.ssTest() return Shadowsocks.ssTest(config)
elif key == 'ssr': elif key == 'ssr':
pass pass
elif key == 'vmess': elif key == 'vmess':

86
test.py

@ -2,65 +2,75 @@
# -*- coding:utf-8 -*- # -*- coding:utf-8 -*-
import os import os
import sys
import time import time
import subprocess import subprocess
import Checker import Checker
import ProxyTester as Tester import ProxyTester as Tester
def testBuild(config: dict): testConfig = {
'port': 12345,
'passwd': 'dnomd343',
'host': 'dns.343.re',
'cert': '/etc/ssl/certs/dns.343.re/certificate.crt',
'key': '/etc/ssl/certs/dns.343.re/private.key'
}
def testBuild(config: dict): # load file and start process
if config['filePath'] is not None: if config['filePath'] is not None:
with open(config['filePath'], 'w') as fileObject: # 保存文件 with open(config['filePath'], 'w') as fileObject: # save file
fileObject.write(config['fileContent']) fileObject.write(config['fileContent'])
return subprocess.Popen( # 进程启动 return subprocess.Popen( # start process
config['startCommand'], config['startCommand'],
env = config['envVar'], env = config['envVar'],
stdout = subprocess.DEVNULL, stdout = subprocess.DEVNULL,
stderr = subprocess.DEVNULL stderr = subprocess.DEVNULL
) )
def testDestroy(config: dict, process): def testDestroy(config: dict, process): # remove file and kill process
if process.poll() is None: # 未死亡 if process.poll() is None: # still alive
while process.poll() is None: # 等待退出 while process.poll() is None: # wait for exit
process.terminate() # SIGTERM process.terminate() # SIGTERM
time.sleep(0.2) time.sleep(0.2)
if config['filePath'] is not None: if config['filePath'] is not None:
os.remove(config['filePath']) # 删除文件 os.remove(config['filePath']) # remove file
testList = Tester.test('ss')
def testObject(option: dict) -> None: # test target object
aiderProcess = None aiderProcess = None
serverProcess = None serverProcess = testBuild(option['server']) # start server process
for testMethod in testList: if option['aider'] is not None:
# print() aiderProcess = testBuild(option['aider']) # start aider process
# print()
# print(testMethod)
# continue
print(testMethod['caption'], end = ' -> ')
serverProcess = testBuild(testMethod['server']) checkResult = Checker.proxyTest({ # http check
if testMethod['aider'] is not None:
aiderProcess = testBuild(testMethod['aider'])
ret = Checker.proxyTest({
'check': ['http'], 'check': ['http'],
'info': testMethod['proxy'] 'info': option['proxy']
}) })
if not ret['success']: print(option['caption'], end=' -> ')
print('check error') if not checkResult['success']: # client build error
delay = ret['check']['http']['delay'] print('\n----------------------------------------------------------------')
print(option)
print('----------------------------------------------------------------\n')
raise Exception('check error')
delay = checkResult['check']['http']['delay'] # get http delay
print(str(delay) + 'ms') print(str(delay) + 'ms')
testDestroy(testMethod['server'], serverProcess) testDestroy(option['server'], serverProcess) # destroy server process
if testMethod['aider'] is not None: if option['aider'] is not None:
testDestroy(testMethod['aider'], aiderProcess) testDestroy(option['aider'], aiderProcess) # destroy aider process
if len(sys.argv) == 1: # no param
print('Unknown test type')
sys.exit(1)
testList = Tester.test(sys.argv[1], testConfig) # get test list
if len(sys.argv) == 2: # test all
for i in range(0, len(testList)):
print(str(i), end = ': ')
testObject(testList[i])
sys.exit(0)
# testName = sys.argv[1] if len(sys.argv) == 3: # test target index
# if testName == 'ss': testObject(testList[int(sys.argv[2])])
# testList = Tester.Shadowsocks(defaultPort, defaultPasswd) else:
# elif testName == 'ssr': print('Too many options')
# testList = Tester.ShadowsocksR(defaultPort, defaultPasswd)
# else:
# print("unknown test name")
# sys.exit(1)
#
# startTest(testList)

Loading…
Cancel
Save