update: xtls params support

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

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
+    https://github.com/XTLS/Xray-core/discussions/716
-        type -> tcp / kcp / ws / http / quic / grpc
+
-        encryption -> auto / aes-128-gcm / chacha20-poly1305
+    FORMAT: aka Decoder.V2ray
-        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
     """
     config = {
-        'type': 'vmess',
+        'type': 'vmess'
-        'info': {}
     }
-    info = config['info']
     logger.debug('VMess url decode -> %s' % url)
     url, config['name'] = splitTag(checkScheme(url, 'vmess', 'VMess'))
-
+    config['info'] = v2ray(url, isXtls = False)
-    params = ''
+    logger.debug('VMess url release -> %s' % config)
-    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
     return config

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'])