chuan
306 lines
14 KiB
Python
306 lines
14 KiB
Python
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)
|