pyxray/tests/libs/test_xray_config.py

from __future__ import annotations

import base64
import json

import pytest

from pyxray.libs.nodes import parse_node_link
from pyxray.libs.xray_config import XrayConfigSettings, XrayConfigSettingsStore, generate_xray_config
from pyxray.libs.xray_config.settings import CustomInboundSettings, CustomRoutingRuleSettings, DnsRuleSettings, validate_settings


def test_settings_store_round_trips_all_sections(tmp_path) -> None:
    settings = XrayConfigSettings()
    settings.core.log_level = "debug"
    settings.core.tcp_fast_open = "yes"
    settings.core.mux_enabled = True
    settings.inbounds.port_sharing = True
    settings.inbounds.api.port = 2017
    settings.inbounds.custom.append(CustomInboundSettings(tag="lan-socks", protocol="socks", port=2080))
    settings.routing.mode = "custom"
    settings.routing.custom_rules.append(
        CustomRoutingRuleSettings(filename="geosite.dat", tags=["cn"], match_type="domain", rule_type="direct")
    )
    settings.transparent.mode = "proxy"
    settings.transparent.type = "tproxy"
    settings.transparent.tproxy_white_country_codes = ["cn"]
    settings.dns.query_strategy = "UseIP"
    settings.dns.rules.append(DnsRuleSettings(server="https://dns.google/dns-query", domains="geosite:google", outbound="proxy"))
    settings.outbounds[0].probe_interval = "30s"
    settings.auto_update.gfwlist_auto_update_mode = "auto_update"

    store = XrayConfigSettingsStore(tmp_path / "settings.toml")
    store.save(settings)
    loaded = store.load()

    assert loaded.core.log_level == "debug"
    assert loaded.core.tcp_fast_open == "yes"
    assert loaded.core.mux_enabled is True
    assert loaded.inbounds.port_sharing is True
    assert loaded.inbounds.api.port == 2017
    assert loaded.inbounds.custom[0].tag == "lan-socks"
    assert loaded.routing.custom_rules[0].filename == "geosite.dat"
    assert loaded.transparent.tproxy_white_country_codes == ["cn"]
    assert loaded.dns.rules[-1].server == "https://dns.google/dns-query"
    assert loaded.outbounds[0].probe_interval == "30s"
    assert loaded.auto_update.gfwlist_auto_update_mode == "auto_update"


def test_settings_store_loads_toml_arrays_inside_dict(tmp_path) -> None:
    store = XrayConfigSettingsStore(tmp_path / "settings.toml")
    store.path.write_text(
        """
[dns.hosts]
"courier.push.apple.com" = ["1-courier.push.apple.com"]
""".strip(),
        encoding="utf-8",
    )

    settings = store.load()

    assert settings.to_dict()["dns"]["hosts"]["courier.push.apple.com"] == ["1-courier.push.apple.com"]


def test_settings_store_migrates_legacy_log_levels(tmp_path) -> None:
    store = XrayConfigSettingsStore(tmp_path / "settings.toml")
    store.path.write_text('[core]\nlog_level = "trace"\n', encoding="utf-8")

    settings = store.load()

    assert settings.core.log_level == "debug"


def test_settings_defaults_match_v2raya_core_values() -> None:
    settings = XrayConfigSettings()

    assert settings.core.log_level == "info"
    assert settings.core.mux_concurrency == 8
    assert settings.inbounds.socks_port == 20170
    assert settings.inbounds.http_port == 20171
    assert settings.inbounds.rule_socks_port == 0
    assert settings.inbounds.rule_http_port == 20172
    assert settings.inbounds.auth_user == ""
    assert settings.inbounds.auth_password == ""
    assert settings.transparent.mode == "close"
    assert settings.transparent.type == "redirect"
    assert settings.transparent.port == 52345
    assert settings.transparent.docker_transparent is True
    assert settings.transparent.docker_transparent_cidrs == "172.16.0.0/12"
    assert settings.transparent.output_bypass_rules == ""
    assert settings.dns.query_strategy == "UseIPv4"
    assert settings.dns.special_mode == "none"
    assert settings.dns.fakedns_domains == "geosite:geolocation-!cn"
    assert settings.dns.rules == [
        DnsRuleSettings(server="localhost", domains="geosite:private", outbound="direct"),
        DnsRuleSettings(server="223.5.5.5", domains="geosite:cn", outbound="direct"),
        DnsRuleSettings(server="8.8.8.8", domains="", outbound="proxy"),
    ]


def test_generate_default_config_matches_v2raya_template_shape() -> None:
    node = parse_node_link(_ss_link())
    config = generate_xray_config(node)

    assert config["log"] == {"loglevel": "info", "access": "", "error": ""}
    assert _inbound(config, "socks")["port"] == 20170
    assert _inbound(config, "socks")["protocol"] == "socks"
    assert _inbound(config, "socks")["settings"]["udp"] is True
    assert _inbound(config, "http")["port"] == 20171
    assert _inbound(config, "rule-mixed")["port"] == 20172
    assert _inbound(config, "rule-mixed")["protocol"] == "mixed"
    assert _inbound(config, "rule-mixed")["settings"] == {"auth": "noauth", "udp": True, "allowTransparent": False}
    assert "rule-http" not in {item["tag"] for item in config["inbounds"]}
    assert "rule-socks" not in {item["tag"] for item in config["inbounds"]}
    assert _outbound(config, "proxy")["protocol"] == "shadowsocks"
    assert _outbound(config, "direct")["protocol"] == "freedom"
    assert _outbound(config, "direct")["settings"]["domainStrategy"] == "UseIP"
    assert _outbound(config, "block")["protocol"] == "blackhole"
    assert _outbound(config, "dns-out")["protocol"] == "dns"
    assert config["routing"]["domainStrategy"] == "IPOnDemand"
    assert config["routing"]["domainMatcher"] == "mph"
    assert config["dns"]["hosts"]["courier.push.apple.com"] == ["1-courier.push.apple.com"]
    assert config["dns"]["queryStrategy"] == "UseIPv4"
    assert "dns-in" not in {item["tag"] for item in config["inbounds"]}


def test_generate_mixed_inbound_with_password_auth() -> None:
    settings = XrayConfigSettings()
    settings.inbounds.auth_user = "alice"
    settings.inbounds.auth_password = "secret"

    mixed = _inbound(generate_xray_config(parse_node_link(_ss_link()), settings), "rule-mixed")

    assert mixed["protocol"] == "mixed"
    assert mixed["settings"] == {
        "auth": "password",
        "udp": True,
        "allowTransparent": False,
        "accounts": [{"user": "alice", "pass": "secret"}],
    }


def test_generate_none_log_level_disables_xray_logs() -> None:
    settings = XrayConfigSettings()
    settings.core.log_level = "none"

    config = generate_xray_config(parse_node_link(_ss_link()), settings)

    assert config["log"] == {"loglevel": "none", "access": "none", "error": "none"}


def test_generate_redirect_dns_inbound_when_transparent_redirect_enabled() -> None:
    settings = XrayConfigSettings()
    settings.transparent.mode = "proxy"
    settings.transparent.type = "redirect"

    config = generate_xray_config(parse_node_link(_ss_link()), settings)

    assert _inbound(config, "transparent")["listen"] == "0.0.0.0"
    assert _inbound(config, "dns-in")["listen"] == "127.2.0.17"
    assert {"type": "field", "inboundTag": ["dns-in"], "outboundTag": "dns-out"} in config["routing"]["rules"]
    assert _outbound(config, "proxy")["streamSettings"]["sockopt"]["mark"] == 128
    assert _outbound(config, "direct")["streamSettings"]["sockopt"]["mark"] == 128


def test_generate_redirect_transparent_inbound_stays_local_when_docker_transparent_disabled() -> None:
    settings = XrayConfigSettings()
    settings.transparent.mode = "proxy"
    settings.transparent.type = "redirect"
    settings.transparent.docker_transparent = False

    config = generate_xray_config(parse_node_link(_ss_link()), settings)

    assert _inbound(config, "transparent")["listen"] == "127.0.0.1"


def test_generate_lan_dns_inbound_when_redirect_and_port_sharing_enabled() -> None:
    settings = XrayConfigSettings()
    settings.inbounds.port_sharing = True
    settings.transparent.mode = "proxy"
    settings.transparent.type = "redirect"

    config = generate_xray_config(parse_node_link(_ss_link()), settings)

    assert _inbound(config, "dns-in")["listen"] == "0.0.0.0"
    assert _inbound(config, "dns-in-local")["listen"] == "127.2.0.17"
    assert {"type": "field", "inboundTag": ["dns-in", "dns-in-local"], "outboundTag": "dns-out"} in config["routing"]["rules"]


def test_skip_local_dns_inbound_for_tproxy_and_tun_like_v2raya() -> None:
    node = parse_node_link(_ss_link())
    settings = XrayConfigSettings()
    settings.transparent.mode = "proxy"
    settings.transparent.type = "tproxy"

    tproxy_config = generate_xray_config(node, settings)
    assert "dns-in" not in {item["tag"] for item in tproxy_config["inbounds"]}

    settings.transparent.type = "tun"
    tun_config = generate_xray_config(node, settings)
    assert "dns-in" not in {item["tag"] for item in tun_config["inbounds"]}


def test_generate_vless_reality_outbound_stream_settings() -> None:
    settings = XrayConfigSettings()
    settings.core.mux_enabled = True
    settings.core.tcp_fast_open = "yes"
    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"
    )

    outbound = _outbound(generate_xray_config(node, settings), "proxy")

    assert outbound["protocol"] == "vless"
    assert outbound["settings"]["vnext"][0]["users"][0]["flow"] == "xtls-rprx-vision"
    assert outbound["streamSettings"]["security"] == "reality"
    assert outbound["streamSettings"]["realitySettings"]["serverName"] == "www.example.com"
    assert outbound["streamSettings"]["realitySettings"]["publicKey"] == "public-key"
    assert outbound["streamSettings"]["sockopt"]["tcpFastOpen"] is True
    assert outbound["mux"] == {"enabled": True, "concurrency": 8}


def test_generate_direct_outbound_uses_shared_sockopt_settings() -> None:
    settings = XrayConfigSettings()
    settings.core.tcp_fast_open = "yes"
    settings.transparent.mode = "proxy"
    settings.transparent.type = "redirect"

    direct = _outbound(generate_xray_config(parse_node_link(_ss_link()), settings), "direct")

    assert direct["streamSettings"]["sockopt"] == {"tcpFastOpen": True, "mark": 128}


def test_generate_vmess_ws_tls_outbound() -> None:
    payload = {
        "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("=")

    outbound = _outbound(generate_xray_config(parse_node_link(f"vmess://{encoded}")), "proxy")

    assert outbound["protocol"] == "vmess"
    assert outbound["settings"]["vnext"][0]["address"] == "vmess.example.net"
    assert outbound["settings"]["vnext"][0]["users"][0]["alterId"] == 0
    assert outbound["streamSettings"]["network"] == "ws"
    assert outbound["streamSettings"]["security"] == "tls"
    assert outbound["streamSettings"]["tlsSettings"]["serverName"] == "sni.example.net"
    assert outbound["streamSettings"]["wsSettings"]["headers"]["Host"] == "cdn.example.net"
    assert outbound["streamSettings"]["wsSettings"]["path"] == "/ray"


def test_generate_trojan_grpc_and_trojan_go_outbound() -> None:
    trojan = parse_node_link(
        "trojan://secret@example.org:443?type=grpc&serviceName=svc&sni=edge.example.org#trojan"
    )
    trojan_go = parse_node_link(
        "trojan-go://secret@example.org:443?type=ws&host=cdn.example.org&path=%2Fgo&sni=edge.example.org#tg"
    )

    trojan_out = _outbound(generate_xray_config(trojan), "proxy")
    trojan_go_out = _outbound(generate_xray_config(trojan_go), "proxy")

    assert trojan_out["protocol"] == "trojan"
    assert trojan_out["settings"]["servers"][0]["password"] == "secret"
    assert trojan_out["streamSettings"]["grpcSettings"]["serviceName"] == "svc"
    assert trojan_out["streamSettings"]["tlsSettings"]["serverName"] == "edge.example.org"
    assert trojan_go_out["protocol"] == "trojan"
    assert trojan_go_out["streamSettings"]["wsSettings"]["path"] == "/go"


def test_generate_vless_ws_early_data_grpc_and_xhttp_settings() -> None:
    ws = parse_node_link(
        "vless://00000000-0000-4000-8000-000000000001@example.com:443"
        "?type=ws&security=tls&host=cdn.example&path=%2Fws&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"
    )
    xhttp = parse_node_link(
        "vless://00000000-0000-4000-8000-000000000001@example.com:443"
        "?type=xhttp&security=tls&host=cdn.example&path=%2Fxhttp#xhttp"
    )

    ws_stream = _outbound(generate_xray_config(ws), "proxy")["streamSettings"]
    grpc_stream = _outbound(generate_xray_config(grpc), "proxy")["streamSettings"]
    xhttp_stream = _outbound(generate_xray_config(xhttp), "proxy")["streamSettings"]

    assert ws_stream["wsSettings"]["maxEarlyData"] == 2048
    assert ws_stream["wsSettings"]["earlyDataHeaderName"] == "Sec-WebSocket-Protocol"
    assert grpc_stream["grpcSettings"]["multiMode"] is True
    assert grpc_stream["grpcSettings"]["idle_timeout"] == 60
    assert grpc_stream["grpcSettings"]["health_check_timeout"] == 20
    assert grpc_stream["grpcSettings"]["permit_without_stream"] is True
    assert grpc_stream["grpcSettings"]["initial_windows_size"] == 65535
    assert xhttp_stream["xhttpSettings"] == {"path": "/xhttp", "host": "cdn.example", "mode": "auto"}


def test_generate_inbounds_custom_api_and_transparent_tproxy() -> None:
    settings = XrayConfigSettings()
    settings.inbounds.port_sharing = True
    settings.inbounds.vmess_port = 20174
    settings.inbounds.api.port = 20175
    settings.inbounds.custom.append(CustomInboundSettings(tag="extra-http", protocol="http", port=20176))
    settings.transparent.mode = "proxy"
    settings.transparent.type = "tproxy"

    config = generate_xray_config(parse_node_link(_ss_link()), settings)

    assert _inbound(config, "socks")["listen"] == "0.0.0.0"
    assert _inbound(config, "rule-mixed")["listen"] == "0.0.0.0"
    assert _inbound(config, "rule-mixed")["port"] == 20172
    assert _inbound(config, "vmess")["protocol"] == "vmess"
    assert _inbound(config, "extra-http")["protocol"] == "http"
    assert _inbound(config, "transparent")["streamSettings"]["sockopt"]["tproxy"] == "tproxy"
    assert _inbound(config, "api-in")["port"] == 20175
    assert config["api"]["services"] == ["LoggerService"]
    assert {"type": "field", "inboundTag": ["transparent"], "outboundTag": "proxy"} in config["routing"]["rules"]


def test_generate_routing_modes_follow_v2raya_rules() -> None:
    settings = XrayConfigSettings()

    whitelist = generate_xray_config(parse_node_link(_ss_link()), settings)["routing"]["rules"]
    assert {"type": "field", "inboundTag": ["rule-mixed"], "domain": ["geosite:cn"], "outboundTag": "direct"} in whitelist
    assert {"type": "field", "inboundTag": ["rule-mixed"], "domain": ["geosite:google"], "outboundTag": "proxy"} in whitelist

    settings.routing.mode = "gfwlist"
    gfwlist = generate_xray_config(parse_node_link(_ss_link()), settings)["routing"]["rules"]
    assert {"type": "field", "inboundTag": ["rule-mixed"], "outboundTag": "direct"} in gfwlist
    assert any("91.108.4.0/22" in rule.get("ip", []) for rule in gfwlist)


def test_generate_custom_routing_a_applies_before_proxy_mode_fallback() -> None:
    settings = XrayConfigSettings()
    settings.routing.mode = "proxy"
    settings.routing.routing_a = "domain(keyword:google)->direct"

    rules = generate_xray_config(parse_node_link(_ss_link()), settings)["routing"]["rules"]

    google_rule = {"type": "field", "inboundTag": ["rule-mixed"], "outboundTag": "direct", "domain": ["keyword:google"]}
    fallback = {"type": "field", "inboundTag": ["rule-mixed"], "outboundTag": "proxy"}
    assert google_rule in rules
    assert fallback in rules
    assert rules.index(google_rule) < rules.index(fallback)


def test_generate_custom_text_rules_before_route_mode_and_default_proxy() -> None:
    settings = XrayConfigSettings()
    settings.routing.mode = "whitelist"
    settings.routing.default_rule = "proxy"
    settings.routing.routing_a = "domain(geosite:google, domain:example.com)->proxy\nip(geoip:cn)->direct"

    rules = generate_xray_config(parse_node_link(_ss_link()), settings)["routing"]["rules"]
    custom_domain = {
        "type": "field",
        "inboundTag": ["rule-mixed"],
        "domain": ["geosite:google", "domain:example.com"],
        "outboundTag": "proxy",
    }
    custom_ip = {
        "type": "field",
        "inboundTag": ["rule-mixed"],
        "ip": ["geoip:cn"],
        "outboundTag": "direct",
    }

    assert rules.index(custom_domain) < rules.index({"type": "field", "inboundTag": ["rule-mixed"], "domain": ["geosite:cn"], "outboundTag": "direct"})
    assert rules.index(custom_ip) < rules.index({"type": "field", "inboundTag": ["rule-mixed"], "ip": ["geoip:private", "geoip:cn"], "outboundTag": "direct"})
    assert {"type": "field", "inboundTag": ["rule-mixed"], "outboundTag": "proxy"} in rules


def test_generate_dns_rules_and_dns_routing() -> None:
    settings = XrayConfigSettings()
    settings.dns.query_strategy = "UseIP"
    settings.dns.disable_fallback = True
    settings.dns.rules = [
        DnsRuleSettings(server="localhost", domains="geosite:private", outbound="direct"),
        DnsRuleSettings(server="223.5.5.5", domains="geosite:cn", outbound="direct"),
        DnsRuleSettings(server="https://dns.google/dns-query", domains="", outbound="proxy"),
    ]

    config = generate_xray_config(parse_node_link(_ss_link()), settings)

    assert config["dns"]["queryStrategy"] == "UseIP"
    assert config["dns"]["disableFallback"] is True
    assert {"address": "localhost", "domains": ["geosite:private"]} in config["dns"]["servers"]
    assert {"address": "223.5.5.5", "domains": ["geosite:cn"]} in config["dns"]["servers"]
    assert "https://dns.google/dns-query" in config["dns"]["servers"]
    assert any(rule.get("domain") == ["dns.google"] and rule.get("outboundTag") == "proxy" for rule in config["routing"]["rules"])


def test_generate_fakedns_adds_dns_server_pool_and_sniffing() -> None:
    settings = XrayConfigSettings()
    settings.transparent.mode = "gfwlist"
    settings.transparent.type = "redirect"
    settings.dns.special_mode = "fakedns"

    config = generate_xray_config(parse_node_link(_ss_link()), settings)

    assert config["fakedns"] == [{"ipPool": "198.18.0.0/15", "poolSize": 65535}]
    assert {"address": "fakedns", "domains": ["geosite:geolocation-!cn"]} in config["dns"]["servers"]
    assert "fakedns" in _inbound(config, "transparent")["sniffing"]["destOverride"]
    assert "fakedns" in _inbound(config, "rule-mixed")["sniffing"]["destOverride"]


def test_generate_fakedns_uses_custom_domain_scope() -> None:
    settings = XrayConfigSettings()
    settings.dns.special_mode = "fakedns"
    settings.dns.fakedns_domains = "geosite:gfw\nkeyword:example"

    config = generate_xray_config(parse_node_link(_ss_link()), settings)

    assert {"address": "fakedns", "domains": ["geosite:gfw", "keyword:example"]} in config["dns"]["servers"]


def test_validate_settings_rejects_invalid_values() -> None:
    settings = XrayConfigSettings()
    settings.transparent.type = "bad"

    with pytest.raises(ValueError, match="transparent.type"):
        validate_settings(settings)


def test_validate_settings_requires_complete_inbound_auth() -> None:
    settings = XrayConfigSettings()
    settings.inbounds.auth_user = "alice"

    with pytest.raises(ValueError, match="auth_user"):
        validate_settings(settings)


def _ss_link() -> str:
    user = base64.urlsafe_b64encode(b"chacha20-ietf-poly1305:secret").decode().rstrip("=")
    return f"ss://{user}@ss.example.net:8388#ss-node"


def _inbound(config: dict, tag: str) -> dict:
    return next(item for item in config["inbounds"] if item.get("tag") == tag)


def _outbound(config: dict, tag: str) -> dict:
    return next(item for item in config["outbounds"] if item.get("tag") == tag)