#!/usr/bin/env python3 # -*- coding: utf-8 -*- import os import json from Tester import Xray from Builder import Trojan from Basis.Logger import logging from Basis.Process import Process from Basis.Functions import md5Sum 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({ 'protocol': 'trojan', 'listen': proxyInfo['server'], 'port': proxyInfo['port'], 'settings': { 'clients': [{**{ 'password': proxyInfo['passwd'], }, **({} if xtlsFlow is None else { 'flow': xtlsFlow })}], }, 'streamSettings': streamConfig }) 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) trojanCommand, trojanConfig, _ = Trojan.load(proxyInfo, socksInfo, clientFile) return Process(settings['workDir'], cmd = trojanCommand, file = { 'path': clientFile, 'content': trojanConfig }, isStart = False) def loadBasicTest(tcpTlsStream: dict) -> dict: proxyInfo = { # connection info 'server': settings['serverBind'], 'port': getAvailablePort(), 'passwd': genFlag(length = 8), # random password 'stream': tcpTlsStream['info'] } socksInfo = { # socks5 interface for test 'addr': settings['clientBind'], 'port': getAvailablePort() } trojanConfig = { 'run_type': 'server', 'local_addr': proxyInfo['server'], 'local_port': proxyInfo['port'], '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'] } } 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) testInfo = { # release test info 'title': 'Trojan test: basic connection', 'client': loadClient('trojan_basic_client.json', proxyInfo, socksInfo), 'server': trojanServer, 'socks': socksInfo, # exposed socks5 address 'interface': { 'addr': proxyInfo['server'], 'port': proxyInfo['port'], } } logging.debug('New trojan test -> %s' % testInfo) return testInfo def loadTest(stream: dict) -> dict: proxyInfo = { # connection info 'server': settings['serverBind'], 'port': getAvailablePort(), 'passwd': genFlag(length = 8), # random password 'stream': stream['info'] } socksInfo = { # socks5 interface for test 'addr': settings['clientBind'], 'port': getAvailablePort() } xtlsFlow = None if stream['info']['secure'] is not None and stream['info']['secure']['type'] == 'xtls': # with XTLS secure xtlsFlow = xtlsFlows[stream['info']['secure']['flow']] xtlsFlow = xtlsFlow.replace('splice', 'direct') # XTLS on server should use xtls-rprx-direct flow configName = 'trojan_%s' % (md5Sum(stream['caption'])[:8]) testInfo = { # release test info 'title': 'Trojan test: %s' % stream['caption'], 'client': loadClient(configName + '_client.json', proxyInfo, socksInfo), 'server': loadServer(configName + '_server.json', proxyInfo, stream['server'], xtlsFlow), 'socks': socksInfo, # exposed socks5 address 'interface': { 'addr': proxyInfo['server'], 'port': proxyInfo['port'], } } logging.debug('New trojan test -> %s' % testInfo) return testInfo def load(): streams = Xray.loadStream() # load xray-core stream list yield loadBasicTest(streams[1]) # Trojan basic test -> TCP stream with TLS for stream in streams: # test all stream cases yield loadTest(stream)