891 lines
37 KiB
Python
891 lines
37 KiB
Python
from __future__ import annotations
|
|
|
|
import base64
|
|
import json
|
|
import os
|
|
import shutil
|
|
import subprocess
|
|
import sys
|
|
import time
|
|
from pathlib import Path
|
|
|
|
from pyxray.libs.xray_assets import XrayAssets, xray_executable_name
|
|
from pyxray.web.server import create_app
|
|
|
|
|
|
def test_index_shows_asset_status(tmp_path: Path) -> None:
|
|
(tmp_path / xray_executable_name()).write_bytes(b"bin")
|
|
|
|
app = create_app(tmp_path)
|
|
client = app.test_client()
|
|
|
|
response = client.get("/")
|
|
|
|
assert response.status_code == 200
|
|
body = response.get_data(as_text=True)
|
|
assert "xray" in body
|
|
assert "geoip.dat" in body
|
|
assert "geosite.dat" in body
|
|
assert "缺失" in body
|
|
assert "操作结果" not in body
|
|
assert "检查文件" not in body
|
|
assert "当前进度" in body
|
|
assert "节点管理" in body
|
|
assert "配置生成" in body
|
|
assert "保存下载设置" not in body
|
|
assert "清除日志" in body
|
|
assert "正在检查并补齐 Xray 资源，请等待后端返回结果" not in body
|
|
|
|
|
|
def test_ensure_api_uses_form_values(monkeypatch, tmp_path: Path) -> None: # noqa: ANN001
|
|
captured = {}
|
|
|
|
def fake_ensure_xray_assets(directory, **options): # noqa: ANN001
|
|
captured["directory"] = directory
|
|
captured["options"] = options
|
|
Path(directory).mkdir(parents=True, exist_ok=True)
|
|
xray = Path(directory) / xray_executable_name()
|
|
xray.write_bytes(b"bin")
|
|
(Path(directory) / "geoip.dat").write_bytes(b"geoip")
|
|
(Path(directory) / "geosite.dat").write_bytes(b"geosite")
|
|
return XrayAssets(
|
|
directory=Path(directory),
|
|
xray=xray,
|
|
geoip=Path(directory) / "geoip.dat",
|
|
geosite=Path(directory) / "geosite.dat",
|
|
downloaded=("archive",),
|
|
)
|
|
|
|
monkeypatch.setattr("pyxray.web.xray_assets.ensure_xray_assets", fake_ensure_xray_assets)
|
|
app = create_app(tmp_path, run_jobs_sync=True)
|
|
client = app.test_client()
|
|
|
|
response = client.post(
|
|
"/api/xray/assets/ensure",
|
|
data={
|
|
"directory": str(tmp_path),
|
|
"version": "v1.2.3",
|
|
"archive_url": "https://mirror.invalid/xray.zip",
|
|
"geoip_url": "",
|
|
"geosite_url": "https://mirror.invalid/geosite.dat",
|
|
"proxy_url": "http://proxy.example.invalid:8080",
|
|
"target": "geosite",
|
|
"force": "on",
|
|
},
|
|
)
|
|
|
|
assert response.status_code == 200
|
|
job_id = response.get_json()["job_id"]
|
|
job = client.get(f"/api/xray/assets/jobs/{job_id}").get_json()
|
|
assert captured["directory"] == str(tmp_path)
|
|
assert captured["options"]["version"] == "v1.2.3"
|
|
assert captured["options"]["archive_url"] == "https://mirror.invalid/xray.zip"
|
|
assert captured["options"]["geoip_url"] is None
|
|
assert captured["options"]["geosite_url"] == "https://mirror.invalid/geosite.dat"
|
|
assert captured["options"]["proxy_url"] == "http://proxy.example.invalid:8080"
|
|
assert captured["options"]["target"] == "geosite"
|
|
assert captured["options"]["force"] is True
|
|
assert callable(captured["options"]["downloader"])
|
|
assert job["state"] == "done"
|
|
assert any(step["name"] == "检查本地文件" for step in job["steps"])
|
|
assert any(step["name"] == "解压 / 写入文件" and "archive" in step["detail"] for step in job["steps"])
|
|
assert job["status"]["ready"] is True
|
|
assert (tmp_path / "download.toml").exists()
|
|
assert 'version = "v1.2.3"' in (tmp_path / "download.toml").read_text(encoding="utf-8")
|
|
assert 'proxy_url = "http://proxy.example.invalid:8080"' in (tmp_path / "download.toml").read_text(encoding="utf-8")
|
|
|
|
|
|
def test_asset_settings_api_persists_download_form_values(tmp_path: Path) -> None:
|
|
app = create_app(tmp_path)
|
|
client = app.test_client()
|
|
|
|
saved = client.post(
|
|
"/api/xray/assets/settings",
|
|
data={
|
|
"directory": str(tmp_path / "custom-xray"),
|
|
"version": "v9.9.9",
|
|
"archive_url": "https://mirror.invalid/xray.zip",
|
|
"geoip_url": "https://mirror.invalid/geoip.dat",
|
|
"geosite_url": "https://mirror.invalid/geosite.dat",
|
|
"proxy_url": "http://127.0.0.1:1080",
|
|
"target": "geoip",
|
|
"force": "on",
|
|
},
|
|
)
|
|
loaded = client.get("/api/xray/assets/settings")
|
|
index = client.get("/")
|
|
|
|
assert saved.status_code == 200
|
|
assert loaded.get_json()["directory"] == str(tmp_path / "custom-xray")
|
|
assert loaded.get_json()["version"] == "v9.9.9"
|
|
assert loaded.get_json()["force"] is True
|
|
body = index.get_data(as_text=True)
|
|
assert str(tmp_path / "custom-xray") in body
|
|
assert "v9.9.9" in body
|
|
|
|
|
|
def test_asset_settings_default_version_uses_latest_release(monkeypatch, tmp_path: Path) -> None: # noqa: ANN001
|
|
monkeypatch.setattr("pyxray.libs.xray_assets._DEFAULT_VERSION_CACHE", None)
|
|
monkeypatch.setattr("pyxray.libs.xray_assets.latest_xray_version", lambda *, timeout: "v99.9.9")
|
|
app = create_app(tmp_path)
|
|
client = app.test_client()
|
|
|
|
payload = client.get("/api/xray/assets/settings").get_json()
|
|
|
|
assert payload["version"] == "v99.9.9"
|
|
|
|
|
|
def test_job_records_real_download_percent(monkeypatch, tmp_path: Path) -> None: # noqa: ANN001
|
|
def fake_download_bytes_stream(url, progress, **options): # noqa: ANN001
|
|
progress(url, 0, 10)
|
|
progress(url, 5, 10)
|
|
progress(url, 10, 10)
|
|
return b"zip"
|
|
|
|
def fake_ensure_xray_assets(directory, *, downloader, **options): # noqa: ANN001, ARG001
|
|
downloader("https://mirror.invalid/xray.zip")
|
|
Path(directory).mkdir(parents=True, exist_ok=True)
|
|
xray = Path(directory) / xray_executable_name()
|
|
xray.write_bytes(b"bin")
|
|
(Path(directory) / "geoip.dat").write_bytes(b"geoip")
|
|
(Path(directory) / "geosite.dat").write_bytes(b"geosite")
|
|
return XrayAssets(
|
|
directory=Path(directory),
|
|
xray=xray,
|
|
geoip=Path(directory) / "geoip.dat",
|
|
geosite=Path(directory) / "geosite.dat",
|
|
downloaded=("archive",),
|
|
)
|
|
|
|
monkeypatch.setattr("pyxray.web.xray_assets.download_bytes_stream", fake_download_bytes_stream)
|
|
monkeypatch.setattr("pyxray.web.xray_assets.ensure_xray_assets", fake_ensure_xray_assets)
|
|
app = create_app(tmp_path, run_jobs_sync=True)
|
|
client = app.test_client()
|
|
|
|
response = client.post("/api/xray/assets/ensure", data={"directory": str(tmp_path), "version": "v1.2.3", "target": "all"})
|
|
job = client.get(f"/api/xray/assets/jobs/{response.get_json()['job_id']}").get_json()
|
|
download_step = next(step for step in job["steps"] if step["name"] == "下载资源")
|
|
|
|
assert download_step["percent"] == 100
|
|
assert download_step["received"] == 3
|
|
assert download_step["total"] == 3
|
|
|
|
|
|
def test_cancel_job_api_marks_job_cancel_requested(tmp_path: Path) -> None:
|
|
app = create_app(tmp_path)
|
|
store = app.extensions["pyxray_jobs"]
|
|
job = store.start(lambda item: item.update({"state": "running"}))
|
|
client = app.test_client()
|
|
|
|
response = client.post(f"/api/xray/assets/jobs/{job['id']}/cancel")
|
|
payload = response.get_json()
|
|
|
|
assert response.status_code == 200
|
|
assert payload["cancel_requested"] is True
|
|
assert store.get(job["id"])["state"] == "cancelled"
|
|
assert store.get(job["id"])["cancel_requested"] is True
|
|
assert any(step["name"] == "任务已停止" for step in store.get(job["id"])["steps"])
|
|
|
|
|
|
def test_xray_service_api_starts_stops_and_reads_logs(tmp_path: Path) -> None:
|
|
_write_fake_xray(tmp_path, stdout="xray-started")
|
|
app = create_app(tmp_path)
|
|
client = app.test_client()
|
|
client.post("/api/nodes/import", data={"links": _ss_link("secret", "ss-node")})
|
|
node = client.get("/api/nodes").get_json()["nodes"][0]
|
|
client.post("/api/nodes/select", data={"node_id": node["id"]})
|
|
|
|
started = client.post("/api/xray/service/start")
|
|
status = client.get("/api/xray/service")
|
|
stopped = client.post("/api/xray/service/stop")
|
|
logs = client.get("/api/xray/service/logs")
|
|
|
|
assert started.status_code == 200
|
|
assert started.get_json()["running"] is True
|
|
assert status.get_json()["pid"] == started.get_json()["pid"]
|
|
assert stopped.status_code == 200
|
|
assert stopped.get_json()["running"] is False
|
|
assert "pyxray start xray" in logs.get_json()["content"]
|
|
assert "xray-started" in logs.get_json()["content"]
|
|
assert json.loads((tmp_path / "config.json").read_text(encoding="utf-8"))["log"]["error"] == ""
|
|
assert json.loads((tmp_path / "service-state.json").read_text(encoding="utf-8")) == {"desired_running": False}
|
|
|
|
|
|
def test_xray_service_restores_desired_running_state_on_app_start(tmp_path: Path) -> None:
|
|
_write_fake_xray(tmp_path, stdout="restored-start")
|
|
app = create_app(tmp_path)
|
|
client = app.test_client()
|
|
client.post("/api/nodes/import", data={"links": _ss_link("secret", "ss-node")})
|
|
node = client.get("/api/nodes").get_json()["nodes"][0]
|
|
client.post("/api/nodes/select", data={"node_id": node["id"]})
|
|
client.post(
|
|
"/api/xray/config/settings",
|
|
data={"settings_toml": '[inbounds]\nsocks_port = 0\nhttp_port = 0\nrule_http_port = 0\n'},
|
|
)
|
|
|
|
started = client.post("/api/xray/service/start")
|
|
app.extensions["pyxray_xray_service"].shutdown()
|
|
|
|
restored_app = create_app(tmp_path)
|
|
restored_client = restored_app.test_client()
|
|
restored_status = restored_client.get("/api/xray/service")
|
|
restored_logs = restored_client.get("/api/xray/service/logs").get_json()["content"]
|
|
restored_app.extensions["pyxray_xray_service"].shutdown()
|
|
|
|
assert started.status_code == 200
|
|
assert json.loads((tmp_path / "service-state.json").read_text(encoding="utf-8")) == {"desired_running": True}
|
|
assert restored_status.get_json()["running"] is True
|
|
assert "pyxray restored desired running state" in restored_logs
|
|
|
|
|
|
def test_xray_service_log_forwarder_flushes_line_output_quickly(tmp_path: Path) -> None:
|
|
_write_fake_xray(tmp_path, stdout="first-line")
|
|
app = create_app(tmp_path)
|
|
client = app.test_client()
|
|
client.post("/api/nodes/import", data={"links": _ss_link("secret", "ss-node")})
|
|
node = client.get("/api/nodes").get_json()["nodes"][0]
|
|
client.post("/api/nodes/select", data={"node_id": node["id"]})
|
|
|
|
started = client.post("/api/xray/service/start")
|
|
time.sleep(0.1)
|
|
logs = client.get("/api/xray/service/logs").get_json()["content"]
|
|
client.post("/api/xray/service/stop")
|
|
|
|
assert started.status_code == 200
|
|
assert "first-line" in logs
|
|
|
|
|
|
def test_xray_service_start_regenerates_config_from_saved_settings(tmp_path: Path) -> None:
|
|
_write_fake_xray(tmp_path)
|
|
app = create_app(tmp_path)
|
|
client = app.test_client()
|
|
client.post("/api/nodes/import", data={"links": _ss_link("secret", "ss-node")})
|
|
node = client.get("/api/nodes").get_json()["nodes"][0]
|
|
client.post("/api/nodes/select", data={"node_id": node["id"]})
|
|
client.post("/api/xray/config/settings", data={"settings_toml": "[core]\nlog_level = \"debug\"\n"})
|
|
(tmp_path / "config.json").write_text('{"old": true}', encoding="utf-8")
|
|
|
|
started = client.post("/api/xray/service/start")
|
|
client.post("/api/xray/service/stop")
|
|
config = json.loads((tmp_path / "config.json").read_text(encoding="utf-8"))
|
|
|
|
assert started.status_code == 200
|
|
assert config["log"]["loglevel"] == "debug"
|
|
assert config["outbounds"][0]["protocol"] == "shadowsocks"
|
|
|
|
|
|
def test_xray_service_applies_transparent_rules_on_start_and_cleans_on_stop(tmp_path: Path) -> None:
|
|
_write_fake_xray(tmp_path)
|
|
app = create_app(tmp_path)
|
|
commands: list[str] = []
|
|
app.extensions["pyxray_transparent_runtime"].executor = _recording_executor(commands)
|
|
app.extensions["pyxray_transparent_runtime"].local_cidrs_provider = lambda: []
|
|
client = app.test_client()
|
|
client.post("/api/nodes/import", data={"links": _ss_link("secret", "ss-node")})
|
|
node = client.get("/api/nodes").get_json()["nodes"][0]
|
|
client.post("/api/nodes/select", data={"node_id": node["id"]})
|
|
client.post(
|
|
"/api/xray/config/settings",
|
|
data={
|
|
"transparent.mode": "proxy",
|
|
"transparent.type": "redirect",
|
|
"transparent.port": "52345",
|
|
"transparent.ipforward": "on",
|
|
"dns.local_dns_listen": "on",
|
|
},
|
|
)
|
|
|
|
started = client.post("/api/xray/service/start")
|
|
stopped = client.post("/api/xray/service/stop")
|
|
|
|
assert started.status_code == 200
|
|
assert stopped.status_code == 200
|
|
assert commands == [
|
|
"resolv-hijack-cleanup.sh",
|
|
"transparent-iptables-cleanup.sh",
|
|
"ip-forward-apply.sh",
|
|
"transparent-iptables-setup.sh",
|
|
"resolv-hijack-setup.sh",
|
|
"resolv-hijack-cleanup.sh",
|
|
"transparent-iptables-cleanup.sh",
|
|
]
|
|
|
|
|
|
def test_xray_service_rolls_back_when_transparent_setup_fails(tmp_path: Path) -> None:
|
|
_write_fake_xray(tmp_path)
|
|
app = create_app(tmp_path)
|
|
commands: list[str] = []
|
|
app.extensions["pyxray_transparent_runtime"].executor = _recording_executor(
|
|
commands,
|
|
failures={"transparent-iptables-setup.sh", "transparent-nft-setup.sh"},
|
|
)
|
|
app.extensions["pyxray_transparent_runtime"].local_cidrs_provider = lambda: []
|
|
client = app.test_client()
|
|
client.post("/api/nodes/import", data={"links": _ss_link("secret", "ss-node")})
|
|
node = client.get("/api/nodes").get_json()["nodes"][0]
|
|
client.post("/api/nodes/select", data={"node_id": node["id"]})
|
|
client.post(
|
|
"/api/xray/config/settings",
|
|
data={
|
|
"transparent.mode": "proxy",
|
|
"transparent.type": "redirect",
|
|
"transparent.port": "52345",
|
|
"transparent.ipforward": "on",
|
|
"dns.local_dns_listen": "on",
|
|
},
|
|
)
|
|
|
|
response = client.post("/api/xray/service/start")
|
|
|
|
assert response.status_code == 400
|
|
assert response.get_json()["status"]["running"] is False
|
|
assert "transparent-iptables-setup.sh" in commands
|
|
assert "transparent-nft-setup.sh" in commands
|
|
assert commands.count("transparent-iptables-cleanup.sh") >= 2
|
|
assert commands.count("transparent-nft-cleanup.sh") >= 1
|
|
assert commands[-2:] == ["resolv-hijack-cleanup.sh", "transparent-nft-cleanup.sh"]
|
|
|
|
|
|
def test_xray_service_shutdown_stops_managed_process(tmp_path: Path) -> None:
|
|
_write_fake_xray(tmp_path)
|
|
app = create_app(tmp_path)
|
|
client = app.test_client()
|
|
client.post("/api/nodes/import", data={"links": _ss_link("secret", "ss-node")})
|
|
node = client.get("/api/nodes").get_json()["nodes"][0]
|
|
client.post("/api/nodes/select", data={"node_id": node["id"]})
|
|
started = client.post("/api/xray/service/start").get_json()
|
|
service = app.extensions["pyxray_xray_service"]
|
|
|
|
service.shutdown()
|
|
|
|
assert started["running"] is True
|
|
assert service.status()["running"] is False
|
|
assert "pyxray shutdown xray" in client.get("/api/xray/service/logs").get_json()["content"]
|
|
|
|
|
|
def test_xray_service_uses_absolute_paths_when_app_created_with_relative_xray_dir(tmp_path: Path, monkeypatch) -> None: # noqa: ANN001
|
|
monkeypatch.chdir(tmp_path)
|
|
xray_dir = tmp_path / "data" / "xray"
|
|
xray_dir.mkdir(parents=True)
|
|
xray = _write_fake_xray(xray_dir, stdout="relative-started")
|
|
app = create_app("data/xray")
|
|
client = app.test_client()
|
|
client.post("/api/nodes/import", data={"links": _ss_link("secret", "ss-node")})
|
|
node = client.get("/api/nodes").get_json()["nodes"][0]
|
|
client.post("/api/nodes/select", data={"node_id": node["id"]})
|
|
|
|
started = client.post("/api/xray/service/start")
|
|
client.post("/api/xray/service/stop")
|
|
logs = client.get("/api/xray/service/logs").get_json()["content"]
|
|
|
|
assert started.status_code == 200
|
|
assert started.get_json()["running"] is True
|
|
assert str(xray) in logs
|
|
|
|
|
|
def test_xray_service_api_reports_missing_config(tmp_path: Path) -> None:
|
|
_write_fake_xray(tmp_path)
|
|
app = create_app(tmp_path)
|
|
client = app.test_client()
|
|
|
|
response = client.post("/api/xray/service/start")
|
|
|
|
assert response.status_code == 400
|
|
assert "未选择节点" in response.get_json()["error"]
|
|
|
|
|
|
def test_xray_service_api_records_immediate_start_failure_output(tmp_path: Path) -> None:
|
|
_write_fake_xray(tmp_path, stderr="bind: permission denied", exit_code=23, sleep_seconds=0)
|
|
app = create_app(tmp_path)
|
|
client = app.test_client()
|
|
client.post("/api/nodes/import", data={"links": _ss_link("secret", "ss-node")})
|
|
node = client.get("/api/nodes").get_json()["nodes"][0]
|
|
client.post("/api/nodes/select", data={"node_id": node["id"]})
|
|
|
|
response = client.post("/api/xray/service/start")
|
|
logs = client.get("/api/xray/service/logs")
|
|
|
|
assert response.status_code == 400
|
|
assert "Xray exited immediately with code 23" in response.get_json()["error"]
|
|
assert "bind: permission denied" in response.get_json()["error"]
|
|
assert "bind: permission denied" in logs.get_json()["content"]
|
|
|
|
|
|
def test_xray_service_api_clears_logs(tmp_path: Path) -> None:
|
|
(tmp_path / "xray.log").write_text("old log", encoding="utf-8")
|
|
app = create_app(tmp_path)
|
|
client = app.test_client()
|
|
|
|
response = client.delete("/api/xray/service/logs")
|
|
logs = client.get("/api/xray/service/logs")
|
|
|
|
assert response.status_code == 200
|
|
assert response.get_json()["content"] == ""
|
|
assert response.get_json()["offset"] == 0
|
|
assert logs.get_json()["content"] == ""
|
|
assert client.get(f"/api/xray/service/logs?offset={response.get_json()['offset']}").get_json()["content"] == ""
|
|
assert (tmp_path / "xray.log").read_text(encoding="utf-8") == ""
|
|
|
|
|
|
def test_xray_service_logs_api_reads_from_offset(tmp_path: Path) -> None:
|
|
log = tmp_path / "xray.log"
|
|
log.write_text("old log\n", encoding="utf-8")
|
|
app = create_app(tmp_path)
|
|
client = app.test_client()
|
|
offset = client.get("/api/xray/service/logs?offset=end").get_json()["offset"]
|
|
with log.open("a", encoding="utf-8") as file:
|
|
file.write("new log\n")
|
|
|
|
payload = client.get(f"/api/xray/service/logs?offset={offset}").get_json()
|
|
|
|
assert payload["content"] == "new log"
|
|
assert payload["offset"] == log.stat().st_size
|
|
|
|
|
|
def test_xray_service_logs_api_returns_compact_route_lines(tmp_path: Path) -> None:
|
|
log = tmp_path / "xray.log"
|
|
log.write_text(
|
|
"\n".join(
|
|
[
|
|
"2026/05/27 04:17:06.829641 [Info] [3636958196] app/dispatcher: sniffed domain: git.pchuan.top",
|
|
"2026/05/27 04:17:06.829662 [Info] [3636958196] app/dispatcher: taking detour [direct] for [tcp:git.pchuan.top:80]",
|
|
"2026/05/27 04:17:06.829733 from 192.168.0.76:53842 accepted tcp:117.72.47.28:80 [transparent -> direct]",
|
|
"2026/05/27 04:17:18.057882 [Info] app/proxyman/outbound: failed to process outbound traffic",
|
|
]
|
|
),
|
|
encoding="utf-8",
|
|
)
|
|
app = create_app(tmp_path)
|
|
client = app.test_client()
|
|
|
|
payload = client.get("/api/xray/service/logs?format=compact").get_json()
|
|
|
|
assert "2026/05/27 04:17:06 git.pchuan.top:80 -> direct" in payload["content"]
|
|
assert "accepted tcp" not in payload["content"]
|
|
assert "failed to process outbound traffic" in payload["content"]
|
|
|
|
|
|
def test_xray_service_logs_api_returns_latest_1000_lines(tmp_path: Path) -> None:
|
|
(tmp_path / "xray.log").write_text("\n".join(f"line-{index}" for index in range(1205)), encoding="utf-8")
|
|
app = create_app(tmp_path)
|
|
client = app.test_client()
|
|
|
|
content = client.get("/api/xray/service/logs").get_json()["content"]
|
|
|
|
assert "line-204" not in content
|
|
assert "line-205" in content
|
|
assert "line-1204" in content
|
|
assert len(content.splitlines()) == 1000
|
|
|
|
|
|
def test_xray_service_prefers_persisted_download_directory(tmp_path: Path) -> None:
|
|
default_dir = tmp_path / "default-xray"
|
|
preferred_dir = tmp_path / "download-xray"
|
|
default_dir.mkdir()
|
|
preferred_dir.mkdir()
|
|
_write_fake_xray(default_dir, stdout="default-xray")
|
|
preferred_xray = _write_fake_xray(preferred_dir, stdout="preferred-xray")
|
|
app = create_app(default_dir, default_data_dir=tmp_path)
|
|
client = app.test_client()
|
|
client.post("/api/nodes/import", data={"links": _ss_link("secret", "ss-node")})
|
|
node = client.get("/api/nodes").get_json()["nodes"][0]
|
|
client.post("/api/nodes/select", data={"node_id": node["id"]})
|
|
client.post("/api/xray/assets/settings", data={"directory": str(preferred_dir), "version": "v1.2.3"})
|
|
|
|
started = client.post("/api/xray/service/start").get_json()
|
|
client.post("/api/xray/service/stop")
|
|
logs = client.get("/api/xray/service/logs").get_json()["content"]
|
|
|
|
assert started["xray"] == str(preferred_xray)
|
|
assert started["xray_dir"] == str(preferred_dir)
|
|
assert str(preferred_xray) in logs
|
|
|
|
|
|
def test_xray_service_falls_back_to_default_directory_when_saved_directory_has_no_xray(tmp_path: Path) -> None:
|
|
default_dir = tmp_path / "default-xray"
|
|
preferred_dir = tmp_path / "download-xray"
|
|
default_dir.mkdir()
|
|
preferred_dir.mkdir()
|
|
default_xray = _write_fake_xray(default_dir, stdout="default-xray")
|
|
app = create_app(default_dir, default_data_dir=tmp_path)
|
|
client = app.test_client()
|
|
client.post("/api/nodes/import", data={"links": _ss_link("secret", "ss-node")})
|
|
node = client.get("/api/nodes").get_json()["nodes"][0]
|
|
client.post("/api/nodes/select", data={"node_id": node["id"]})
|
|
client.post("/api/xray/assets/settings", data={"directory": str(preferred_dir), "version": "v1.2.3"})
|
|
|
|
started = client.post("/api/xray/service/start").get_json()
|
|
client.post("/api/xray/service/stop")
|
|
logs = client.get("/api/xray/service/logs").get_json()["content"]
|
|
|
|
assert started["xray"] == str(default_xray)
|
|
assert started["xray_dir"] == str(default_dir)
|
|
assert started["fallback_xray_dir"] == str(default_dir)
|
|
assert str(default_xray) in logs
|
|
|
|
|
|
def test_nodes_api_imports_lists_selects_and_deletes_node(tmp_path: Path) -> None:
|
|
app = create_app(tmp_path)
|
|
client = app.test_client()
|
|
|
|
imported = client.post("/api/nodes/import", data={"links": _ss_link("secret", "ss-node")})
|
|
listed = client.get("/api/nodes")
|
|
node = listed.get_json()["nodes"][0]
|
|
selected = client.post("/api/nodes/select", data={"node_id": node["id"]})
|
|
deleted = client.delete(f"/api/nodes/{node['id']}")
|
|
|
|
assert imported.status_code == 200
|
|
assert imported.get_json()["results"][0]["ok"] is True
|
|
assert listed.status_code == 200
|
|
assert node["name"] == "ss-node"
|
|
assert selected.status_code == 200
|
|
assert selected.get_json()["node"]["id"] == node["id"]
|
|
assert deleted.status_code == 200
|
|
assert deleted.get_json()["removed"] is True
|
|
assert client.get("/api/nodes").get_json()["nodes"] == []
|
|
|
|
|
|
def test_selecting_node_restarts_running_xray(tmp_path: Path) -> None:
|
|
_write_fake_xray(tmp_path, stdout="started", echo_args=True)
|
|
app = create_app(tmp_path)
|
|
client = app.test_client()
|
|
client.post("/api/nodes/import", data={"links": "\n".join([_ss_link("one", "one"), _ss_link("two", "two")])})
|
|
nodes = client.get("/api/nodes").get_json()["nodes"]
|
|
client.post("/api/nodes/select", data={"node_id": nodes[0]["id"]})
|
|
client.post(
|
|
"/api/xray/config/settings",
|
|
data={"settings_toml": '[inbounds]\nsocks_port = 0\nhttp_port = 0\nrule_http_port = 0\n'},
|
|
)
|
|
first = client.post("/api/xray/service/start").get_json()
|
|
|
|
selected = client.post("/api/nodes/select", data={"node_id": nodes[1]["id"]})
|
|
config = json.loads((tmp_path / "config.json").read_text(encoding="utf-8"))
|
|
client.post("/api/xray/service/stop")
|
|
|
|
assert selected.status_code == 200
|
|
assert selected.get_json()["service"]["running"] is True
|
|
assert selected.get_json()["service"]["pid"] != first["pid"]
|
|
assert config["outbounds"][0]["settings"]["servers"][0]["password"] == "two"
|
|
|
|
|
|
def test_xray_config_api_saves_settings_and_generates_config(tmp_path: Path) -> None:
|
|
app = create_app(tmp_path)
|
|
client = app.test_client()
|
|
client.post("/api/nodes/import", data={"links": _ss_link("secret", "ss-node")})
|
|
node = client.get("/api/nodes").get_json()["nodes"][0]
|
|
client.post("/api/nodes/select", data={"node_id": node["id"]})
|
|
|
|
saved = client.post(
|
|
"/api/xray/config/settings",
|
|
data={"settings_toml": "[core]\nlog_level = \"debug\"\n"},
|
|
)
|
|
generated = client.post("/api/xray/config/generate")
|
|
payload = generated.get_json()
|
|
|
|
assert saved.status_code == 200
|
|
assert "log_level = \"debug\"" in saved.get_json()["settings_toml"]
|
|
assert generated.status_code == 200
|
|
assert payload["config"]["log"]["loglevel"] == "debug"
|
|
assert payload["config"]["outbounds"][0]["protocol"] == "shadowsocks"
|
|
assert (tmp_path / "config.json").exists()
|
|
assert json.loads((tmp_path / "config.json").read_text(encoding="utf-8"))["log"]["loglevel"] == "debug"
|
|
assert (tmp_path / "transparent" / "transparent-iptables-setup.sh").exists()
|
|
|
|
|
|
def test_saving_settings_restarts_running_xray(tmp_path: Path) -> None:
|
|
_write_fake_xray(tmp_path, stdout="settings-started")
|
|
app = create_app(tmp_path)
|
|
client = app.test_client()
|
|
client.post("/api/nodes/import", data={"links": _ss_link("secret", "ss-node")})
|
|
node = client.get("/api/nodes").get_json()["nodes"][0]
|
|
client.post("/api/nodes/select", data={"node_id": node["id"]})
|
|
client.post(
|
|
"/api/xray/config/settings",
|
|
data={"settings_toml": '[inbounds]\nsocks_port = 0\nhttp_port = 0\nrule_http_port = 0\n'},
|
|
)
|
|
first = client.post("/api/xray/service/start").get_json()
|
|
|
|
saved = client.post(
|
|
"/api/xray/config/settings",
|
|
data={"settings_toml": '[core]\nlog_level = "debug"\n[inbounds]\nsocks_port = 0\nhttp_port = 0\nrule_http_port = 0\n'},
|
|
)
|
|
config = json.loads((tmp_path / "config.json").read_text(encoding="utf-8"))
|
|
client.post("/api/xray/service/stop")
|
|
|
|
assert saved.status_code == 200
|
|
assert saved.get_json()["service"]["running"] is True
|
|
assert saved.get_json()["service"]["pid"] != first["pid"]
|
|
assert config["log"]["loglevel"] == "debug"
|
|
|
|
|
|
def test_xray_config_api_saves_settings_from_form_controls(tmp_path: Path) -> None:
|
|
app = create_app(tmp_path)
|
|
client = app.test_client()
|
|
client.post("/api/nodes/import", data={"links": _ss_link("secret", "ss-node")})
|
|
node = client.get("/api/nodes").get_json()["nodes"][0]
|
|
client.post("/api/nodes/select", data={"node_id": node["id"]})
|
|
|
|
saved = client.post(
|
|
"/api/xray/config/settings",
|
|
data={
|
|
"core.log_level": "error",
|
|
"core.tcp_fast_open": "default",
|
|
"core.mux_concurrency": "16",
|
|
"inbounds.listen": "127.0.0.1",
|
|
"inbounds.socks_port": "20180",
|
|
"inbounds.http_port": "0",
|
|
"inbounds.rule_socks_port": "0",
|
|
"inbounds.rule_http_port": "20181",
|
|
"inbounds.auth_user": "alice",
|
|
"inbounds.auth_password": "secret",
|
|
"inbounds.vmess_port": "0",
|
|
"inbounds.inbound_sniffing": "http,tls",
|
|
"inbounds.route_only": "on",
|
|
"inbounds.api.port": "0",
|
|
"inbounds.api.services": "LoggerService",
|
|
"routing.mode": "proxy",
|
|
"routing.default_rule": "proxy",
|
|
"transparent.mode": "close",
|
|
"transparent.type": "redirect",
|
|
"transparent.port": "52345",
|
|
"transparent.socks_port": "52306",
|
|
"transparent.ipforward": "off",
|
|
"transparent.tun_auto_route": "on",
|
|
"transparent.output_bypass_rules": "tcp 117.72.47.28:33010",
|
|
"dns.query_strategy": "UseIPv4",
|
|
"dns.local_dns_listen": "on",
|
|
"dns.antipollution": "closed",
|
|
"dns.special_mode": "fakedns",
|
|
"dns.fakedns_domains": "geosite:gfw\nkeyword:example",
|
|
"dns.rules": "localhost|geosite:private|direct\n8.8.8.8||proxy",
|
|
"outbounds.0.tag": "proxy",
|
|
"outbounds.0.probe_url": "https://www.gstatic.com/generate_204",
|
|
"outbounds.0.probe_interval": "30s",
|
|
"outbounds.0.type": "leastping",
|
|
"auto_update.gfwlist_auto_update_mode": "none",
|
|
"auto_update.gfwlist_auto_update_interval_hour": "0",
|
|
"auto_update.subscription_auto_update_mode": "none",
|
|
"auto_update.subscription_auto_update_interval_hour": "0",
|
|
"auto_update.proxy_mode_when_subscribe": "direct",
|
|
},
|
|
)
|
|
generated = client.post("/api/xray/config/generate")
|
|
config = generated.get_json()["config"]
|
|
|
|
assert saved.status_code == 200
|
|
assert "log_level = \"error\"" in saved.get_json()["settings_toml"]
|
|
assert "socks_port = 0" in saved.get_json()["settings_toml"]
|
|
assert "http_port = 0" in saved.get_json()["settings_toml"]
|
|
assert "auth_user = \"alice\"" in saved.get_json()["settings_toml"]
|
|
assert "auth_password = \"secret\"" in saved.get_json()["settings_toml"]
|
|
assert "route_only = true" in saved.get_json()["settings_toml"]
|
|
assert "output_bypass_rules = \"tcp 117.72.47.28:33010\"" in saved.get_json()["settings_toml"]
|
|
assert "special_mode = \"fakedns\"" in saved.get_json()["settings_toml"]
|
|
assert "fakedns_domains = \"geosite:gfw\\nkeyword:example\"" in saved.get_json()["settings_toml"]
|
|
assert generated.status_code == 200
|
|
assert config["log"]["loglevel"] == "error"
|
|
assert any(inbound["tag"] == "rule-mixed" and inbound["port"] == 20181 for inbound in config["inbounds"])
|
|
assert next(inbound for inbound in config["inbounds"] if inbound["tag"] == "rule-mixed")["settings"]["accounts"] == [
|
|
{"user": "alice", "pass": "secret"}
|
|
]
|
|
assert config["outbounds"][0]["mux"] == {"enabled": True, "concurrency": 16}
|
|
assert config["dns"]["queryStrategy"] == "UseIPv4"
|
|
assert config["fakedns"] == [{"ipPool": "198.18.0.0/15", "poolSize": 65535}]
|
|
assert {"address": "fakedns", "domains": ["geosite:gfw", "keyword:example"]} in config["dns"]["servers"]
|
|
|
|
|
|
def test_xray_config_api_mux_zero_disables_mux(tmp_path: Path) -> None:
|
|
app = create_app(tmp_path)
|
|
client = app.test_client()
|
|
|
|
saved = client.post(
|
|
"/api/xray/config/settings",
|
|
data={
|
|
"core.log_level": "info",
|
|
"core.tcp_fast_open": "default",
|
|
"core.mux_concurrency": "0",
|
|
"inbounds.listen": "127.0.0.1",
|
|
"inbounds.socks_port": "20170",
|
|
"inbounds.http_port": "20171",
|
|
"inbounds.rule_socks_port": "0",
|
|
"inbounds.rule_http_port": "20172",
|
|
"inbounds.vmess_port": "0",
|
|
"inbounds.inbound_sniffing": "http,tls,quic",
|
|
"inbounds.api.port": "0",
|
|
"routing.mode": "whitelist",
|
|
"routing.default_rule": "proxy",
|
|
"transparent.mode": "close",
|
|
"transparent.type": "redirect",
|
|
"transparent.port": "52345",
|
|
"transparent.socks_port": "52306",
|
|
"transparent.ipforward": "on",
|
|
"transparent.tun_auto_route": "off",
|
|
"dns.query_strategy": "",
|
|
"dns.local_dns_listen": "on",
|
|
"dns.antipollution": "closed",
|
|
"dns.special_mode": "none",
|
|
"outbounds.0.tag": "proxy",
|
|
"outbounds.0.probe_url": "https://www.gstatic.com/generate_204",
|
|
"outbounds.0.probe_interval": "60s",
|
|
"outbounds.0.type": "leastping",
|
|
"auto_update.gfwlist_auto_update_mode": "none",
|
|
"auto_update.gfwlist_auto_update_interval_hour": "0",
|
|
"auto_update.subscription_auto_update_mode": "none",
|
|
"auto_update.subscription_auto_update_interval_hour": "0",
|
|
"auto_update.proxy_mode_when_subscribe": "direct",
|
|
},
|
|
)
|
|
|
|
assert saved.status_code == 200
|
|
assert "mux_enabled = false" in saved.get_json()["settings_toml"]
|
|
assert "ipforward = true" in saved.get_json()["settings_toml"]
|
|
assert "tun_auto_route = false" in saved.get_json()["settings_toml"]
|
|
|
|
|
|
def test_xray_config_api_requires_selected_node(tmp_path: Path) -> None:
|
|
app = create_app(tmp_path)
|
|
client = app.test_client()
|
|
|
|
response = client.post("/api/xray/config/generate")
|
|
|
|
assert response.status_code == 400
|
|
assert "未选择节点" in response.get_json()["error"]
|
|
|
|
|
|
def test_xray_config_api_generates_transparent_rule_files(tmp_path: Path) -> None:
|
|
app = create_app(tmp_path)
|
|
client = app.test_client()
|
|
client.post("/api/nodes/import", data={"links": _ss_link("secret", "ss-node")})
|
|
node = client.get("/api/nodes").get_json()["nodes"][0]
|
|
client.post("/api/nodes/select", data={"node_id": node["id"]})
|
|
|
|
client.post(
|
|
"/api/xray/config/settings",
|
|
data={
|
|
"transparent.mode": "proxy",
|
|
"transparent.type": "tproxy",
|
|
"transparent.port": "52345",
|
|
"transparent.socks_port": "52306",
|
|
"transparent.ipforward": "off",
|
|
"transparent.docker_transparent": "off",
|
|
"transparent.docker_transparent_cidrs": "172.16.0.0/12;172.18.0.0/16",
|
|
"transparent.tproxy_excluded_interfaces": "docker*,veth*",
|
|
"transparent.tun_auto_route": "on",
|
|
"dns.disable_fallback": "off",
|
|
"dns.local_dns_listen": "on",
|
|
},
|
|
)
|
|
response = client.post("/api/xray/config/generate")
|
|
payload = response.get_json()
|
|
|
|
assert response.status_code == 200
|
|
assert payload["transparent_rule_paths"]["ip_forward"] == str(tmp_path / "transparent" / "ip-forward-apply.sh")
|
|
assert payload["transparent_rule_paths"]["resolv_setup"] == str(tmp_path / "transparent" / "resolv-hijack-setup.sh")
|
|
assert payload["transparent_rule_paths"]["nftables"] == str(tmp_path / "transparent" / "v2raya.nft")
|
|
assert "printf '%s' 0 > /proc/sys/net/ipv4/ip_forward" in (tmp_path / "transparent" / "ip-forward-apply.sh").read_text(encoding="utf-8")
|
|
assert "TPROXY --on-port 52345" in (tmp_path / "transparent" / "transparent-iptables-setup.sh").read_text(encoding="utf-8")
|
|
assert "ip rule add fwmark 0x40/0xc0 table 100" in (tmp_path / "transparent" / "transparent-nft-setup.sh").read_text(encoding="utf-8")
|
|
assert "table inet v2raya" in (tmp_path / "transparent" / "v2raya.nft").read_text(encoding="utf-8")
|
|
|
|
|
|
def test_xray_config_api_generates_tinytun_config_for_tun_mode(tmp_path: Path) -> None:
|
|
app = create_app(tmp_path)
|
|
client = app.test_client()
|
|
client.post("/api/nodes/import", data={"links": _ss_link("secret", "ss-node")})
|
|
node = client.get("/api/nodes").get_json()["nodes"][0]
|
|
client.post("/api/nodes/select", data={"node_id": node["id"]})
|
|
|
|
client.post(
|
|
"/api/xray/config/settings",
|
|
data={
|
|
"transparent.mode": "proxy",
|
|
"transparent.type": "tun",
|
|
"transparent.port": "52345",
|
|
"transparent.socks_port": "52306",
|
|
"transparent.ipforward": "off",
|
|
"transparent.tun_auto_route": "off",
|
|
"transparent.tun_bypass_interfaces": "172.17.0.0/16",
|
|
"transparent.tun_exclude_processes": "xray",
|
|
"dns.disable_fallback": "off",
|
|
"dns.local_dns_listen": "on",
|
|
},
|
|
)
|
|
response = client.post("/api/xray/config/generate")
|
|
payload = response.get_json()
|
|
tinytun_path = tmp_path / "transparent" / "tinytun.yaml"
|
|
|
|
assert response.status_code == 200
|
|
assert payload["transparent_rule_paths"]["tinytun"] == str(tinytun_path)
|
|
assert "ip: 198.18.0.1" in tinytun_path.read_text(encoding="utf-8")
|
|
assert "address: 127.0.0.1:52345" in tinytun_path.read_text(encoding="utf-8")
|
|
assert f"geosite_file: {tmp_path / 'geosite.dat'}" in tinytun_path.read_text(encoding="utf-8")
|
|
|
|
|
|
def _ss_link(password: str, name: str) -> str:
|
|
user = base64.urlsafe_b64encode(f"chacha20-ietf-poly1305:{password}".encode()).decode().rstrip("=")
|
|
return f"ss://{user}@ss.example.net:8388#{name}"
|
|
|
|
|
|
def _write_fake_xray(
|
|
directory: Path,
|
|
*,
|
|
stdout: str = "",
|
|
stderr: str = "",
|
|
exit_code: int = 0,
|
|
sleep_seconds: float = 30,
|
|
echo_args: bool = False,
|
|
) -> Path:
|
|
xray = directory / xray_executable_name()
|
|
code = _fake_xray_code(stdout=stdout, stderr=stderr, exit_code=exit_code, sleep_seconds=sleep_seconds, echo_args=echo_args)
|
|
if os.name == "nt":
|
|
try:
|
|
os.link(sys.executable, xray)
|
|
except OSError:
|
|
shutil.copy2(sys.executable, xray)
|
|
(directory / "run").write_text(code, encoding="utf-8")
|
|
return xray
|
|
xray.write_text(f"#!{sys.executable}\n{code}", encoding="utf-8")
|
|
xray.chmod(0o755)
|
|
return xray
|
|
|
|
|
|
def _fake_xray_code(
|
|
*,
|
|
stdout: str,
|
|
stderr: str,
|
|
exit_code: int,
|
|
sleep_seconds: float,
|
|
echo_args: bool,
|
|
) -> str:
|
|
lines = ["import sys", "import time"]
|
|
if stdout:
|
|
if echo_args:
|
|
lines.append(f"print({stdout!r} + ' ' + ' '.join(sys.argv[1:]), flush=True)")
|
|
else:
|
|
lines.append(f"print({stdout!r}, flush=True)")
|
|
if stderr:
|
|
lines.append(f"print({stderr!r}, file=sys.stderr, flush=True)")
|
|
if sleep_seconds:
|
|
lines.append(f"time.sleep({sleep_seconds!r})")
|
|
lines.append(f"raise SystemExit({exit_code})")
|
|
return "\n".join(lines) + "\n"
|
|
|
|
|
|
def _recording_executor(
|
|
commands: list[str],
|
|
failures: set[str] | None = None,
|
|
):
|
|
failures = failures or set()
|
|
|
|
def execute(command: list[str]) -> subprocess.CompletedProcess[str]:
|
|
script = Path(command[-1]).name
|
|
commands.append(script)
|
|
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
|