feat: 提供命令行功能

README.md

@@ -37,7 +37,7 @@ flowchart LR
 | `data/settings.toml` | 保存配置页设置。 |
 | `data/download.toml` | 保存下载页设置。 |
 | `data/config.json` | 生成给 Xray 使用的配置。 |
-| `data/xray/` | 保存 `xray`、`geoip.dat`、`geosite.dat`。 |
+| `data/xray/` | 保存 `xray` / `xray.exe`、`geoip.dat`、`geosite.dat`。 |
 | `data/transparent/` | 保存透明代理脚本、nftables 配置和 `tinytun.yaml`。 |
 | `data/xray.log` | 保存 Xray 输出和 pyxray 运行日志。 |
 
@@ -82,6 +82,35 @@ http://<host-ip>:8080
 7. 点击“启动 Xray”。
 8. 在“日志”页确认 Xray 和透明代理脚本执行结果。
 
+## CLI
+
+启动 Web 控制台：
+
+```bash
+pyxray web --host 127.0.0.1 --port 3309 --xray-dir data/xray
+```
+
+查看配置文件：
+
+```bash
+pyxray configs --download
+pyxray configs --settings
+```
+
+清理配置/下载资源：
+
+```bash
+pyxray clear --download
+pyxray clear --all
+```
+
+直接下载或补齐 Xray 资源：
+
+```bash
+pyxray download --target all --directory data/xray --force
+pyxray download --target geoip --geoip-url https://example.invalid/geoip.dat
+```
+
 ## 透明代理建议
 
 | 场景 | 建议 |

docs/config/assets.md

@@ -5,7 +5,7 @@
 | 设置 | UI | 默认值 | 可选值 | 作用 | 什么时候修改 |
 | --- | --- | --- | --- | --- | --- |
 | `directory` | 下载页 | `data/xray`；Docker 中通常为 `/config/xray` | 路径 | Xray 资源保存目录。 | Docker 部署保持 `/config/xray`；本机运行可用默认值。 |
-| `version` | 下载页 | `v26.5.9` | Xray release tag | 官方 release 版本。 | 需要固定或升级 Xray 版本时修改。 |
+| `version` | 下载页 | 首次打开时优先使用最新 release；获取失败回退 `v26.5.9` | Xray release tag | 官方 release 版本。 | 需要固定或升级 Xray 版本时修改。 |
 | `archive_url` | 下载页 | `""` | URL | 自定义 Xray release zip 地址；为空时用官方地址。 | 官方下载慢或使用镜像时修改。 |
 | `geoip_url` | 下载页 | `""` | URL | 自定义 `geoip.dat` 下载地址。 | 需要替换 geoip 数据源时修改。 |
 | `geosite_url` | 下载页 | `""` | URL | 自定义 `geosite.dat` 下载地址。 | 需要替换 geosite 数据源时修改。 |
@@ -17,7 +17,7 @@
 
 | 文件 | 作用 |
 | --- | --- |
-| `xray` | Xray 可执行文件。 |
+| `xray` / `xray.exe` | Xray 可执行文件；Windows 下为 `xray.exe`，其它平台为 `xray`。 |
 | `geoip.dat` | IP 地理库，用于 `geoip:*` 规则。 |
 | `geosite.dat` | 域名分类库，用于 `geosite:*` 规则。 |
 
@@ -26,7 +26,9 @@
 | 条件 | 行为 |
 | --- | --- |
 | `target = all` | 确保三个必需文件都存在。 |
+| 首次无 `download.toml` | 尝试在 5 秒内获取 Xray-core 最新 release tag；失败则使用内置回退版本。 |
 | `force = false` 且文件存在 | 跳过已有文件。 |
 | `force = true` | 覆盖目标文件。 |
+| `archive_url` 为空 | 按当前平台选择官方 release zip；Windows x64 使用 `Xray-windows-64.zip`，Linux x64 使用 `Xray-linux-64.zip`。 |
 | `geoip_url` / `geosite_url` 非空 | 对应 dat 文件使用自定义 URL，优先于 release zip 内置版本。 |
 | Docker 部署 | 资源仍由 Web 下载页处理，不在 Dockerfile 中下载。 |

pyxray/cli.py

@@ -1,30 +1,123 @@
 from __future__ import annotations
 
 import argparse
+from pathlib import Path
 
+from pyxray.libs.app_data import (
+    CONFIG_FILES,
+    clear_all_data,
+    clear_download_data,
+    download_xray_assets,
+    read_config_file,
+    resolve_app_data_paths,
+)
+from pyxray.libs.xray_assets import ASSET_TARGETS
 from pyxray.web.server import run_web
 
 
+DEFAULT_HOST = "0.0.0.0"
+DEFAULT_PORT = 8000
+DEFAULT_XRAY_DIR = "data/xray"
+
+
 def main(argv: list[str] | None = None) -> None:
     """pyxray 命令行入口。"""
 
     parser = argparse.ArgumentParser(prog="pyxray")
     subparsers = parser.add_subparsers(dest="command")
     _add_web_parser(subparsers)
+    _add_clear_parser(subparsers)
+    _add_configs_parser(subparsers)
+    _add_download_parser(subparsers)
 
     args = parser.parse_args(argv)
-    if args.command in (None, "web"):
+    if args.command is None:
-        run_web(args.host, args.port, args.xray_dir)
+        run_web(DEFAULT_HOST, DEFAULT_PORT, DEFAULT_XRAY_DIR)
+        return
+    args.func(args)
 
 
 def _add_web_parser(subparsers: argparse._SubParsersAction) -> None:
     """注册 Web 服务启动参数。"""
 
     parser = subparsers.add_parser("web", help="启动 Web 控制台")
-    parser.add_argument("--host", default="0.0.0.0", help="监听地址，默认 0.0.0.0")
+    parser.add_argument("--host", default=DEFAULT_HOST, help=f"监听地址，默认 {DEFAULT_HOST}")
-    parser.add_argument("--port", default=8000, type=int, help="监听端口，默认 8000")
+    parser.add_argument("--port", default=DEFAULT_PORT, type=int, help=f"监听端口，默认 {DEFAULT_PORT}")
-    parser.add_argument("--xray-dir", default="data/xray", help="Xray 资源目录，默认 data/xray")
+    parser.add_argument("--xray-dir", default=DEFAULT_XRAY_DIR, help=f"Xray 资源目录，默认 {DEFAULT_XRAY_DIR}")
     parser.set_defaults(command="web")
+    parser.set_defaults(func=_run_web_command)
+
+
+def _add_clear_parser(subparsers: argparse._SubParsersAction) -> None:
+    parser = subparsers.add_parser("clear", help="清除 pyxray 配置/数据文件")
+    scope = parser.add_mutually_exclusive_group(required=True)
+    scope.add_argument("--all", action="store_true", help="清除所有 pyxray 配置/数据文件和已知生成产物")
+    scope.add_argument("--download", action="store_true", help="清除下载设置和已知 Xray 下载产物")
+    parser.add_argument("--xray-dir", default=DEFAULT_XRAY_DIR, help=f"Xray 资源目录，默认 {DEFAULT_XRAY_DIR}")
+    parser.set_defaults(command="clear", func=_run_clear_command)
+
+
+def _add_configs_parser(subparsers: argparse._SubParsersAction) -> None:
+    parser = subparsers.add_parser("configs", help="显示指定 pyxray 配置文件内容")
+    group = parser.add_mutually_exclusive_group(required=True)
+    for option in CONFIG_FILES:
+        group.add_argument(f"--{option.replace('_', '-')}", action="store_const", const=option, dest="config_name")
+    parser.add_argument("--xray-dir", default=DEFAULT_XRAY_DIR, help=f"Xray 资源目录，默认 {DEFAULT_XRAY_DIR}")
+    parser.set_defaults(command="configs", func=_run_configs_command)
+
+
+def _add_download_parser(subparsers: argparse._SubParsersAction) -> None:
+    parser = subparsers.add_parser("download", help="下载或补齐 Xray 运行资源")
+    parser.add_argument("--target", choices=ASSET_TARGETS, default=None, help="下载目标：all、xray、geoip、geosite")
+    parser.add_argument("--directory", default=None, help=f"Xray 资源目录，默认读取 download.toml 或 {DEFAULT_XRAY_DIR}")
+    parser.add_argument("--version", default=None, help="Xray-core release 版本，例如 v26.5.9")
+    parser.add_argument("--force", action="store_true", default=None, help="覆盖并重新下载已存在文件")
+    parser.add_argument("--archive-url", default=None, help="自定义 Xray release zip URL")
+    parser.add_argument("--geoip-url", default=None, help="自定义 geoip.dat URL")
+    parser.add_argument("--geosite-url", default=None, help="自定义 geosite.dat URL")
+    parser.add_argument("--proxy-url", default=None, help="下载代理 URL")
+    parser.set_defaults(command="download", func=_run_download_command)
+
+
+def _run_web_command(args: argparse.Namespace) -> None:
+    run_web(args.host, args.port, args.xray_dir)
+
+
+def _run_clear_command(args: argparse.Namespace) -> None:
+    paths = resolve_app_data_paths(args.xray_dir)
+    result = clear_all_data(paths) if args.all else clear_download_data(paths)
+    for path in result.removed:
+        print(f"removed {path}")
+    if not result.removed:
+        print("nothing removed")
+
+
+def _run_configs_command(args: argparse.Namespace) -> None:
+    paths = resolve_app_data_paths(args.xray_dir)
+    try:
+        print(read_config_file(paths, args.config_name), end="")
+    except FileNotFoundError as exc:
+        raise SystemExit(f"config file not found: {Path(exc.filename)}") from exc
+
+
+def _run_download_command(args: argparse.Namespace) -> None:
+    directory = args.directory or DEFAULT_XRAY_DIR
+    paths = resolve_app_data_paths(directory)
+    overrides = {
+        "directory": args.directory,
+        "version": args.version,
+        "archive_url": args.archive_url,
+        "geoip_url": args.geoip_url,
+        "geosite_url": args.geosite_url,
+        "proxy_url": args.proxy_url,
+        "target": args.target,
+        "force": args.force,
+    }
+    assets = download_xray_assets(paths.download_settings, default_directory=directory, overrides=overrides)
+    print(f"directory: {assets.directory}")
+    print(f"downloaded: {', '.join(assets.downloaded) if assets.downloaded else 'none'}")
+    print(f"skipped: {', '.join(assets.skipped) if assets.skipped else 'none'}")
+    print(f"ready: {assets.ready}")
 
 
 if __name__ == "__main__":

pyxray/libs/app_data.py

@@ -0,0 +1,149 @@
+from __future__ import annotations
+
+import shutil
+from dataclasses import dataclass
+from pathlib import Path
+
+from pyxray.libs.xray_asset_settings import XrayAssetSettings, XrayAssetSettingsStore
+from pyxray.libs.xray_assets import ensure_xray_assets
+
+
+CONFIG_FILES = {
+    "download": "download.toml",
+    "nodes": "nodes.toml",
+    "settings": "settings.toml",
+    "config": "config.json",
+    "service_state": "service-state.json",
+    "log": "xray.log",
+}
+GENERATED_DIRS = ("transparent",)
+DOWNLOAD_ASSET_FILES = ("xray", "xray.exe", "geoip.dat", "geosite.dat")
+
+
+@dataclass(frozen=True, slots=True)
+class AppDataPaths:
+    """Filesystem layout shared by the Web app and CLI."""
+
+    data_dir: Path
+    xray_dir: Path
+
+    @property
+    def download_settings(self) -> Path:
+        return self.data_dir / CONFIG_FILES["download"]
+
+    @property
+    def nodes(self) -> Path:
+        return self.data_dir / CONFIG_FILES["nodes"]
+
+    @property
+    def settings(self) -> Path:
+        return self.data_dir / CONFIG_FILES["settings"]
+
+    @property
+    def generated_config(self) -> Path:
+        return self.data_dir / CONFIG_FILES["config"]
+
+    @property
+    def service_state(self) -> Path:
+        return self.data_dir / CONFIG_FILES["service_state"]
+
+    @property
+    def log(self) -> Path:
+        return self.data_dir / CONFIG_FILES["log"]
+
+    @property
+    def transparent_dir(self) -> Path:
+        return self.data_dir / "transparent"
+
+    def config_path(self, name: str) -> Path:
+        try:
+            return self.data_dir / CONFIG_FILES[name]
+        except KeyError as exc:
+            raise ValueError(f"unsupported config name: {name}") from exc
+
+
+@dataclass(frozen=True, slots=True)
+class ClearResult:
+    removed: tuple[Path, ...]
+    missing: tuple[Path, ...]
+
+
+def resolve_app_data_paths(xray_dir: str | Path = "data/xray", *, data_dir: str | Path | None = None) -> AppDataPaths:
+    """Resolve the pyxray data directory from the Xray asset directory."""
+
+    resolved_xray_dir = Path(xray_dir)
+    resolved_data_dir = Path(data_dir) if data_dir is not None else default_data_dir(resolved_xray_dir)
+    return AppDataPaths(data_dir=resolved_data_dir, xray_dir=resolved_xray_dir)
+
+
+def default_data_dir(xray_dir: str | Path) -> Path:
+    """Use data/xray -> data, matching the Web app's default layout."""
+
+    path = Path(xray_dir)
+    return path.parent if path.name == "xray" else path
+
+
+def clear_download_data(paths: AppDataPaths) -> ClearResult:
+    """Remove persisted download settings and known downloaded Xray assets only."""
+
+    return _remove_known_paths([paths.download_settings, *_download_asset_paths(paths.xray_dir)])
+
+
+def clear_all_data(paths: AppDataPaths) -> ClearResult:
+    """Remove pyxray-owned config/data files and generated artifacts."""
+
+    targets = [
+        *(paths.data_dir / filename for filename in CONFIG_FILES.values()),
+        *(paths.data_dir / name for name in GENERATED_DIRS),
+        *_download_asset_paths(paths.xray_dir),
+    ]
+    return _remove_known_paths(targets)
+
+
+def read_config_file(paths: AppDataPaths, name: str) -> str:
+    path = paths.config_path(name)
+    if not path.exists():
+        raise FileNotFoundError(path)
+    return path.read_text(encoding="utf-8")
+
+
+def download_xray_assets(settings_path: str | Path, *, default_directory: str | Path, overrides: dict[str, object]):
+    """Merge CLI options with download.toml, persist them, and ensure assets exist."""
+
+    store = XrayAssetSettingsStore(settings_path, default_directory=default_directory)
+    current = store.load()
+    values = current.to_dict()
+    for key, value in overrides.items():
+        if value is not None:
+            values[key] = value
+    settings = XrayAssetSettings.from_dict(values)
+    store.save(settings)
+    return ensure_xray_assets(
+        settings.directory,
+        version=settings.version,
+        archive_url=settings.archive_url or None,
+        geoip_url=settings.geoip_url or None,
+        geosite_url=settings.geosite_url or None,
+        proxy_url=settings.proxy_url or None,
+        target=settings.target,
+        force=settings.force,
+    )
+
+
+def _download_asset_paths(xray_dir: Path) -> tuple[Path, ...]:
+    return tuple(xray_dir / name for name in DOWNLOAD_ASSET_FILES)
+
+
+def _remove_known_paths(paths: list[Path]) -> ClearResult:
+    removed: list[Path] = []
+    missing: list[Path] = []
+    for path in dict.fromkeys(paths):
+        if not path.exists() and not path.is_symlink():
+            missing.append(path)
+            continue
+        if path.is_dir() and not path.is_symlink():
+            shutil.rmtree(path)
+        else:
+            path.unlink()
+        removed.append(path)
+    return ClearResult(removed=tuple(removed), missing=tuple(missing))

pyxray/libs/xray_asset_settings.py

@@ -8,7 +8,7 @@ from typing import Any
 
 import tomlkit
 
-from pyxray.libs.xray_assets import DEFAULT_VERSION
+from pyxray.libs.xray_assets import default_xray_version
 
 
 @dataclass(slots=True)
@@ -16,7 +16,7 @@ class XrayAssetSettings:
     """Xray 资源下载页面的可持久化设置。"""
 
     directory: str = "data/xray"
-    version: str = DEFAULT_VERSION
+    version: str = ""
     archive_url: str = ""
     geoip_url: str = ""
     geosite_url: str = ""
@@ -51,7 +51,7 @@ class XrayAssetSettingsStore:
 
     def load(self) -> XrayAssetSettings:
         if not self.path.exists():
-            return XrayAssetSettings(directory=self.default_directory)
+            return XrayAssetSettings(directory=self.default_directory, version=default_xray_version())
         values = dict(tomlkit.parse(self.path.read_text(encoding="utf-8")))
         if "directory" not in values:
             values["directory"] = self.default_directory

pyxray/libs/xray_assets.py

@@ -1,7 +1,9 @@
 from __future__ import annotations
 
 import os
+import platform
 import stat
+import json
 import urllib.parse
 import urllib.request
 import zipfile
@@ -12,13 +14,17 @@ from typing import Callable
 
 
 OFFICIAL_RELEASE_BASE = "https://github.com/XTLS/Xray-core/releases/download"
+OFFICIAL_LATEST_RELEASE_API = "https://api.github.com/repos/XTLS/Xray-core/releases/latest"
 DEFAULT_VERSION = "v26.5.9"
 DEFAULT_ARCHIVE_NAME = "Xray-linux-64.zip"
-REQUIRED_FILES = ("xray", "geoip.dat", "geosite.dat")
+REQUIRED_DATA_FILES = ("geoip.dat", "geosite.dat")
 ASSET_TARGETS = ("all", "xray", "geoip", "geosite")
 
 Downloader = Callable[[str], bytes]
 DownloadProgress = Callable[[str, int, int | None], None]
+VersionFetcher = Callable[[str, float], str]
+
+_DEFAULT_VERSION_CACHE: str | None = None
 
 
 @dataclass(frozen=True, slots=True)
@@ -59,10 +65,57 @@ class XrayAssetStatus:
         return tuple(name for name, exists in self.files.items() if not exists)
 
 
-def official_archive_url(version: str = DEFAULT_VERSION, archive_name: str = DEFAULT_ARCHIVE_NAME) -> str:
+def xray_executable_name(os_name: str | None = None) -> str:
+    """返回当前平台的 Xray 可执行文件名。"""
+
+    return "xray.exe" if (os_name or os.name) == "nt" else "xray"
+
+
+def required_files(os_name: str | None = None) -> tuple[str, ...]:
+    """返回当前平台运行 Xray 所需的核心文件。"""
+
+    return (xray_executable_name(os_name), *REQUIRED_DATA_FILES)
+
+
+def default_archive_name(os_name: str | None = None, machine: str | None = None) -> str:
+    """返回当前平台默认使用的 Xray-core release zip 文件名。"""
+
+    resolved_os = os_name or os.name
+    resolved_machine = (machine or platform.machine()).lower()
+    is_arm64 = resolved_machine in {"arm64", "aarch64"}
+    if resolved_os == "nt":
+        return "Xray-windows-arm64-v8a.zip" if is_arm64 else "Xray-windows-64.zip"
+    return "Xray-linux-arm64-v8a.zip" if is_arm64 else DEFAULT_ARCHIVE_NAME
+
+
+def official_archive_url(version: str = DEFAULT_VERSION, archive_name: str | None = None) -> str:
     """返回官方 Xray-core release zip 下载地址。"""
 
-    return f"{OFFICIAL_RELEASE_BASE}/{version}/{archive_name}"
+    return f"{OFFICIAL_RELEASE_BASE}/{version}/{archive_name or default_archive_name()}"
+
+
+def latest_xray_version(*, timeout: float = 5.0, fetcher: VersionFetcher | None = None) -> str:
+    """从 GitHub release API 获取最新 Xray-core 版本。"""
+
+    payload = (fetcher or _fetch_url_text)(OFFICIAL_LATEST_RELEASE_API, timeout)
+    values = json.loads(payload)
+    tag = str(values.get("tag_name") or "").strip()
+    if not tag:
+        raise ValueError("latest Xray release response has no tag_name")
+    return tag
+
+
+def default_xray_version(*, timeout: float = 5.0) -> str:
+    """返回默认 Xray 版本；优先远程最新版本，失败时回退到内置版本。"""
+
+    global _DEFAULT_VERSION_CACHE
+    if _DEFAULT_VERSION_CACHE:
+        return _DEFAULT_VERSION_CACHE
+    try:
+        _DEFAULT_VERSION_CACHE = latest_xray_version(timeout=timeout)
+    except Exception:  # noqa: BLE001
+        _DEFAULT_VERSION_CACHE = DEFAULT_VERSION
+    return _DEFAULT_VERSION_CACHE
 
 
 def check_xray_assets(directory: str | Path) -> XrayAssetStatus:
@@ -71,7 +124,7 @@ def check_xray_assets(directory: str | Path) -> XrayAssetStatus:
     directory = Path(directory)
     return XrayAssetStatus(
         directory=directory,
-        files={name: (directory / name).exists() for name in REQUIRED_FILES},
+        files={name: (directory / name).exists() for name in required_files()},
     )
 
 
@@ -116,17 +169,18 @@ def ensure_xray_assets(
     if target not in ASSET_TARGETS:
         raise ValueError(f"unsupported xray asset target: {target}")
 
-    xray = directory / "xray"
+    xray = directory / xray_executable_name()
     geoip = directory / "geoip.dat"
     geosite = directory / "geosite.dat"
     downloaded: list[str] = []
     skipped: list[str] = []
 
     requested = _requested_files(target)
+    xray_name = xray_executable_name()
     archive_names = {
         name
         for name in requested
-        if name == "xray"
+        if name == xray_name
         or (name == "geoip.dat" and geoip_url is None)
         or (name == "geosite.dat" and geosite_url is None)
     }
@@ -174,6 +228,11 @@ def download_bytes(url: str, *, proxy_url: str | None = None) -> bytes:
         return response.read()
 
 
+def _fetch_url_text(url: str, timeout: float) -> str:
+    with urllib.request.urlopen(url, timeout=timeout) as response:  # noqa: S310
+        return response.read().decode("utf-8")
+
+
 def download_bytes_stream(
     url: str,
     progress: DownloadProgress,
@@ -250,9 +309,9 @@ def _requested_files(target: str) -> tuple[str, ...]:
     """把用户选择的下载目标转换为实际文件名。"""
 
     if target == "all":
-        return REQUIRED_FILES
+        return required_files()
     if target == "xray":
-        return ("xray",)
+        return (xray_executable_name(),)
     if target == "geoip":
         return ("geoip.dat",)
     if target == "geosite":

pyxray/web/dashboard.py

@@ -4,7 +4,7 @@ from dataclasses import asdict
 
 from flask import Blueprint, Flask, current_app, render_template
 
-from pyxray.libs.xray_assets import DEFAULT_VERSION, check_xray_assets, official_archive_url
+from pyxray.libs.xray_assets import check_xray_assets, default_xray_version, official_archive_url
 from pyxray.libs.xray_config.store import dump_settings_toml
 from pyxray.web.nodes import get_node_manager
 from pyxray.web.xray_assets import asset_form_from_settings, get_asset_settings_store
@@ -42,7 +42,7 @@ def index():  # noqa: ANN202
         status=asdict(status),
         ready=status.ready,
         missing=status.missing,
-        official_url=official_archive_url(form["version"] or DEFAULT_VERSION),
+        official_url=official_archive_url(form["version"] or default_xray_version()),
         nodes=nodes,
         selected_id=selected_id,
         selected_name=selected_node.name if selected_node is not None else "",

pyxray/web/server.py

@@ -11,6 +11,7 @@ from pathlib import Path
 from flask import Flask
 
 from pyxray import __version__
+from pyxray.libs.app_data import resolve_app_data_paths
 from pyxray.web.dashboard import register_dashboard
 from pyxray.web.jobs import init_job_store
 from pyxray.web.nodes import register_nodes
@@ -28,13 +29,13 @@ def create_app(
     """创建 pyxray Web 应用。"""
 
     app = Flask(__name__)
-    data_dir = Path(default_data_dir) if default_data_dir is not None else _default_data_dir(default_xray_dir)
+    paths = resolve_app_data_paths(default_xray_dir, data_dir=default_data_dir)
-    config_path = data_dir / "config.json"
+    config_path = paths.generated_config
     init_job_store(app, run_sync=run_jobs_sync)
-    register_xray_assets(app, default_xray_dir, data_dir / "download.toml")
+    register_xray_assets(app, paths.xray_dir, paths.download_settings)
-    register_nodes(app, data_dir / "nodes.toml")
+    register_nodes(app, paths.nodes)
-    register_xray_config(app, data_dir / "settings.toml", config_path)
+    register_xray_config(app, paths.settings, config_path)
-    register_xray_service(app, xray_dir=default_xray_dir, config_path=config_path, log_path=data_dir / "xray.log")
+    register_xray_service(app, xray_dir=paths.xray_dir, config_path=config_path, log_path=paths.log)
     _bind_xray_lifecycle(app)
     register_dashboard(app)
     return app
@@ -51,13 +52,6 @@ def run_web(host: str, port: int, default_xray_dir: str | Path = "data/xray") ->
     create_app(default_xray_dir).run(host=host, port=port)
 
 
-def _default_data_dir(default_xray_dir: str | Path) -> Path:
-    """根据资源目录推导默认数据目录。"""
-
-    path = Path(default_xray_dir)
-    return path.parent if path.name == "xray" else path
-
-
 def _print_listen_urls(host: str, port: int) -> None:
     if host in {"0.0.0.0", "::"}:
         print(f" * Pyxray URL: http://127.0.0.1:{port}", flush=True)

pyxray/web/xray_assets.py

@@ -7,6 +7,7 @@ from flask import Blueprint, Flask, current_app, jsonify, request
 from pyxray.libs.xray_assets import (
     DEFAULT_VERSION,
     check_xray_assets,
+    default_xray_version,
     download_bytes_stream,
     ensure_xray_assets,
 )
@@ -78,7 +79,7 @@ def default_asset_form(directory: str) -> dict[str, str]:
 
     return {
         "directory": directory,
-        "version": DEFAULT_VERSION,
+        "version": default_xray_version(),
         "archive_url": "",
         "geoip_url": "",
         "geosite_url": "",
@@ -105,7 +106,7 @@ def _form_values() -> dict[str, str]:
 def _settings_from_form(form: dict[str, str]) -> XrayAssetSettings:
     return XrayAssetSettings(
         directory=form["directory"],
-        version=form["version"] or DEFAULT_VERSION,
+        version=form["version"] or default_xray_version(),
         archive_url=form["archive_url"],
         geoip_url=form["geoip_url"],
         geosite_url=form["geosite_url"],

tests/libs/test_xray_assets.py

@@ -6,7 +6,17 @@ from urllib.response import addinfourl
 
 import pytest
 
-from pyxray.libs.xray_assets import DEFAULT_VERSION, download_bytes, download_bytes_stream, ensure_xray_assets, official_archive_url
+from pyxray.libs.xray_assets import (
+    DEFAULT_VERSION,
+    default_archive_name,
+    default_xray_version,
+    download_bytes,
+    download_bytes_stream,
+    ensure_xray_assets,
+    latest_xray_version,
+    official_archive_url,
+    required_files,
+)
 
 
 def _zip_bytes(files: dict[str, bytes]) -> bytes:
@@ -19,13 +29,46 @@ def _zip_bytes(files: dict[str, bytes]) -> bytes:
 
 def test_official_archive_url_defaults_to_xray_core_v26_5_9() -> None:
     assert DEFAULT_VERSION == "v26.5.9"
-    assert official_archive_url() == "https://github.com/XTLS/Xray-core/releases/download/v26.5.9/Xray-linux-64.zip"
+    assert official_archive_url(archive_name="Xray-linux-64.zip") == (
+        "https://github.com/XTLS/Xray-core/releases/download/v26.5.9/Xray-linux-64.zip"
+    )
+
+
+def test_default_archive_name_is_platform_specific() -> None:
+    assert default_archive_name(os_name="posix", machine="x86_64") == "Xray-linux-64.zip"
+    assert default_archive_name(os_name="nt", machine="AMD64") == "Xray-windows-64.zip"
+    assert default_archive_name(os_name="nt", machine="ARM64") == "Xray-windows-arm64-v8a.zip"
+
+
+def test_required_files_are_platform_specific() -> None:
+    assert required_files(os_name="posix") == ("xray", "geoip.dat", "geosite.dat")
+    assert required_files(os_name="nt") == ("xray.exe", "geoip.dat", "geosite.dat")
+
+
+def test_latest_xray_version_reads_github_release_tag() -> None:
+    def fetcher(url: str, timeout: float) -> str:
+        assert url == "https://api.github.com/repos/XTLS/Xray-core/releases/latest"
+        assert timeout == 5.0
+        return '{"tag_name": "v99.1.2"}'
+
+    assert latest_xray_version(fetcher=fetcher) == "v99.1.2"
+
+
+def test_default_xray_version_falls_back_to_pinned_version(monkeypatch) -> None:  # noqa: ANN001
+    monkeypatch.setattr("pyxray.libs.xray_assets._DEFAULT_VERSION_CACHE", None)
+    monkeypatch.setattr(
+        "pyxray.libs.xray_assets.latest_xray_version",
+        lambda *, timeout: (_ for _ in ()).throw(TimeoutError("slow")),
+    )
+
+    assert default_xray_version(timeout=5.0) == DEFAULT_VERSION
 
 
 def test_ensure_xray_assets_extracts_official_archive_files(tmp_path) -> None:  # noqa: ANN001
+    xray_name = required_files()[0]
     archive = _zip_bytes(
         {
-            "xray": b"bin",
+            xray_name: b"bin",
             "geoip.dat": b"geoip",
             "geosite.dat": b"geosite",
             "README.md": b"ignored",
@@ -42,13 +85,13 @@ def test_ensure_xray_assets_extracts_official_archive_files(tmp_path) -> None:
     assert result.ready is True
     assert result.downloaded == ("archive",)
     assert calls == [official_archive_url()]
-    assert (tmp_path / "xray").read_bytes() == b"bin"
+    assert (tmp_path / xray_name).read_bytes() == b"bin"
     assert (tmp_path / "geoip.dat").read_bytes() == b"geoip"
     assert (tmp_path / "geosite.dat").read_bytes() == b"geosite"
 
 
 def test_version_and_archive_url_can_be_overridden(tmp_path) -> None:  # noqa: ANN001
-    archive = _zip_bytes({"xray": b"bin", "geoip.dat": b"geoip", "geosite.dat": b"geosite"})
+    archive = _zip_bytes({required_files()[0]: b"bin", "geoip.dat": b"geoip", "geosite.dat": b"geosite"})
     calls = []
 
     def downloader(url: str) -> bytes:
@@ -61,7 +104,8 @@ def test_version_and_archive_url_can_be_overridden(tmp_path) -> None:  # noqa: A
 
 
 def test_dat_urls_override_archive_dat_files(tmp_path) -> None:  # noqa: ANN001
-    archive = _zip_bytes({"xray": b"bin", "geoip.dat": b"old-geoip", "geosite.dat": b"old-geosite"})
+    xray_name = required_files()[0]
+    archive = _zip_bytes({xray_name: b"bin", "geoip.dat": b"old-geoip", "geosite.dat": b"old-geosite"})
     payloads = {
         official_archive_url(): archive,
         "https://mirror.invalid/geoip.dat": b"new-geoip",
@@ -76,13 +120,14 @@ def test_dat_urls_override_archive_dat_files(tmp_path) -> None:  # noqa: ANN001
     )
 
     assert result.downloaded == ("archive", "geoip.dat", "geosite.dat")
-    assert (tmp_path / "xray").read_bytes() == b"bin"
+    assert (tmp_path / xray_name).read_bytes() == b"bin"
     assert (tmp_path / "geoip.dat").read_bytes() == b"new-geoip"
     assert (tmp_path / "geosite.dat").read_bytes() == b"new-geosite"
 
 
 def test_existing_files_skip_download(tmp_path) -> None:  # noqa: ANN001
-    (tmp_path / "xray").write_bytes(b"bin")
+    xray_name = required_files()[0]
+    (tmp_path / xray_name).write_bytes(b"bin")
     (tmp_path / "geoip.dat").write_bytes(b"geoip")
     (tmp_path / "geosite.dat").write_bytes(b"geosite")
 
@@ -90,7 +135,7 @@ def test_existing_files_skip_download(tmp_path) -> None:  # noqa: ANN001
 
     assert result.ready is True
     assert result.downloaded == ()
-    assert result.skipped == ("xray", "geoip.dat", "geosite.dat")
+    assert result.skipped == (xray_name, "geoip.dat", "geosite.dat")
 
 
 def test_force_redownloads_selected_existing_file(tmp_path) -> None:  # noqa: ANN001
@@ -116,7 +161,7 @@ def test_force_redownloads_selected_existing_file(tmp_path) -> None:  # noqa: AN
 
 
 def test_missing_required_file_raises_after_bad_archive(tmp_path) -> None:  # noqa: ANN001
-    archive = _zip_bytes({"xray": b"bin", "geoip.dat": b"geoip"})
+    archive = _zip_bytes({required_files()[0]: b"bin", "geoip.dat": b"geoip"})
 
     with pytest.raises(FileNotFoundError, match="geosite.dat"):
         ensure_xray_assets(tmp_path, downloader=lambda url: archive)

tests/libs/test_xray_runtime.py

@@ -2,8 +2,12 @@ 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
@@ -58,9 +62,7 @@ def test_transparent_runtime_auto_backend_falls_back_to_nft_when_iptables_setup_
 
 
 def test_xray_service_manager_reports_inbound_port_conflict(tmp_path: Path) -> None:
-    xray = tmp_path / "xray"
+    _write_fake_xray(tmp_path)
-    xray.write_text("#!/bin/sh\nsleep 30\n", encoding="utf-8")
-    xray.chmod(0o755)
     listener = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
     listener.bind(("127.0.0.1", 0))
     listener.listen(1)
@@ -101,3 +103,17 @@ def _executor(commands: list[list[str]], failures: set[str] | None = None):
         )
 
     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

tests/test_cli.py

@@ -0,0 +1,138 @@
+from __future__ import annotations
+
+from pathlib import Path
+
+from pyxray import cli
+from pyxray.libs.xray_assets import XrayAssets
+
+
+def test_no_subcommand_defaults_to_web(monkeypatch) -> None:  # noqa: ANN001
+    captured = {}
+
+    def fake_run_web(host: str, port: int, xray_dir: str) -> None:
+        captured.update({"host": host, "port": port, "xray_dir": xray_dir})
+
+    monkeypatch.setattr(cli, "run_web", fake_run_web)
+
+    cli.main([])
+
+    assert captured == {"host": "0.0.0.0", "port": 8000, "xray_dir": "data/xray"}
+
+
+def test_web_subcommand_still_uses_web_runner(monkeypatch) -> None:  # noqa: ANN001
+    captured = {}
+
+    def fake_run_web(host: str, port: int, xray_dir: str) -> None:
+        captured.update({"host": host, "port": port, "xray_dir": xray_dir})
+
+    monkeypatch.setattr(cli, "run_web", fake_run_web)
+
+    cli.main(["web", "--host", "127.0.0.1", "--port", "9000", "--xray-dir", "runtime/xray"])
+
+    assert captured == {"host": "127.0.0.1", "port": 9000, "xray_dir": "runtime/xray"}
+
+
+def test_configs_download_prints_download_toml(tmp_path: Path, capsys, monkeypatch) -> None:  # noqa: ANN001
+    monkeypatch.chdir(tmp_path)
+    data = tmp_path / "data"
+    data.mkdir()
+    (data / "download.toml").write_text('version = "v1.2.3"\n', encoding="utf-8")
+
+    cli.main(["configs", "--download", "--xray-dir", "data/xray"])
+
+    assert capsys.readouterr().out == 'version = "v1.2.3"\n'
+
+
+def test_clear_download_removes_only_download_settings_and_known_assets(tmp_path: Path, monkeypatch) -> None:  # noqa: ANN001
+    monkeypatch.chdir(tmp_path)
+    data = tmp_path / "data"
+    xray_dir = data / "xray"
+    xray_dir.mkdir(parents=True)
+    (data / "download.toml").write_text("download", encoding="utf-8")
+    (data / "nodes.toml").write_text("nodes", encoding="utf-8")
+    (xray_dir / "xray").write_bytes(b"xray")
+    (xray_dir / "xray.exe").write_bytes(b"xray.exe")
+    (xray_dir / "geoip.dat").write_bytes(b"geoip")
+    (xray_dir / "geosite.dat").write_bytes(b"geosite")
+    (xray_dir / "user-file.txt").write_text("keep", encoding="utf-8")
+
+    cli.main(["clear", "--download", "--xray-dir", "data/xray"])
+
+    assert not (data / "download.toml").exists()
+    assert not (xray_dir / "xray").exists()
+    assert not (xray_dir / "xray.exe").exists()
+    assert not (xray_dir / "geoip.dat").exists()
+    assert not (xray_dir / "geosite.dat").exists()
+    assert (data / "nodes.toml").exists()
+    assert (xray_dir / "user-file.txt").exists()
+
+
+def test_clear_all_removes_known_data_and_keeps_unrelated_files(tmp_path: Path, monkeypatch) -> None:  # noqa: ANN001
+    monkeypatch.chdir(tmp_path)
+    data = tmp_path / "data"
+    xray_dir = data / "xray"
+    transparent_dir = data / "transparent"
+    transparent_dir.mkdir(parents=True)
+    xray_dir.mkdir()
+    for name in ("download.toml", "nodes.toml", "settings.toml", "config.json", "service-state.json", "xray.log"):
+        (data / name).write_text(name, encoding="utf-8")
+    (transparent_dir / "transparent-iptables-setup.sh").write_text("script", encoding="utf-8")
+    (xray_dir / "xray").write_bytes(b"xray")
+    (xray_dir / "geoip.dat").write_bytes(b"geoip")
+    (data / "notes.txt").write_text("keep", encoding="utf-8")
+    (xray_dir / "custom.dat").write_text("keep", encoding="utf-8")
+
+    cli.main(["clear", "--all", "--xray-dir", "data/xray"])
+
+    for name in ("download.toml", "nodes.toml", "settings.toml", "config.json", "service-state.json", "xray.log"):
+        assert not (data / name).exists()
+    assert not transparent_dir.exists()
+    assert not (xray_dir / "xray").exists()
+    assert not (xray_dir / "geoip.dat").exists()
+    assert (data / "notes.txt").exists()
+    assert (xray_dir / "custom.dat").exists()
+
+
+def test_download_command_persists_settings_and_reuses_ensure(monkeypatch, tmp_path: Path) -> None:  # noqa: ANN001
+    monkeypatch.chdir(tmp_path)
+    captured = {}
+
+    def fake_ensure_xray_assets(directory, **options):  # noqa: ANN001
+        captured["directory"] = directory
+        captured["options"] = options
+        path = Path(directory)
+        path.mkdir(parents=True, exist_ok=True)
+        xray = path / "xray"
+        xray.write_bytes(b"xray")
+        geoip = path / "geoip.dat"
+        geoip.write_bytes(b"geoip")
+        geosite = path / "geosite.dat"
+        geosite.write_bytes(b"geosite")
+        return XrayAssets(directory=path, xray=xray, geoip=geoip, geosite=geosite, downloaded=("geoip.dat",))
+
+    monkeypatch.setattr("pyxray.libs.app_data.ensure_xray_assets", fake_ensure_xray_assets)
+
+    cli.main(
+        [
+            "download",
+            "--target",
+            "geoip",
+            "--directory",
+            "data/xray",
+            "--version",
+            "v1.2.3",
+            "--geoip-url",
+            "https://mirror.example.invalid/geoip.dat",
+            "--force",
+        ]
+    )
+
+    assert captured["directory"] == "data/xray"
+    assert captured["options"]["target"] == "geoip"
+    assert captured["options"]["version"] == "v1.2.3"
+    assert captured["options"]["geoip_url"] == "https://mirror.example.invalid/geoip.dat"
+    assert captured["options"]["force"] is True
+    content = (tmp_path / "data" / "download.toml").read_text(encoding="utf-8")
+    assert 'directory = "data/xray"' in content
+    assert 'target = "geoip"' in content
+    assert 'version = "v1.2.3"' in content

tests/web/test_xray_assets_web.py

@@ -3,16 +3,18 @@ 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
+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").write_bytes(b"bin")
+    (tmp_path / xray_executable_name()).write_bytes(b"bin")
 
     app = create_app(tmp_path)
     client = app.test_client()
@@ -42,12 +44,13 @@ def test_ensure_api_uses_form_values(monkeypatch, tmp_path: Path) -> None:  # no
         captured["directory"] = directory
         captured["options"] = options
         Path(directory).mkdir(parents=True, exist_ok=True)
-        (Path(directory) / "xray").write_bytes(b"bin")
+        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=Path(directory) / "xray",
+            xray=xray,
             geoip=Path(directory) / "geoip.dat",
             geosite=Path(directory) / "geosite.dat",
             downloaded=("archive",),
@@ -121,6 +124,17 @@ def test_asset_settings_api_persists_download_form_values(tmp_path: Path) -> Non
     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)
@@ -131,12 +145,13 @@ def test_job_records_real_download_percent(monkeypatch, tmp_path: Path) -> None:
     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")
+        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=Path(directory) / "xray",
+            xray=xray,
             geoip=Path(directory) / "geoip.dat",
             geosite=Path(directory) / "geosite.dat",
             downloaded=("archive",),
@@ -173,9 +188,7 @@ def test_cancel_job_api_marks_job_cancel_requested(tmp_path: Path) -> None:
 
 
 def test_xray_service_api_starts_stops_and_reads_logs(tmp_path: Path) -> None:
-    xray = tmp_path / "xray"
+    _write_fake_xray(tmp_path, stdout="xray-started")
-    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")})
@@ -199,9 +212,7 @@ def test_xray_service_api_starts_stops_and_reads_logs(tmp_path: Path) -> None:
 
 
 def test_xray_service_restores_desired_running_state_on_app_start(tmp_path: Path) -> None:
-    xray = tmp_path / "xray"
+    _write_fake_xray(tmp_path, stdout="restored-start")
-    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")})
@@ -228,9 +239,7 @@ def test_xray_service_restores_desired_running_state_on_app_start(tmp_path: Path
 
 
 def test_xray_service_log_forwarder_flushes_line_output_quickly(tmp_path: Path) -> None:
-    xray = tmp_path / "xray"
+    _write_fake_xray(tmp_path, stdout="first-line")
-    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")})
@@ -247,9 +256,7 @@ def test_xray_service_log_forwarder_flushes_line_output_quickly(tmp_path: Path)
 
 
 def test_xray_service_start_regenerates_config_from_saved_settings(tmp_path: Path) -> None:
-    xray = tmp_path / "xray"
+    _write_fake_xray(tmp_path)
-    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")})
@@ -268,9 +275,7 @@ def test_xray_service_start_regenerates_config_from_saved_settings(tmp_path: Pat
 
 
 def test_xray_service_applies_transparent_rules_on_start_and_cleans_on_stop(tmp_path: Path) -> None:
-    xray = tmp_path / "xray"
+    _write_fake_xray(tmp_path)
-    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)
@@ -307,9 +312,7 @@ def test_xray_service_applies_transparent_rules_on_start_and_cleans_on_stop(tmp_
 
 
 def test_xray_service_rolls_back_when_transparent_setup_fails(tmp_path: Path) -> None:
-    xray = tmp_path / "xray"
+    _write_fake_xray(tmp_path)
-    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(
@@ -344,9 +347,7 @@ def test_xray_service_rolls_back_when_transparent_setup_fails(tmp_path: Path) ->
 
 
 def test_xray_service_shutdown_stops_managed_process(tmp_path: Path) -> None:
-    xray = tmp_path / "xray"
+    _write_fake_xray(tmp_path)
-    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")})
@@ -366,9 +367,7 @@ def test_xray_service_uses_absolute_paths_when_app_created_with_relative_xray_di
     monkeypatch.chdir(tmp_path)
     xray_dir = tmp_path / "data" / "xray"
     xray_dir.mkdir(parents=True)
-    xray = xray_dir / "xray"
+    xray = _write_fake_xray(xray_dir, stdout="relative-started")
-    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")})
@@ -385,9 +384,7 @@ def test_xray_service_uses_absolute_paths_when_app_created_with_relative_xray_di
 
 
 def test_xray_service_api_reports_missing_config(tmp_path: Path) -> None:
-    xray = tmp_path / "xray"
+    _write_fake_xray(tmp_path)
-    xray.write_text("#!/bin/sh\nsleep 30\n", encoding="utf-8")
-    os.chmod(xray, 0o755)
     app = create_app(tmp_path)
     client = app.test_client()
 
@@ -398,9 +395,7 @@ def test_xray_service_api_reports_missing_config(tmp_path: Path) -> None:
 
 
 def test_xray_service_api_records_immediate_start_failure_output(tmp_path: Path) -> None:
-    xray = tmp_path / "xray"
+    _write_fake_xray(tmp_path, stderr="bind: permission denied", exit_code=23, sleep_seconds=0)
-    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")})
@@ -488,10 +483,8 @@ def test_xray_service_prefers_persisted_download_directory(tmp_path: Path) -> No
     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")
+    _write_fake_xray(default_dir, stdout="default-xray")
-    (preferred_dir / "xray").write_text("#!/bin/sh\necho preferred-xray\nsleep 30\n", encoding="utf-8")
+    preferred_xray = _write_fake_xray(preferred_dir, stdout="preferred-xray")
-    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")})
@@ -503,9 +496,9 @@ def test_xray_service_prefers_persisted_download_directory(tmp_path: Path) -> No
     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"] == str(preferred_xray)
     assert started["xray_dir"] == str(preferred_dir)
-    assert str(preferred_dir / "xray") in logs
+    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:
@@ -513,8 +506,7 @@ def test_xray_service_falls_back_to_default_directory_when_saved_directory_has_n
     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")
+    default_xray = _write_fake_xray(default_dir, stdout="default-xray")
-    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")})
@@ -526,10 +518,10 @@ def test_xray_service_falls_back_to_default_directory_when_saved_directory_has_n
     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"] == str(default_xray)
     assert started["xray_dir"] == str(default_dir)
     assert started["fallback_xray_dir"] == str(default_dir)
-    assert str(default_dir / "xray") in logs
+    assert str(default_xray) in logs
 
 
 def test_nodes_api_imports_lists_selects_and_deletes_node(tmp_path: Path) -> None:
@@ -554,9 +546,7 @@ def test_nodes_api_imports_lists_selects_and_deletes_node(tmp_path: Path) -> Non
 
 
 def test_selecting_node_restarts_running_xray(tmp_path: Path) -> None:
-    xray = tmp_path / "xray"
+    _write_fake_xray(tmp_path, stdout="started", echo_args=True)
-    xray.write_text("#!/bin/sh\necho 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": "\n".join([_ss_link("one", "one"), _ss_link("two", "two")])})
@@ -603,9 +593,7 @@ def test_xray_config_api_saves_settings_and_generates_config(tmp_path: Path) ->
 
 
 def test_saving_settings_restarts_running_xray(tmp_path: Path) -> None:
-    xray = tmp_path / "xray"
+    _write_fake_xray(tmp_path, stdout="settings-started")
-    xray.write_text("#!/bin/sh\necho settings-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")})
@@ -838,6 +826,51 @@ def _ss_link(password: str, name: str) -> str:
     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,

uv.lock

@@ -154,7 +154,7 @@ wheels = [
 
 [[package]]
 name = "pyxray"
-version = "1.0.4"
+version = "1.0.5"
 source = { editable = "." }
 dependencies = [
     { name = "flask" },