178 lines
6.3 KiB
Python
178 lines
6.3 KiB
Python
#!/usr/bin/env python3
|
||
"""Expose a static file's parent directory through a Tailscale Serve HTTP route."""
|
||
|
||
from __future__ import annotations
|
||
|
||
import argparse
|
||
import ipaddress
|
||
import json
|
||
import re
|
||
import shutil
|
||
import subprocess
|
||
import sys
|
||
from pathlib import Path, PurePosixPath
|
||
from urllib.parse import quote
|
||
|
||
|
||
class PreviewError(RuntimeError):
|
||
pass
|
||
|
||
|
||
def run_tailscale(*args: str) -> subprocess.CompletedProcess[str]:
|
||
try:
|
||
return subprocess.run(
|
||
["tailscale", *args],
|
||
check=False,
|
||
capture_output=True,
|
||
text=True,
|
||
)
|
||
except OSError as exc:
|
||
raise PreviewError(f"无法运行 tailscale：{exc}") from exc
|
||
|
||
|
||
def require_tailscale() -> None:
|
||
if shutil.which("tailscale") is None:
|
||
raise PreviewError("未找到 tailscale，请先安装并登录 Tailscale")
|
||
|
||
|
||
def tailscale_ipv4() -> str:
|
||
result = run_tailscale("ip", "-4")
|
||
if result.returncode != 0:
|
||
detail = (result.stderr or result.stdout).strip()
|
||
raise PreviewError(f"Tailscale 节点不可用：{detail or 'tailscale ip -4 失败'}")
|
||
for line in result.stdout.splitlines():
|
||
candidate = line.strip()
|
||
try:
|
||
address = ipaddress.ip_address(candidate)
|
||
except ValueError:
|
||
continue
|
||
if address.version == 4:
|
||
return candidate
|
||
raise PreviewError("tailscale 没有返回 IPv4 地址；请确认节点已连接 tailnet")
|
||
|
||
|
||
def serve_status() -> tuple[dict, str]:
|
||
result = run_tailscale("serve", "status", "--json")
|
||
if result.returncode != 0:
|
||
detail = (result.stderr or result.stdout).strip()
|
||
raise PreviewError(f"无法读取 Tailscale Serve 状态：{detail or 'status 失败'}")
|
||
try:
|
||
data = json.loads(result.stdout or "{}")
|
||
except json.JSONDecodeError as exc:
|
||
raise PreviewError("Tailscale Serve 返回了无法解析的 JSON 状态") from exc
|
||
if not isinstance(data, dict):
|
||
raise PreviewError("Tailscale Serve 状态不是 JSON 对象")
|
||
|
||
plain = run_tailscale("serve", "status")
|
||
if plain.returncode != 0:
|
||
detail = (plain.stderr or plain.stdout).strip()
|
||
raise PreviewError(f"无法读取 Tailscale Serve 文本状态：{detail or 'status 失败'}")
|
||
return data, plain.stdout
|
||
|
||
|
||
def status_blob(data: dict, plain: str) -> str:
|
||
return json.dumps(data, ensure_ascii=False, sort_keys=True) + "\n" + plain
|
||
|
||
|
||
def normalize_mount_path(value: str) -> str:
|
||
path = value.strip()
|
||
if not path.startswith("/") or path == "/":
|
||
raise PreviewError("mount path 必须以 / 开头且不能是根路径")
|
||
if "?" in path or "#" in path or "//" in path:
|
||
raise PreviewError("mount path 不能包含 query、fragment 或空路径段")
|
||
parts = PurePosixPath(path).parts
|
||
if any(part in {".", ".."} for part in parts):
|
||
raise PreviewError("mount path 不能包含 . 或 ..")
|
||
return "/" + "/".join(parts[1:])
|
||
|
||
|
||
def default_mount_path(directory: Path) -> str:
|
||
slug = re.sub(r"[^a-zA-Z0-9._-]+", "-", directory.name).strip("-")
|
||
if not slug:
|
||
raise PreviewError("无法从文件目录生成 mount path，请显式提供 --mount-path")
|
||
return f"/preview/{slug}"
|
||
|
||
|
||
def route_matches(blob: str, mount_path: str, directory: Path) -> bool:
|
||
return mount_path in blob and str(directory) in blob
|
||
|
||
|
||
def route_conflicts(blob: str, mount_path: str, directory: Path) -> bool:
|
||
return mount_path in blob and str(directory) not in blob
|
||
|
||
|
||
def configure_route(directory: Path, mount_path: str, port: int, data: dict, plain: str) -> None:
|
||
blob = status_blob(data, plain)
|
||
if route_conflicts(blob, mount_path, directory):
|
||
raise PreviewError(
|
||
f"mount path {mount_path} 已被其它目录占用；请改用唯一的 --mount-path，不会覆盖已有路由"
|
||
)
|
||
|
||
if route_matches(blob, mount_path, directory):
|
||
return
|
||
|
||
result = run_tailscale(
|
||
"serve",
|
||
"--bg",
|
||
f"--http={port}",
|
||
f"--set-path={mount_path}",
|
||
"--yes",
|
||
str(directory),
|
||
)
|
||
if result.returncode != 0:
|
||
detail = (result.stderr or result.stdout).strip()
|
||
raise PreviewError(f"Tailscale Serve 配置失败：{detail or 'serve 命令失败'}")
|
||
|
||
data, plain = serve_status()
|
||
blob = status_blob(data, plain)
|
||
if route_conflicts(blob, mount_path, directory) or not route_matches(blob, mount_path, directory):
|
||
raise PreviewError("Serve 命令返回成功，但状态中没有确认目标路由；停止输出 URL")
|
||
|
||
|
||
def build_result(file_path: Path, directory: Path, mount_path: str, ip: str, port: int) -> dict[str, object]:
|
||
relative = file_path.relative_to(directory).as_posix()
|
||
suffix = f":{port}" if port != 80 else ""
|
||
url = f"http://{ip}{suffix}{mount_path}/{quote(relative, safe='/')}"
|
||
return {
|
||
"file": str(file_path),
|
||
"directory": str(directory),
|
||
"mountPath": mount_path,
|
||
"ip": ip,
|
||
"port": port,
|
||
"url": url,
|
||
"stopCommand": f"tailscale serve --http={port} --set-path={mount_path} off",
|
||
}
|
||
|
||
|
||
def parse_args(argv: list[str]) -> argparse.Namespace:
|
||
parser = argparse.ArgumentParser(description=__doc__)
|
||
parser.add_argument("file", type=Path, help="要展示的静态文件")
|
||
parser.add_argument("--mount-path", help="Tailscale Serve 的 URL 路径")
|
||
parser.add_argument("--port", type=int, default=80, help="HTTP 端口，默认 80")
|
||
return parser.parse_args(argv)
|
||
|
||
|
||
def main(argv: list[str] | None = None) -> int:
|
||
args = parse_args(argv or sys.argv[1:])
|
||
try:
|
||
if not 1 <= args.port <= 65535:
|
||
raise PreviewError("port 必须在 1 到 65535 之间")
|
||
require_tailscale()
|
||
file_path = args.file.expanduser().resolve()
|
||
if not file_path.is_file():
|
||
raise PreviewError(f"输入不是普通文件或不存在：{file_path}")
|
||
directory = file_path.parent
|
||
mount_path = normalize_mount_path(args.mount_path) if args.mount_path else default_mount_path(directory)
|
||
ip = tailscale_ipv4()
|
||
data, plain = serve_status()
|
||
configure_route(directory, mount_path, args.port, data, plain)
|
||
print(json.dumps(build_result(file_path, directory, mount_path, ip, args.port), ensure_ascii=False))
|
||
return 0
|
||
except PreviewError as exc:
|
||
print(f"previewer: {exc}", file=sys.stderr)
|
||
return 2
|
||
|
||
|
||
if __name__ == "__main__":
|
||
raise SystemExit(main())
|