implement chaining support for xray and sing-box config generation

tests/test_sing_box.py

@@ -0,0 +1,40 @@
+import json
+import uuid
+
+from v2share import SingBoxConfig, V2Data
+
+
+def test_sing_box_chaining():
+    detour_config = V2Data(
+        "vmess",
+        "detour_outbound",
+        "127.0.0.1",
+        1234,
+        uuid=uuid.UUID("bb34fc3a-529d-473a-a3d9-1749b2116f2a"),
+    )
+    main_config = V2Data(
+        "vmess",
+        "main_outbound",
+        "127.0.0.1",
+        1234,
+        uuid=uuid.UUID("bb34fc3a-529d-473a-a3d9-1749b2116f2a"),
+        next=detour_config,
+    )
+    sb = SingBoxConfig()
+    sb.add_proxies([main_config])
+
+    result = json.loads(sb.render())
+
+    main_outbound, selector_outbounds = None, []
+    for outbound in result["outbounds"]:
+        if outbound["tag"] == "main_outbound":
+            main_outbound = outbound
+
+        if outbound["type"] == "selector" or outbound["type"] == "urltest":
+            selector_outbounds.append(outbound)
+
+    assert main_outbound["detour"] == "detour_outbound"
+    for outbound in selector_outbounds:
+        assert (
+            "detour_outbound" not in outbound["outbounds"]
+        )  # detour outbound should not be in the selectors

tests/test_xray.py

@@ -0,0 +1,34 @@
+import json
+import uuid
+
+from v2share import XrayConfig, V2Data
+
+
+def test_xray_chaining():
+    detour_config = V2Data(
+        "vmess",
+        "detour_outbound",
+        "127.0.0.1",
+        1234,
+        uuid=uuid.UUID("bb34fc3a-529d-473a-a3d9-1749b2116f2a"),
+    )
+    main_config = V2Data(
+        "vmess",
+        "main_outbound",
+        "127.0.0.1",
+        1234,
+        uuid=uuid.UUID("bb34fc3a-529d-473a-a3d9-1749b2116f2a"),
+        next=detour_config,
+    )
+    x = XrayConfig()
+    x.add_proxies([main_config])
+    result = json.loads(x.render())
+
+    main_outbound = None
+    for outbound in result[0]["outbounds"]:
+        if outbound["tag"] == "main_outbound":
+            main_outbound = outbound
+            break
+    assert (
+        main_outbound["streamSettings"]["sockopt"]["dialerProxy"] == "detour_outbound"
+    )

v2share/base.py

@@ -5,6 +5,10 @@
 
 
 class BaseConfig(ABC):
+    chaining_support: bool = False
+    supported_transports: List
+    supported_protocols: List
+
     @abstractmethod
     def render(self, sort: bool, shuffle: bool) -> str:
         pass

v2share/clash.py

@@ -8,11 +8,11 @@
 from v2share.data import V2Data
 from v2share.exceptions import TransportNotSupportedError, ProtocolNotSupportedError
 
-supported_transports = ["tcp", "http", "ws", "grpc", "h2"]
-supported_protocols = ["vmess", "trojan", "shadowsocks"]
-
 
 class ClashConfig(BaseConfig):
+    supported_transports = ["tcp", "http", "ws", "grpc", "h2", None]
+    supported_protocols = ["vmess", "trojan", "shadowsocks"]
+
     def __init__(self, template_path: Optional[str] = None, swallow_errors=True):
         if not template_path:
             template_path = resources.files("v2share.templates") / "clash.yml"
@@ -26,8 +26,10 @@ def add_proxies(self, proxies: List[V2Data]):
             # validation
             if (
                 unsupported_transport := proxy.transport_type
-                not in supported_transports
-            ) or (unsupported_protocol := proxy.protocol not in supported_protocols):
+                not in self.supported_transports
+            ) or (
+                unsupported_protocol := proxy.protocol not in self.supported_protocols
+            ):
                 if self._swallow_errors:
                     continue
                 if unsupported_transport:

v2share/clashmeta.py

@@ -4,11 +4,11 @@
 from v2share.data import V2Data
 from v2share.exceptions import TransportNotSupportedError, ProtocolNotSupportedError
 
-supported_transports = ["tcp", "http", "ws", "grpc", "h2"]
-supported_protocols = ["vmess", "trojan", "shadowsocks", "vless"]
-
 
 class ClashMetaConfig(ClashConfig):
+    supported_transports = ["tcp", "http", "ws", "grpc", "h2", None]
+    supported_protocols = ["vmess", "trojan", "shadowsocks", "vless"]
+
     def _make_node(
         self,
         name: str,
@@ -93,8 +93,10 @@ def add_proxies(self, proxies: List[V2Data]):
             # validation
             if (
                 unsupported_transport := proxy.transport_type
-                not in supported_transports
-            ) or (unsupported_protocol := proxy.protocol not in supported_protocols):
+                not in self.supported_transports
+            ) or (
+                unsupported_protocol := proxy.protocol not in self.supported_protocols
+            ):
                 if self._swallow_errors:
                     continue
                 if unsupported_transport:

v2share/data.py

@@ -97,6 +97,7 @@ class V2Data:
     enable_mux: bool = False
     allow_insecure: bool = False
     weight: int = 1
+    next: Optional["V2Data"] = None
     splithttp_settings: Optional[SplitHttpSettings] = None
     mux_settings: MuxSettings = field(default_factory=MuxSettings)
 

v2share/links.py

@@ -5,29 +5,29 @@
 from v2share.data import V2Data
 from v2share.exceptions import TransportNotSupportedError, ProtocolNotSupportedError
 
-supported_transports = [
-    "tcp",
-    "kcp",
-    "ws",
-    "http",
-    "quic",
-    "grpc",
-    "httpupgrade",
-    "splithttp",
-    None,
-]
-supported_protocols = [
-    "vmess",
-    "vless",
-    "trojan",
-    "shadowsocks",
-    "hysteria2",
-    "wireguard",
-    "tuic",
-]
-
 
 class LinksConfig(BaseConfig):
+    supported_transports = [
+        "tcp",
+        "kcp",
+        "ws",
+        "http",
+        "quic",
+        "grpc",
+        "httpupgrade",
+        "splithttp",
+        None,
+    ]
+    supported_protocols = [
+        "vmess",
+        "vless",
+        "trojan",
+        "shadowsocks",
+        "hysteria2",
+        "wireguard",
+        "tuic",
+    ]
+
     def __init__(self, swallow_errors=True):
         self._configs: List[V2Data] = []
         self._swallow_errors = swallow_errors
@@ -48,8 +48,10 @@ def add_proxies(self, proxies: List[V2Data]):
             # validation
             if (
                 unsupported_transport := proxy.transport_type
-                not in supported_transports
-            ) or (unsupported_protocol := proxy.protocol not in supported_protocols):
+                not in self.supported_transports
+            ) or (
+                unsupported_protocol := proxy.protocol not in self.supported_protocols
+            ):
                 if self._swallow_errors:
                     continue
                 if unsupported_transport:

v2share/singbox.py

@@ -7,20 +7,21 @@
 from v2share.data import V2Data
 from v2share.exceptions import ProtocolNotSupportedError, TransportNotSupportedError
 
-supported_protocols = [
-    "shadowsocks",
-    "vmess",
-    "trojan",
-    "vless",
-    "hysteria2",
-    "wireguard",
-    "shadowtls",
-    "tuic",
-]
-supported_transports = ["tcp", "ws", "quic", "httpupgrade", "grpc", "http", None]
-
 
 class SingBoxConfig(BaseConfig):
+    chaining_support = True
+    supported_protocols = [
+        "shadowsocks",
+        "vmess",
+        "trojan",
+        "vless",
+        "hysteria2",
+        "wireguard",
+        "shadowtls",
+        "tuic",
+    ]
+    supported_transports = ["tcp", "ws", "quic", "httpupgrade", "grpc", "http", None]
+
     def __init__(self, template_path: str = None, swallow_errors=True):
         if not template_path:
             template_path = resources.files("v2share.templates") / "singbox.json"
@@ -34,12 +35,25 @@ def render(self, sort: bool = True, shuffle: bool = False):
         if shuffle is True:
             configs = random.sample(self._configs, len(self._configs))
         elif sort is True:
-            configs = sorted(self._configs, key=lambda config: config.weight)
+            configs = sorted(self._configs, key=lambda c: c.weight)
         else:
             configs = self._configs
 
         result = json.loads(self._template_data)
-        result["outbounds"].extend([self.create_outbound(config) for config in configs])
+
+        blackset = set()
+        for config in configs:
+            c = config
+            while True:
+                outbound = self.create_outbound(c)
+                if c.next:
+                    outbound["detour"] = config.next.remark
+                    blackset.add(c.next.remark)
+                    c = config.next
+                    result["outbounds"].append(outbound)
+                else:
+                    result["outbounds"].append(outbound)
+                    break
 
         urltest_types = [
             "hysteria2",
@@ -54,7 +68,7 @@ def render(self, sort: bool = True, shuffle: bool = False):
         urltest_tags = [
             outbound["tag"]
             for outbound in result["outbounds"]
-            if outbound["type"] in urltest_types
+            if outbound["type"] in urltest_types and outbound["tag"] not in blackset
         ]
         selector_types = [
             "hysteria2",
@@ -70,7 +84,7 @@ def render(self, sort: bool = True, shuffle: bool = False):
         selector_tags = [
             outbound["tag"]
             for outbound in result["outbounds"]
-            if outbound["type"] in selector_types
+            if outbound["type"] in selector_types and outbound["tag"] not in blackset
         ]
 
         for outbound in result["outbounds"]:
@@ -254,8 +268,10 @@ def add_proxies(self, proxies: List[V2Data]):
             # validation
             if (
                 unsupported_transport := proxy.transport_type
-                not in supported_transports
-            ) or (unsupported_protocol := proxy.protocol not in supported_protocols):
+                not in self.supported_transports
+            ) or (
+                unsupported_protocol := proxy.protocol not in self.supported_protocols
+            ):
                 if self._swallow_errors:
                     continue
                 if unsupported_transport: