768 lines
32 KiB
Python
768 lines
32 KiB
Python
from __future__ import annotations
|
|
|
|
import base64
|
|
import json
|
|
import os
|
|
import subprocess
|
|
import time
|
|
from pathlib import Path
|
|
|
|
from pyxray.libs.xray_assets import XrayAssets
|
|
from pyxray.web.server import create_app
|
|
|
|
|
|
def test_index_shows_asset_status(tmp_path: Path) -> None:
|
|
(tmp_path / "xray").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)
|
|
(Path(directory) / "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=Path(directory) / "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_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)
|
|
(Path(directory) / "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=Path(directory) / "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:
|
|
xray = tmp_path / "xray"
|
|
xray.write_text("#!/bin/sh\necho xray-started\nsleep 30\n", encoding="utf-8")
|
|
os.chmod(xray, 0o755)
|
|
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:
|
|
xray = tmp_path / "xray"
|
|
xray.write_text("#!/bin/sh\necho restored-start\nsleep 30\n", encoding="utf-8")
|
|
os.chmod(xray, 0o755)
|
|
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:
|
|
xray = tmp_path / "xray"
|
|
xray.write_text("#!/bin/sh\necho first-line\nsleep 30\n", encoding="utf-8")
|
|
os.chmod(xray, 0o755)
|
|
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:
|
|
xray = tmp_path / "xray"
|
|
xray.write_text("#!/bin/sh\nsleep 30\n", encoding="utf-8")
|
|
os.chmod(xray, 0o755)
|
|
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:
|
|
xray = tmp_path / "xray"
|
|
xray.write_text("#!/bin/sh\nsleep 30\n", encoding="utf-8")
|
|
os.chmod(xray, 0o755)
|
|
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:
|
|
xray = tmp_path / "xray"
|
|
xray.write_text("#!/bin/sh\nsleep 30\n", encoding="utf-8")
|
|
os.chmod(xray, 0o755)
|
|
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:
|
|
xray = tmp_path / "xray"
|
|
xray.write_text("#!/bin/sh\nsleep 30\n", encoding="utf-8")
|
|
os.chmod(xray, 0o755)
|
|
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 = xray_dir / "xray"
|
|
xray.write_text("#!/bin/sh\necho relative-started\nsleep 30\n", encoding="utf-8")
|
|
os.chmod(xray, 0o755)
|
|
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:
|
|
xray = tmp_path / "xray"
|
|
xray.write_text("#!/bin/sh\nsleep 30\n", encoding="utf-8")
|
|
os.chmod(xray, 0o755)
|
|
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:
|
|
xray = tmp_path / "xray"
|
|
xray.write_text("#!/bin/sh\necho 'bind: permission denied' >&2\nexit 23\n", encoding="utf-8")
|
|
os.chmod(xray, 0o755)
|
|
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"] == len("old log")
|
|
assert logs.get_json()["content"] == "old log"
|
|
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") == "old log"
|
|
|
|
|
|
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_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()
|
|
(default_dir / "xray").write_text("#!/bin/sh\necho default-xray\nsleep 30\n", encoding="utf-8")
|
|
(preferred_dir / "xray").write_text("#!/bin/sh\necho preferred-xray\nsleep 30\n", encoding="utf-8")
|
|
os.chmod(default_dir / "xray", 0o755)
|
|
os.chmod(preferred_dir / "xray", 0o755)
|
|
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_dir / "xray")
|
|
assert started["xray_dir"] == str(preferred_dir)
|
|
assert str(preferred_dir / "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_dir / "xray").write_text("#!/bin/sh\necho default-xray\nsleep 30\n", encoding="utf-8")
|
|
os.chmod(default_dir / "xray", 0o755)
|
|
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_dir / "xray")
|
|
assert started["xray_dir"] == str(default_dir)
|
|
assert started["fallback_xray_dir"] == str(default_dir)
|
|
assert str(default_dir / "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_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_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.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",
|
|
"dns.query_strategy": "UseIPv4",
|
|
"dns.local_dns_listen": "on",
|
|
"dns.antipollution": "closed",
|
|
"dns.special_mode": "none",
|
|
"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 "route_only = true" in saved.get_json()["settings_toml"]
|
|
assert generated.status_code == 200
|
|
assert config["log"]["loglevel"] == "error"
|
|
assert config["inbounds"][0]["port"] == 20180
|
|
assert config["outbounds"][0]["mux"] == {"enabled": True, "concurrency": 16}
|
|
assert config["dns"]["queryStrategy"] == "UseIPv4"
|
|
|
|
|
|
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 _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
|