From 4d14d62f1c5b1748f3054ee1db9275bb19ad29eb Mon Sep 17 00:00:00 2001 From: dnomd343 Date: Thu, 8 Sep 2022 00:23:37 +0800 Subject: [PATCH] update: xtls params support --- Decoder/V2ray.py | 116 +++++++++++++++++++++++++++++++++++++++++++++++ Decoder/VMess.py | 84 ++++------------------------------ test.py | 4 +- 3 files changed, 128 insertions(+), 76 deletions(-) create mode 100644 Decoder/V2ray.py diff --git a/Decoder/V2ray.py b/Decoder/V2ray.py new file mode 100644 index 0000000..d749c1d --- /dev/null +++ b/Decoder/V2ray.py @@ -0,0 +1,116 @@ +#!/usr/bin/env python3 +# -*- coding: utf-8 -*- + +from Utils.Logger import logger +from Utils.Exception import decodeException +from Utils.Common import splitParam, splitEdParam + + +def tlsSecure(params: dict) -> dict or None: + if params['security'] == 'tls': + return { + 'sni': params['sni'] if 'sni' in params else '', # sni option + 'alpn': params['alpn'] if 'alpn' in params and params['alpn'] != '' else None # alpn option + } + elif params['security'] in ['', 'none']: + return None + logger.error('V2ray url with unknown secure type -> %s' % params['security']) + raise decodeException('Unknown v2ray secure type') + + +def xtlsSecure(params: dict) -> dict or None: + if params['security'] in ['tls', 'xtls']: + secure = { + 'type': params['security'], + 'sni': params['sni'] if 'sni' in params else '', # sni option + 'alpn': params['alpn'] if 'alpn' in params and params['alpn'] != '' else None # alpn option + } + if params['security'] == 'xtls' and 'flow' in params: # XTLS flow + if 'origin' in params['flow']: + secure['flow'] = 'xtls-origin' + elif 'direct' in params['flow']: + secure['flow'] = 'xtls-direct' + elif 'splice' in params['flow']: + secure['flow'] = 'xtls-splice' + else: + logger.error('XTLS with unknown flow type -> %s' % params['flow']) + raise decodeException('Unknown xtls flow type') + return secure + elif params['security'] in ['', 'none']: + return None + logger.error('V2ray url with unknown secure type -> %s' % params['security']) + raise decodeException('Unknown v2ray secure type') + + +def v2ray(url: str, isXtls: bool) -> dict: # include VMess and VLESS + """ + https://github.com/XTLS/Xray-core/discussions/716 + + FORMAT: scheme://UUID@server:port?fields#remark + scheme -> vmess / vless + type -> tcp / kcp / ws / http / quic / grpc + encryption -> auto / aes-128-gcm / chacha20-poly1305 (VMess) + -> none (VLESS) + security -> none / tls / xtls (VMess without xtls) + path -> WebSocket / HTTP/2 / http obfs + host -> WebSocket / HTTP/2 / http obfs + headerType -> mKCP / QUIC UDP obfs -> none / srtp / utp / wechat-video / dtls / wireguard + -> TCP (http obfs) -> http + serviceName -> gRPC Service Name + mode -> gRPC transport mode -> gun / multi / guna + sni -> TLS / XTLS SNI (VMess without XTLS) + alpn -> TLS / XTLS ALPN (VMess without XTLS) + flow -> XTLS flow type -> xtls-rprx-origin / xtls-rprx-direct / xtls-rprx-splice (VMess without xtls) + """ + params = '' + if '?' in url: + url, params = url.replace('/?', '?').split('?') # `.../?...` or `...?...` + params = splitParam(params) + + info = {} + info['id'], url = url.rsplit('@', 1) + info['server'], info['port'] = url.rsplit(':', 1) + if 'encryption' in params: + info['method'] = params['encryption'] + + stream = { + 'type': params['type'] if 'type' in params else 'tcp' # stream type (default = tcp) + } + stream['type'] = 'h2' if stream['type'] == 'http' else stream['type'] # http -> h2 + if stream['type'] == 'tcp': + if 'headerType' in params and params['headerType'] == 'http': + stream['obfs'] = { + 'host': params['host'] if 'host' in params else '', + 'path': params['path'] if 'path' in params else '/', + } + elif stream['type'] == 'kcp': + stream['obfs'] = params['headerType'] if 'headerType' in params else 'none' + stream['seed'] = params['seed'] if 'seed' in params else None + elif stream['type'] == 'ws': + stream['host'] = params['host'] if 'host' in params else '' + if 'path' in params: + try: + stream['path'], stream['ed'] = splitEdParam(params['path']) + except: + stream['path'] = params['path'] + elif stream['type'] == 'h2': + stream['host'] = params['host'] if 'host' in params else '' + stream['path'] = params['path'] if 'path' in params else '/' + elif stream['type'] == 'quic': + stream['obfs'] = params['headerType'] if 'headerType' in params else 'none' + stream['method'] = params['quicSecurity'] if 'quicSecurity' in params else 'none' + stream['passwd'] = params['key'] if 'key' in params else '' + elif stream['type'] == 'grpc': + stream['service'] = params['serviceName'] + stream['mode'] = 'multi' if 'mode' in params and params['mode'] == 'multi' else 'gun' + else: + logger.error('V2ray url with unknown network type -> %s' % stream['type']) + raise decodeException('Unknown v2ray network type') + + if 'security' in params: # enable TLS / XTLS + if not isXtls: + stream['secure'] = tlsSecure(params) + else: + stream['secure'] = xtlsSecure(params) + info['stream'] = stream + return info diff --git a/Decoder/VMess.py b/Decoder/VMess.py index c82dc54..fdbaa4c 100644 --- a/Decoder/VMess.py +++ b/Decoder/VMess.py @@ -1,17 +1,18 @@ #!/usr/bin/env python3 # -*- coding: utf-8 -*- -# VMess / VLESS: https://github.com/XTLS/Xray-core/discussions/716 -# V2rayN: https://github.com/2dust/v2rayN/wiki/%E5%88%86%E4%BA%AB%E9%93%BE%E6%8E%A5%E6%A0%BC%E5%BC%8F%E8%AF%B4%E6%98%8E(ver-2) - import json from Utils.Logger import logger from Utils.Exception import decodeException from Utils.Common import b64Decode, checkScheme from Utils.Common import hostFormat, splitTag, splitEdParam, splitParam +from Decoder.V2ray import v2ray + def v2rayN(url: str) -> dict: """ + https://github.com/2dust/v2rayN/wiki/%E5%88%86%E4%BA%AB%E9%93%BE%E6%8E%A5%E6%A0%BC%E5%BC%8F%E8%AF%B4%E6%98%8E(ver-2) + FORMAT: vmess://BASE64-ENCODED-JSON-STRING fields => v(=2) / ps / add / port / id / aid / scy / net / type / host / path / tls / sni / alpn """ @@ -79,80 +80,15 @@ def v2rayN(url: str) -> dict: def vmess(url: str) -> dict: """ - FORMAT: vmess://UUID@server:port?fields#remark - type -> tcp / kcp / ws / http / quic / grpc - encryption -> auto / aes-128-gcm / chacha20-poly1305 - security -> none / tls - path -> WebSocket / HTTP/2 / http obfs - host -> WebSocket / HTTP/2 / http obfs - headerType -> mKCP / QUIC UDP obfs -> none / srtp / utp / wechat-video / dtls / wireguard - -> TCP (http obfs) -> http - seed -> mKCP seed - quicSecurity -> QUIC method - key -> QUIC key - serviceName -> gRPC Service Name - mode -> gRPC transport mode -> gun / multi / guna - sni -> TLS SNI - alpn -> TLS ALPN + https://github.com/XTLS/Xray-core/discussions/716 + + FORMAT: aka Decoder.V2ray """ config = { - 'type': 'vmess', - 'info': {} + 'type': 'vmess' } - info = config['info'] logger.debug('VMess url decode -> %s' % url) url, config['name'] = splitTag(checkScheme(url, 'vmess', 'VMess')) - - params = '' - if '?' in url: - url, params = url.replace('/?', '?').split('?') # `.../?...` or `...?...` - params = splitParam(params) - - config['info']['id'], url = url.rsplit('@', 1) - config['info']['server'], config['info']['port'] = url.rsplit(':', 1) - - # split params - if 'encryption' in params: - info['method'] = params['encryption'] - stream = { - 'type': params['type'] if 'type' in params else 'tcp' # stream type (default = tcp) - } - stream['type'] = 'h2' if stream['type'] == 'http' else stream['type'] # http -> h2 - - if stream['type'] == 'tcp': - if 'headerType' in params and params['headerType'] == 'http': - stream['obfs'] = { - 'host': params['host'] if 'host' in params else '', - 'path': params['path'] if 'path' in params else '/', - } - elif stream['type'] == 'kcp': - stream['obfs'] = params['headerType'] if 'headerType' in params else 'none' - stream['seed'] = params['seed'] if 'seed' in params else None - elif stream['type'] == 'ws': - stream['host'] = params['host'] if 'host' in params else '' - if 'path' in params: - try: - stream['path'], stream['ed'] = splitEdParam(params['path']) - except: - stream['path'] = params['path'] - elif stream['type'] == 'h2': - stream['host'] = params['host'] if 'host' in params else '' - stream['path'] = params['path'] if 'path' in params else '/' - elif stream['type'] == 'quic': - stream['obfs'] = params['headerType'] if 'headerType' in params else 'none' - stream['method'] = params['quicSecurity'] if 'quicSecurity' in params else 'none' - stream['passwd'] = params['key'] if 'key' in params else '' - elif stream['type'] == 'grpc': - stream['service'] = params['serviceName'] - stream['mode'] = 'multi' if 'mode' in params and params['mode'] == 'multi' else 'gun' - else: - logger.error('VMess url with unknown network type -> %s' % stream['type']) - raise decodeException('Unknown vmess network type') - - if 'security' in params and params['security'] == 'tls': # enable TLS - stream['secure'] = { - 'sni': params['sni'] if 'sni' in params else '', # sni option - 'alpn': params['alpn'] if 'alpn' in params and params['alpn'] != '' else None # alpn option - } - config['info']['stream'] = stream + config['info'] = v2ray(url, isXtls = False) + logger.debug('VMess url release -> %s' % config) return config diff --git a/test.py b/test.py index 090cb51..e70ca0d 100755 --- a/test.py +++ b/test.py @@ -22,8 +22,8 @@ from Filter import Filter # ret = Decoder.v2rayN('vmess://eyJhZGQiOiJmbnlkdXpheW92dnNxaTMyZ2kucTc1NDMudG9wIiwicHMiOiJ2MXzpppnmuK8wM3zljp_nlJ984piF4piF4piFICgyKSIsInNjeSI6ImF1dG8iLCJ0eXBlIjoiaHR0cCIsInNuaSI6IiIsInBhdGgiOiIvIiwicG9ydCI6MzgzMzcsInYiOjIsImhvc3QiOiJ4aGY0eHE3ZDVjYjRnc3kzeW1teWR3b2kuc2luYS5jbiIsInRscyI6IiIsImlkIjoiOTAxY2UyNTUtOTM2OS1kMWUyLTk1ODQtZGE1YTdqZjA1NDdrIiwibmV0IjoidGNwIiwiYWlkIjowfQ') # ret = Decoder.v2rayN('vmess://ew0KICAidiI6ICIyIiwNCiAgInBzIjogIummmea4rzLnur8iLA0KICAiYWRkIjogIjExOS4yOC44OC4yMzAiLA0KICAicG9ydCI6ICI0NDMiLA0KICAiaWQiOiAiODRhOGU2ZDItMWFkMy00NjEyLTgzNjItYTdjMjNlOWU5MzEyIiwNCiAgImFpZCI6ICIwIiwNCiAgInNjeSI6ICJhdXRvIiwNCiAgIm5ldCI6ICJ3cyIsDQogICJ0eXBlIjogIm5vbmUiLA0KICAiaG9zdCI6ICJ0ZXN0LnNjdXRyb2JvdC5jb20iLA0KICAicGF0aCI6ICIvdm1lc3M/ZGVtbz10cnVlJmVkPTIwNDgmdGVzdD1vayIsDQogICJ0bHMiOiAidGxzIiwNCiAgInNuaSI6ICJoay5zY3V0cm9ib3QuY29tIiwNCiAgImFscG4iOiAiIg0KfQ==') -ret = Decoder.vmess('vmess://901ce214-9369-d1e2-9584-da5a7ff0547d@pqxsiziyovtsqljrhq.q7543.top:38337/?type=tcp&encryption=auto&headerType=http&host=xhf4xq7d5cb4gsyjpmmydwoi.sina.cn&path=%2F#v1%7C%E9%A6%99%E6%B8%AF01%7CMPTCP%7C%E2%98%85%E2%98%85%E2%98%85%20%282%29') - +# ret = Decoder.vmess('vmess://901ce214-9369-d1e2-9584-da5a7ff0547d@pqxsiziyovtsqljrhq.q7543.top:38337/?type=tcp&encryption=auto&headerType=http&host=xhf4xq7d5cb4gsyjpmmydwoi.sina.cn&path=%2F#v1%7C%E9%A6%99%E6%B8%AF01%7CMPTCP%7C%E2%98%85%E2%98%85%E2%98%85%20%282%29') +ret = Decoder.vmess('vmess://5f50d808-8281-45d9-a0c2-69e039099da7@119.28.88.230:443/?type=ws&encryption=auto&host=hk.scutrobot.com&path=%2Fvmess%3Fed%3D2048&security=tls&sni=hk.scutrobot.com#%E9%A6%99%E6%B8%AF2%E7%BA%BF') # ret['info']['server'] = '[%s]' % ret['info']['server'] ret['info'] = Filter(ret['type'], ret['info'])