diff --git a/tests/test_sing_box.py b/tests/test_sing_box.py new file mode 100644 index 0000000..370cc9d --- /dev/null +++ b/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 diff --git a/tests/test_xray.py b/tests/test_xray.py new file mode 100644 index 0000000..a8d3562 --- /dev/null +++ b/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" + ) diff --git a/v2share/base.py b/v2share/base.py index ca8dca8..b8fe294 100644 --- a/v2share/base.py +++ b/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 diff --git a/v2share/clash.py b/v2share/clash.py index 673c9f5..2c162af 100644 --- a/v2share/clash.py +++ b/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: diff --git a/v2share/clashmeta.py b/v2share/clashmeta.py index 6eaf9d4..c2a9015 100644 --- a/v2share/clashmeta.py +++ b/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: diff --git a/v2share/data.py b/v2share/data.py index d12c485..7a6e18a 100644 --- a/v2share/data.py +++ b/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) diff --git a/v2share/links.py b/v2share/links.py index afefd73..762d636 100644 --- a/v2share/links.py +++ b/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: diff --git a/v2share/singbox.py b/v2share/singbox.py index cfa32b7..21ef10a 100644 --- a/v2share/singbox.py +++ b/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: diff --git a/v2share/xray.py b/v2share/xray.py index bd3f72d..6c05774 100644 --- a/v2share/xray.py +++ b/v2share/xray.py @@ -9,25 +9,22 @@ from v2share.data import V2Data, XrayNoise, SplitHttpSettings from v2share.exceptions import ProtocolNotSupportedError, TransportNotSupportedError -supported_transports = [ - "tcp", - "kcp", - "mkcp", - "ws", - "websocket", - "http", - "h2", - "quic", - "grpc", - "gun", - "httpupgrade", - "splithttp", - None, -] -supported_protocols = ["vmess", "vless", "trojan", "shadowsocks", "wireguard"] - class XrayConfig(BaseConfig): + chaining_support = True + supported_transports = [ + "tcp", + "kcp", + "ws", + "http", + "quic", + "grpc", + "httpupgrade", + "splithttp", + None, + ] + supported_protocols = ["vmess", "vless", "trojan", "shadowsocks", "wireguard"] + def __init__( self, template_path: str = None, @@ -51,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: @@ -73,49 +72,67 @@ def render(self, sort: bool = True, shuffle: bool = False): xray_configs = [] for data in configs: - outbound = {"tag": data.remark, "protocol": data.protocol} - - if data.protocol == "vmess": - outbound["settings"] = XrayConfig.vmess_config( - address=data.address, port=data.port, uuid=str(data.uuid) - ) - elif data.protocol == "vless": - if data.tls in {"reality", "tls"}: - flow = data.flow or "" - else: - flow = "" - outbound["settings"] = XrayConfig.vless_config( - address=data.address, port=data.port, uuid=str(data.uuid), flow=flow - ) + outbounds = self.create_outbounds(data, self._mux_template) + json_template = json.loads(self._template) + complete_config = { + **json_template, + **{ + "remarks": data.remark, + "outbounds": outbounds + json_template["outbounds"], + }, + } + xray_configs.append(complete_config) + return json.dumps(xray_configs, indent=4) - elif data.protocol == "trojan": - outbound["settings"] = XrayConfig.trojan_config( - address=data.address, port=data.port, password=data.password - ) + @staticmethod + def create_outbounds(data: V2Data, mux_template: str): + outbound = {"tag": data.remark, "protocol": data.protocol} + outbounds = [outbound] + dialer_proxy = None + + if data.protocol == "vmess": + outbound["settings"] = XrayConfig.vmess_config( + address=data.address, port=data.port, uuid=str(data.uuid) + ) + elif data.protocol == "vless": + if data.tls in {"reality", "tls"}: + flow = data.flow or "" + else: + flow = "" + outbound["settings"] = XrayConfig.vless_config( + address=data.address, port=data.port, uuid=str(data.uuid), flow=flow + ) - elif data.protocol == "shadowsocks": - outbound["settings"] = XrayConfig.shadowsocks_config( - address=data.address, - port=data.port, - password=data.password, - method=data.shadowsocks_method, - ) - elif data.protocol == "wireguard": - outbound["settings"] = XrayConfig.wireguard_config( - address=data.address, - port=data.port, - peer_public_key=data.path, - private_key=data.ed25519, - mtu=data.mtu, - allowed_ips=data.allowed_ips or ["0.0.0.0/0", "::/0"], - keepalive=25, - client_address=[data.client_address], - ) + elif data.protocol == "trojan": + outbound["settings"] = XrayConfig.trojan_config( + address=data.address, port=data.port, password=data.password + ) - outbounds = [outbound] - dialer_proxy = None + elif data.protocol == "shadowsocks": + outbound["settings"] = XrayConfig.shadowsocks_config( + address=data.address, + port=data.port, + password=data.password, + method=data.shadowsocks_method, + ) + elif data.protocol == "wireguard": + outbound["settings"] = XrayConfig.wireguard_config( + address=data.address, + port=data.port, + peer_public_key=data.path, + private_key=data.ed25519, + mtu=data.mtu, + allowed_ips=data.allowed_ips or ["0.0.0.0/0", "::/0"], + keepalive=25, + client_address=[data.client_address], + ) - if data.fragment: + if data.protocol != "wireguard": + if data.next: + next_outbounds = XrayConfig.create_outbounds(data.next, mux_template) + outbounds.extend(next_outbounds) + dialer_proxy = next_outbounds[0]["tag"] + elif data.fragment: fragment_outbound = XrayConfig.make_fragment_outbound( data.fragment_packets, data.fragment_length, data.fragment_interval ) @@ -126,39 +143,29 @@ def render(self, sort: bool = True, shuffle: bool = False): outbounds.append(noisy_outbound) dialer_proxy = noisy_outbound["tag"] - if data.protocol != "wireguard": - outbound["streamSettings"] = XrayConfig.make_stream_settings( - net=data.transport_type, - tls=data.tls, - sni=data.sni, - host=data.host, - path=data.path, - alpn=data.alpn, - fp=data.fingerprint, - pbk=data.reality_pbk, - sid=data.reality_sid, - ais=data.allow_insecure, - header_type=data.header_type, - grpc_multi_mode=data.grpc_multi_mode, - dialer_proxy=dialer_proxy, - headers=data.http_headers, - ) + outbound["streamSettings"] = XrayConfig.make_stream_settings( + net=data.transport_type, + tls=data.tls, + sni=data.sni, + host=data.host, + path=data.path, + alpn=data.alpn, + fp=data.fingerprint, + pbk=data.reality_pbk, + sid=data.reality_sid, + ais=data.allow_insecure, + header_type=data.header_type, + grpc_multi_mode=data.grpc_multi_mode, + dialer_proxy=dialer_proxy, + headers=data.http_headers, + ) - if data.mux_settings.protocol == "mux_cool": - mux_config = json.loads(self._mux_template) - mux_config["enabled"] = data.enable_mux - outbound["mux"] = mux_config + if data.mux_settings.protocol == "mux_cool": + mux_config = json.loads(mux_template) + mux_config["enabled"] = data.enable_mux + outbound["mux"] = mux_config - json_template = json.loads(self._template) - complete_config = { - **json_template, - **{ - "remarks": data.remark, - "outbounds": outbounds + json_template["outbounds"], - }, - } - xray_configs.append(complete_config) - return json.dumps(xray_configs, indent=4) + return outbounds @staticmethod def tls_config(sni, fingerprint, alpn=None, ais=False):