Browse Source

feat: VLESS builder

master
Dnomd343 2 years ago
parent
commit
4eb430f406
  1. 45
      ProxyBuilder/Shadowsocks.py
  2. 33
      ProxyBuilder/ShadowsocksR.py
  3. 163
      ProxyBuilder/V2ray.py
  4. 46
      ProxyBuilder/VLESS.py
  5. 187
      ProxyBuilder/VMess.py
  6. 45
      ProxyBuilder/Xray.py
  7. 103
      ProxyBuilder/builder.py
  8. 43
      demo.py

45
ProxyBuilder/Shadowsocks.py

@ -202,31 +202,24 @@ def load(proxyInfo: dict, socksPort: int, configFile: str) -> tuple[list or None
socksPort: 本地通讯端口
configFile: 配置文件路径
节点有误:
return None, None, None
载入成功:
return startCommand, fileContent, envVar
"""
try:
if proxyInfo['plugin'] is None: # 无插件时启用UDP
isUdp = True
else:
isUdp = not __pluginWithUdp( # 获取插件UDP冲突状态
proxyInfo['plugin']['type'], proxyInfo['plugin']['param']
)
if proxyInfo['method'] in ssMethodList['ss-libev']: # 按序匹配客户端
config, ssFile = __ssLibev(proxyInfo, socksPort, isUdp)
elif proxyInfo['method'] in ssMethodList['ss-libev-legacy']:
config, ssFile = __ssLibev(proxyInfo, socksPort, isUdp, isLegacy = True)
elif proxyInfo['method'] in ssMethodList['ss-python']:
config, ssFile = __ssPython(proxyInfo, socksPort, isUdp)
elif proxyInfo['method'] in ssMethodList['ss-python-legacy']:
config, ssFile = __ssPython(proxyInfo, socksPort, isUdp, isLegacy = True)
elif proxyInfo['method'] in ssMethodList['ss-rust']:
config, ssFile = __ssRust(proxyInfo, socksPort, isUdp)
else:
return None, None, None # 无匹配加密方式
return [ssFile, '-c', configFile], json.dumps(config), {}
except:
return None, None, None # 节点配置错误
if proxyInfo['plugin'] is None: # 无插件时启用UDP
isUdp = True
else:
isUdp = not __pluginWithUdp( # 获取插件UDP冲突状态
proxyInfo['plugin']['type'], proxyInfo['plugin']['param']
)
if proxyInfo['method'] in ssMethodList['ss-libev']: # 按序匹配客户端
config, ssFile = __ssLibev(proxyInfo, socksPort, isUdp)
elif proxyInfo['method'] in ssMethodList['ss-libev-legacy']:
config, ssFile = __ssLibev(proxyInfo, socksPort, isUdp, isLegacy = True)
elif proxyInfo['method'] in ssMethodList['ss-python']:
config, ssFile = __ssPython(proxyInfo, socksPort, isUdp)
elif proxyInfo['method'] in ssMethodList['ss-python-legacy']:
config, ssFile = __ssPython(proxyInfo, socksPort, isUdp, isLegacy = True)
elif proxyInfo['method'] in ssMethodList['ss-rust']:
config, ssFile = __ssRust(proxyInfo, socksPort, isUdp)
else:
raise Exception('Unknown Shadowsocks method') # 无匹配加密方式
return [ssFile, '-c', configFile], json.dumps(config), {}

33
ProxyBuilder/ShadowsocksR.py

@ -10,25 +10,18 @@ def load(proxyInfo: dict, socksPort: int, configFile: str) -> tuple[list or None
socksPort: 本地通讯端口
configFile: 配置文件路径
节点有误:
return None, None, None
载入成功:
return startCommand, fileContent, envVar
"""
try:
config = {
'server': proxyInfo['server'],
'server_port': proxyInfo['port'],
'local_address': '127.0.0.1',
'local_port': socksPort,
'password': proxyInfo['passwd'],
'method': proxyInfo['method'],
'protocol': proxyInfo['protocol'],
'protocol_param': proxyInfo['protocolParam'],
'obfs': proxyInfo['obfs'],
'obfs_param': proxyInfo['obfsParam']
}
return ['ssr-local', '-c', configFile], json.dumps(config), {}
except:
return None, None, None
config = {
'server': proxyInfo['server'],
'server_port': proxyInfo['port'],
'local_address': '127.0.0.1',
'local_port': socksPort,
'password': proxyInfo['passwd'],
'method': proxyInfo['method'],
'protocol': proxyInfo['protocol'],
'protocol_param': proxyInfo['protocolParam'],
'obfs': proxyInfo['obfs'],
'obfs_param': proxyInfo['obfsParam']
}
return ['ssr-local', '-c', configFile], json.dumps(config), {}

163
ProxyBuilder/V2ray.py

@ -0,0 +1,163 @@
#!/usr/bin/python
# -*- coding:utf-8 -*-
logLevel = 'warning'
httpHeader = {
'type': 'http',
'request': {
'version': '1.1',
'method': 'GET',
'path': [],
'headers': {
'Host': [],
'User-Agent': [
'Mozilla/5.0 (Windows NT 10.0; WOW64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/55.0.2883.75 Safari/537.36',
'Mozilla/5.0 (iPhone; CPU iPhone OS 10_0_2 like Mac OS X) AppleWebKit/601.1 (KHTML, like Gecko) CriOS/53.0.2785.109 Mobile/14A456 Safari/601.1.46'
],
'Accept-Encoding': [
'gzip, deflate'
],
'Connection': [
'keep-alive'
],
'Pragma': 'no-cache'
}
}
}
kcpSettings = {
'mtu': 1350,
'tti': 50,
'uplinkCapacity': 12,
'downlinkCapacity': 100,
'congestion': False,
'readBufferSize': 2,
'writeBufferSize': 2,
'header': {}
}
def __secureConfig(secureInfo: dict or None) -> dict: # TLS加密传输配置
if secureInfo is None:
return {}
tlsObject = {
'allowInsecure': not secureInfo['verify'],
'alpn': secureInfo['alpn'].split(',')
}
if secureInfo['sni'] != '':
tlsObject['serverName'] = secureInfo['sni']
return {
'security': 'tls',
'tlsSettings': tlsObject
}
def tcpConfig(streamInfo: dict, secureFunc) -> dict: # TCP传输方式配置
tcpObject = {}
if streamInfo['obfs'] is not None:
httpHeader['request']['path'].append(streamInfo['obfs']['path'])
httpHeader['request']['headers']['Host'] = streamInfo['obfs']['host'].split(',')
tcpObject['header'] = httpHeader
return {**{
'network': 'tcp',
'tcpSettings': tcpObject
}, **secureFunc(streamInfo['secure'])}
def kcpConfig(streamInfo: dict, secureFunc) -> dict: # mKCP传输方式配置
kcpObject = kcpSettings
kcpObject['header']['type'] = streamInfo['obfs']
if streamInfo['seed'] is not None:
kcpObject['seed'] = streamInfo['seed']
return {**{
'network': 'kcp',
'kcpSettings': kcpObject
}, **secureFunc(streamInfo['secure'])}
def wsConfig(streamInfo: dict, edInPath: bool, secureFunc) -> dict: # WebSocket传输方式配置
wsObject = {
'path': streamInfo['path']
}
if streamInfo['host'] != '':
wsObject['headers'] = {}
wsObject['headers']['Host'] = streamInfo['host']
if streamInfo['ed'] is not None:
if not edInPath:
wsObject['maxEarlyData'] = streamInfo['ed']
wsObject['earlyDataHeaderName'] = 'Sec-WebSocket-Protocol'
else: # ed参数写入路径 -> /...?ed=xxx
if wsObject['path'].find('?') == -1: # 原路径不带参数
wsObject['path'] += '?ed=' + str(streamInfo['ed'])
else:
wsObject['path'] += '&ed=' + str(streamInfo['ed'])
return {**{
'network': 'ws',
'wsSettings': wsObject
}, **secureFunc(streamInfo['secure'])}
def h2Config(streamInfo: dict, secureFunc) -> dict: # HTTP/2传输方式配置
h2Object = {
'path': streamInfo['path']
}
if streamInfo['host'] != '':
h2Object['host'] = streamInfo['host'].split(',')
return {**{
'network': 'h2',
'httpSettings': h2Object
}, **secureFunc(streamInfo['secure'])}
def quicConfig(streamInfo: dict, secureFunc) -> dict: # QUIC传输方式配置
return {**{
'network': 'quic',
'quicSettings': {
'security': streamInfo['method'],
'key': streamInfo['passwd'],
'header': {
'type': streamInfo['obfs']
}
}
}, **secureFunc(streamInfo['secure'])}
def grpcConfig(streamInfo: dict, secureFunc) -> dict: # gRPC传输方式配置
return {**{
'network': 'grpc',
'grpcSettings': {
'serviceName': streamInfo['service']
}
}, **secureFunc(streamInfo['secure'])}
def v2rayStreamConfig(streamInfo: dict) -> dict: # 生成v2ray传输方式配置
streamType = streamInfo['type']
if streamType == 'tcp':
return tcpConfig(streamInfo, __secureConfig)
elif streamType == 'kcp':
return kcpConfig(streamInfo, __secureConfig)
elif streamType == 'ws':
return wsConfig(streamInfo, False, __secureConfig)
elif streamType == 'h2':
return h2Config(streamInfo, __secureConfig)
elif streamType == 'quic':
return quicConfig(streamInfo, __secureConfig)
elif streamType == 'grpc':
return grpcConfig(streamInfo, __secureConfig)
else:
raise Exception('Unknown stream type')
def baseConfig(socksPort: int, outboundObject: dict) -> dict: # 基础配置生成
return {
'log': {
'loglevel': logLevel
},
'inbounds': [
{
'port': socksPort,
'listen': '127.0.0.1',
'protocol': 'socks',
'settings': {
'udp': True,
'auth': 'noauth'
}
}
],
'outbounds': [
outboundObject
]
}

46
ProxyBuilder/VLESS.py

@ -0,0 +1,46 @@
#!/usr/bin/python
# -*- coding:utf-8 -*-
import json
from ProxyBuilder import Xray
def load(proxyInfo: dict, socksPort: int, configFile: str) -> tuple[list, str, dict]:
"""
VLESS配置载入
proxyInfo: 节点信息
socksPort: 本地通讯端口
configFile: 配置文件路径
return startCommand, fileContent, envVar
"""
user = {
'id': proxyInfo['id'],
'encryption': proxyInfo['method']
}
if proxyInfo['stream']['secure'] is not None and proxyInfo['stream']['secure']['type'] == 'xtls':
flowType = proxyInfo['stream']['secure']['flow']
if flowType == 'xtls-origin':
user['flow'] = 'xtls-rprx-origin'
elif flowType == 'xtls-direct':
user['flow'] = 'xtls-rprx-direct'
elif flowType == 'xtls-splice':
user['flow'] = 'xtls-rprx-splice'
else:
raise Exception('Unknown XTLS flow')
if proxyInfo['stream']['secure']['udp443']:
user['flow'] += '-udp443'
outboundConfig = {
'protocol': 'vless',
'settings': {
'vnext': [
{
'address': proxyInfo['server'],
'port': proxyInfo['port'],
'users': [user]
}
]
},
'streamSettings': Xray.xrayStreamConfig(proxyInfo['stream'])
}
config = Xray.baseConfig(socksPort, outboundConfig) # VLESS节点配置
return ['xray', '-c', configFile], json.dumps(config), {}

187
ProxyBuilder/VMess.py

@ -2,162 +2,18 @@
# -*- coding:utf-8 -*-
import json
from ProxyBuilder import V2ray
logLevel = 'warning'
httpHeader = {
'type': 'http',
'request': {
'version': '1.1',
'method': 'GET',
'path': [],
'headers': {
'Host': [],
'User-Agent': [
'Mozilla/5.0 (Windows NT 10.0; WOW64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/55.0.2883.75 Safari/537.36',
'Mozilla/5.0 (iPhone; CPU iPhone OS 10_0_2 like Mac OS X) AppleWebKit/601.1 (KHTML, like Gecko) CriOS/53.0.2785.109 Mobile/14A456 Safari/601.1.46'
],
'Accept-Encoding': [
'gzip, deflate'
],
'Connection': [
'keep-alive'
],
'Pragma': 'no-cache'
}
}
}
kcpSettings = {
'mtu': 1350,
'tti': 50,
'uplinkCapacity': 12,
'downlinkCapacity': 100,
'congestion': False,
'readBufferSize': 2,
'writeBufferSize': 2,
'header': {}
}
def __baseConfig(socksPort: int, outboundObject: dict) -> dict: # v2ray配置生成
return {
'log': {
'loglevel': logLevel
},
'inbounds': [
{
'port': socksPort,
'listen': '127.0.0.1',
'protocol': 'socks',
'settings': {
'udp': True,
'auth': 'noauth'
}
}
],
'outbounds': [
outboundObject
]
}
def __secureConfig(secureInfo: dict or None) -> dict: # TLS加密传输配置
if secureInfo is None:
return {}
tlsObject = {
'allowInsecure': not secureInfo['verify'],
'alpn': secureInfo['alpn'].split(',')
}
if secureInfo['sni'] != '':
tlsObject['serverName'] = secureInfo['sni']
return {
'security': 'tls',
'tlsSettings': tlsObject
}
def __tcpConfig(streamInfo: dict) -> dict: # TCP传输方式配置
tcpObject = {}
if streamInfo['obfs'] is not None:
httpHeader['request']['path'].append(streamInfo['obfs']['path'])
httpHeader['request']['headers']['Host'] = streamInfo['obfs']['host'].split(',')
tcpObject['header'] = httpHeader
return {**{
'network': 'tcp',
'tcpSettings': tcpObject
}, **__secureConfig(streamInfo['secure'])}
def __kcpConfig(streamInfo: dict) -> dict: # mKCP传输方式配置
kcpObject = kcpSettings
kcpObject['header']['type'] = streamInfo['obfs']
if streamInfo['seed'] is not None:
kcpObject['seed'] = streamInfo['seed']
return {**{
'network': 'kcp',
'kcpSettings': kcpObject
}, **__secureConfig(streamInfo['secure'])}
def __wsConfig(streamInfo: dict) -> dict: # WebSocket传输方式配置
wsObject = {}
if streamInfo['path'] != '':
wsObject['path'] = streamInfo['path']
if streamInfo['host'] != '':
wsObject['headers'] = {}
wsObject['headers']['Host'] = streamInfo['host']
if streamInfo['ed'] is not None:
wsObject['maxEarlyData'] = streamInfo['ed']
wsObject['earlyDataHeaderName'] = 'Sec-WebSocket-Protocol'
return {**{
'network': 'ws',
'wsSettings': wsObject
}, **__secureConfig(streamInfo['secure'])}
def __h2Config(streamInfo: dict) -> dict: # HTTP/2传输方式配置
h2Object = {
'path': streamInfo['path']
}
if streamInfo['host'] != '':
h2Object['host'] = streamInfo['host'].split(',')
return {**{
'network': 'h2',
'httpSettings': h2Object
}, **__secureConfig(streamInfo['secure'])}
def __quicConfig(streamInfo: dict) -> dict: # QUIC传输方式配置
return {**{
'network': 'quic',
'quicSettings': {
'security': streamInfo['method'],
'key': streamInfo['passwd'],
'header': {
'type': streamInfo['obfs']
}
}
}, **__secureConfig(streamInfo['secure'])}
def __grpcConfig(streamInfo: dict) -> dict: # gRPC传输方式配置
return {**{
'network': 'grpc',
'grpcSettings': {
'serviceName': streamInfo['service']
}
}, **__secureConfig(streamInfo['secure'])}
def load(proxyInfo: dict, socksPort: int, configFile: str) -> tuple[list, str, dict]:
"""
VMess配置载入
proxyInfo: 节点信息
socksPort: 本地通讯端口
configFile: 配置文件路径
def __vmessConfig(proxyInfo: dict) -> dict: # VMess节点配置
streamType = proxyInfo['stream']['type']
if streamType == 'tcp':
streamObject = __tcpConfig(proxyInfo['stream'])
elif streamType == 'kcp':
streamObject = __kcpConfig(proxyInfo['stream'])
elif streamType == 'ws':
streamObject = __wsConfig(proxyInfo['stream'])
elif streamType == 'h2':
streamObject = __h2Config(proxyInfo['stream'])
elif streamType == 'quic':
streamObject = __quicConfig(proxyInfo['stream'])
elif streamType == 'grpc':
streamObject = __grpcConfig(proxyInfo['stream'])
else:
raise Exception('Unknown stream type')
return {
return startCommand, fileContent, envVar
"""
outboundConfig = {
'protocol': 'vmess',
'settings': {
'vnext': [
@ -174,24 +30,7 @@ def __vmessConfig(proxyInfo: dict) -> dict: # VMess节点配置
}
]
},
'streamSettings': streamObject
'streamSettings': V2ray.v2rayStreamConfig(proxyInfo['stream'])
}
def load(proxyInfo: dict, socksPort: int, configFile: str) -> tuple[list or None, str or None, dict or None]:
"""
VMess配置载入
proxyInfo: 节点信息
socksPort: 本地通讯端口
configFile: 配置文件路径
节点有误:
return None, None, None
载入成功:
return startCommand, fileContent, envVar
"""
try:
config = __baseConfig(socksPort, __vmessConfig(proxyInfo))
return ['v2ray', '-c', configFile], json.dumps(config), {}
except:
return None, None, None
config = V2ray.baseConfig(socksPort, outboundConfig) # VMess节点配置
return ['v2ray', '-c', configFile], json.dumps(config), {}

45
ProxyBuilder/Xray.py

@ -0,0 +1,45 @@
#!/usr/bin/python
# -*- coding:utf-8 -*-
from ProxyBuilder import V2ray
baseConfig = V2ray.baseConfig
def __secureConfig(secureInfo: dict or None) -> dict: # TLS/XTLS加密传输配置
if secureInfo is None:
return {}
secureObject = {
'allowInsecure': not secureInfo['verify'],
'alpn': secureInfo['alpn'].split(',')
}
if secureInfo['sni'] != '':
secureObject['serverName'] = secureInfo['sni']
if secureInfo['type'] == 'tls':
return {
'security': 'tls',
'tlsSettings': secureObject
}
elif secureInfo['type'] == 'xtls':
return {
'security': 'xtls',
'xtlsSettings': secureObject
}
else:
raise Exception('Unknown secure type')
def xrayStreamConfig(streamInfo: dict) -> dict: # 生成xray传输方式配置
streamType = streamInfo['type']
if streamType == 'tcp':
return V2ray.tcpConfig(streamInfo, __secureConfig)
elif streamType == 'kcp':
return V2ray.kcpConfig(streamInfo, __secureConfig)
elif streamType == 'ws':
return V2ray.wsConfig(streamInfo, True, __secureConfig)
elif streamType == 'h2':
return V2ray.h2Config(streamInfo, __secureConfig)
elif streamType == 'quic':
return V2ray.quicConfig(streamInfo, __secureConfig)
elif streamType == 'grpc':
return V2ray.grpcConfig(streamInfo, __secureConfig)
else:
raise Exception('Unknown stream type')

103
ProxyBuilder/builder.py

@ -11,6 +11,7 @@ import subprocess
from ProxyBuilder import Shadowsocks
from ProxyBuilder import ShadowsocksR
from ProxyBuilder import VMess
from ProxyBuilder import VLESS
libcPaths = [
'/usr/lib64/libc.so.6', # CentOS
@ -93,60 +94,68 @@ def build(proxyInfo: dict, configDir: str,
'process': process
}
"""
taskFlag = __genTaskFlag() # 生成测试标志
socksPort = __getAvailablePort(portRangeStart, portRangeEnd) # 获取Socks5测试端口
if 'type' not in proxyInfo: # 未指定节点类型
return False, 'Proxy type not specified'
if proxyInfo['type'] == 'ss': # Shadowsocks节点
clientObj = Shadowsocks
elif proxyInfo['type'] == 'ssr': # ShadowsocksR节点
clientObj = ShadowsocksR
elif proxyInfo['type'] == 'vmess': # VMess节点
clientObj = VMess
else: # 未知类型
return False, 'Unknown proxy type'
configFile = configDir + '/' + taskFlag + '.json' # 配置文件路径
startCommand, fileContent, envVar = clientObj.load(proxyInfo, socksPort, configFile) # 载入配置
if startCommand is None: # 格式出错
return False, 'Format error with ' + str(proxyInfo['type'])
try:
with open(configFile, 'w') as fileObject: # 保存配置文件
fileObject.write(fileContent)
except: # 配置文件写入失败
return None, "Unable write to file " + str(configFile)
process = None
try: # 子进程形式启动
for libcPath in libcPaths:
if os.path.exists(libcPath): # 定位libc.so文件
break
process = subprocess.Popen( # 启动子进程
startCommand,
env = envVar,
stdout = subprocess.DEVNULL,
stderr = subprocess.DEVNULL,
preexec_fn = lambda: ctypes.CDLL(libcPath).prctl(1, 15) # 子进程跟随退出
)
except:
taskFlag = __genTaskFlag() # 生成测试标志
socksPort = __getAvailablePort(portRangeStart, portRangeEnd) # 获取Socks5测试端口
if 'type' not in proxyInfo: # 未指定节点类型
return False, 'Proxy type not specified'
if proxyInfo['type'] == 'ss': # Shadowsocks节点
clientObj = Shadowsocks
elif proxyInfo['type'] == 'ssr': # ShadowsocksR节点
clientObj = ShadowsocksR
elif proxyInfo['type'] == 'vmess': # VMess节点
clientObj = VMess
elif proxyInfo['type'] == 'vless': # VLESS节点
clientObj = VLESS
else: # 未知类型
return False, 'Unknown proxy type'
configFile = configDir + '/' + taskFlag + '.json' # 配置文件路径
try:
startCommand, fileContent, envVar = clientObj.load(proxyInfo, socksPort, configFile) # 载入配置
except: # 格式出错
return False, 'Format error with ' + str(proxyInfo['type'])
try:
process = subprocess.Popen( # prctl失败 回退正常启动
with open(configFile, 'w') as fileObject: # 保存配置文件
fileObject.write(fileContent)
except: # 配置文件写入失败
return None, "Unable write to file " + str(configFile)
process = None
try: # 子进程形式启动
for libcPath in libcPaths:
if os.path.exists(libcPath): # 定位libc.so文件
break
process = subprocess.Popen( # 启动子进程
startCommand,
env = envVar,
stdout = subprocess.DEVNULL,
stderr = subprocess.DEVNULL
stderr = subprocess.DEVNULL,
preexec_fn = lambda: ctypes.CDLL(libcPath).prctl(1, 15) # 子进程跟随退出
)
except:
pass
if process is None: # 启动失败
return None, 'Subprocess start failed by `' + ' '.join(startCommand) + '`'
return True, { # 返回连接参数
'flag': taskFlag,
'port': socksPort,
'file': configFile,
'process': process
}
try:
process = subprocess.Popen( # prctl失败 回退正常启动
startCommand,
env = envVar,
stdout = subprocess.DEVNULL,
stderr = subprocess.DEVNULL
)
except:
pass
if process is None: # 启动失败
return None, 'Subprocess start failed by `' + ' '.join(startCommand) + '`'
return True, { # 返回连接参数
'flag': taskFlag,
'port': socksPort,
'file': configFile,
'process': process
}
except:
return None, 'Unknown error'
def check(client: dict) -> bool or None:
"""

43
demo.py

@ -1,26 +1,49 @@
#!/usr/bin/python
# -*- coding:utf-8 -*-
import copy
import time
import ProxyFilter as Filter
import ProxyBuilder as Builder
# info = {
# 'type': 'vless',
# 'server': '127.0.0.1',
# 'port': '12345',
# 'id': 'dnomd343',
# 'stream': {
# 'type': 'grpc',
# 'service': 'dnomd343',
# 'secure': {
# 'type': 'tls',
# 'sni': '',
# 'flow': 'xtls-origin',
# 'udp443': True
# }
# }
# }
#
# ret = Filter.filte(info)
#
# print(ret[0])
# print(ret[1])
info = {
'type': 'vless',
'server': '127.0.0.1',
'port': '12345',
'id': 'dnomd343',
'id': '58c0f2eb-5d47-45d0-8f5f-ebae5c2cfdd9',
'stream': {
'type': 'grpc',
'service': 'dnomd343',
'type': 'tcp',
'secure': {
'type': 'tls',
'sni': '',
'flow': 'xtls-origin',
'type': 'xtls',
'udp443': True
}
}
}
ret = Filter.filte(info)
print(ret[0])
print(ret[1])
info = copy.deepcopy(Filter.filte(info)[1])
print(info)
Builder.build(info, '/tmp/ProxyC')
time.sleep(5)
Builder.destroy(info)

Loading…
Cancel
Save