chuan
213 lines
7.6 KiB
Python
213 lines
7.6 KiB
Python
from __future__ import annotations
|
|
|
|
import base64
|
|
import json
|
|
|
|
from pyxray.libs.nodes import import_node_links, parse_node_link
|
|
|
|
|
|
def test_parse_vless_reality_link_normalizes_node() -> None:
|
|
node = parse_node_link(
|
|
"vless://00000000-0000-4000-8000-000000000001@example.com:443"
|
|
"?type=tcp&security=reality&encryption=none&flow=xtls-rprx-vision"
|
|
"&sni=www.example.com&fp=chrome&pbk=public-key&sid=abcd&spx=%2F#hk"
|
|
)
|
|
|
|
assert node.protocol == "vless"
|
|
assert node.name == "hk"
|
|
assert node.server == "example.com"
|
|
assert node.port == 443
|
|
assert node.settings["uuid"] == "00000000-0000-4000-8000-000000000001"
|
|
assert node.settings["flow"] == "xtls-rprx-vision"
|
|
assert node.transport == {
|
|
"network": "tcp",
|
|
"header_type": "none",
|
|
"host": "",
|
|
"path": "",
|
|
}
|
|
assert node.security["type"] == "reality"
|
|
assert node.security["server_name"] == "www.example.com"
|
|
assert node.security["public_key"] == "public-key"
|
|
assert node.security["short_id"] == "abcd"
|
|
assert node.fingerprint
|
|
assert node.id.startswith("vless_")
|
|
assert node.canonical_link.startswith("vless://")
|
|
|
|
|
|
def test_parse_trojan_ws_tls_link_normalizes_node() -> None:
|
|
node = parse_node_link(
|
|
"trojan://secret@example.org:443"
|
|
"?type=ws&host=cdn.example.org&path=%2Fws&sni=edge.example.org&allowInsecure=1#trojan"
|
|
)
|
|
|
|
assert node.protocol == "trojan"
|
|
assert node.name == "trojan"
|
|
assert node.server == "example.org"
|
|
assert node.port == 443
|
|
assert node.settings["password"] == "secret"
|
|
assert node.transport["network"] == "ws"
|
|
assert node.transport["path"] == "/ws"
|
|
assert node.transport["host"] == "cdn.example.org"
|
|
assert node.security["type"] == "tls"
|
|
assert node.security["server_name"] == "edge.example.org"
|
|
assert node.security["allow_insecure"] is True
|
|
|
|
|
|
def test_parse_vmess_base64_json_link_normalizes_node() -> None:
|
|
payload = {
|
|
"v": "2",
|
|
"ps": "vmess-node",
|
|
"add": "vmess.example.net",
|
|
"port": "443",
|
|
"id": "00000000-0000-4000-8000-000000000002",
|
|
"aid": "0",
|
|
"scy": "auto",
|
|
"net": "websocket",
|
|
"type": "none",
|
|
"host": "cdn.example.net",
|
|
"path": "/ray",
|
|
"tls": "tls",
|
|
"sni": "sni.example.net",
|
|
}
|
|
encoded = base64.urlsafe_b64encode(json.dumps(payload).encode()).decode().rstrip("=")
|
|
|
|
node = parse_node_link(f"vmess://{encoded}")
|
|
|
|
assert node.protocol == "vmess"
|
|
assert node.name == "vmess-node"
|
|
assert node.server == "vmess.example.net"
|
|
assert node.port == 443
|
|
assert node.settings["uuid"] == "00000000-0000-4000-8000-000000000002"
|
|
assert node.transport["network"] == "ws"
|
|
assert node.security["type"] == "tls"
|
|
|
|
|
|
def test_parse_shadowsocks_sip002_link_normalizes_node() -> None:
|
|
user = base64.urlsafe_b64encode(b"chacha20-ietf-poly1305:secret").decode().rstrip("=")
|
|
|
|
node = parse_node_link(f"ss://{user}@ss.example.net:8388#ss-node")
|
|
|
|
assert node.protocol == "shadowsocks"
|
|
assert node.name == "ss-node"
|
|
assert node.server == "ss.example.net"
|
|
assert node.port == 8388
|
|
assert node.settings["method"] == "chacha20-ietf-poly1305"
|
|
assert node.settings["password"] == "secret"
|
|
assert node.security["type"] == "none"
|
|
|
|
|
|
def test_import_node_links_parses_multiline_input_without_subscription_groups() -> None:
|
|
user = base64.urlsafe_b64encode(b"aes-128-gcm:secret").decode().rstrip("=")
|
|
results = import_node_links(
|
|
"\n"
|
|
"# comment\n"
|
|
"bad://example\n"
|
|
f"ss://{user}@ss.example.net:8388#ss-node\n"
|
|
)
|
|
|
|
assert len(results) == 2
|
|
assert results[0].ok is False
|
|
assert results[0].error == "unsupported node link scheme: bad"
|
|
assert results[1].ok is True
|
|
assert results[1].node is not None
|
|
assert results[1].node.protocol == "shadowsocks"
|
|
|
|
|
|
def test_parse_trojan_go_keeps_protocol_and_specific_fields() -> None:
|
|
node = parse_node_link(
|
|
"trojan-go://secret@example.org:443"
|
|
"?type=ws&host=cdn.example.org&path=%2Fgo&sni=edge.example.org"
|
|
"&encryption=ss%3Baes-128-gcm%3Asecret&allowInsecure=1#tg"
|
|
)
|
|
|
|
assert node.protocol == "trojan-go"
|
|
assert node.settings["password"] == "secret"
|
|
assert node.settings["encryption"] == "ss;aes-128-gcm:secret"
|
|
assert node.transport["network"] == "ws"
|
|
assert node.transport["host"] == "cdn.example.org"
|
|
assert node.security["allow_insecure"] is False
|
|
assert node.canonical_link.startswith("trojan-go://")
|
|
|
|
|
|
def test_parse_vless_xhttp_ws_and_grpc_extension_fields() -> None:
|
|
xhttp = parse_node_link(
|
|
"vless://00000000-0000-4000-8000-000000000001@example.com:443"
|
|
"?type=xhttp&security=tls&sni=www.example.com#xhttp"
|
|
)
|
|
ws = parse_node_link(
|
|
"vless://00000000-0000-4000-8000-000000000001@example.com:443"
|
|
"?type=ws&security=tls&maxEarlyData=2048&earlyDataHeaderName=Sec-WebSocket-Protocol#ws"
|
|
)
|
|
grpc = parse_node_link(
|
|
"vless://00000000-0000-4000-8000-000000000001@example.com:443"
|
|
"?type=grpc&security=tls&serviceName=svc&multiMode=true&idleTimeout=60"
|
|
"&healthCheckTimeout=20&permitWithoutStream=true&initialWindowsSize=65535#grpc"
|
|
)
|
|
|
|
assert xhttp.transport["network"] == "xhttp"
|
|
assert xhttp.transport["xhttp_mode"] == "auto"
|
|
assert ws.transport["max_early_data"] == "2048"
|
|
assert ws.transport["early_data_header_name"] == "Sec-WebSocket-Protocol"
|
|
assert grpc.transport["service_name"] == "svc"
|
|
assert grpc.transport["multi_mode"] == "true"
|
|
assert grpc.transport["idle_timeout"] == "60"
|
|
assert grpc.transport["health_check_timeout"] == "20"
|
|
assert grpc.transport["permit_without_stream"] == "true"
|
|
assert grpc.transport["initial_windows_size"] == "65535"
|
|
|
|
|
|
def test_parse_vmess_legacy_link() -> None:
|
|
payload = base64.urlsafe_b64encode(b"auto:00000000-0000-4000-8000-000000000003@legacy.example:443").decode()
|
|
|
|
node = parse_node_link(
|
|
f"vmess://{payload}?remarks=legacy&obfs=websocket&obfsParam=cdn.example"
|
|
"&path=%2Fray&alterId=0&tls=1&sni=sni.example"
|
|
)
|
|
|
|
assert node.protocol == "vmess"
|
|
assert node.name == "legacy"
|
|
assert node.server == "legacy.example"
|
|
assert node.port == 443
|
|
assert node.settings["uuid"] == "00000000-0000-4000-8000-000000000003"
|
|
assert node.transport["network"] == "ws"
|
|
assert node.transport["host"] == "cdn.example"
|
|
assert node.transport["path"] == "/ray"
|
|
assert node.security["type"] == "tls"
|
|
assert node.security["server_name"] == "sni.example"
|
|
|
|
|
|
def test_parse_vmess_moves_path_from_host_when_needed() -> None:
|
|
payload = {
|
|
"ps": "host-path",
|
|
"add": "vmess.example.net",
|
|
"port": "443",
|
|
"id": "00000000-0000-4000-8000-000000000004",
|
|
"host": "/ray",
|
|
"net": "ws",
|
|
}
|
|
encoded = base64.urlsafe_b64encode(json.dumps(payload).encode()).decode().rstrip("=")
|
|
|
|
node = parse_node_link(f"vmess://{encoded}")
|
|
|
|
assert node.transport["host"] == ""
|
|
assert node.transport["path"] == "/ray"
|
|
|
|
|
|
def test_parse_shadowsocks_plugin_as_structured_sip003_options() -> None:
|
|
user = base64.urlsafe_b64encode(b"chacha20-ietf-poly1305:secret").decode().rstrip("=")
|
|
|
|
node = parse_node_link(
|
|
f"ss://{user}@ss.example.net:8388?plugin=simpleobfs%3Bobfs%3Dhttp%3Bobfs-host%3Dcdn.example"
|
|
"%3Bobfs-uri%3Dws#ss-plugin"
|
|
)
|
|
|
|
assert node.settings["plugin_options"] == {
|
|
"name": "simple-obfs",
|
|
"raw": "simpleobfs;obfs=http;obfs-host=cdn.example;obfs-uri=ws",
|
|
"tls": "",
|
|
"obfs": "http",
|
|
"host": "cdn.example",
|
|
"path": "/ws",
|
|
"impl": "",
|
|
}
|