from __future__ import annotations import ipaddress from pyxray.libs.xray_config import XrayConfigSettings, generate_transparent_rules from pyxray.libs.xray_config.transparent_rules import write_transparent_rule_files def test_redirect_iptables_rules_match_v2raya_shape() -> None: settings = XrayConfigSettings() settings.transparent.mode = "proxy" settings.transparent.type = "redirect" settings.transparent.port = 52345 settings.transparent.docker_transparent = False rules = generate_transparent_rules(settings, backend="iptables") assert rules.mode == "redirect" assert "iptables -w 2 -t nat -N TP_OUT" in rules.setup assert "iptables -w 2 -t nat -A TP_RULE -d 10.0.0.0/8 -j RETURN" in rules.setup assert "ip -o -4 addr show" in rules.setup assert 'iptables -w 2 -t nat -A TP_RULE -d "$cidr" -j RETURN' in rules.setup assert "iptables -w 2 -t nat -A TP_RULE -i docker+ -j RETURN" in rules.setup assert "iptables -w 2 -t nat -A TP_RULE -p tcp -j REDIRECT --to-ports 52345" in rules.setup assert "iptables -w 2 -t nat -I PREROUTING -p tcp -j TP_PRE" in rules.setup assert "iptables -w 2 -t nat -D OUTPUT -p tcp -j TP_OUT" in rules.cleanup def test_tproxy_iptables_rules_match_v2raya_shape() -> None: settings = XrayConfigSettings() settings.transparent.mode = "proxy" settings.transparent.type = "tproxy" settings.transparent.port = 52345 settings.transparent.docker_transparent = False settings.transparent.tproxy_white_custom_ips = ["203.0.113.88/32"] rules = generate_transparent_rules(settings, backend="iptables") assert rules.mode == "tproxy" assert "ip rule add fwmark 0x40/0xc0 table 100" in rules.setup assert "ip route add local 0.0.0.0/0 dev lo table 100" in rules.setup assert "iptables -w 2 -t mangle -N TP_MARK" in rules.setup assert "iptables -w 2 -t mangle -A TP_RULE -p udp --dport 53 -j TP_MARK" in rules.setup assert "ip -o -4 addr show" in rules.setup assert 'iptables -w 2 -t mangle -A TP_RULE -d "$cidr" -j RETURN' in rules.setup assert "iptables -w 2 -t mangle -A TP_RULE -d 10.0.0.0/8 -j RETURN" in rules.setup assert "iptables -w 2 -t mangle -A TP_RULE -d 172.16.0.0/12 -j RETURN" in rules.setup assert "iptables -w 2 -t mangle -A TP_RULE -d 203.0.113.88/32 -j RETURN" in rules.setup assert "iptables -w 2 -t mangle -A TP_PRE -p udp -m mark --mark 0x40/0xc0 -j TPROXY --on-port 52345 --on-ip 127.0.0.1" in rules.setup assert "iptables -w 2 -t mangle -F TP_MARK" in rules.cleanup def test_cleanup_scripts_are_best_effort_like_v2raya() -> None: settings = XrayConfigSettings() settings.transparent.mode = "proxy" settings.transparent.type = "redirect" redirect = generate_transparent_rules(settings, backend="iptables") settings.transparent.type = "tproxy" tproxy = generate_transparent_rules(settings, backend="iptables") assert "iptables -w 2 -t nat -F TP_OUT 2>/dev/null || true" in redirect.cleanup assert "iptables -w 2 -t mangle -F TP_OUT 2>/dev/null || true" in tproxy.cleanup assert "ip rule del fwmark 0x40/0xc0 table 100 2>/dev/null || true" in tproxy.cleanup def test_redirect_nft_rules_include_table_and_loader_command() -> None: settings = XrayConfigSettings() settings.transparent.mode = "proxy" settings.transparent.type = "redirect" settings.transparent.docker_transparent = False rules = generate_transparent_rules(settings, backend="nft", nftables_path="/tmp/pyxray.nft") assert rules.setup == "nft -f /tmp/pyxray.nft" assert rules.cleanup == "nft delete table inet v2raya" assert "table inet v2raya" in rules.nftables assert "set whitelist" in rules.nftables assert "meta mark & 0x80 == 0x80 return" in rules.nftables assert "iifname \"docker*\" return" in rules.nftables assert "meta l4proto tcp redirect to :52345" in rules.nftables def test_tproxy_nft_rules_return_reserved_destinations() -> None: settings = XrayConfigSettings() settings.transparent.mode = "proxy" settings.transparent.type = "tproxy" rules = generate_transparent_rules(settings, backend="nft") assert "set whitelist" in rules.nftables assert "10.0.0.0/8" in rules.nftables assert "172.16.0.0/12" in rules.nftables assert "ip daddr @whitelist return" in rules.nftables def test_tproxy_country_code_whitelist_reads_geoip_dat(tmp_path) -> None: # noqa: ANN001 geoip = tmp_path / "geoip.dat" geoip.write_bytes(_geoip_dat({"cn": ["1.2.3.0/24", "2001:db8::/32"], "private": ["198.18.0.0/15"]})) settings = XrayConfigSettings() settings.transparent.mode = "proxy" settings.transparent.type = "tproxy" settings.transparent.tproxy_white_country_codes = ["cn", "private"] settings.transparent.tproxy_white_custom_ips = ["203.0.113.88/32"] iptables_rules = generate_transparent_rules(settings, backend="iptables", geoip_file=geoip) nft_rules = generate_transparent_rules(settings, backend="nft", geoip_file=geoip) assert "iptables -w 2 -t mangle -A TP_RULE -d 1.2.3.0/24 -j RETURN" in iptables_rules.setup assert "iptables -w 2 -t mangle -A TP_RULE -d 203.0.113.88/32 -j RETURN" in iptables_rules.setup assert "198.18.0.0/15" not in iptables_rules.setup assert "ip daddr 1.2.3.0/24 return" in nft_rules.nftables def test_docker_transparent_removes_docker_interface_exclusions() -> None: settings = XrayConfigSettings() settings.transparent.mode = "proxy" settings.transparent.type = "redirect" settings.transparent.docker_transparent = True settings.transparent.tproxy_excluded_interfaces = "docker*,veth*,wg*,ppp*,br-*" iptables_rules = generate_transparent_rules(settings, backend="iptables") nft_rules = generate_transparent_rules(settings, backend="nft") assert "iptables -w 2 -t nat -A TP_RULE -i docker+ -j RETURN" not in iptables_rules.setup assert "iptables -w 2 -t nat -A TP_RULE -i veth+ -j RETURN" not in iptables_rules.setup assert "iptables -w 2 -t nat -A TP_RULE -i br-+ -j RETURN" not in iptables_rules.setup assert "iptables -w 2 -t nat -A TP_RULE -i wg+ -j RETURN" in iptables_rules.setup assert "iifname \"docker*\" return" not in nft_rules.nftables assert "iifname \"veth*\" return" not in nft_rules.nftables assert "iifname \"br-*\" return" not in nft_rules.nftables assert "iifname \"wg*\" return" in nft_rules.nftables def test_docker_transparent_keeps_configured_docker_cidrs_as_destination_returns() -> None: settings = XrayConfigSettings() settings.transparent.mode = "proxy" settings.transparent.type = "redirect" settings.transparent.docker_transparent = True settings.transparent.docker_transparent_cidrs = "172.16.0.0/12" iptables_rules = generate_transparent_rules(settings, backend="iptables") nft_rules = generate_transparent_rules(settings, backend="nft") assert "iptables -w 2 -t nat -A TP_RULE -d 172.16.0.0/12 -j RETURN" in iptables_rules.setup assert "172.16.0.0/12" in nft_rules.nftables def test_docker_transparent_limits_prerouting_to_configured_source_cidrs() -> None: settings = XrayConfigSettings() settings.transparent.mode = "proxy" settings.transparent.type = "redirect" settings.transparent.docker_transparent = True settings.transparent.docker_transparent_cidrs = "172.16.0.0/12;172.30.250.0/24" iptables_rules = generate_transparent_rules(settings, backend="iptables") nft_rules = generate_transparent_rules(settings, backend="nft") assert "iptables -w 2 -t nat -A TP_PRE -s 172.16.0.0/12 -j TP_RULE" in iptables_rules.setup assert "iptables -w 2 -t nat -A TP_PRE -s 172.30.250.0/24 -j TP_RULE" in iptables_rules.setup assert "iptables -w 2 -t nat -A TP_PRE -j TP_RULE" not in iptables_rules.setup assert "ip saddr 172.16.0.0/12 meta l4proto tcp jump tp_rule" in nft_rules.nftables assert "ip saddr 172.30.250.0/24 meta l4proto tcp jump tp_rule" in nft_rules.nftables def test_docker_transparent_limits_tproxy_prerouting_to_configured_source_cidrs() -> None: settings = XrayConfigSettings() settings.transparent.mode = "proxy" settings.transparent.type = "tproxy" settings.transparent.docker_transparent = True settings.transparent.docker_transparent_cidrs = "172.16.0.0/12" iptables_rules = generate_transparent_rules(settings, backend="iptables") nft_rules = generate_transparent_rules(settings, backend="nft") assert "iptables -w 2 -t mangle -A TP_PRE -s 172.16.0.0/12 -p tcp" in iptables_rules.setup assert "iptables -w 2 -t mangle -A TP_PRE -s 172.16.0.0/12 -p udp" in iptables_rules.setup assert "iptables -w 2 -t mangle -A TP_PRE -p tcp -m addrtype" not in iptables_rules.setup assert "ip saddr 172.16.0.0/12 meta l4proto { tcp, udp }" in nft_rules.nftables def test_redirect_output_bypass_rules_return_before_transparent_rule() -> None: settings = XrayConfigSettings() settings.transparent.mode = "proxy" settings.transparent.type = "redirect" settings.transparent.output_bypass_rules = "tcp 117.72.47.28:33010\nall 192.168.0.0/24\nudp 198.51.100.10:3478" iptables_rules = generate_transparent_rules(settings, backend="iptables") nft_rules = generate_transparent_rules(settings, backend="nft") assert "iptables -w 2 -t nat -A TP_OUT -p tcp -d 117.72.47.28 --dport 33010 -j RETURN" in iptables_rules.setup assert "iptables -w 2 -t nat -A TP_OUT -p tcp -d 192.168.0.0/24 -j RETURN" in iptables_rules.setup assert "iptables -w 2 -t nat -A TP_OUT -p udp" not in iptables_rules.setup assert iptables_rules.setup.index("-d 117.72.47.28 --dport 33010") < iptables_rules.setup.index("iptables -w 2 -t nat -A TP_OUT -j TP_RULE") assert "ip daddr 117.72.47.28 meta l4proto tcp th dport 33010 return" in nft_rules.nftables assert "meta l4proto udp" not in nft_rules.nftables def test_tproxy_output_bypass_rules_support_udp_and_all() -> None: settings = XrayConfigSettings() settings.transparent.mode = "proxy" settings.transparent.type = "tproxy" settings.transparent.output_bypass_rules = "udp 198.51.100.10:3478\nall 192.168.0.0/24" iptables_rules = generate_transparent_rules(settings, backend="iptables") nft_rules = generate_transparent_rules(settings, backend="nft") assert "iptables -w 2 -t mangle -A TP_OUT -p udp -d 198.51.100.10 --dport 3478 -j RETURN" in iptables_rules.setup assert "iptables -w 2 -t mangle -A TP_OUT -p tcp -d 192.168.0.0/24 -j RETURN" in iptables_rules.setup assert "iptables -w 2 -t mangle -A TP_OUT -p udp -d 192.168.0.0/24 -j RETURN" in iptables_rules.setup assert "ip daddr 198.51.100.10 meta l4proto udp th dport 3478 return" in nft_rules.nftables def test_close_mode_has_no_system_rules() -> None: settings = XrayConfigSettings() settings.transparent.mode = "close" rules = generate_transparent_rules(settings) assert rules.setup == "" assert rules.cleanup == "" def test_write_transparent_rule_files_outputs_auditable_scripts(tmp_path) -> None: settings = XrayConfigSettings() settings.transparent.mode = "proxy" settings.transparent.type = "redirect" settings.transparent.docker_transparent = False files = write_transparent_rule_files(settings, tmp_path) assert files.iptables_setup.read_text(encoding="utf-8").startswith("#!/bin/sh\nset -eu\n") assert "printf '%s' 0 > /proc/sys/net/ipv4/ip_forward" in files.ip_forward.read_text(encoding="utf-8") assert "nameserver 127.2.0.17" in files.resolv_setup.read_text(encoding="utf-8") assert "REDIRECT --to-ports 52345" in files.iptables_setup.read_text(encoding="utf-8") assert "nft -f" in files.nft_setup.read_text(encoding="utf-8") assert files.nftables is not None assert "table inet v2raya" in files.nftables.read_text(encoding="utf-8") def test_ip_forward_script_follows_setting_like_v2raya(tmp_path) -> None: settings = XrayConfigSettings() settings.transparent.ipforward = True files = write_transparent_rule_files(settings, tmp_path) content = files.ip_forward.read_text(encoding="utf-8") assert "printf '%s' 1 > /proc/sys/net/ipv4/ip_forward" in content assert "printf '%s' 1 > /proc/sys/net/ipv6/conf/all/forwarding" in content def test_docker_transparent_forces_ip_forward_script(tmp_path) -> None: settings = XrayConfigSettings() settings.transparent.ipforward = False settings.transparent.docker_transparent = True files = write_transparent_rule_files(settings, tmp_path) content = files.ip_forward.read_text(encoding="utf-8") assert "printf '%s' 1 > /proc/sys/net/ipv4/ip_forward" in content def test_resolv_hijack_scripts_follow_v2raya_redirect_dns_behavior(tmp_path) -> None: settings = XrayConfigSettings() settings.transparent.mode = "proxy" settings.transparent.type = "redirect" files = write_transparent_rule_files(settings, tmp_path) assert "nameserver 127.2.0.17" in files.resolv_setup.read_text(encoding="utf-8") assert "nameserver 119.29.29.29" in files.resolv_setup.read_text(encoding="utf-8") assert "nameserver 223.6.6.6" in files.resolv_cleanup.read_text(encoding="utf-8") def _geoip_dat(entries: dict[str, list[str]]) -> bytes: return b"".join(_field_bytes(1, _geoip_entry(code, cidrs)) for code, cidrs in entries.items()) def _geoip_entry(country_code: str, cidrs: list[str]) -> bytes: content = _field_bytes(1, country_code.encode()) for cidr in cidrs: content += _field_bytes(2, _cidr(cidr)) return content def _cidr(cidr: str) -> bytes: network = ipaddress.ip_network(cidr) return _field_bytes(1, network.network_address.packed) + _field_varint(2, network.prefixlen) def _field_bytes(field: int, value: bytes) -> bytes: return _varint((field << 3) | 2) + _varint(len(value)) + value def _field_varint(field: int, value: int) -> bytes: return _varint(field << 3) + _varint(value) def _varint(value: int) -> bytes: out = bytearray() while value >= 0x80: out.append((value & 0x7F) | 0x80) value >>= 7 out.append(value) return bytes(out)