#!/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()
