Browse Source

feat: test of Trojan-Go

master
Dnomd343 3 years ago
parent
commit
3144dbb937
  1. 2
      ProxyBuilder/TrojanGo.py
  2. 13
      ProxyBuilder/builder.py
  3. 164
      ProxyTester/TrojanGo.py
  4. 3
      ProxyTester/tester.py
  5. 12
      Test.py
  6. 24
      demo.py

2
ProxyBuilder/TrojanGo.py

@ -73,4 +73,4 @@ def load(proxyInfo: dict, socksPort: int, configFile: str) -> tuple[list, str, d
'shadowsocks': __ssConfig(proxyInfo), 'shadowsocks': __ssConfig(proxyInfo),
'transport_plugin': __pluginConfig(proxyInfo) 'transport_plugin': __pluginConfig(proxyInfo)
} }
return ['trojan-go', '-config', configFile], json.dumps(config), {} return ['trojan-go', '-config', configFile], json.dumps(config), {'PATH': '/usr/bin'}

13
ProxyBuilder/builder.py

@ -6,6 +6,7 @@ import time
import ctypes import ctypes
import random import random
import socket import socket
import signal
import subprocess import subprocess
from ProxyBuilder import Shadowsocks from ProxyBuilder import Shadowsocks
@ -163,6 +164,9 @@ def check(client: dict) -> bool or None:
""" """
return client['process'].poll() is None return client['process'].poll() is None
def skipDestroy() -> None:
pass
def destroy(client: dict) -> bool: def destroy(client: dict) -> bool:
""" """
结束客户端并清理 结束客户端并清理
@ -171,12 +175,21 @@ def destroy(client: dict) -> bool:
销毁成功: return True 销毁成功: return True
""" """
signal.signal(signal.SIGINT, skipDestroy) # 捕获SIGINT
signal.signal(signal.SIGTERM, skipDestroy) # 捕获SIGTERM
try: try:
process = client['process'] process = client['process']
os.killpg(os.getpgid(process.pid), signal.SIGTERM) # 杀进程组
time.sleep(0.2)
maxWait = 100
if process.poll() is None: # 未死亡 if process.poll() is None: # 未死亡
while process.poll() is None: # 等待退出 while process.poll() is None: # 等待退出
process.terminate() # SIGTERM process.terminate() # SIGTERM
time.sleep(0.2) time.sleep(0.2)
maxWait -= 1
if maxWait < 0:
process.kill() # SIGKILL
time.sleep(0.5)
except: except:
return False return False

164
ProxyTester/TrojanGo.py

@ -0,0 +1,164 @@
#!/usr/bin/python
# -*- coding:utf-8 -*-
import copy
import json
from ProxyTester import Plugin
config = {}
trojanGoMethod = [
'AES-128-GCM',
'AES-256-GCM',
'CHACHA20-IETF-POLY1305'
]
sip003PluginList = [ # SIP003插件列表
'obfs-local',
'simple-tls',
'v2ray-plugin',
'xray-plugin',
'kcptun-client',
'gost-plugin',
'ck-client',
'gq-client',
'mtt-client',
'rabbit-plugin',
'qtun-client',
'gun-plugin'
]
def loadTrojanGo(isWs: bool, ssMethod: str or None) -> dict:
caption = 'Trojan-Go original'
serverConfig = {
'run_type': 'server',
'local_addr': '127.0.0.1',
'local_port': config['port'],
'remote_addr': '127.0.0.1', # only for shadowsocks fallback
'remote_port': 343,
'password': [
config['passwd']
],
'disable_http_check': True,
'ssl': {
'cert': config['cert'],
'key': config['key']
}
}
proxyInfo = {
'type': 'trojan-go',
'server': '127.0.0.1',
'port': config['port'],
'passwd': config['passwd'],
'sni': config['host'],
}
if ssMethod is not None: # add Shadowsocks encrypt
caption += ' ' + ssMethod + ' encrypt'
serverConfig['shadowsocks'] = {
'enabled': True,
'method': ssMethod,
'password': config['passwd']
}
proxyInfo['ss'] = {
'method': ssMethod,
'passwd': config['passwd']
}
if isWs: # add WebSocket config
caption += ' (websocket)'
serverConfig['websocket'] = {
'enabled': True,
'host': config['host'],
'path': config['path']
}
proxyInfo['ws'] = {
'host': config['host'],
'path': config['path']
}
return {
'caption': caption,
'client': proxyInfo,
'server': serverConfig,
'file': None,
'path': None
}
def loadTrojanGoPlugin(plugin: str) -> list:
result = []
rabbitPort = 20191
trojanBaseConfig = loadTrojanGo(False, None)
if plugin == 'rabbit-plugin': # rabbit-tcp
trojanBaseConfig['caption'] = 'Trojan-Go rabbit-plugin (basic mode)'
trojanBaseConfig['client']['port'] = rabbitPort
trojanBaseConfig['client']['plugin'] = {
'type': 'rabbit-plugin',
'param': 'serviceAddr=127.0.0.1:' + str(config['port']) + ';password=' + config['passwd']
}
trojanBaseConfig['server']['transport_plugin'] = {
'enabled': True,
'type': 'other',
'command': 'rabbit',
'arg': [
'-mode', 's',
'-password', config['passwd'],
'-rabbit-addr', ':' + str(rabbitPort)
]
}
trojanBaseConfig['file'] = None
trojanBaseConfig['path'] = None
return [trojanBaseConfig]
# other plugin
pluginConfig = Plugin.loadPluginConfig(plugin, config['host'], config['cert'], config['key']) # 载入插件配置
for pluginOption in pluginConfig:
trojanConfig = copy.deepcopy(trojanBaseConfig)
trojanConfig['caption'] = 'Trojan-Go plugin ' + plugin + ' (' + pluginOption['caption'] + ')'
trojanConfig['client']['plugin'] = pluginOption['client']
trojanConfig['server']['transport_plugin'] = {
'enabled': True,
'type': 'shadowsocks',
'command': pluginOption['server']['type'],
'option': pluginOption['server']['param']
}
trojanConfig['file'] = pluginOption['file']
trojanConfig['path'] = pluginOption['path']
result.append(trojanConfig)
return result
def loadTrojanGoConfig(trojanGoConfigList: list) -> list:
result = []
for trojanGoConfig in trojanGoConfigList:
result.append({
'caption': trojanGoConfig['caption'],
'proxy': trojanGoConfig['client'],
'server': {
'startCommand': ['trojan-go', '-config', config['file']],
'fileContent': json.dumps(trojanGoConfig['server']),
'filePath': config['file'],
'envVar': {'PATH': '/usr/bin'}
},
'aider': {
'startCommand': None,
'fileContent': trojanGoConfig['file'],
'filePath': trojanGoConfig['path'],
'envVar': {}
}
})
return result
def trojanGoTest(trojanGoConfig: dict) -> list:
result = []
for key, value in trojanGoConfig.items(): # trojanGoConfig -> config
config[key] = value
result += loadTrojanGoConfig([loadTrojanGo(False, None)])
result += loadTrojanGoConfig([loadTrojanGo(True, None)])
for ssMethod in trojanGoMethod:
result += loadTrojanGoConfig([loadTrojanGo(False, ssMethod)])
result += loadTrojanGoConfig([loadTrojanGo(True, ssMethod)])
for plugin in sip003PluginList:
result += loadTrojanGoConfig(loadTrojanGoPlugin(plugin))
return result

3
ProxyTester/tester.py

@ -6,6 +6,7 @@ from ProxyTester import ShadowsocksR
from ProxyTester import VMess from ProxyTester import VMess
from ProxyTester import VLESS from ProxyTester import VLESS
from ProxyTester import Trojan from ProxyTester import Trojan
from ProxyTester import TrojanGo
def test(key: str, config: dict) -> list: def test(key: str, config: dict) -> list:
if key in ['ss', 'shadowsocks']: if key in ['ss', 'shadowsocks']:
@ -18,5 +19,7 @@ def test(key: str, config: dict) -> list:
return VLESS.vlessTest(config) return VLESS.vlessTest(config)
elif key == 'trojan': elif key == 'trojan':
return Trojan.trojanTest(config) return Trojan.trojanTest(config)
elif key == 'trojan-go':
return TrojanGo.trojanGoTest(config)
else: else:
return [] return []

12
Test.py

@ -12,12 +12,12 @@ import ProxyTester as Tester
testConfig = { testConfig = {
'port': 12345, 'port': 12345,
'passwd': 'dnomd343', 'passwd': 'dnomd343',
'host': 'dns.343.re', 'host': 'local.343.re',
'path': '/test', 'path': '/test',
'service': 'dnomd343', 'service': 'dnomd343',
'file': '/tmp/proxyc-test.json', 'file': '/tmp/proxycTest.json',
'cert': '/etc/ssl/certs/dns.343.re/certificate.crt', 'cert': '/etc/ssl/certs/343.re/fullchain.pem',
'key': '/etc/ssl/certs/dns.343.re/private.key', 'key': '/etc/ssl/certs/343.re/privkey.pem',
'id': '1f7aa040-94d8-4b53-ae85-af6946d550bb', 'id': '1f7aa040-94d8-4b53-ae85-af6946d550bb',
} }
@ -25,6 +25,8 @@ 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: # save file with open(config['filePath'], 'w') as fileObject: # save file
fileObject.write(config['fileContent']) fileObject.write(config['fileContent'])
if config['startCommand'] is None:
return None
return subprocess.Popen( # start process return subprocess.Popen( # start process
config['startCommand'], config['startCommand'],
env = config['envVar'], env = config['envVar'],
@ -33,7 +35,7 @@ def testBuild(config: dict): # load file and start process
) )
def testDestroy(config: dict, process): # remove file and kill process def testDestroy(config: dict, process): # remove file and kill process
if process.poll() is None: # still alive if process is not None and process.poll() is None: # still alive
while process.poll() is None: # wait for exit while process.poll() is None: # wait for exit
process.terminate() # SIGTERM process.terminate() # SIGTERM
time.sleep(0.2) time.sleep(0.2)

24
demo.py

@ -6,31 +6,21 @@ import Check as Checker
info = { info = {
'type': 'trojan-go', 'type': 'trojan-go',
'server': '127.0.0.1', 'server': '127.0.0.1',
'port': '12345', 'port': 12345,
'passwd': 'dnomd343', 'passwd': 'dnomd343',
'sni': 'local.343.re', 'sni': 'local.343.re',
'alpn': 'h2', 'plugin': {
'verify': False, # 'type': 'obfs-local',
# 'ws': { # 'param': 'obfs=http;obfs-host=www.bing.com'
# 'host': 'local.343.re', 'type': 'simple-tls',
# 'path': '/test' 'param': 'n=local.343.re;no-verify'
# }, }
# 'ss': {
# 'method': 'chacha20-ietf-poly1305',
# 'passwd': 'dnomd343'
# },
# 'plugin': {
# 'type': 'obfs-local',
# 'param': 'obfs=http'
# }
} }
status, ret = Filter.filte(info, isExtra = True) status, ret = Filter.filte(info, isExtra = True)
print(status) print(status)
print(ret) print(ret)
# Builder.build(ret, '/tmp/ProxyC')
data = Checker.proxyTest({ data = Checker.proxyTest({
'check': ['http'], 'check': ['http'],
'info': ret 'info': ret

Loading…
Cancel
Save