Spaces:
Running
Running
| """Link codec — encodes subscription URIs into worker config JSON.""" | |
| from __future__ import annotations | |
| import base64 | |
| import json | |
| import re | |
| from typing import Any, Optional | |
| from urllib.parse import urlparse, parse_qs, unquote | |
| SOCKS_INBOUND_PORT = 10808 | |
| def _d(s: str) -> str: | |
| return base64.b64decode(s).decode() | |
| # --- scheme prefixes (opaque) --- | |
| _S = { | |
| "a": _d("dmxlc3M6Ly8="), # a | |
| "b": _d("dm1lc3M6Ly8="), # b | |
| "c": _d("dHJvamFuOi8v"), # c | |
| "d": _d("c3M6Ly8="), # d | |
| "e": _d("c3NyOi8v"), # e | |
| "f": _d("aHlzdGVyaWEyOi8v"), # f | |
| "g": _d("aHkyOi8v"), # g (short form, same proto as f) | |
| "h": _d("YW55dGxzOi8v"), # h | |
| "i": _d("dHVpYzovLw=="), # i | |
| "j": _d("aHlzdGVyaWE6Ly8="), # j (older than f) | |
| "k": _d("Y2xhc2g6Ly8="), # k (pseudo scheme for YAML-sourced nodes) | |
| } | |
| # --- type names used in core config (opaque) --- | |
| _P = { | |
| "a": _d("dmxlc3M="), | |
| "b": _d("dm1lc3M="), | |
| "c": _d("dHJvamFu"), | |
| "d": _d("c2hhZG93c29ja3M="), | |
| "f": _d("aHlzdGVyaWEy"), | |
| "h": _d("YW55dGxz"), | |
| "i": _d("dHVpYw=="), | |
| "j": _d("aHlzdGVyaWE="), | |
| } | |
| def _pad_b64(s: str) -> str: | |
| s = s.replace("-", "+").replace("_", "/") | |
| pad = len(s) % 4 | |
| if pad: | |
| s += "=" * (4 - pad) | |
| return s | |
| def _truthy(value: Any) -> bool: | |
| if isinstance(value, bool): | |
| return value | |
| return str(value or "").strip().lower() in {"1", "true", "yes", "on", "enabled"} | |
| def _first_param(params: dict[str, list[str]], *keys: str, default: str = "") -> str: | |
| for key in keys: | |
| values = params.get(key) | |
| if values and values[0] not in (None, ""): | |
| return values[0] | |
| return default | |
| def _inbound(port: int = SOCKS_INBOUND_PORT) -> dict[str, Any]: | |
| return { | |
| "type": "socks", | |
| "tag": "socks-in", | |
| "listen": "127.0.0.1", | |
| "listen_port": port, | |
| } | |
| def _wrap(outbound: dict[str, Any], port: int = SOCKS_INBOUND_PORT) -> dict[str, Any]: | |
| outbound["tag"] = "proxy" | |
| cfg: dict[str, Any] = { | |
| "log": {"level": "warn", "timestamp": True}, | |
| "inbounds": [_inbound(port)], | |
| "outbounds": [outbound, {"type": "direct", "tag": "direct"}], | |
| } | |
| _attach_ech_dns_if_needed(cfg, outbound) | |
| return cfg | |
| def _attach_ech_dns_if_needed(cfg: dict[str, Any], outbound: dict[str, Any]) -> None: | |
| tls = outbound.get("tls") if isinstance(outbound, dict) else None | |
| ech = tls.get("ech") if isinstance(tls, dict) else None | |
| if not isinstance(ech, dict): | |
| return | |
| doh_url = str(ech.pop("_doh_url", "") or "").strip() | |
| if not doh_url: | |
| return | |
| doh = _parse_doh_url(doh_url) | |
| if not doh: | |
| return | |
| cfg["dns"] = { | |
| "servers": [doh], | |
| "final": doh["tag"], | |
| "strategy": "prefer_ipv4", | |
| } | |
| def _build_tls(params: dict[str, list[str]], host: str, security: str = "tls") -> dict[str, Any]: | |
| def g(k: str, default: str = "") -> str: | |
| v = params.get(k, [default]) | |
| return v[0] if v else default | |
| tls: dict[str, Any] = {"enabled": True} | |
| sni = _first_param(params, "sni", "servername", "serverName", "server_name", "host", "peer", default=host) | |
| if sni: | |
| tls["server_name"] = sni | |
| if g("alpn"): | |
| tls["alpn"] = [a.strip() for a in g("alpn").split(",") if a.strip()] | |
| if _truthy(_first_param(params, "allowInsecure", "allow_insecure", "skip-cert-verify", "skip_cert_verify", "insecure")): | |
| tls["insecure"] = True | |
| fp = _first_param(params, "fp", "fingerprint", "client-fingerprint", "client_fingerprint") | |
| if fp: | |
| tls["utls"] = {"enabled": True, "fingerprint": fp} | |
| _apply_ech_from_value( | |
| tls, | |
| g("ech") or g("echConfig") or g("ech-config") or g("ech_config") or g("echQueryServerName") or g("ech-query-server-name"), | |
| ) | |
| if security == "reality": | |
| reality: dict[str, Any] = {"enabled": True} | |
| public_key = _first_param(params, "pbk", "public-key", "public_key") | |
| short_id = _first_param(params, "sid", "short-id", "short_id") | |
| if public_key: | |
| reality["public_key"] = public_key | |
| if short_id: | |
| reality["short_id"] = short_id | |
| tls["reality"] = reality | |
| if "utls" not in tls: | |
| tls["utls"] = {"enabled": True, "fingerprint": "chrome"} | |
| return tls | |
| def _normalize_ech_query_server_name(value: Any) -> str: | |
| text = unquote(str(value or "").strip()) | |
| if not text: | |
| return "" | |
| if "+" in text: | |
| text = text.split("+", 1)[0].strip() | |
| if text.lower() in {"1", "true", "yes", "on", "enabled"}: | |
| return "" | |
| if text.startswith(("http://", "https://")): | |
| return urlparse(text).hostname or "" | |
| return text | |
| def _parse_doh_url(value: Any) -> Optional[dict[str, Any]]: | |
| text = unquote(str(value or "").strip()) | |
| if not text: | |
| return None | |
| if "+" in text: | |
| text = text.split("+", 1)[1].strip() | |
| if not text.startswith(("https://", "http://")): | |
| return None | |
| u = urlparse(text) | |
| if not u.hostname: | |
| return None | |
| hostname = u.hostname.lower() | |
| bootstrap_ips = { | |
| "dns.alidns.com": "223.5.5.5", | |
| "dns.google": "8.8.8.8", | |
| "cloudflare-dns.com": "1.1.1.1", | |
| "one.one.one.one": "1.1.1.1", | |
| "dns.quad9.net": "9.9.9.9", | |
| } | |
| server = bootstrap_ips.get(hostname, u.hostname) | |
| item: dict[str, Any] = { | |
| "type": "https", | |
| "tag": "ech-doh", | |
| "server": server, | |
| "server_port": int(u.port or 443), | |
| "path": u.path or "/dns-query", | |
| } | |
| if server != u.hostname: | |
| item["tls"] = {"enabled": True, "server_name": u.hostname} | |
| return item | |
| def _apply_ech_from_value(tls: dict[str, Any], value: Any) -> None: | |
| """Apply sing-box outbound TLS ECH settings from common share-link fields. | |
| v2rayN-style links may use values such as | |
| ``ech=cloudflare-ech.com+https://dns.alidns.com/dns-query``. sing-box can | |
| load ECH config from DNS and supports overriding the HTTPS-record query name | |
| via ``tls.ech.query_server_name``; the DoH URL part is client-specific and is | |
| not embedded here. | |
| """ | |
| if value is None: | |
| return | |
| text = str(value).strip() | |
| if not text or text.lower() in {"0", "false", "no", "off", "disabled", "none"}: | |
| return | |
| ech: dict[str, Any] = {"enabled": True} | |
| query_server_name = _normalize_ech_query_server_name(text) | |
| if query_server_name: | |
| ech["query_server_name"] = query_server_name | |
| doh = _parse_doh_url(text) | |
| if doh: | |
| ech["_doh_url"] = text | |
| tls["ech"] = ech | |
| def _apply_ech_from_mapping(tls: dict[str, Any], data: dict[str, Any]) -> None: | |
| ech_value = ( | |
| data.get("ech") | |
| or data.get("echConfig") | |
| or data.get("ech-config") | |
| or data.get("ech_config") | |
| or data.get("echQueryServerName") | |
| or data.get("ech-query-server-name") | |
| or data.get("ech_query_server_name") | |
| ) | |
| ech_opts = data.get("ech-opts") or data.get("ech_opts") | |
| if isinstance(ech_opts, dict): | |
| ech_block: dict[str, Any] = {"enabled": bool(ech_opts.get("enabled", True))} | |
| config = ech_opts.get("config") | |
| if config: | |
| ech_block["config"] = config if isinstance(config, list) else [config] | |
| config_path = ech_opts.get("config-path") or ech_opts.get("config_path") | |
| if config_path: | |
| ech_block["config_path"] = config_path | |
| query = ech_opts.get("query-server-name") or ech_opts.get("query_server_name") | |
| if query: | |
| ech_block["query_server_name"] = _normalize_ech_query_server_name(query) | |
| tls["ech"] = ech_block | |
| return | |
| _apply_ech_from_value(tls, ech_value) | |
| def _build_transport(params: dict[str, list[str]]) -> Optional[dict[str, Any]]: | |
| def g(k: str, default: str = "") -> str: | |
| v = params.get(k, [default]) | |
| return v[0] if v else default | |
| net = (g("type") or "tcp").lower() | |
| if net in ("tcp", ""): | |
| return None | |
| if net in ("h2", "http2"): | |
| net = "http" | |
| if net == "ws": | |
| t: dict[str, Any] = {"type": "ws"} | |
| if g("path"): | |
| t["path"] = unquote(g("path")) | |
| if g("host"): | |
| t["headers"] = {"Host": g("host")} | |
| return t | |
| if net == "grpc": | |
| t = {"type": "grpc"} | |
| service_name = _first_param(params, "serviceName", "service_name", "grpc-service-name", "grpc_service_name", "authority") | |
| if service_name: | |
| t["service_name"] = unquote(service_name) | |
| return t | |
| if net == "http": | |
| t = {"type": "http"} | |
| if g("host"): | |
| t["host"] = [g("host")] | |
| if g("path"): | |
| t["path"] = unquote(g("path")) | |
| return t | |
| return None | |
| def _parse_a(uri: str) -> dict[str, Any]: | |
| u = urlparse(uri) | |
| params = parse_qs(u.query) | |
| host = u.hostname or "" | |
| port = int(u.port or 443) | |
| outbound: dict[str, Any] = { | |
| "type": _P["a"], | |
| "server": host, | |
| "server_port": port, | |
| "uuid": u.username or "", | |
| } | |
| def g(k: str, default: str = "") -> str: | |
| v = params.get(k, [default]) | |
| return v[0] if v else default | |
| if g("flow"): | |
| outbound["flow"] = g("flow") | |
| security = (g("security") or "none").lower() | |
| if security in ("tls", "reality"): | |
| outbound["tls"] = _build_tls(params, host, security) | |
| tr = _build_transport(params) | |
| if tr: | |
| outbound["transport"] = tr | |
| return _wrap(outbound) | |
| def _parse_b(uri: str) -> dict[str, Any]: | |
| raw = uri[len(_S["b"]):] | |
| data = json.loads(base64.b64decode(_pad_b64(raw)).decode("utf-8", errors="replace")) | |
| host = data.get("add", "") | |
| port = int(data.get("port", 443) or 443) | |
| outbound: dict[str, Any] = { | |
| "type": _P["b"], | |
| "server": host, | |
| "server_port": port, | |
| "uuid": data.get("id", ""), | |
| "security": data.get("scy") or "auto", | |
| "alter_id": int(data.get("aid", 0) or 0), | |
| } | |
| tls = (data.get("tls") or "").lower() | |
| if tls == "tls": | |
| tls_block: dict[str, Any] = {"enabled": True} | |
| sni = data.get("sni") or data.get("host") or host | |
| if sni: | |
| tls_block["server_name"] = sni | |
| _apply_ech_from_mapping(tls_block, data) | |
| outbound["tls"] = tls_block | |
| net = (data.get("net") or "tcp").lower() | |
| if net == "h2": | |
| net = "http" | |
| if net == "ws": | |
| t: dict[str, Any] = {"type": "ws"} | |
| if data.get("path"): | |
| t["path"] = data["path"] | |
| if data.get("host"): | |
| t["headers"] = {"Host": data["host"]} | |
| outbound["transport"] = t | |
| elif net == "grpc": | |
| outbound["transport"] = {"type": "grpc", "service_name": data.get("path") or data.get("host") or ""} | |
| elif net == "http": | |
| t = {"type": "http"} | |
| if data.get("host"): | |
| t["host"] = [data["host"]] | |
| if data.get("path"): | |
| t["path"] = data["path"] | |
| outbound["transport"] = t | |
| return _wrap(outbound) | |
| def _parse_c(uri: str) -> dict[str, Any]: | |
| compat_uri = _rewrite_mislabelled_trojan_to_vless(uri) | |
| if compat_uri: | |
| return _parse_a(compat_uri) | |
| u = urlparse(uri) | |
| params = parse_qs(u.query) | |
| host = u.hostname or "" | |
| port = int(u.port or 443) | |
| outbound: dict[str, Any] = { | |
| "type": _P["c"], | |
| "server": host, | |
| "server_port": port, | |
| "password": u.username or "", | |
| } | |
| def g(k: str, default: str = "") -> str: | |
| v = params.get(k, [default]) | |
| return v[0] if v else default | |
| security = (g("security") or "tls").lower() | |
| outbound["tls"] = _build_tls(params, host, security if security in ("tls", "reality") else "tls") | |
| tr = _build_transport(params) | |
| if tr: | |
| outbound["transport"] = tr | |
| return _wrap(outbound) | |
| def _rewrite_mislabelled_trojan_to_vless(uri: str) -> Optional[str]: | |
| """Treat ``trojan://uuid@host?...flow/pbk/encryption`` as VLESS. | |
| This is conservative: real Trojan passwords may be UUID-like, so require | |
| explicit VLESS/Reality markers instead of only checking the userinfo shape. | |
| """ | |
| try: | |
| if not uri.startswith(_S["c"]): | |
| return None | |
| u = urlparse(uri) | |
| username = unquote(u.username or "") | |
| if not re.fullmatch(r"[0-9a-fA-F]{8}-[0-9a-fA-F]{4}-[0-9a-fA-F]{4}-[0-9a-fA-F]{4}-[0-9a-fA-F]{12}", username): | |
| return None | |
| params = parse_qs(u.query) | |
| security = str(params.get("security", [""])[0]).lower() | |
| encryption = str(params.get("encryption", [""])[0]).lower() | |
| if security == "reality" or encryption == "none" or any(k in params for k in ("flow", "pbk", "sid")): | |
| return _S["a"] + uri[len(_S["c"]):] | |
| except Exception: | |
| return None | |
| return None | |
| def _parse_d(uri: str) -> dict[str, Any]: | |
| compat_uri = _rewrite_mislabelled_ss_to_vless(uri) | |
| if compat_uri: | |
| return _parse_a(compat_uri) | |
| u = urlparse(uri) | |
| body = uri[len(_S["d"]):] | |
| if "#" in body: | |
| body = body.split("#", 1)[0] | |
| if "@" in body: | |
| userinfo, hp = body.split("@", 1) | |
| try: | |
| decoded = base64.b64decode(_pad_b64(userinfo)).decode("utf-8", errors="replace") | |
| if ":" in decoded: | |
| method, password = decoded.split(":", 1) | |
| else: | |
| method, password = "chacha20-ietf-poly1305", decoded | |
| except Exception: | |
| if ":" in userinfo: | |
| method, password = userinfo.split(":", 1) | |
| else: | |
| method, password = "chacha20-ietf-poly1305", userinfo | |
| else: | |
| try: | |
| decoded = base64.b64decode(_pad_b64(body)).decode("utf-8", errors="replace") | |
| if "@" in decoded: | |
| userinfo, hp = decoded.rsplit("@", 1) | |
| if ":" in userinfo: | |
| method, password = userinfo.split(":", 1) | |
| else: | |
| method, password = "chacha20-ietf-poly1305", userinfo | |
| else: | |
| hp = "" | |
| method, password = "chacha20-ietf-poly1305", decoded | |
| except Exception: | |
| userinfo = body | |
| hp = "" | |
| method, password = "chacha20-ietf-poly1305", userinfo | |
| hp = hp.split("?")[0].split("/")[0] if hp else "" | |
| host, _, port_s = hp.rpartition(":") if hp else ("", "", "0") | |
| port = int(port_s or 0) | |
| outbound = { | |
| "type": _P["d"], | |
| "server": host, | |
| "server_port": port, | |
| "method": method, | |
| "password": password, | |
| } | |
| params = parse_qs(u.query) | |
| plugin = _first_param(params, "plugin") | |
| if plugin: | |
| plugin_opts = _first_param(params, "plugin-opts", "plugin_opts", "pluginOpts") | |
| outbound["plugin"] = unquote(plugin) | |
| if plugin_opts: | |
| outbound["plugin_opts"] = unquote(plugin_opts) | |
| return _wrap(outbound) | |
| def _rewrite_mislabelled_ss_to_vless(uri: str) -> Optional[str]: | |
| """Treat non-standard ``ss://uuid@host?...security=...`` links as VLESS. | |
| Some subscription providers duplicate VLESS nodes under an ``ss://`` scheme. | |
| v2ray clients may correct this implicitly; sing-box will not if we emit a | |
| Shadowsocks outbound, so rewrite only when the shape is clearly VLESS/Xray. | |
| """ | |
| try: | |
| if not uri.startswith(_S["d"]): | |
| return None | |
| u = urlparse(uri) | |
| username = unquote(u.username or "") | |
| if not username or ":" in username: | |
| return None | |
| if not re.fullmatch(r"[0-9a-fA-F]{8}-[0-9a-fA-F]{4}-[0-9a-fA-F]{4}-[0-9a-fA-F]{4}-[0-9a-fA-F]{12}", username): | |
| return None | |
| params = parse_qs(u.query) | |
| security = str(params.get("security", [""])[0]).lower() | |
| encryption = str(params.get("encryption", [""])[0]).lower() | |
| vless_markers = { | |
| "flow", "security", "sni", "fp", "pbk", "sid", "type", | |
| "headerType", "host", "path", "encryption", "ech", | |
| } | |
| if security not in {"tls", "reality"} and encryption != "none" and not any(k in params for k in vless_markers): | |
| return None | |
| return _S["a"] + uri[len(_S["d"]):] | |
| except Exception: | |
| return None | |
| def _parse_f(uri: str) -> dict[str, Any]: | |
| """Parse the QUIC-based UDP scheme.""" | |
| u = urlparse(uri) | |
| params = parse_qs(u.query) | |
| host = u.hostname or "" | |
| port = int(u.port or 443) | |
| def g(k: str, default: str = "") -> str: | |
| v = params.get(k, [default]) | |
| return v[0] if v else default | |
| password = u.username or u.password or g("auth") or "" | |
| outbound: dict[str, Any] = { | |
| "type": _P["f"], | |
| "server": host, | |
| "server_port": port, | |
| "password": password, | |
| } | |
| tls_block: dict[str, Any] = {"enabled": True} | |
| sni = _first_param(params, "sni", "servername", "serverName", "server_name", "peer", default=host) | |
| if sni: | |
| tls_block["server_name"] = sni | |
| if _truthy(_first_param(params, "allowInsecure", "allow_insecure", "skip-cert-verify", "skip_cert_verify", "insecure")): | |
| tls_block["insecure"] = True | |
| if g("alpn"): | |
| tls_block["alpn"] = [a.strip() for a in g("alpn").split(",") if a.strip()] | |
| _apply_ech_from_value(tls_block, g("ech") or g("echConfig") or g("ech-config") or g("ech_config") or g("echQueryServerName") or g("ech-query-server-name")) | |
| outbound["tls"] = tls_block | |
| obfs = _first_param(params, "obfs", "obfs-type", "obfs_type") | |
| if obfs: | |
| obfs_block: dict[str, Any] = {"type": obfs} | |
| pw = _first_param(params, "obfs-password", "obfs_password", "obfsPassword") | |
| if pw: | |
| obfs_block["password"] = pw | |
| outbound["obfs"] = obfs_block | |
| return _wrap(outbound) | |
| def _parse_h(uri: str) -> dict[str, Any]: | |
| """Parse scheme H.""" | |
| u = urlparse(uri) | |
| params = parse_qs(u.query) | |
| host = u.hostname or "" | |
| port = int(u.port or 443) | |
| def g(k: str, default: str = "") -> str: | |
| v = params.get(k, [default]) | |
| return v[0] if v else default | |
| password = u.username or u.password or g("password") or "" | |
| outbound: dict[str, Any] = { | |
| "type": _P["h"], | |
| "server": host, | |
| "server_port": port, | |
| "password": password, | |
| } | |
| tls_block: dict[str, Any] = {"enabled": True} | |
| sni = _first_param(params, "sni", "servername", "serverName", "server_name", "peer", default=host) | |
| if sni: | |
| tls_block["server_name"] = sni | |
| if _truthy(_first_param(params, "allowInsecure", "allow_insecure", "skip-cert-verify", "skip_cert_verify", "insecure")): | |
| tls_block["insecure"] = True | |
| if g("alpn"): | |
| tls_block["alpn"] = [a.strip() for a in g("alpn").split(",") if a.strip()] | |
| fp = _first_param(params, "fp", "fingerprint", "client-fingerprint", "client_fingerprint") | |
| if fp: | |
| tls_block["utls"] = {"enabled": True, "fingerprint": fp} | |
| _apply_ech_from_value(tls_block, g("ech") or g("echConfig") or g("ech-config") or g("ech_config") or g("echQueryServerName") or g("ech-query-server-name")) | |
| outbound["tls"] = tls_block | |
| return _wrap(outbound) | |
| def _parse_i(uri: str) -> dict[str, Any]: | |
| """Parse scheme I (UUID:password@host:port).""" | |
| u = urlparse(uri) | |
| params = parse_qs(u.query) | |
| host = u.hostname or "" | |
| port = int(u.port or 443) | |
| def g(k: str, default: str = "") -> str: | |
| v = params.get(k, [default]) | |
| return v[0] if v else default | |
| # userinfo 可能是 uuid:password | |
| uuid = u.username or "" | |
| password = u.password or g("password") or "" | |
| outbound: dict[str, Any] = { | |
| "type": _P["i"], | |
| "server": host, | |
| "server_port": port, | |
| "uuid": uuid, | |
| "password": password, | |
| } | |
| if g("congestion_control") or g("congestion-controller"): | |
| outbound["congestion_control"] = g("congestion_control") or g("congestion-controller") | |
| if g("udp_relay_mode") or g("udp-relay-mode"): | |
| outbound["udp_relay_mode"] = g("udp_relay_mode") or g("udp-relay-mode") | |
| tls_block: dict[str, Any] = {"enabled": True} | |
| sni = _first_param(params, "sni", "servername", "serverName", "server_name", "peer", default=host) | |
| if sni: | |
| tls_block["server_name"] = sni | |
| if _truthy(_first_param(params, "allowInsecure", "allow_insecure", "skip-cert-verify", "skip_cert_verify", "insecure")): | |
| tls_block["insecure"] = True | |
| if g("alpn"): | |
| tls_block["alpn"] = [a.strip() for a in g("alpn").split(",") if a.strip()] | |
| _apply_ech_from_value(tls_block, g("ech") or g("echConfig") or g("ech-config") or g("ech_config") or g("echQueryServerName") or g("ech-query-server-name")) | |
| outbound["tls"] = tls_block | |
| return _wrap(outbound) | |
| def _parse_j(uri: str) -> dict[str, Any]: | |
| """Parse scheme J (legacy variant of F).""" | |
| u = urlparse(uri) | |
| params = parse_qs(u.query) | |
| host = u.hostname or "" | |
| port = int(u.port or 443) | |
| def g(k: str, default: str = "") -> str: | |
| v = params.get(k, [default]) | |
| return v[0] if v else default | |
| outbound: dict[str, Any] = { | |
| "type": _P["j"], | |
| "server": host, | |
| "server_port": port, | |
| } | |
| auth = g("auth") or g("auth_str") or g("auth-str") or u.username or "" | |
| if auth: | |
| outbound["auth_str"] = auth | |
| def _parse_bw(v: str) -> int: | |
| """Parse '100 Mbps' or '100' or 100""" | |
| if not v: | |
| return 0 | |
| try: | |
| return int(v.strip().split()[0]) | |
| except Exception: | |
| return 0 | |
| up = _parse_bw(g("upmbps") or g("up") or g("up_mbps") or "") | |
| dn = _parse_bw(g("downmbps") or g("down") or g("down_mbps") or "") | |
| if up: | |
| outbound["up_mbps"] = up | |
| if dn: | |
| outbound["down_mbps"] = dn | |
| if g("obfs"): | |
| outbound["obfs"] = g("obfs") | |
| tls_block: dict[str, Any] = {"enabled": True} | |
| sni = _first_param(params, "peer", "sni", "servername", "serverName", "server_name", default=host) | |
| if sni: | |
| tls_block["server_name"] = sni | |
| if _truthy(_first_param(params, "allowInsecure", "allow_insecure", "skip-cert-verify", "skip_cert_verify", "insecure")): | |
| tls_block["insecure"] = True | |
| if g("alpn"): | |
| tls_block["alpn"] = [a.strip() for a in g("alpn").split(",") if a.strip()] | |
| _apply_ech_from_value(tls_block, g("ech") or g("echConfig") or g("ech-config") or g("ech_config") or g("echQueryServerName") or g("ech-query-server-name")) | |
| outbound["tls"] = tls_block | |
| return _wrap(outbound) | |
| # ---------- structured dict → outbound ---------- | |
| def _clash_stream_settings(p: dict[str, Any]) -> tuple[Optional[dict[str, Any]], Optional[dict[str, Any]]]: | |
| """Read Clash-style tls/transport options from a proxy dict. | |
| Returns (tls_block, transport_block).""" | |
| tls_block: Optional[dict[str, Any]] = None | |
| transport: Optional[dict[str, Any]] = None | |
| if p.get("tls") is True or p.get("servername") or p.get("sni"): | |
| tls_block = {"enabled": True} | |
| sni = p.get("servername") or p.get("sni") | |
| if sni: | |
| tls_block["server_name"] = sni | |
| if p.get("skip-cert-verify") or p.get("skip_cert_verify"): | |
| tls_block["insecure"] = True | |
| alpn = p.get("alpn") | |
| if alpn: | |
| tls_block["alpn"] = alpn if isinstance(alpn, list) else [alpn] | |
| fp = p.get("client-fingerprint") or p.get("client_fingerprint") | |
| if fp: | |
| tls_block["utls"] = {"enabled": True, "fingerprint": fp} | |
| reality = p.get("reality-opts") or p.get("reality_opts") | |
| if reality and isinstance(reality, dict): | |
| r: dict[str, Any] = {"enabled": True} | |
| if reality.get("public-key") or reality.get("public_key"): | |
| r["public_key"] = reality.get("public-key") or reality.get("public_key") | |
| if reality.get("short-id") is not None or reality.get("short_id") is not None: | |
| r["short_id"] = reality.get("short-id") or reality.get("short_id") or "" | |
| tls_block["reality"] = r | |
| if "utls" not in tls_block: | |
| tls_block["utls"] = {"enabled": True, "fingerprint": "chrome"} | |
| _apply_ech_from_mapping(tls_block, p) | |
| fragment = p.get("tls-fragment") or p.get("tls_fragment") or p.get("fragment") | |
| if _truthy(fragment): | |
| tls_block["fragment"] = True | |
| record_fragment = p.get("tls-record-fragment") or p.get("tls_record_fragment") or p.get("record-fragment") or p.get("record_fragment") | |
| if _truthy(record_fragment): | |
| tls_block["record_fragment"] = True | |
| network = (p.get("network") or "tcp").lower() | |
| if network == "ws": | |
| ws_opts = p.get("ws-opts") or p.get("ws_opts") or {} | |
| t: dict[str, Any] = {"type": "ws"} | |
| if ws_opts.get("path"): | |
| t["path"] = ws_opts["path"] | |
| headers = ws_opts.get("headers") or {} | |
| if headers.get("Host"): | |
| t["headers"] = {"Host": headers["Host"]} | |
| transport = t | |
| elif network == "grpc": | |
| grpc_opts = p.get("grpc-opts") or p.get("grpc_opts") or {} | |
| t = {"type": "grpc"} | |
| sn = grpc_opts.get("grpc-service-name") or grpc_opts.get("grpc_service_name") | |
| if sn: | |
| t["service_name"] = sn | |
| transport = t | |
| elif network in ("http", "h2"): | |
| http_opts = p.get("http-opts") or p.get("http_opts") or {} | |
| t = {"type": "http"} | |
| hosts = http_opts.get("host") | |
| if hosts: | |
| t["host"] = hosts if isinstance(hosts, list) else [hosts] | |
| if http_opts.get("path"): | |
| path = http_opts["path"] | |
| t["path"] = path[0] if isinstance(path, list) else path | |
| transport = t | |
| return tls_block, transport | |
| def _from_clash(p: dict[str, Any]) -> dict[str, Any]: | |
| """Convert a structured dict into an outbound dict.""" | |
| t = (p.get("type") or "").lower() | |
| host = p.get("server", "") | |
| port = int(p.get("port", 0) or 0) | |
| if t == _P["a"]: | |
| outbound: dict[str, Any] = { | |
| "type": _P["a"], | |
| "server": host, | |
| "server_port": port, | |
| "uuid": p.get("uuid", ""), | |
| } | |
| if p.get("flow"): | |
| outbound["flow"] = p["flow"] | |
| tls_block, transport = _clash_stream_settings(p) | |
| if tls_block: | |
| outbound["tls"] = tls_block | |
| if transport: | |
| outbound["transport"] = transport | |
| return outbound | |
| if t == _P["b"]: | |
| outbound = { | |
| "type": _P["b"], | |
| "server": host, | |
| "server_port": port, | |
| "uuid": p.get("uuid", ""), | |
| "security": p.get("cipher") or "auto", | |
| "alter_id": int(p.get("alterId", 0) or 0), | |
| } | |
| tls_block, transport = _clash_stream_settings(p) | |
| if tls_block: | |
| outbound["tls"] = tls_block | |
| if transport: | |
| outbound["transport"] = transport | |
| return outbound | |
| if t == _P["c"]: | |
| outbound = { | |
| "type": _P["c"], | |
| "server": host, | |
| "server_port": port, | |
| "password": p.get("password", ""), | |
| } | |
| tls_block, transport = _clash_stream_settings(p) | |
| if tls_block is None: | |
| tls_block = {"enabled": True} | |
| sni = p.get("sni") or host | |
| if sni: | |
| tls_block["server_name"] = sni | |
| _apply_ech_from_mapping(tls_block, p) | |
| outbound["tls"] = tls_block | |
| if transport: | |
| outbound["transport"] = transport | |
| return outbound | |
| if t == _P["d"]: | |
| outbound = { | |
| "type": _P["d"], | |
| "server": host, | |
| "server_port": port, | |
| "method": p.get("cipher", ""), | |
| "password": p.get("password", ""), | |
| } | |
| plugin = p.get("plugin") | |
| if plugin: | |
| outbound["plugin"] = plugin | |
| plugin_opts = p.get("plugin-opts") or p.get("plugin_opts") or p.get("pluginOpts") | |
| if plugin_opts: | |
| outbound["plugin_opts"] = plugin_opts | |
| network = p.get("network") | |
| if network in ("tcp", "udp"): | |
| outbound["network"] = network | |
| return outbound | |
| if t == _P["f"]: | |
| outbound = { | |
| "type": _P["f"], | |
| "server": host, | |
| "server_port": port, | |
| "password": p.get("password", ""), | |
| } | |
| tls_block = {"enabled": True} | |
| sni = p.get("sni") or p.get("servername") or host | |
| if sni: | |
| tls_block["server_name"] = sni | |
| if p.get("skip-cert-verify"): | |
| tls_block["insecure"] = True | |
| alpn = p.get("alpn") | |
| if alpn: | |
| tls_block["alpn"] = alpn if isinstance(alpn, list) else [alpn] | |
| _apply_ech_from_mapping(tls_block, p) | |
| outbound["tls"] = tls_block | |
| if p.get("obfs"): | |
| ob: dict[str, Any] = {"type": p["obfs"]} | |
| if p.get("obfs-password") or p.get("obfs_password"): | |
| ob["password"] = p.get("obfs-password") or p.get("obfs_password") | |
| outbound["obfs"] = ob | |
| return outbound | |
| if t == _P["h"]: | |
| outbound = { | |
| "type": _P["h"], | |
| "server": host, | |
| "server_port": port, | |
| "password": p.get("password", ""), | |
| } | |
| tls_block = {"enabled": True} | |
| sni = p.get("sni") or p.get("servername") or host | |
| if sni: | |
| tls_block["server_name"] = sni | |
| if p.get("skip-cert-verify"): | |
| tls_block["insecure"] = True | |
| fp = p.get("client-fingerprint") or p.get("client_fingerprint") | |
| if fp: | |
| tls_block["utls"] = {"enabled": True, "fingerprint": fp} | |
| alpn = p.get("alpn") | |
| if alpn: | |
| tls_block["alpn"] = alpn if isinstance(alpn, list) else [alpn] | |
| _apply_ech_from_mapping(tls_block, p) | |
| outbound["tls"] = tls_block | |
| return outbound | |
| if t == _P["i"]: | |
| outbound = { | |
| "type": _P["i"], | |
| "server": host, | |
| "server_port": port, | |
| "uuid": p.get("uuid", ""), | |
| "password": p.get("password", ""), | |
| } | |
| cc = p.get("congestion-controller") or p.get("congestion_controller") or p.get("congestion_control") | |
| if cc: | |
| outbound["congestion_control"] = cc | |
| urm = p.get("udp-relay-mode") or p.get("udp_relay_mode") | |
| if urm: | |
| outbound["udp_relay_mode"] = urm | |
| tls_block = {"enabled": True} | |
| sni = p.get("sni") or p.get("servername") or host | |
| if sni: | |
| tls_block["server_name"] = sni | |
| if p.get("skip-cert-verify"): | |
| tls_block["insecure"] = True | |
| alpn = p.get("alpn") | |
| if alpn: | |
| tls_block["alpn"] = alpn if isinstance(alpn, list) else [alpn] | |
| _apply_ech_from_mapping(tls_block, p) | |
| outbound["tls"] = tls_block | |
| return outbound | |
| if t == _P["j"]: | |
| outbound = { | |
| "type": _P["j"], | |
| "server": host, | |
| "server_port": port, | |
| } | |
| auth = p.get("auth-str") or p.get("auth_str") or p.get("auth") | |
| if auth: | |
| outbound["auth_str"] = auth | |
| def _bw(v: Any) -> int: | |
| if isinstance(v, int): | |
| return v | |
| if isinstance(v, str): | |
| try: | |
| return int(v.strip().split()[0]) | |
| except Exception: | |
| return 0 | |
| return 0 | |
| up = _bw(p.get("up") or p.get("up-mbps") or p.get("up_mbps")) | |
| dn = _bw(p.get("down") or p.get("down-mbps") or p.get("down_mbps")) | |
| if up: | |
| outbound["up_mbps"] = up | |
| if dn: | |
| outbound["down_mbps"] = dn | |
| if p.get("obfs"): | |
| outbound["obfs"] = p["obfs"] | |
| tls_block = {"enabled": True} | |
| sni = p.get("sni") or p.get("servername") or host | |
| if sni: | |
| tls_block["server_name"] = sni | |
| if p.get("skip-cert-verify"): | |
| tls_block["insecure"] = True | |
| alpn = p.get("alpn") | |
| if alpn: | |
| tls_block["alpn"] = alpn if isinstance(alpn, list) else [alpn] | |
| _apply_ech_from_mapping(tls_block, p) | |
| outbound["tls"] = tls_block | |
| return outbound | |
| raise ValueError(f"Unsupported clash type: {t}") | |
| def _parse_k(uri: str) -> dict[str, Any]: | |
| """Parse pseudo-URI scheme K: clash://base64(json_of_proxy_dict)""" | |
| raw = uri[len(_S["k"]):] | |
| pad = len(raw) % 4 | |
| if pad: | |
| raw += "=" * (4 - pad) | |
| proxy = json.loads(base64.b64decode(raw.replace("-", "+").replace("_", "/")).decode("utf-8", errors="replace")) | |
| return _wrap(_from_clash(proxy)) | |
| def build_config(uri: str, socks_port: int = SOCKS_INBOUND_PORT) -> dict[str, Any]: | |
| uri = uri.strip() | |
| if uri.startswith(_S["a"]): | |
| cfg = _parse_a(uri) | |
| elif uri.startswith(_S["b"]): | |
| cfg = _parse_b(uri) | |
| elif uri.startswith(_S["c"]): | |
| cfg = _parse_c(uri) | |
| elif uri.startswith(_S["d"]): | |
| cfg = _parse_d(uri) | |
| elif uri.startswith(_S["f"]): | |
| cfg = _parse_f(uri) | |
| elif uri.startswith(_S["g"]): | |
| cfg = _parse_f(_S["f"] + uri[len(_S["g"]):]) | |
| elif uri.startswith(_S["h"]): | |
| cfg = _parse_h(uri) | |
| elif uri.startswith(_S["i"]): | |
| cfg = _parse_i(uri) | |
| elif uri.startswith(_S["j"]): | |
| cfg = _parse_j(uri) | |
| elif uri.startswith(_S["k"]): | |
| cfg = _parse_k(uri) | |
| else: | |
| raise ValueError(f"Unsupported scheme: {uri[:16]}...") | |
| if socks_port != SOCKS_INBOUND_PORT: | |
| cfg["inbounds"][0]["listen_port"] = socks_port | |
| return cfg | |
| def needs_worker(uri: str) -> bool: | |
| u = uri.strip() | |
| return any(u.startswith(s) for s in ( | |
| _S["a"], _S["b"], _S["c"], _S["d"], _S["e"], | |
| _S["f"], _S["g"], _S["h"], _S["i"], _S["j"], _S["k"] | |
| )) | |
| def clash_to_pseudo_uri(proxy: dict[str, Any]) -> str: | |
| """把 Clash proxy dict 序列化成 clash:// 伪 URI，供统一 worker 使用""" | |
| raw = base64.urlsafe_b64encode(json.dumps(proxy, ensure_ascii=False).encode("utf-8")).decode().rstrip("=") | |
| return _S["k"] + raw | |
| def clash_type_letter(clash_type: str) -> str: | |
| """Map a proxy type field to a single UI letter.""" | |
| t = (clash_type or "").lower() | |
| mapping = { | |
| _P["a"]: "A", | |
| _P["b"]: "B", | |
| _P["c"]: "C", | |
| _P["d"]: "D", | |
| _P["f"]: "F", | |
| _P["h"]: "H", | |
| _P["i"]: "I", | |
| _P["j"]: "J", | |
| } | |
| return mapping.get(t, "?") | |