add support for new sub format

xray_manager.py

@@ -1,16 +1,16 @@
+from traceback import print_exc
+from math import ceil
+from xray_config import xray_manager_config, xray_local_config, xray_config
+from binascii import a2b_base64
+from re import match
+from urllib.parse import unquote_plus, parse_qs
+from urllib import request
+import sys
 import json
 from os import path as _pth
 CURRENT_DIR = _pth.abspath(_pth.dirname(__file__))
-import sys
 sys.path.append(CURRENT_DIR)
 
-from urllib import request
-from urllib.parse import unquote_plus
-from re import match
-from binascii import a2b_base64
-from xray_config import xray_manager_config, xray_local_config, xray_config
-from math import ceil
-from traceback import print_exc
 
 def fill_padding(base64_encode_str):
     need_padding = len(base64_encode_str) % 4 != 0
@@ -19,11 +19,13 @@ def fill_padding(base64_encode_str):
         base64_encode_str += '=' * missing_padding
     return base64_encode_str
 
+
 def base64_decode(base64_encode_str):
     base64_encode_str = base64_encode_str.replace("-", "+").replace("_", "/")
     base64_encode_str = fill_padding(base64_encode_str)
     return a2b_base64(base64_encode_str).decode('utf8')
 
+
 def parse_format_sip002(line):
     b64data = match(r"^ss:\/\/(.+?)@(.+):(.+?)(?:[\/\?].*?)?(#(.+))?$", line)
     if b64data == None:
@@ -33,7 +35,8 @@ def parse_format_sip002(line):
     server_addr = b64data[2]
     server_port = b64data[3]
     remarks = b64data[5].strip()
-    remarks = server_addr if remarks == None else unquote_plus(remarks, "utf8", errors='ignore')
+    remarks = server_addr if remarks == None else unquote_plus(
+        remarks, "utf8", errors='ignore')
     return {
         "protocol": "shadowsocks",
         "settings": {
@@ -49,14 +52,140 @@ def parse_format_sip002(line):
         "remarks": remarks,
     }
 
-def parse_format_default(line):
+
+def parse_format_sip002_like_protocol(line):
+    mt = match(
+        r"^(\w+):\/\/(.+?)@(.+):(.+?)(?:(?:[\/\?]{1,2})(.*?))?(?:#(.+))?$", line)
+    if mt == None:
+        return None
+    protocol = mt[1]
+    uuid = mt[2]
+    server_addr = mt[3]
+    server_port = mt[4]
+    query_string = mt[5]
+    try:
+        cfg = parse_qs(query_string, strict_parsing=True)
+    except:
+        return None
+    remarks = mt[6].strip() if mt[6] else None
+    remarks = server_addr if remarks == None else unquote_plus(
+        remarks, "utf8", errors='ignore')
+    server = {
+        "remarks": remarks,
+        "protocol": protocol.lower(),
+    }
+    settings = {
+        "address": server_addr,
+        "port": int(server_port),
+    }
+    match server["protocol"]:
+        case "vless":
+            settings["id"] = uuid
+            settings["encryption"] = cfg.get("encryption", ["none"])[0]
+            settings["flow"] = cfg.get("flow", [""])[0]
+            settings["level"] = int(cfg.get("level", ["0"])[0])
+        case "vmess":
+            settings["id"] = uuid
+            settings["security"] = "auto"
+            settings["level"] = int(cfg.get("level", ["0"])[0])
+        case "hysteria2":
+            server["protocol"] = "hysteria"
+            settings["version"] = 2
+            cfg.setdefault("type", ["hysteria"])  # default network hysteria
+        case _:
+            print(f"protocol {server['protocol']} not support:", line)
+            return None
+    server["settings"] = settings
+    stream_settings = {}
+    match cfg.get("type", ["raw"])[0]:
+        case "tcp" | "raw":
+            stream_settings["network"] = "raw"
+            stream_settings["rawSettings"] = {}
+            if "headerType" in cfg:
+                headerType = cfg["headerType"][0]
+                match headerType:
+                    case "none":
+                        stream_settings["rawSettings"]["header"] = {
+                            "type": "none"}
+                    case _:
+                        print(f"headerType {headerType} not support:", line)
+                        return None
+        case "ws":
+            stream_settings["network"] = "ws"
+            stream_settings["wsSettings"] = {}
+            stream_settings["wsSettings"]["heartbeatPeriod"] = 20
+            if "host" in cfg:
+                stream_settings["wsSettings"]["host"] = cfg["host"][0]
+            if "path" in cfg:
+                stream_settings["wsSettings"]["path"] = cfg["path"][0]
+            if "headerType" in cfg and cfg["headerType"][0] != "none":
+                print(f"headerType {cfg['headerType'][0]} not support:", line)
+                return None
+        case "hysteria":
+            del cfg["type"]
+            stream_settings["network"] = "hysteria"
+            stream_settings["hysteriaSettings"] = {}
+            stream_settings["hysteriaSettings"]["auth"] = uuid
+            stream_settings["hysteriaSettings"]["version"] = 2
+            # default security hysteria
+            cfg.setdefault("security", ["hysteria"])
+        case _:
+            print(
+                f"Streaming type {cfg.get('type', ['tcp'])[0]} not support:", line)
+            return None
+    match cfg.get("security", ["none"])[0]:
+        case "tls":
+            stream_settings["security"] = "tls"
+            stream_settings["tlsSettings"] = {
+                "serverName": cfg.get("sni", [server_addr])[0],
+                "allowInsecure": False,
+                "fingerprint": cfg.get("fp", ["chrome"])[0],
+            }
+        case "reality":
+            stream_settings["security"] = "reality"
+            stream_settings["realitySettings"] = {
+                "serverName": cfg.get("sni", [server_addr])[0],
+                "fingerprint": cfg.get("fp", ["chrome"])[0],
+                "password": cfg.get("pbk", [""])[0],
+                "shortId": cfg.get("sid", [""])[0],
+            }
+        case "none":
+            stream_settings["security"] = "none"
+        case "hysteria":
+            del cfg["security"]
+            stream_settings["security"] = "tls"
+            stream_settings["tlsSettings"] = {
+                "serverName": cfg.get("sni", [server_addr])[0],
+                "alpn": ["h3"],
+            }
+            if "insecure" in cfg and cfg["insecure"][0] == "1":
+                stream_settings["tlsSettings"]["allowInsecure"] = True
+            stream_settings["finalmask"] = {
+                "quicParams": {},
+            }
+            if "mport" in cfg:
+                stream_settings["finalmask"]["quicParams"]["udpHop"] = {
+                    "ports": cfg["mport"][0]
+                }
+        case _:
+            print(
+                f"Streaming security {cfg.get('security', ['none'])[0]} not support:", line)
+            return None
+    server["streamSettings"] = stream_settings
+    return server
+
+
+def parse_format_encoded_json_url(line):
     b64data = match(r"^(\w+?):\/\/(.*)$", line)
     if b64data == None:
         return None
     protocol = b64data[1]
     b64data = b64data[2]
+    try:
         data = base64_decode(b64data)
         server_json = json.loads(data)
+    except:
+        return None
     if protocol.lower() == "vless":
         user = {
             "id": server_json["id"],
@@ -72,7 +201,7 @@ def parse_format_default(line):
         "network": server_json["net"],
     }
     if server_json["net"].lower() == "tcp":
-        if server_json["type"].lower() == "http":
+        if server_json["type"] == "http":
             stream["tcpSettings"] = {
                 "header": {
                     "type": "http",
@@ -104,7 +233,7 @@ def parse_format_default(line):
     else:
         print(f"net {server_json['net'].lower()} not support:", server_json)
         return None
-    if server_json["tls"] == "tls":
+    if "tls" in server_json and server_json["tls"] == "tls":
         stream["security"] = "tls"
         stream["tlsSettings"] = {
             "serverName": server_json["sni"]
@@ -133,6 +262,7 @@ def parse_format_default(line):
     server["remarks"] = server_json["ps"]
     return server
 
+
 def get_servers_from_subscribe_url(url):
     servers = []
     try:
@@ -159,14 +289,18 @@ def get_servers_from_subscribe_url(url):
     for line in text.split("\n"):
         try:
             server = parse_format_sip002(line)
-            server = parse_format_default(line) if server == None else server
+            server = parse_format_sip002_like_protocol(
+                line) if server == None else server
+            server = parse_format_encoded_json_url(
+                line) if server == None else server
             if server != None:
                 servers.append(server)
         except:
             print(line)
-            print_exc()
+            print_exc(limit=2)
     return servers
 
+
 def main():
     loop = True
     while loop:
@@ -220,7 +354,8 @@ def main():
                             break
                         print("{:< 4d}: {}".format(i, servers[i]["remarks"]))
                     print(": PAGE {}/{}".format(pages+1, page_count))
-                    num_sel = input("> Please input a number or operators to continue: ")
+                    num_sel = input(
+                        "> Please input a number or operators to continue: ")
                     if num_sel.startswith("<"):
                         pages -= num_sel.count("<")
                         if pages < 0:
@@ -235,7 +370,8 @@ def main():
                         index = -1
                         try:
                             index = int(num_sel)
-                        except: pass
+                        except:
+                            pass
                         if index >= 0 and index < page_size:
                             server = servers[index]
                             out = {}
@@ -260,5 +396,8 @@ def main():
             print_exc()
         print()
 
+
 if __name__ == "__main__":
     main()
+    # debug
+    # get_servers_from_subscribe_url("")