loop-uh
d286239e0c
Some checks failed
Windows project source guards / test (push) Failing after 33s
«Подключается, но трафика нет» (0.8.45–0.8.47), проверено вживую против сервера Xray: - alpn=h2,http/1.1 из ссылки переносился в sing-box как есть, сервер выбирал h2, и WebSocket поверх него не поднимался (EOF). Для ws и httpupgrade остаётся только http/1.1, как делает сам Xray. - «?ed=2048» в пути WebSocket уходил в запрос (404). Теперь это max_early_data + Sec-WebSocket-Protocol. - httpupgrade не переносился ни в Xray, ни в sing-box; host из нового поля wsSettings.host тоже учитывается. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
257 lines
11 KiB
Python
257 lines
11 KiB
Python
from __future__ import annotations
|
||
|
||
from copy import deepcopy
|
||
from typing import Any
|
||
|
||
from xray_fluent.application.protocol_core import ProtocolCore, protocol_core
|
||
from xray_fluent.engines.tls_fingerprint import BROWSER_TLS_FINGERPRINT, tls_over_quic
|
||
|
||
_SUPPORTED_NATIVE_PROTOCOLS = {"vmess", "trojan", "shadowsocks", "socks", "http"}
|
||
|
||
# Типы, которые sing-box 1.13+ принимает только в top-level массиве `endpoints[]`,
|
||
# а не в `outbounds[]`.
|
||
_SINGBOX_ENDPOINT_TYPES = {"wireguard"}
|
||
|
||
|
||
def is_singbox_endpoint_outbound(outbound: dict[str, Any] | None) -> bool:
|
||
"""True, если конфиг ноды — sing-box endpoint (например, WireGuard/AWG)."""
|
||
if not isinstance(outbound, dict):
|
||
return False
|
||
native_type = str(outbound.get("type") or "").strip().lower()
|
||
return native_type in _SINGBOX_ENDPOINT_TYPES and not outbound.get("protocol")
|
||
|
||
|
||
def is_singbox_endpoint_node(node) -> bool:
|
||
return node is not None and is_singbox_endpoint_outbound(node.outbound)
|
||
|
||
|
||
def unenforced_tls_pin_warning(node) -> str:
|
||
"""Предупреждение, если сервер с pcs уходит в sing-box, где пин не проверяется."""
|
||
outbound = node.outbound if node is not None and isinstance(node.outbound, dict) else {}
|
||
if outbound.get("type") and not outbound.get("protocol"):
|
||
return ""
|
||
if protocol_core(node) is not ProtocolCore.SINGBOX:
|
||
return ""
|
||
stream = outbound.get("streamSettings")
|
||
tls_settings = stream.get("tlsSettings") if isinstance(stream, dict) else None
|
||
if not isinstance(tls_settings, dict) or not str(tls_settings.get("pinnedPeerCertSha256") or "").strip():
|
||
return ""
|
||
name = str(getattr(node, "name", "") or getattr(node, "server", "") or "сервер").strip()
|
||
return (
|
||
f"[warning] {name}: ссылка закрепляет сертификат (pcs), но sing-box не умеет сверять "
|
||
"SHA-256 всего сертификата, поэтому сертификат сервера не проверяется (tls.insecure). "
|
||
"Проверка отпечатка работает только в ядре Xray (VLESS)."
|
||
)
|
||
|
||
|
||
def build_singbox_outbound(node, *, tag: str = "proxy") -> dict[str, Any]:
|
||
"""Convert a stored node outbound into a native sing-box outbound."""
|
||
source = deepcopy(node.outbound or {})
|
||
protocol = str(source.get("protocol") or "").lower()
|
||
native_type = str(source.get("type") or "").lower()
|
||
if protocol_core(node) is not ProtocolCore.SINGBOX:
|
||
raise ValueError("VLESS belongs to Xray; Hysteria2 belongs to the official Hysteria core.")
|
||
if native_type and not protocol:
|
||
# Служебные ключи приложения (`_dns` и прочие с префиксом `_`) не входят
|
||
# в схему sing-box, строгий декодер их отвергает — отбрасываем.
|
||
source = {key: value for key, value in source.items() if not str(key).startswith("_")}
|
||
source["tag"] = tag
|
||
return source
|
||
|
||
if protocol not in _SUPPORTED_NATIVE_PROTOCOLS:
|
||
raise ValueError(
|
||
f"Текущий сервер нельзя конвертировать в native sing-box outbound: protocol `{protocol or 'unknown'}`"
|
||
)
|
||
|
||
outbound = _convert_outbound(source)
|
||
unsupported_transport = str(outbound.pop("_unsupported_transport", "") or "").strip()
|
||
if unsupported_transport:
|
||
raise ValueError(
|
||
f"Текущий сервер нельзя конвертировать в native sing-box outbound: transport `{unsupported_transport}` не поддерживается"
|
||
)
|
||
|
||
outbound["tag"] = tag
|
||
return outbound
|
||
|
||
|
||
def _convert_outbound(xray_ob: dict[str, Any]) -> dict[str, Any]:
|
||
protocol = str(xray_ob.get("protocol") or "").lower()
|
||
xray_settings = dict(xray_ob.get("settings") or {})
|
||
stream = dict(xray_ob.get("streamSettings") or {})
|
||
|
||
sb: dict[str, Any] = {"type": protocol}
|
||
|
||
if protocol == "vmess":
|
||
vnext = (xray_settings.get("vnext") or [{}])[0]
|
||
sb["server"] = str(vnext.get("address") or "")
|
||
sb["server_port"] = int(vnext.get("port") or 0)
|
||
users = (vnext.get("users") or [{}])[0]
|
||
sb["uuid"] = str(users.get("id") or "")
|
||
sb["alter_id"] = int(users.get("alterId") or 0)
|
||
sb["security"] = str(users.get("security") or "auto")
|
||
|
||
elif protocol == "trojan":
|
||
servers = (xray_settings.get("servers") or [{}])[0]
|
||
sb["server"] = str(servers.get("address") or "")
|
||
sb["server_port"] = int(servers.get("port") or 0)
|
||
sb["password"] = str(servers.get("password") or "")
|
||
|
||
elif protocol == "shadowsocks":
|
||
servers = (xray_settings.get("servers") or [{}])[0]
|
||
sb["server"] = str(servers.get("address") or "")
|
||
sb["server_port"] = int(servers.get("port") or 0)
|
||
sb["method"] = str(servers.get("method") or "")
|
||
sb["password"] = str(servers.get("password") or "")
|
||
|
||
elif protocol in ("socks", "http"):
|
||
servers = (xray_settings.get("servers") or [{}])[0]
|
||
sb["server"] = str(servers.get("address") or "")
|
||
sb["server_port"] = int(servers.get("port") or 0)
|
||
user_list = servers.get("users") or []
|
||
if user_list:
|
||
sb["username"] = str(user_list[0].get("user") or "")
|
||
sb["password"] = str(user_list[0].get("pass") or "")
|
||
|
||
_apply_tls(sb, stream)
|
||
_apply_transport(sb, stream)
|
||
return sb
|
||
|
||
|
||
def _apply_tls(sb: dict[str, Any], stream: dict[str, Any]) -> None:
|
||
security = str(stream.get("security") or "").lower()
|
||
if security not in ("tls", "reality"):
|
||
return
|
||
|
||
tls: dict[str, Any] = {"enabled": True}
|
||
# Без fp в ссылке — рукопожатие Chrome, а не ClientHello библиотеки Go.
|
||
default_fingerprint = "" if tls_over_quic(stream) else BROWSER_TLS_FINGERPRINT
|
||
|
||
if security == "reality":
|
||
reality_settings = dict(stream.get("realitySettings") or {})
|
||
tls["server_name"] = str(reality_settings.get("serverName") or "")
|
||
fingerprint = str(reality_settings.get("fingerprint") or "") or default_fingerprint
|
||
if fingerprint:
|
||
tls["utls"] = {"enabled": True, "fingerprint": fingerprint}
|
||
public_key = str(reality_settings.get("publicKey") or "")
|
||
short_id = str(reality_settings.get("shortId") or "")
|
||
tls["reality"] = {"enabled": True, "public_key": public_key, "short_id": short_id}
|
||
else:
|
||
tls_settings = dict(stream.get("tlsSettings") or {})
|
||
server_name = str(tls_settings.get("serverName") or "")
|
||
if server_name:
|
||
tls["server_name"] = server_name
|
||
alpn = tls_settings.get("alpn")
|
||
if alpn:
|
||
tls["alpn"] = list(alpn)
|
||
fingerprint = str(tls_settings.get("fingerprint") or "") or default_fingerprint
|
||
if fingerprint:
|
||
tls["utls"] = {"enabled": True, "fingerprint": fingerprint}
|
||
# sing-box не умеет закреплять SHA-256 всего сертификата (его
|
||
# certificate_public_key_sha256 это хэш SPKI, другое значение), поэтому
|
||
# пин Xray здесь не проверяется. Без insecure самоподписанный сертификат
|
||
# не пройдёт, а ссылка с pcs рассчитана ровно на него; об этом
|
||
# предупреждает unenforced_tls_pin_warning().
|
||
if tls_settings.get("allowInsecure", False) or str(tls_settings.get("pinnedPeerCertSha256") or "").strip():
|
||
tls["insecure"] = True
|
||
|
||
sb["tls"] = tls
|
||
|
||
|
||
def _split_early_data(path: str) -> tuple[str, int]:
|
||
"""``/ws?ed=2048`` → (``/ws``, 2048); остальные параметры пути остаются."""
|
||
|
||
base, mark, query = path.partition("?")
|
||
if not mark:
|
||
return path, 0
|
||
kept: list[str] = []
|
||
early = 0
|
||
for part in query.split("&"):
|
||
key, _, value = part.partition("=")
|
||
if key == "ed" and value.isdigit():
|
||
early = int(value)
|
||
elif part:
|
||
kept.append(part)
|
||
return (f"{base}?{'&'.join(kept)}" if kept else base), early
|
||
|
||
|
||
def _http1_only(sb: dict[str, Any]) -> None:
|
||
"""WebSocket и HTTP upgrade работают только поверх HTTP/1.1.
|
||
|
||
Ссылка с ``alpn=h2,http/1.1`` давала серверу выбрать h2: TLS
|
||
устанавливался, а запрос WebSocket — нет, и трафик не шёл (Xray в этом
|
||
случае сам оставляет только http/1.1).
|
||
"""
|
||
|
||
tls = sb.get("tls")
|
||
if not isinstance(tls, dict) or not tls.get("alpn"):
|
||
return
|
||
alpn = [value for value in tls["alpn"] if str(value).lower() not in ("h2", "h3")]
|
||
if alpn:
|
||
tls["alpn"] = alpn
|
||
else:
|
||
tls.pop("alpn")
|
||
|
||
|
||
def _apply_transport(sb: dict[str, Any], stream: dict[str, Any]) -> None:
|
||
network = str(stream.get("network") or "tcp").lower()
|
||
if network == "tcp":
|
||
return
|
||
|
||
if network == "ws":
|
||
ws_settings = dict(stream.get("wsSettings") or {})
|
||
transport: dict[str, Any] = {"type": "ws"}
|
||
path, early_data = _split_early_data(str(ws_settings.get("path") or ""))
|
||
if path:
|
||
transport["path"] = path
|
||
if early_data:
|
||
# Xray читает «?ed=2048» из пути как ранние данные в заголовке
|
||
# Sec-WebSocket-Protocol; sing-box отправил бы его в пути, и сервер
|
||
# ответил бы 404.
|
||
transport["max_early_data"] = early_data
|
||
transport["early_data_header_name"] = "Sec-WebSocket-Protocol"
|
||
headers = dict(ws_settings.get("headers") or {})
|
||
host = str(ws_settings.get("host") or "")
|
||
if host and not any(str(key).lower() == "host" for key in headers):
|
||
headers["Host"] = host
|
||
if headers:
|
||
transport["headers"] = headers
|
||
sb["transport"] = transport
|
||
_http1_only(sb)
|
||
return
|
||
|
||
if network == "httpupgrade":
|
||
upgrade_settings = dict(stream.get("httpupgradeSettings") or {})
|
||
transport = {"type": "httpupgrade"}
|
||
host = str(upgrade_settings.get("host") or "")
|
||
if host:
|
||
transport["host"] = host
|
||
path = str(upgrade_settings.get("path") or "")
|
||
if path:
|
||
transport["path"] = path
|
||
sb["transport"] = transport
|
||
_http1_only(sb)
|
||
return
|
||
|
||
if network in ("http", "h2"):
|
||
http_settings = dict(stream.get("httpSettings") or stream.get("h2Settings") or {})
|
||
transport = {"type": "http"}
|
||
host = http_settings.get("host")
|
||
if host:
|
||
transport["host"] = list(host) if isinstance(host, list) else [str(host)]
|
||
path = str(http_settings.get("path") or "")
|
||
if path:
|
||
transport["path"] = path
|
||
sb["transport"] = transport
|
||
return
|
||
|
||
if network == "grpc":
|
||
grpc_settings = dict(stream.get("grpcSettings") or {})
|
||
transport = {"type": "grpc"}
|
||
service_name = str(grpc_settings.get("serviceName") or "")
|
||
if service_name:
|
||
transport["service_name"] = service_name
|
||
sb["transport"] = transport
|
||
return
|
||
|
||
if network == "xhttp":
|
||
sb["_unsupported_transport"] = "xhttp"
|