120 lines
4.1 KiB
Python
120 lines
4.1 KiB
Python
from __future__ import annotations
|
|
|
|
import subprocess
|
|
import socket
|
|
import os
|
|
import shutil
|
|
import sys
|
|
from pathlib import Path
|
|
|
|
from pyxray.libs.xray_assets import xray_executable_name
|
|
from pyxray.libs.xray_config import XrayConfigSettings, write_transparent_rule_files
|
|
from pyxray.libs.xray_runtime import XrayServiceManager
|
|
from pyxray.libs.xray_transparent_runtime import TransparentRuntime
|
|
|
|
|
|
def test_transparent_runtime_auto_backend_prefers_iptables_and_starts_local_ip_watcher(tmp_path: Path) -> None:
|
|
settings = XrayConfigSettings()
|
|
settings.transparent.mode = "proxy"
|
|
settings.transparent.type = "redirect"
|
|
write_transparent_rule_files(settings, tmp_path)
|
|
commands: list[list[str]] = []
|
|
|
|
runtime = TransparentRuntime(
|
|
transparent_dir=tmp_path,
|
|
log_path=tmp_path / "xray.log",
|
|
backend="auto",
|
|
executor=_executor(commands),
|
|
local_cidrs_provider=lambda: ["198.51.100.10/32"],
|
|
watcher_interval=60,
|
|
)
|
|
|
|
runtime.setup(settings)
|
|
runtime.cleanup()
|
|
|
|
assert runtime.backend == "iptables"
|
|
assert ["/bin/sh", str(tmp_path / "transparent-iptables-setup.sh")] in commands
|
|
assert ["iptables", "-w", "2", "-t", "nat", "-I", "TP_RULE", "1", "-d", "198.51.100.10/32", "-j", "RETURN"] in commands
|
|
assert ["iptables", "-w", "2", "-t", "nat", "-D", "TP_RULE", "-d", "198.51.100.10/32", "-j", "RETURN"] in commands
|
|
|
|
|
|
def test_transparent_runtime_auto_backend_falls_back_to_nft_when_iptables_setup_fails(tmp_path: Path) -> None:
|
|
settings = XrayConfigSettings()
|
|
settings.transparent.mode = "proxy"
|
|
settings.transparent.type = "redirect"
|
|
write_transparent_rule_files(settings, tmp_path)
|
|
commands: list[list[str]] = []
|
|
|
|
runtime = TransparentRuntime(
|
|
transparent_dir=tmp_path,
|
|
log_path=tmp_path / "xray.log",
|
|
backend="auto",
|
|
executor=_executor(commands, failures={"transparent-iptables-setup.sh"}),
|
|
local_cidrs_provider=lambda: [],
|
|
)
|
|
|
|
runtime.setup(settings)
|
|
runtime.cleanup()
|
|
|
|
assert runtime.backend == "nft"
|
|
assert ["/bin/sh", str(tmp_path / "transparent-iptables-setup.sh")] in commands
|
|
assert ["/bin/sh", str(tmp_path / "transparent-nft-setup.sh")] in commands
|
|
|
|
|
|
def test_xray_service_manager_reports_inbound_port_conflict(tmp_path: Path) -> None:
|
|
_write_fake_xray(tmp_path)
|
|
listener = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
|
|
listener.bind(("127.0.0.1", 0))
|
|
listener.listen(1)
|
|
port = listener.getsockname()[1]
|
|
config = tmp_path / "config.json"
|
|
config.write_text(
|
|
f'{{"inbounds":[{{"tag":"conflict-http","protocol":"http","listen":"127.0.0.1","port":{port}}}]}}',
|
|
encoding="utf-8",
|
|
)
|
|
manager = XrayServiceManager(xray_dir=tmp_path, config_path=config, log_path=tmp_path / "xray.log")
|
|
|
|
try:
|
|
try:
|
|
manager.start()
|
|
except RuntimeError as exc:
|
|
message = str(exc)
|
|
else:
|
|
raise AssertionError("expected port conflict")
|
|
finally:
|
|
listener.close()
|
|
|
|
assert "inbound port conflict" in message
|
|
assert f"conflict-http 127.0.0.1:{port}/tcp" in message
|
|
assert "inbound port conflict" in (tmp_path / "xray.log").read_text(encoding="utf-8")
|
|
|
|
|
|
def _executor(commands: list[list[str]], failures: set[str] | None = None):
|
|
failures = failures or set()
|
|
|
|
def execute(command: list[str]) -> subprocess.CompletedProcess[str]:
|
|
commands.append(command)
|
|
script = Path(command[-1]).name
|
|
return subprocess.CompletedProcess(
|
|
args=command,
|
|
returncode=1 if script in failures else 0,
|
|
stdout="",
|
|
stderr=f"{script} failed" if script in failures else "",
|
|
)
|
|
|
|
return execute
|
|
|
|
|
|
def _write_fake_xray(directory: Path) -> Path:
|
|
xray = directory / xray_executable_name()
|
|
if os.name == "nt":
|
|
try:
|
|
os.link(sys.executable, xray)
|
|
except OSError:
|
|
shutil.copy2(sys.executable, xray)
|
|
(directory / "run").write_text("import time\ntime.sleep(30)\n", encoding="utf-8")
|
|
return xray
|
|
xray.write_text("#!/bin/sh\nsleep 30\n", encoding="utf-8")
|
|
xray.chmod(0o755)
|
|
return xray
|