From c43ab4a16e143d03ced0c2d31823fee4e3e78fe9 Mon Sep 17 00:00:00 2001
From: typist <1003659191@qq.com>
Date: Sun, 1 Nov 2026 04:21:07 +0800
Subject: [PATCH] =?UTF-8?q?=E5=88=9D=E5=A7=8B=E7=89=88=E6=9C=AC:=20sbp=20H?=
 =?UTF-8?q?ysteria2=20=E6=9C=8D=E5=8A=A1=E7=AE=A1=E7=90=86=E5=91=BD?=
 =?UTF-8?q?=E4=BB=A4?=
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

---
 README.md  |  112 ++++++
 install.sh |   33 ++
 sbp        | 1098 ++++++++++++++++++++++++++++++++++++++++++++++++++++
 3 files changed, 1243 insertions(+)
 create mode 100644 README.md
 create mode 100755 install.sh
 create mode 100755 sbp

diff --git a/README.md b/README.md
new file mode 100644
index 0000000..1a47efb
--- /dev/null
+++ b/README.md
@@ -0,0 +1,112 @@
+# sbp
+
+`sbp` 在用户级别搭建和管理 sing-box Hysteria2 服务. 不需要 root. 服务由 systemd user 管理.
+
+## 部署到 VPS
+
+一条命令下载并安装 `sbp`:
+
+```bash
+curl -fsSL https://raw.githubusercontent.com/justypist/sbp/main/install.sh | bash
+sbp install
+```
+
+下载后直接开始搭建:
+
+```bash
+curl -fsSL https://raw.githubusercontent.com/justypist/sbp/main/install.sh | bash -s -- --run
+```
+
+`install.sh` 只安装 `sbp` 命令. 环境变量 `SBP_BIN` 可以改变安装目录 (默认 `~/.local/bin`).
+
+也可以手动复制:
+
+```bash
+install -m 755 sbp ~/.local/bin/sbp
+sbp --version
+```
+
+依赖: Linux, systemd user, openssl, python3 (标准库). 网络可访问 GitHub.
+
+## 用法
+
+```bash
+sbp install                 # 下载最新 sing-box, 生成证书和用户, 启动服务, 打印配置
+sbp install --port 8443 --sni www.example.com --host 203.0.113.10
+sbp install --cert fullchain.pem --key privkey.pem --sni example.com
+sbp new phone               # 新增用户并打印配置
+sbp list                    # 列出用户和分享链接
+sbp show phone              # 打印用户的完整配置
+sbp show phone --uri        # 只打印分享链接
+sbp show phone --json       # 只打印 sing-box 客户端配置
+sbp show --cert             # 打印服务端证书
+sbp del phone               # 删除用户
+sbp status                  # 服务状态
+sbp log -f                  # 查看日志
+sbp upgrade                 # 升级 sing-box
+sbp uninstall               # 停止并删除服务和程序, 保留配置
+sbp uninstall --purge       # 同时删除配置和证书
+```
+
+`install` 的选项:
+
+| 选项 | 默认值 | 说明 |
+| --- | --- | --- |
+| `--port` | 随机空闲端口 | UDP 监听端口, 必须 >= 1024 |
+| `--listen` | `::` | 监听地址, 默认同时监听 IPv4 和 IPv6 |
+| `--sni` | `--host` 是域名时用它, 否则 `www.bing.com` | TLS server name 和证书名 |
+| `--host` | 自动探测公网 IP | 打印给客户端使用的地址 |
+| `--cert` / `--key` | 自签名证书 | 使用已有证书, 证书和私钥必须匹配 |
+| `--masquerade` | `https://<sni>` | 伪装地址 |
+| `--no-obfs` | 关闭 | 关闭 salamander 混淆 (不推荐) |
+| `--sing-box-version` | 最新版 | 指定 sing-box 版本 |
+| `--force` | 关闭 | 重装并保留已有用户 |
+
+## 安全设计
+
+- 密码和混淆密码由 `secrets.token_urlsafe(32)` 生成, 每个用户独立.
+- 默认自签名 ECDSA P-256 证书, 有效期 10 年, 只含一个 SAN.
+- TLS 最低版本 1.3, ALPN 只用 `h3`.
+- 客户端配置使用 `certificate_public_key_sha256` 固定证书公钥. 不开启 `insecure`.
+- 服务以非特权用户运行, 只监听 >= 1024 的端口.
+- 下载的 sing-box 压缩包与 GitHub 发布的 `sha256` 摘要比对.
+- systemd 加固: `NoNewPrivileges`, `ProtectSystem=strict`, `ProtectHome=read-only`,
+  `PrivateTmp`, `PrivateDevices`, `ProtectProc=invisible`, `ProcSubset=pid`,
+  `ProtectKernelTunables`, `ProtectKernelModules`, `ProtectKernelLogs`,
+  `RestrictAddressFamilies` (含 `AF_NETLINK`), `RestrictNamespaces`,
+  `MemoryDenyWriteExecute`, `SystemCallFilter=@system-service`, 空 capability 集等.
+- `InaccessiblePaths` 屏蔽家目录中的常见凭据目录 (`~/.ssh`, `~/.gnupg`, `~/.aws`,
+  `~/.kube`, `~/.docker`, 浏览器配置等). 只屏蔽已存在的路径.
+- 配置和状态文件权限 `0600`, 证书目录权限 `0700`.
+- 日志级别 `warn`, 不记录访问日志.
+
+## 分享链接与 pinSHA256
+
+分享链接中的 `pinSHA256` 是证书 DER 的 SHA-256, 十六进制. 这与 Hysteria 2 URI 规范一致.
+
+部分客户端不支持 `pinSHA256`. 这些客户端会拒绝自签名证书. 请改用 `sbp show <name> --json`
+输出的 sing-box 客户端配置. 不要为了兼容而关闭证书校验.
+
+客户端配置使用 `certificate_public_key_sha256`. 该字段要求 sing-box 1.13 或更高版本.
+旧客户端无法解析.
+
+## 文件位置
+
+| 路径 | 内容 |
+| --- | --- |
+| `~/.config/sbp/state.json` | 用户, 端口, 密码, 证书指纹 |
+| `~/.config/sbp/sing-box.json` | 生成的 sing-box 服务端配置 |
+| `~/.config/sbp/certs/` | 证书和私钥 |
+| `~/.local/share/sbp/bin/sing-box` | sing-box 程序 |
+| `~/.config/systemd/user/sbp.service` | systemd user 单元 |
+
+## 说明
+
+- 主机需要公网可达的 UDP 端口. 路由器 NAT 需要端口转发.
+- 防火墙需要放行 UDP 端口. `install` 会打印示例命令.
+- 用户级服务依赖 lingering. 确认 `loginctl show-user $USER -p Linger` 输出 `Linger=yes`.
+  `sbp install` 在未开启 lingering 时会给出警告.
+- 主机没有 IPv6 时, 默认监听 `0.0.0.0`, 不会监听 `::`.
+- 首次安装后, `sbp` 会等待服务进入 `active`. 如果失败, 用 `sbp log` 查看原因.
+- Python 的 `urllib` 不读取 `ALL_PROXY`. 使用 SOCKS 代理时请设置 `HTTPS_PROXY`.
+- 文件位置遵循 `XDG_CONFIG_HOME` 和 `XDG_DATA_HOME`. 上表按默认值列出.
diff --git a/install.sh b/install.sh
new file mode 100755
index 0000000..79dc881
--- /dev/null
+++ b/install.sh
@@ -0,0 +1,33 @@
+#!/usr/bin/env bash
+set -euo pipefail
+
+repo="${SBP_REPO:-justypist/sbp}"
+branch="${SBP_BRANCH:-main}"
+dest="${SBP_BIN:-$HOME/.local/bin}"
+url="https://raw.githubusercontent.com/${repo}/${branch}/sbp"
+
+command -v curl >/dev/null 2>&1 || { echo "error: curl is required" >&2; exit 1; }
+command -v python3 >/dev/null 2>&1 || { echo "error: python3 is required" >&2; exit 1; }
+
+mkdir -p "$dest"
+tmp="$(mktemp)"
+trap 'rm -f "$tmp"' EXIT
+
+curl -fsSL "$url" -o "$tmp"
+head -n 1 "$tmp" | grep -q '^#!/usr/bin/env python3' || {
+  echo "error: unexpected content from $url" >&2
+  exit 1
+}
+chmod 755 "$tmp"
+mv "$tmp" "$dest/sbp"
+trap - EXIT
+
+echo "installed $dest/sbp"
+case ":$PATH:" in
+  *":$dest:"*) ;;
+  *) echo "note: add $dest to PATH, for example: export PATH=\"$dest:\$PATH\"" ;;
+esac
+
+if [ "${1:-}" = "--run" ]; then
+  exec "$dest/sbp" install "${@:2}"
+fi
diff --git a/sbp b/sbp
new file mode 100755
index 0000000..08033f5
--- /dev/null
+++ b/sbp
@@ -0,0 +1,1098 @@
+#!/usr/bin/env python3
+"""sbp - manage a user-level sing-box Hysteria2 server."""
+
+from __future__ import annotations
+
+import argparse
+import base64
+import fcntl
+import hashlib
+import ipaddress
+import json
+import os
+import platform
+import re
+import secrets
+import shutil
+import socket
+import stat
+import subprocess
+import sys
+import tarfile
+import tempfile
+import urllib.error
+import urllib.parse
+import urllib.request
+from contextlib import contextmanager
+from datetime import datetime, timezone
+from pathlib import Path
+
+APP = "sbp"
+VERSION = "1.0.0"
+DEFAULT_SNI = "www.bing.com"
+DEFAULT_LISTEN = "::"
+DEFAULT_CLIENT_PORT = 1080
+RELEASE_API = "https://api.github.com/repos/SagerNet/sing-box/releases/latest"
+RELEASE_TAG_API = "https://api.github.com/repos/SagerNet/sing-box/releases/tags/{tag}"
+
+def xdg(env: str, fallback: str) -> Path:
+  return Path(os.environ.get(env) or fallback).expanduser()
+
+
+CONFIG_DIR = Path(os.environ.get("SBP_CONFIG_DIR") or xdg("XDG_CONFIG_HOME", "~/.config") / "sbp")
+DATA_DIR = Path(os.environ.get("SBP_DATA_DIR") or xdg("XDG_DATA_HOME", "~/.local/share") / "sbp")
+UNIT_DIR = Path(os.environ.get("SBP_UNIT_DIR") or xdg("XDG_CONFIG_HOME", "~/.config") / "systemd" / "user")
+UNIT_NAME = os.environ.get("SBP_UNIT_NAME", "sbp")
+
+STATE_FILE = CONFIG_DIR / "state.json"
+SERVER_CONFIG = CONFIG_DIR / "sing-box.json"
+CERT_FILE = CONFIG_DIR / "certs" / "cert.pem"
+KEY_FILE = CONFIG_DIR / "certs" / "key.pem"
+BINARY = DATA_DIR / "bin" / "sing-box"
+UNIT_FILE = UNIT_DIR / f"{UNIT_NAME}.service"
+
+TTY = sys.stdout.isatty() and not os.environ.get("NO_COLOR")
+
+
+def paint(code: str, text: str) -> str:
+  return f"\x1b[{code}m{text}\x1b[0m" if TTY else text
+
+
+def info(msg: str) -> None:
+  print(f"{paint('36', '==>')} {msg}")
+
+
+def ok(msg: str) -> None:
+  print(f"{paint('32', '==>')} {msg}")
+
+
+def warn(msg: str) -> None:
+  print(f"{paint('33', 'warning:')} {msg}", file=sys.stderr)
+
+
+def die(msg: str, code: int = 1) -> None:
+  print(f"{paint('31', 'error:')} {msg}", file=sys.stderr)
+  sys.exit(code)
+
+
+def now_iso() -> str:
+  return datetime.now(timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ")
+
+
+def token() -> str:
+  return secrets.token_urlsafe(32)
+
+
+def run(cmd: list[str], check: bool = True, input: str | None = None) -> subprocess.CompletedProcess:
+  try:
+    proc = subprocess.run(cmd, check=False, input=input, capture_output=True, text=True)
+  except OSError as exc:
+    die(f"{cmd[0]}: {exc}")
+  if check and proc.returncode != 0:
+    detail = (proc.stderr or proc.stdout or "").strip()
+    die(f"{' '.join(cmd)}: {detail}")
+  return proc
+
+
+def run_bytes(cmd: list[str], input: bytes | None = None) -> bytes:
+  try:
+    proc = subprocess.run(cmd, check=False, input=input, capture_output=True)
+  except OSError as exc:
+    die(f"{cmd[0]}: {exc}")
+  if proc.returncode != 0:
+    detail = (proc.stderr or proc.stdout).decode(errors="replace").strip()
+    die(f"{' '.join(cmd)}: {detail}")
+  return proc.stdout
+
+
+def open_exclusive(path: Path, mode: int) -> int:
+  flags = os.O_WRONLY | os.O_CREAT | os.O_EXCL
+  if hasattr(os, "O_NOFOLLOW"):
+    flags |= os.O_NOFOLLOW
+  return os.open(path, flags, mode)
+
+
+def write_file(path: Path, data: str, mode: int) -> None:
+  path.parent.mkdir(parents=True, exist_ok=True)
+  tmp = path.with_name(f"{path.name}.{os.getpid()}.{secrets.token_hex(4)}.tmp")
+  fd = open_exclusive(tmp, mode)
+  with os.fdopen(fd, "w") as handle:
+    os.fchmod(handle.fileno(), mode)
+    handle.write(data)
+  os.replace(tmp, path)
+
+
+def load_state(strict: bool = True) -> dict | None:
+  if not STATE_FILE.exists():
+    return None
+  try:
+    return json.loads(STATE_FILE.read_text())
+  except (json.JSONDecodeError, OSError) as exc:
+    if strict:
+      die(f"cannot read state file {STATE_FILE}: {exc}")
+    warn(f"ignoring unreadable state file {STATE_FILE}: {exc}")
+    return None
+
+
+def save_state(state: dict) -> None:
+  write_file(STATE_FILE, json.dumps(state, indent=2) + "\n", 0o600)
+
+
+def require_state() -> dict:
+  state = load_state()
+  if state is None:
+    die("sbp is not installed; run 'sbp install' first")
+  return state
+
+
+def is_ip(value: str) -> bool:
+  try:
+    ipaddress.ip_address(value)
+    return True
+  except ValueError:
+    return False
+
+
+def is_private(value: str) -> bool:
+  try:
+    addr = ipaddress.ip_address(value)
+  except ValueError:
+    return False
+  return addr.is_private or addr.is_loopback
+
+
+def validate_name(name: str) -> str:
+  if not re.fullmatch(r"[A-Za-z0-9_-]{1,32}", name):
+    die("user name must match [A-Za-z0-9_-]{1,32}")
+  return name
+
+
+def port_free(port: int, listen: str, retries: int = 1) -> bool:
+  import time
+  family = socket.AF_INET6 if ":" in listen else socket.AF_INET
+  for attempt in range(retries):
+    sock = socket.socket(family, socket.SOCK_DGRAM)
+    try:
+      sock.bind((listen, port))
+      return True
+    except OSError:
+      if attempt + 1 < retries:
+        time.sleep(0.2)
+    finally:
+      sock.close()
+  return False
+
+
+def pick_port(listen: str) -> int:
+  for _ in range(128):
+    port = 20000 + secrets.randbelow(40000)
+    if port_free(port, listen):
+      return port
+  die("cannot find a free UDP port")
+
+
+def ipv6_available() -> bool:
+  try:
+    sock = socket.socket(socket.AF_INET6, socket.SOCK_DGRAM)
+    sock.bind(("::", 0))
+    sock.close()
+    return True
+  except OSError:
+    return False
+
+
+def format_address(host: str, port: int) -> str:
+  if ":" in host and not host.startswith("["):
+    host = f"[{host}]"
+  return f"{host}:{port}"
+
+
+def proxy_env() -> bool:
+  keys = ("HTTP_PROXY", "HTTPS_PROXY", "ALL_PROXY", "http_proxy", "https_proxy", "all_proxy")
+  return any(os.environ.get(key) for key in keys)
+
+
+def linger_enabled() -> bool | None:
+  if shutil.which("loginctl") is None:
+    return None
+  proc = run(["loginctl", "show-user", str(os.getuid()), "-p", "Linger", "--value"], check=False)
+  if proc.returncode != 0:
+    return None
+  return proc.stdout.strip() == "yes"
+
+
+@contextmanager
+def state_lock():
+  CONFIG_DIR.mkdir(parents=True, exist_ok=True)
+  os.chmod(CONFIG_DIR, 0o700)
+  fd = os.open(CONFIG_DIR / ".lock", os.O_WRONLY | os.O_CREAT, 0o600)
+  try:
+    fcntl.flock(fd, fcntl.LOCK_EX)
+    yield
+  finally:
+    fcntl.flock(fd, fcntl.LOCK_UN)
+    os.close(fd)
+
+
+def locked(func):
+  def wrapper(args: argparse.Namespace) -> None:
+    with state_lock():
+      func(args)
+  return wrapper
+
+
+def valid_host(value: str) -> bool:
+  if is_ip(value):
+    return True
+  if not value or len(value) > 253:
+    return False
+  return re.fullmatch(r"[A-Za-z0-9]([A-Za-z0-9.-]*[A-Za-z0-9])?", value) is not None
+
+
+def detect_host() -> str:
+  if proxy_env():
+    warn("proxy environment detected; the detected address may be the proxy exit, use --host to override")
+  try:
+    with urllib.request.urlopen("https://api.ipify.org", timeout=4) as resp:
+      value = resp.read().decode().strip()
+      if is_ip(value):
+        return value
+  except Exception:
+    pass
+  try:
+    sock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
+    sock.connect(("8.8.8.8", 53))
+    value = sock.getsockname()[0]
+    sock.close()
+    return value
+  except Exception:
+    return "127.0.0.1"
+
+
+def check_platform() -> None:
+  if sys.platform != "linux":
+    die("sbp supports Linux only")
+  if os.geteuid() == 0:
+    die("run sbp as your normal user, not as root")
+  if shutil.which("openssl") is None:
+    die("openssl is required but not found in PATH")
+  if shutil.which("systemctl") is None:
+    die("systemctl is required but not found in PATH")
+  if "XDG_RUNTIME_DIR" not in os.environ:
+    runtime = Path(f"/run/user/{os.getuid()}")
+    if runtime.is_dir():
+      os.environ["XDG_RUNTIME_DIR"] = str(runtime)
+  proc = run(["systemctl", "--user", "show-environment"], check=False)
+  if proc.returncode != 0:
+    detail = (proc.stderr or proc.stdout).strip()
+    die(f"systemd user manager is unavailable: {detail}\n"
+        "run 'loginctl enable-linger $USER' as root, then log in again")
+  if linger_enabled() is False:
+    warn(f"lingering is disabled; {UNIT_NAME} stops when you log out\n"
+         f"        run: sudo loginctl enable-linger {os.environ.get('USER', os.getuid())}")
+  if os.environ.get("ALL_PROXY") and not (os.environ.get("HTTPS_PROXY") or os.environ.get("https_proxy")):
+    warn("ALL_PROXY is not used by Python downloads; set HTTPS_PROXY if you need a proxy")
+
+
+def systemctl(*args: str, check: bool = True) -> subprocess.CompletedProcess:
+  return run(["systemctl", "--user", *args], check=check)
+
+
+def service_active() -> bool:
+  proc = systemctl("is-active", UNIT_NAME, check=False)
+  return proc.stdout.strip() == "active"
+
+
+def http_json(url: str) -> dict:
+  request = urllib.request.Request(url, headers={
+    "User-Agent": f"{APP}/{VERSION}",
+    "Accept": "application/vnd.github+json",
+  })
+  try:
+    with urllib.request.urlopen(request, timeout=30) as resp:
+      return json.load(resp)
+  except urllib.error.HTTPError as exc:
+    die(f"GitHub API returned HTTP {exc.code} for {url}")
+  except Exception as exc:
+    die(f"cannot reach GitHub: {exc}")
+  raise AssertionError
+
+
+def get_release(version: str | None) -> tuple[str, list[dict]]:
+  if version in (None, "", "latest"):
+    data = http_json(RELEASE_API)
+  else:
+    tag = version if version.startswith("v") else f"v{version}"
+    data = http_json(RELEASE_TAG_API.format(tag=tag))
+  return data["tag_name"], data["assets"]
+
+
+def asset_for(assets: list[dict], tag: str) -> dict:
+  machine = platform.machine()
+  arch = {
+    "x86_64": "amd64", "amd64": "amd64",
+    "aarch64": "arm64", "arm64": "arm64",
+    "armv7l": "armv7", "armv6l": "armv6",
+    "i686": "386", "i386": "386",
+  }.get(machine)
+  if arch is None:
+    die(f"unsupported architecture: {machine}")
+  libc = "glibc" if platform.libc_ver()[0] == "glibc" else "musl"
+  prefix = f"sing-box-{tag.lstrip('v')}-linux-{arch}"
+  candidates = [a for a in assets if a["name"].startswith(prefix) and a["name"].endswith(".tar.gz")]
+  if not candidates:
+    die(f"no release asset for linux-{arch}")
+  preferred = [a for a in candidates if libc in a["name"]]
+  return (preferred or candidates)[0]
+
+
+def download(url: str, dest: Path) -> None:
+  request = urllib.request.Request(url, headers={"User-Agent": f"{APP}/{VERSION}"})
+  try:
+    with urllib.request.urlopen(request, timeout=180) as resp, open(dest, "wb") as handle:
+      shutil.copyfileobj(resp, handle, 1 << 20)
+  except Exception as exc:
+    die(f"download failed: {exc}")
+
+
+def file_sha256(path: Path) -> str:
+  digest = hashlib.sha256()
+  with open(path, "rb") as handle:
+    for chunk in iter(lambda: handle.read(1 << 20), b""):
+      digest.update(chunk)
+  return digest.hexdigest()
+
+
+def verify_digest(path: Path, asset: dict) -> None:
+  expected = (asset.get("digest") or "").strip().lower()
+  if not expected.startswith("sha256:"):
+    warn(f"{asset['name']} has no sha256 digest; skipping integrity check")
+    return
+  actual = file_sha256(path)
+  if actual != expected.removeprefix("sha256:"):
+    die(f"checksum mismatch for {asset['name']}: expected {expected}, got sha256:{actual}")
+  ok("release checksum verified")
+
+
+def install_binary(tag: str, assets: list[dict]) -> None:
+  asset = asset_for(assets, tag)
+  size = asset.get("size", 0) / 1048576
+  info(f"downloading sing-box {tag} ({size:.1f} MiB)")
+  with tempfile.TemporaryDirectory(prefix="sbp-") as tmpdir:
+    tarball = Path(tmpdir) / "sing-box.tar.gz"
+    download(asset["browser_download_url"], tarball)
+    verify_digest(tarball, asset)
+    try:
+      with tarfile.open(tarball, "r:gz") as archive:
+        members = [m for m in archive.getmembers() if m.isfile() and m.name.endswith("/sing-box")]
+        if not members:
+          die("sing-box binary not found in archive")
+        source = archive.extractfile(members[0])
+        if source is None:
+          die("cannot read sing-box binary from archive")
+        BINARY.parent.mkdir(parents=True, exist_ok=True)
+        staged = BINARY.with_name(f"sing-box.{os.getpid()}.{secrets.token_hex(4)}.tmp")
+        fd = open_exclusive(staged, 0o600)
+        with os.fdopen(fd, "wb") as handle:
+          os.fchmod(handle.fileno(), 0o600)
+          shutil.copyfileobj(source, handle)
+    except tarfile.TarError as exc:
+      die(f"invalid release archive: {exc}")
+    os.chmod(staged, 0o755)
+    proc = run([str(staged), "version"], check=False)
+    if proc.returncode != 0:
+      staged.unlink(missing_ok=True)
+      die("downloaded sing-box binary does not run")
+    if SERVER_CONFIG.exists():
+      check = run([str(staged), "check", "-c", str(SERVER_CONFIG)], check=False)
+      if check.returncode != 0:
+        staged.unlink(missing_ok=True)
+        detail = (check.stderr or check.stdout).strip()
+        die(f"new sing-box rejects the current config; not replacing the binary: {detail}")
+    os.replace(staged, BINARY)
+  ok(f"installed sing-box {tag} at {BINARY}")
+
+
+def gen_cert(sni: str, cert: Path, key: Path) -> None:
+  cert.parent.mkdir(parents=True, exist_ok=True)
+  os.chmod(cert.parent, 0o700)
+  san = f"IP:{sni}" if is_ip(sni) else f"DNS:{sni}"
+  run([
+    "openssl", "req", "-x509", "-newkey", "ec",
+    "-pkeyopt", "ec_paramgen_curve:P-256", "-nodes",
+    "-keyout", str(key), "-out", str(cert), "-days", "3650",
+    "-subj", f"/CN={sni}",
+    "-addext", f"subjectAltName={san}",
+    "-addext", "basicConstraints=critical,CA:FALSE",
+    "-addext", "keyUsage=critical,digitalSignature",
+    "-addext", "extendedKeyUsage=serverAuth",
+  ])
+  os.chmod(key, 0o600)
+  os.chmod(cert, 0o644)
+
+
+def cert_pins(cert: Path) -> tuple[str, str]:
+  der = run_bytes(["openssl", "x509", "-in", str(cert), "-outform", "DER"])
+  public_pem = run_bytes(["openssl", "x509", "-in", str(cert), "-pubkey", "-noout"])
+  spki = run_bytes(["openssl", "pkey", "-pubin", "-outform", "DER"], input=public_pem)
+  spki_pin = base64.b64encode(hashlib.sha256(spki).digest()).decode()
+  cert_sha256 = hashlib.sha256(der).hexdigest()
+  return spki_pin, cert_sha256
+
+
+def cert_expiry(cert: Path) -> str:
+  proc = run(["openssl", "x509", "-in", str(cert), "-noout", "-enddate"], check=False)
+  if proc.returncode != 0:
+    return "unknown"
+  value = proc.stdout.strip().split("=", 1)[-1]
+  try:
+    return datetime.strptime(value, "%b %d %H:%M:%S %Y %Z").strftime("%Y-%m-%d")
+  except ValueError:
+    return value
+
+
+def cert_names(cert: Path) -> list[str]:
+  proc = run(["openssl", "x509", "-in", str(cert), "-noout", "-ext", "subjectAltName"], check=False)
+  names = []
+  for part in proc.stdout.replace("\n", ",").split(","):
+    part = part.strip()
+    if part.startswith("DNS:"):
+      names.append(part[4:])
+    elif part.startswith("IP Address:"):
+      names.append(part[len("IP Address:"):])
+  return names
+
+
+def cert_self_signed(cert: Path) -> bool:
+  subject = run(["openssl", "x509", "-in", str(cert), "-noout", "-subject"], check=False).stdout
+  issuer = run(["openssl", "x509", "-in", str(cert), "-noout", "-issuer"], check=False).stdout
+  return subject.strip().split("=", 1)[-1] == issuer.strip().split("=", 1)[-1]
+
+
+def verify_cert_pair(cert: Path, key: Path) -> None:
+  if not cert.is_file():
+    die(f"certificate not found: {cert}")
+  if not key.is_file():
+    die(f"private key not found: {key}")
+  mode = key.stat().st_mode
+  if not stat.S_ISREG(mode):
+    die(f"private key is not a regular file: {key}")
+  if mode & 0o077:
+    warn(f"{key} is readable by group or others; run: chmod 600 {key}")
+  cert_pub = run_bytes(["openssl", "x509", "-in", str(cert), "-pubkey", "-noout"])
+  key_pub = run_bytes(["openssl", "pkey", "-in", str(key), "-pubout"])
+  if cert_pub.strip() != key_pub.strip():
+    die("certificate and private key do not match")
+
+
+def setup_tls(state: dict, args: argparse.Namespace, force_regen: bool) -> None:
+  if args.cert or args.key:
+    if not (args.cert and args.key):
+      die("--cert and --key must be used together")
+    cert, key = Path(args.cert).expanduser().resolve(), Path(args.key).expanduser().resolve()
+    verify_cert_pair(cert, key)
+    names = cert_names(cert)
+    if args.sni and names and args.sni not in names:
+      warn(f"certificate names are {', '.join(names)}; {args.sni} is not listed")
+    state["sni"] = args.sni or (names[0] if names else DEFAULT_SNI)
+    entry: dict = {"mode": "custom", "cert": str(cert), "key": str(key)}
+    if cert_self_signed(cert):
+      spki_pin, cert_sha256 = cert_pins(cert)
+      entry.update({"self_signed": True, "spki_pin": spki_pin, "cert_sha256": cert_sha256})
+    state["tls"] = entry
+    ok("using the provided certificate")
+    return
+
+  sni = state["sni"]
+  stale = state.get("tls", {}).get("sni") != sni
+  if force_regen or stale or not (CERT_FILE.is_file() and KEY_FILE.is_file()):
+    info(f"generating a self-signed certificate for {sni}")
+    gen_cert(sni, CERT_FILE, KEY_FILE)
+  spki_pin, cert_sha256 = cert_pins(CERT_FILE)
+  state["tls"] = {
+    "mode": "self-signed",
+    "self_signed": True,
+    "cert": str(CERT_FILE),
+    "key": str(KEY_FILE),
+    "sni": sni,
+    "spki_pin": spki_pin,
+    "cert_sha256": cert_sha256,
+  }
+
+
+def tls_pinned(state: dict) -> bool:
+  return bool(state["tls"].get("self_signed"))
+
+
+def masquerade_for(sni: str) -> str:
+  return "https://www.bing.com" if is_ip(sni) else f"https://{sni}"
+
+
+def server_config(state: dict) -> dict:
+  tls = {
+    "enabled": True,
+    "server_name": state["sni"],
+    "alpn": ["h3"],
+    "min_version": "1.3",
+    "certificate_path": state["tls"]["cert"],
+    "key_path": state["tls"]["key"],
+  }
+  inbound: dict = {
+    "type": "hysteria2",
+    "tag": "hy2-in",
+    "listen": state["listen"],
+    "listen_port": state["port"],
+    "users": [{"name": user["name"], "password": user["password"]} for user in state["users"]],
+    "masquerade": state.get("masquerade") or masquerade_for(state["sni"]),
+    "tls": tls,
+  }
+  if state["obfs"]["enabled"]:
+    inbound["obfs"] = {"type": "salamander", "password": state["obfs"]["password"]}
+  return {
+    "log": {"level": "warn", "timestamp": True},
+    "inbounds": [inbound],
+    "outbounds": [{"type": "direct", "tag": "direct"}],
+    "route": {"rules": [], "final": "direct"},
+  }
+
+
+def client_config(state: dict, user: dict) -> dict:
+  tls: dict = {
+    "enabled": True,
+    "server_name": state["sni"],
+    "alpn": ["h3"],
+    "min_version": "1.3",
+  }
+  if tls_pinned(state):
+    tls["certificate_public_key_sha256"] = [state["tls"]["spki_pin"]]
+  outbound: dict = {
+    "type": "hysteria2",
+    "tag": "hy2-out",
+    "server": state["host"],
+    "server_port": state["port"],
+    "password": user["password"],
+    "tls": tls,
+  }
+  if state["obfs"]["enabled"]:
+    outbound["obfs"] = {"type": "salamander", "password": state["obfs"]["password"]}
+  return {
+    "log": {"level": "warn"},
+    "inbounds": [{
+      "type": "mixed",
+      "tag": "mixed-in",
+      "listen": "127.0.0.1",
+      "listen_port": DEFAULT_CLIENT_PORT,
+    }],
+    "outbounds": [outbound],
+    "route": {"rules": [], "final": "hy2-out"},
+  }
+
+
+def share_uri(state: dict, user: dict) -> str:
+  host = state["host"]
+  if ":" in host and not host.startswith("["):
+    host = f"[{host}]"
+  query: dict[str, str] = {"sni": state["sni"]}
+  if state["obfs"]["enabled"]:
+    query["obfs"] = "salamander"
+    query["obfs-password"] = state["obfs"]["password"]
+  if tls_pinned(state):
+    query["pinSHA256"] = state["tls"]["cert_sha256"]
+  auth = urllib.parse.quote(user["password"], safe="")
+  name = urllib.parse.quote(user["name"], safe="")
+  return f"hysteria2://{auth}@{host}:{state['port']}/?{urllib.parse.urlencode(query)}#{name}"
+
+
+def render_server(state: dict) -> None:
+  if not BINARY.is_file():
+    die(f"sing-box binary missing: {BINARY}; run 'sbp install --force'")
+  for path in (state["tls"]["cert"], state["tls"]["key"]):
+    if not Path(path).is_absolute() or not Path(path).is_file():
+      die(f"TLS file is missing or not absolute: {path}")
+  staged = SERVER_CONFIG.with_name(f".sing-box.{os.getpid()}.{secrets.token_hex(4)}.json")
+  write_file(staged, json.dumps(server_config(state), indent=2) + "\n", 0o600)
+  proc = run([str(BINARY), "check", "-c", str(staged)], check=False)
+  if proc.returncode != 0:
+    staged.unlink(missing_ok=True)
+    detail = (proc.stderr or proc.stdout).strip()
+    die(f"generated server config is invalid: {detail}")
+  os.replace(staged, SERVER_CONFIG)
+
+
+def unit_quote(path: Path) -> str:
+  escaped = str(path).replace("%", "%%").replace('"', '\\"')
+  return f'"{escaped}"'
+
+
+SENSITIVE_HOME_PATHS = [
+  ".ssh", ".gnupg", ".aws", ".azure", ".kube", ".docker",
+  ".netrc", ".git-credentials", ".password-store", ".mozilla",
+  ".config/gcloud", ".config/gh", ".config/hub", ".config/op",
+  ".config/google-chrome", ".config/chromium", ".config/BraveSoftware",
+  ".local/share/keyrings",
+]
+
+
+def inaccessible_paths() -> list[Path]:
+  home = Path.home()
+  return [home / name for name in SENSITIVE_HOME_PATHS if (home / name).exists()]
+
+
+def render_unit(state: dict) -> None:
+  binary = unit_quote(BINARY)
+  config = unit_quote(SERVER_CONFIG)
+  inaccessible = "\n".join(
+    f"InaccessiblePaths={unit_quote(path)}" for path in inaccessible_paths()
+  )
+  unit = f"""[Unit]
+Description=sbp sing-box Hysteria2 server
+Documentation=https://sing-box.sagernet.org
+After=network.target
+
+[Service]
+Type=simple
+ExecStart={binary} run -c {config}
+Restart=on-failure
+RestartSec=3s
+LimitNOFILE=65535
+UMask=0077
+SystemCallArchitectures=native
+NoNewPrivileges=yes
+PrivateTmp=yes
+PrivateDevices=yes
+ProtectSystem=strict
+ProtectHome=read-only
+{inaccessible}
+ProtectProc=invisible
+ProcSubset=pid
+ProtectKernelTunables=yes
+ProtectKernelModules=yes
+ProtectKernelLogs=yes
+ProtectClock=yes
+ProtectHostname=yes
+RestrictAddressFamilies=AF_INET AF_INET6 AF_UNIX AF_NETLINK
+RestrictNamespaces=yes
+RestrictRealtime=yes
+RestrictSUIDSGID=yes
+LockPersonality=yes
+MemoryDenyWriteExecute=yes
+SystemCallFilter=@system-service
+SystemCallErrorNumber=EPERM
+CapabilityBoundingSet=
+AmbientCapabilities=
+
+[Install]
+WantedBy=default.target
+"""
+  write_file(UNIT_FILE, unit, 0o644)
+
+
+def wait_active(timeout: float = 8.0) -> bool:
+  import time
+  deadline = time.monotonic() + timeout
+  while time.monotonic() < deadline:
+    if service_active():
+      time.sleep(0.5)
+      return service_active()
+    time.sleep(0.2)
+  return False
+
+
+def fail_inactive(action: str) -> None:
+  detail = systemctl("status", UNIT_NAME, "--no-pager", "-n", "15", check=False)
+  print(detail.stdout + detail.stderr, file=sys.stderr)
+  die(f"{UNIT_NAME} did not become active after {action}\n"
+      f"check 'journalctl --user -u {UNIT_NAME}'")
+
+
+def restart_service() -> None:
+  systemctl("daemon-reload", check=False)
+  proc = systemctl("restart", UNIT_NAME, check=False)
+  if proc.returncode != 0:
+    detail = (proc.stderr or proc.stdout).strip()
+    die(f"cannot restart {UNIT_NAME}: {detail}\ncheck 'journalctl --user -u {UNIT_NAME}'")
+  if not wait_active():
+    fail_inactive("restart")
+
+
+def enable_service() -> None:
+  systemctl("daemon-reload", check=False)
+  proc = systemctl("enable", "--now", UNIT_NAME, check=False)
+  if proc.returncode != 0:
+    detail = (proc.stderr or proc.stdout).strip()
+    die(f"cannot start {UNIT_NAME}: {detail}\ncheck 'journalctl --user -u {UNIT_NAME}'")
+  if not wait_active():
+    fail_inactive("start")
+
+
+def print_user(state: dict, user: dict) -> None:
+  config = client_config(state, user)
+  uri = share_uri(state, user)
+  print()
+  print(paint("1", f"User {user['name']}"))
+  print(f"  password  {user['password']}")
+  print()
+  print(paint("1", "Share URI"))
+  print(f"  {uri}")
+  if tls_pinned(state):
+    print("  note: pinSHA256 pins the certificate. Clients that cannot pin must")
+    print("        import the certificate or use the JSON config below.")
+  print()
+  print(paint("1", "sing-box client config"))
+  for line in json.dumps(config, indent=2).splitlines():
+    print(f"  {line}")
+  print()
+  print(paint("1", "Client"))
+  print("  save the JSON above as client.json, then run:")
+  print("    sing-box run -c client.json")
+  print(f"  SOCKS5 and HTTP proxy: 127.0.0.1:{DEFAULT_CLIENT_PORT}")
+
+
+def print_server(state: dict) -> None:
+  print()
+  print(paint("1", "Server"))
+  print(f"  address    {format_address(state['host'], state['port'])}/udp")
+  print(f"  sni        {state['sni']}")
+  print(f"  tls        {'self-signed (pinned)' if tls_pinned(state) else 'custom'}")
+  print(f"  obfs       {'salamander' if state['obfs']['enabled'] else 'disabled'}")
+  print(f"  masquerade {state.get('masquerade') or masquerade_for(state['sni'])}")
+  print(f"  users      {len(state['users'])}")
+  print(f"  config     {SERVER_CONFIG}")
+  if is_private(state["host"]):
+    warn(f"advertised address {state['host']} is private; pass --host if clients connect elsewhere")
+
+
+def firewall_hint(state: dict) -> None:
+  port = state["port"]
+  print()
+  print(paint("1", "Firewall"))
+  print(f"  allow UDP {port} if a firewall is active, for example:")
+  print(f"    sudo ufw allow {port}/udp")
+  print(f"    sudo firewall-cmd --permanent --add-port={port}/udp && sudo firewall-cmd --reload")
+
+
+def new_user(name: str) -> dict:
+  return {"name": name, "password": token(), "created": now_iso()}
+
+
+def safe_rmtree(path: Path, marker: Path) -> None:
+  resolved = path.resolve()
+  home = Path.home().resolve()
+  if resolved == Path("/") or resolved == home or home.is_relative_to(resolved):
+    die(f"refusing to delete {resolved}")
+  if not marker.exists():
+    die(f"refusing to delete {resolved}: {marker} not found")
+  shutil.rmtree(resolved)
+
+
+def build_state(args: argparse.Namespace, existing: dict | None) -> dict:
+  state = existing or {"schema": 1, "created": now_iso(), "users": []}
+  if args.listen:
+    listen = args.listen
+  elif state.get("listen"):
+    listen = state["listen"]
+  else:
+    listen = DEFAULT_LISTEN if ipv6_available() else "0.0.0.0"
+  if listen == "::" and not ipv6_available():
+    die("IPv6 is unavailable on this host; use --listen 0.0.0.0")
+  port = args.port or state.get("port") or pick_port(listen)
+  own_port = (existing is not None and existing.get("port") == port
+              and existing.get("listen") == listen and service_active())
+  if not own_port and not port_free(port, listen, retries=10):
+    die(f"UDP port {port} is already in use on {listen}")
+  state["listen"] = listen
+  state["port"] = port
+  state["host"] = args.host or state.get("host") or detect_host()
+  if not valid_host(state["host"]):
+    die(f"invalid --host value: {state['host']}")
+  if args.sni:
+    if not valid_host(args.sni):
+      die(f"invalid --sni value: {args.sni}")
+    state["sni"] = args.sni
+  elif not state.get("sni"):
+    state["sni"] = state["host"] if not is_ip(state["host"]) else DEFAULT_SNI
+  if "obfs" not in state:
+    state["obfs"] = {"enabled": not args.no_obfs, "password": token()}
+  if args.no_obfs:
+    state["obfs"]["enabled"] = False
+  if args.masquerade:
+    state["masquerade"] = args.masquerade
+  if not state["users"]:
+    state["users"] = [new_user("u1")]
+  return state
+
+
+@locked
+def cmd_install(args: argparse.Namespace) -> None:
+  check_platform()
+  existing = load_state()
+  if existing and not args.force:
+    die(f"already installed in {CONFIG_DIR}; use --force to reinstall")
+  if bool(args.cert) != bool(args.key):
+    die("--cert and --key must be used together")
+
+  tag, assets = get_release(args.singbox_version)
+  install_binary(tag, assets)
+
+  state = build_state(args, existing)
+  state["version"] = tag
+  sni_changed = existing is not None and existing.get("sni") != state["sni"]
+  setup_tls(state, args, force_regen=sni_changed or existing is None)
+  render_server(state)
+  render_unit(state)
+  save_state(state)
+  if existing:
+    systemctl("stop", UNIT_NAME, check=False)
+  enable_service()
+  ok(f"{UNIT_NAME} is active")
+
+  print_server(state)
+  for user in state["users"]:
+    print_user(state, user)
+  firewall_hint(state)
+
+
+@locked
+def cmd_new(args: argparse.Namespace) -> None:
+  state = require_state()
+  name = args.name or f"u{secrets.token_hex(3)}"
+  validate_name(name)
+  if any(user["name"] == name for user in state["users"]):
+    die(f"user {name} already exists")
+  user = new_user(name)
+  state["users"].append(user)
+  render_server(state)
+  save_state(state)
+  restart_service()
+  ok(f"added user {name}")
+  print_user(state, user)
+
+
+@locked
+def cmd_del(args: argparse.Namespace) -> None:
+  state = require_state()
+  validate_name(args.name)
+  remaining = [user for user in state["users"] if user["name"] != args.name]
+  if len(remaining) == len(state["users"]):
+    die(f"user {args.name} does not exist")
+  state["users"] = remaining
+  render_server(state)
+  save_state(state)
+  restart_service()
+  ok(f"deleted user {args.name}")
+  if not remaining:
+    warn("no users left; all clients are rejected")
+
+
+def cmd_list(args: argparse.Namespace) -> None:
+  state = require_state()
+  if args.json:
+    payload = {
+      "installed": True,
+      "service": "active" if service_active() else "inactive",
+      "address": format_address(state["host"], state["port"]),
+      "sni": state["sni"],
+      "users": [
+        {"name": u["name"], "created": u["created"], "uri": share_uri(state, u)}
+        for u in state["users"]
+      ],
+    }
+    print(json.dumps(payload, indent=2))
+    return
+  if not state["users"]:
+    print("no users")
+    return
+  rows = [(u["name"], u["created"][:10], share_uri(state, u)) for u in state["users"]]
+  name_width = max(len("NAME"), *(len(r[0]) for r in rows))
+  date_width = max(len("CREATED"), *(len(r[1]) for r in rows))
+  print(f"{'NAME':<{name_width}}  {'CREATED':<{date_width}}  URI")
+  for name, created, uri in rows:
+    print(f"{name:<{name_width}}  {created:<{date_width}}  {uri}")
+
+
+def cmd_show(args: argparse.Namespace) -> None:
+  state = require_state()
+  if args.cert:
+    cert = Path(state["tls"]["cert"])
+    if not cert.is_file():
+      die(f"certificate not found: {cert}")
+    print(cert.read_text(), end="")
+    return
+  if args.uri:
+    if not args.name:
+      die("--uri requires a user name")
+    user = find_user(state, args.name)
+    print(share_uri(state, user))
+    return
+  if args.json:
+    if args.name:
+      print(json.dumps(client_config(state, find_user(state, args.name)), indent=2))
+    else:
+      if not state["users"]:
+        die("no users; run 'sbp new' first")
+      print(json.dumps(client_config(state, state["users"][0]), indent=2))
+    return
+  if args.name:
+    print_user(state, find_user(state, args.name))
+  else:
+    print_server(state)
+    for user in state["users"]:
+      print_user(state, user)
+
+
+def find_user(state: dict, name: str) -> dict:
+  for user in state["users"]:
+    if user["name"] == name:
+      return user
+  die(f"user {name} does not exist")
+  raise AssertionError
+
+
+def cmd_status(args: argparse.Namespace) -> None:
+  state = require_state()
+  active = service_active()
+  if BINARY.is_file():
+    version = run([str(BINARY), "version"], check=False)
+    version_text = version.stdout.splitlines()[0].replace("sing-box version ", "").strip() \
+      if version.returncode == 0 else "unknown"
+  else:
+    version_text = "missing (run 'sbp install --force')"
+  print(f"service    {'active' if active else 'inactive'}")
+  print(f"sing-box   {version_text}")
+  listen = f"[{state['listen']}]" if ":" in state["listen"] else state["listen"]
+  print(f"listen     {listen}:{state['port']}/udp")
+  print(f"sni        {state['sni']}")
+  print(f"tls        {'self-signed (pinned)' if tls_pinned(state) else 'custom'}")
+  print(f"users      {len(state['users'])}")
+  print(f"created    {state['created'][:10]}")
+  print(f"cert until {cert_expiry(Path(state['tls']['cert']))}")
+  if not active:
+    print(f"hint       journalctl --user -u {UNIT_NAME} -n 50")
+
+
+def cmd_log(args: argparse.Namespace) -> None:
+  require_state()
+  cmd = ["journalctl", "--user", "-u", UNIT_NAME, "-n", str(args.lines), "--no-pager"]
+  if args.follow:
+    cmd.append("-f")
+  os.execvp("journalctl", cmd)
+
+
+@locked
+def cmd_upgrade(args: argparse.Namespace) -> None:
+  state = require_state()
+  tag, assets = get_release(args.singbox_version)
+  current = state.get("version")
+  if current == tag and not args.force:
+    ok(f"already on {tag}")
+    return
+  install_binary(tag, assets)
+  state["version"] = tag
+  render_server(state)
+  save_state(state)
+  restart_service()
+  ok(f"upgraded {current or 'unknown'} -> {tag}")
+
+
+@locked
+def cmd_uninstall(args: argparse.Namespace) -> None:
+  state = load_state(strict=False)
+  if state is None and not UNIT_FILE.exists() and not BINARY.exists():
+    die("sbp is not installed")
+  systemctl("disable", "--now", UNIT_NAME, check=False)
+  if UNIT_FILE.exists():
+    UNIT_FILE.unlink()
+  systemctl("daemon-reload", check=False)
+  same_dir = CONFIG_DIR.resolve() == DATA_DIR.resolve()
+  if DATA_DIR.exists():
+    marker = BINARY if BINARY.exists() else STATE_FILE
+    safe_rmtree(DATA_DIR, marker)
+  if args.purge:
+    if CONFIG_DIR.exists() and not same_dir:
+      safe_rmtree(CONFIG_DIR, STATE_FILE)
+    ok("removed the service, binary, config, and certificates")
+  else:
+    ok(f"removed the service and binary; kept {CONFIG_DIR} (use --purge to delete)")
+
+
+def build_parser() -> argparse.ArgumentParser:
+  parser = argparse.ArgumentParser(
+    prog=APP,
+    description="Manage a user-level sing-box Hysteria2 server.",
+    formatter_class=argparse.RawDescriptionHelpFormatter,
+    epilog="""examples:
+  sbp install
+  sbp install --port 23456 --sni www.example.com --host 203.0.113.10
+  sbp install --cert /path/fullchain.pem --key /path/privkey.pem --sni example.com
+  sbp new phone
+  sbp list
+  sbp show phone --uri
+  sbp del phone
+  sbp uninstall --purge
+""",
+  )
+  parser.add_argument("-V", "--version", action="version", version=f"{APP} {VERSION}")
+  sub = parser.add_subparsers(dest="command", required=True)
+
+  install = sub.add_parser("install", help="download sing-box, create a server, and start it")
+  install.add_argument("--port", type=int, help="UDP listen port (default: random free port)")
+  install.add_argument("--listen", help="listen address (default: ::)")
+  install.add_argument("--sni", help=f"TLS server name and certificate CN (default: {DEFAULT_SNI})")
+  install.add_argument("--host", help="address printed for clients (default: detected public IP)")
+  install.add_argument("--cert", help="use an existing certificate (PEM)")
+  install.add_argument("--key", help="use an existing private key (PEM)")
+  install.add_argument("--masquerade", help="masquerade URL (default: https://<sni>)")
+  install.add_argument("--no-obfs", action="store_true", help="disable salamander obfuscation")
+  install.add_argument("--sing-box-version", dest="singbox_version", help="sing-box release tag (default: latest)")
+  install.add_argument("--force", action="store_true", help="reinstall over an existing setup")
+  install.set_defaults(func=cmd_install)
+
+  new = sub.add_parser("new", help="add a user and print its config")
+  new.add_argument("name", nargs="?", help="user name (default: random)")
+  new.set_defaults(func=cmd_new)
+
+  delete = sub.add_parser("del", help="delete a user")
+  delete.add_argument("name")
+  delete.set_defaults(func=cmd_del)
+
+  listing = sub.add_parser("list", help="list users")
+  listing.add_argument("--json", action="store_true", help="print JSON")
+  listing.set_defaults(func=cmd_list)
+
+  show = sub.add_parser("show", help="print server or user config")
+  show.add_argument("name", nargs="?", help="user name")
+  show.add_argument("--uri", action="store_true", help="print the share URI only")
+  show.add_argument("--json", action="store_true", help="print the sing-box client config only")
+  show.add_argument("--cert", action="store_true", help="print the server certificate (PEM)")
+  show.set_defaults(func=cmd_show)
+
+  status = sub.add_parser("status", help="show service status")
+  status.set_defaults(func=cmd_status)
+
+  log = sub.add_parser("log", help="show service logs")
+  log.add_argument("-n", "--lines", type=int, default=100, help="number of lines")
+  log.add_argument("-f", "--follow", action="store_true", help="follow the log")
+  log.set_defaults(func=cmd_log)
+
+  upgrade = sub.add_parser("upgrade", help="upgrade sing-box to the latest release")
+  upgrade.add_argument("--sing-box-version", dest="singbox_version", help="release tag (default: latest)")
+  upgrade.add_argument("--force", action="store_true", help="reinstall the same version")
+  upgrade.set_defaults(func=cmd_upgrade)
+
+  uninstall = sub.add_parser("uninstall", help="stop and remove the service and binary")
+  uninstall.add_argument("--purge", action="store_true", help="also delete config and certificates")
+  uninstall.set_defaults(func=cmd_uninstall)
+
+  return parser
+
+
+def main() -> None:
+  os.umask(0o077)
+  args = build_parser().parse_args()
+  if args.command == "install" and args.port is not None and not 1024 <= args.port <= 65535:
+    die("--port must be between 1024 and 65535")
+  args.func(args)
+
+
+if __name__ == "__main__":
+  main()