feat: 支持托管中转主机和节点接入线路

2026-09-30 17:36:17 +08:00
parent ac5fed695c
commit b6aa2bf30d
15 changed files with 1338 additions and 5 deletions
@@ -210,6 +210,32 @@ const MIGRATIONS: (string | ((db: Database.Database) => void))[] = [
    PRIMARY KEY(node_id, epoch, auth_id)
  );
  `,
+  `
+  CREATE TABLE relay_hosts (
+    id TEXT PRIMARY KEY,
+    name TEXT NOT NULL,
+    server_host TEXT NOT NULL,
+    token_hash TEXT NOT NULL,
+    token_encrypted TEXT NOT NULL,
+    enabled INTEGER NOT NULL DEFAULT 1,
+    last_seen_at TEXT,
+    applied_hash TEXT NOT NULL DEFAULT '',
+    service_active INTEGER NOT NULL DEFAULT 0,
+    error TEXT NOT NULL DEFAULT '',
+    created_at TEXT NOT NULL
+  );
+  CREATE TABLE relay_routes (
+    id TEXT PRIMARY KEY,
+    host_id TEXT NOT NULL REFERENCES relay_hosts(id) ON DELETE CASCADE,
+    node_id INTEGER NOT NULL REFERENCES hy2_nodes(id) ON DELETE RESTRICT,
+    name TEXT NOT NULL,
+    public_port INTEGER NOT NULL CHECK(public_port BETWEEN 1 AND 65535),
+    listen_port INTEGER NOT NULL CHECK(listen_port BETWEEN 1 AND 65535),
+    enabled INTEGER NOT NULL DEFAULT 1,
+    created_at TEXT NOT NULL
+  );
+  CREATE INDEX relay_routes_host ON relay_routes(host_id);
+  `,
 ];

 export function migrateDatabase(db: Database.Database) {
@@ -0,0 +1,307 @@
+import {
+  createHash,
+  randomBytes,
+  randomUUID,
+  timingSafeEqual,
+} from "node:crypto";
+import { isIP } from "node:net";
+import { z } from "zod";
+import type Database from "better-sqlite3";
+import { ensureDb } from "@/lib/db";
+import { encryptText } from "@/lib/crypto";
+import type { Hy2NodeRow } from "@/lib/store";
+
+export type RelayHost = {
+  id: string;
+  name: string;
+  server_host: string;
+  enabled: number;
+  token_hash: string;
+  token_encrypted: string;
+  last_seen_at: string | null;
+  applied_hash: string;
+  service_active: number;
+  error: string;
+  created_at: string;
+};
+export type RelayRoute = {
+  id: string;
+  host_id: string;
+  node_id: number;
+  name: string;
+  public_port: number;
+  listen_port: number;
+  enabled: number;
+  created_at: string;
+};
+export class RelayError extends Error {}
+const hostname = z
+  .string()
+  .trim()
+  .min(1)
+  .max(253)
+  .refine(
+    (v) =>
+      Boolean(isIP(v)) ||
+      /^(?=.{1,253}$)[a-zA-Z0-9](?:[a-zA-Z0-9.-]*[a-zA-Z0-9])?$/.test(v),
+    "请输入域名或 IP，不含协议和端口。",
+  );
+const hostSchema = z.object({
+  id: z.string().uuid().optional(),
+  name: z.string().trim().min(1).max(100),
+  server_host: hostname,
+  enabled: z.boolean(),
+});
+const routeSchema = z.object({
+  id: z.string().uuid().optional(),
+  host_id: z.string().uuid(),
+  node_id: z.coerce.number().int().positive(),
+  name: z.string().trim().min(1).max(100),
+  public_port: z.coerce.number().int().min(1).max(65535),
+  listen_port: z.coerce.number().int().min(1).max(65535),
+  enabled: z.boolean(),
+});
+const digest = (value: string) =>
+  createHash("sha256").update(value).digest("hex");
+export const relayReportSchema = z
+  .object({
+    appliedHash: z.string().regex(/^([a-f0-9]{64})?$/),
+    active: z.boolean(),
+    error: z.enum(["", "port_conflict", "apply_failed", "service_stopped"]),
+  })
+  .strict();
+export async function relayInventory() {
+  const db = await ensureDb();
+  return {
+    hosts: db
+      .prepare("SELECT * FROM relay_hosts ORDER BY created_at DESC")
+      .all() as RelayHost[],
+    routes: db
+      .prepare("SELECT * FROM relay_routes ORDER BY created_at DESC")
+      .all() as RelayRoute[],
+  };
+}
+export async function saveRelayHost(input: unknown) {
+  const parsed = hostSchema.safeParse(input);
+  if (!parsed.success)
+    throw new RelayError("请填写主机名称和有效的公网域名/IP。");
+  const data = parsed.data,
+    db = await ensureDb();
+  if (data.id) {
+    if (
+      !db
+        .prepare(
+          "UPDATE relay_hosts SET name=?,server_host=?,enabled=? WHERE id=?",
+        )
+        .run(data.name, data.server_host, +data.enabled, data.id).changes
+    )
+      throw new RelayError("中转主机不存在。");
+    return data.id;
+  }
+  const id = randomUUID(),
+    token = randomBytes(32).toString("hex");
+  db.prepare(
+    "INSERT INTO relay_hosts(id,name,server_host,enabled,token_hash,token_encrypted,created_at) VALUES(?,?,?,?,?,?,?)",
+  ).run(
+    id,
+    data.name,
+    data.server_host,
+    +data.enabled,
+    digest(token),
+    encryptText(token),
+    new Date().toISOString(),
+  );
+  return id;
+}
+export async function saveRelayRoute(input: unknown) {
+  const parsed = routeSchema.safeParse(input);
+  if (!parsed.success)
+    throw new RelayError("请填写线路名称、主机、目标节点及 1–65535 的端口。");
+  const data = parsed.data,
+    db = await ensureDb();
+  return db
+    .transaction(() => {
+      const host = db
+        .prepare("SELECT * FROM relay_hosts WHERE id=?")
+        .get(data.host_id) as RelayHost | undefined;
+      const node = db
+        .prepare("SELECT * FROM hy2_nodes WHERE id=?")
+        .get(data.node_id) as Hy2NodeRow | undefined;
+      if (
+        !host ||
+        !node ||
+        !["hy2", "anytls"].includes(node.node_type) ||
+        !node.server_host ||
+        !node.server_port
+      )
+        throw new RelayError("请选择托管 HY2 或 AnyTLS 节点作为目标。");
+      if (
+        data.enabled &&
+        db
+          .prepare(
+            `SELECT r.id FROM relay_routes r JOIN hy2_nodes n ON n.id=r.node_id WHERE r.host_id=? AND r.enabled=1 AND n.node_type=? AND (r.listen_port=? OR r.public_port=?) AND r.id<>?`,
+          )
+          .get(
+            data.host_id,
+            node.node_type,
+            data.listen_port,
+            data.public_port,
+            data.id || "",
+          )
+      )
+        throw new RelayError(
+          "该主机同协议的监听端口或公网端口已被其他线路使用。",
+        );
+      if (
+        db
+          .prepare("SELECT id FROM relay_routes WHERE name=? AND id<>?")
+          .get(data.name, data.id || "") ||
+        db
+          .prepare("SELECT id FROM hy2_nodes WHERE name=? OR client_name=?")
+          .get(data.name, data.name)
+      )
+        throw new RelayError("订阅名称与已有节点或线路重复。");
+      if (
+        host.server_host.toLowerCase() === node.server_host.toLowerCase() &&
+        [data.public_port, data.listen_port].includes(node.server_port)
+      )
+        throw new RelayError("中转入口不能指向自身。");
+      const id = data.id || randomUUID();
+      if (data.id) {
+        if (
+          !db
+            .prepare(
+              "UPDATE relay_routes SET host_id=?,node_id=?,name=?,public_port=?,listen_port=?,enabled=? WHERE id=?",
+            )
+            .run(
+              data.host_id,
+              data.node_id,
+              data.name,
+              data.public_port,
+              data.listen_port,
+              +data.enabled,
+              id,
+            ).changes
+        )
+          throw new RelayError("线路不存在。");
+      } else
+        db.prepare(
+          "INSERT INTO relay_routes(id,host_id,node_id,name,public_port,listen_port,enabled,created_at) VALUES(?,?,?,?,?,?,?,?)",
+        ).run(
+          id,
+          data.host_id,
+          data.node_id,
+          data.name,
+          data.public_port,
+          data.listen_port,
+          +data.enabled,
+          new Date().toISOString(),
+        );
+      return id;
+    })
+    .immediate();
+}
+export async function deleteRelay(kind: "host" | "route", id: string) {
+  const db = await ensureDb();
+  db.prepare(
+    kind === "host"
+      ? "DELETE FROM relay_hosts WHERE id=?"
+      : "DELETE FROM relay_routes WHERE id=?",
+  ).run(id);
+}
+const address = (host: string, port: number) =>
+  `${isIP(host) === 6 ? `[${host}]` : host}:${port}`;
+export function relayDesired(db: Database.Database, host: RelayHost) {
+  const rows = db
+    .prepare(
+      `SELECT r.*, n.node_type,n.server_host,n.server_port FROM relay_routes r JOIN hy2_nodes n ON n.id=r.node_id WHERE r.host_id=? AND r.enabled=1 AND n.enabled=1 AND n.node_type IN ('hy2','anytls') ORDER BY r.id`,
+    )
+    .all(host.id) as (RelayRoute &
+    Pick<Hy2NodeRow, "node_type" | "server_host" | "server_port">)[];
+  const active = host.enabled ? rows : [];
+  const config = {
+    log: { level: "warn" },
+    endpoints: active.map((r) => ({
+      listen: `0.0.0.0:${r.listen_port}`,
+      remote: address(r.server_host, r.server_port),
+      network: {
+        no_tcp: r.node_type === "hy2",
+        use_udp: r.node_type === "hy2",
+        udp_timeout: 300,
+      },
+    })),
+  };
+  return {
+    config,
+    hash: digest(JSON.stringify(config)),
+    leaseSeconds: 90,
+    intervalSeconds: 15,
+  };
+}
+export async function syncRelay(id: string, token: string, input: unknown) {
+  const db = await ensureDb();
+  const host = db.prepare("SELECT * FROM relay_hosts WHERE id=?").get(id) as
+    RelayHost | undefined;
+  if (
+    !host ||
+    !/^[a-f0-9]{64}$/.test(token) ||
+    !timingSafeEqual(Buffer.from(digest(token)), Buffer.from(host.token_hash))
+  )
+    throw new RelayError("凭据无效。");
+  const data = relayReportSchema.parse(input);
+  db.prepare(
+    "UPDATE relay_hosts SET last_seen_at=?,applied_hash=?,service_active=?,error=? WHERE id=?",
+  ).run(
+    new Date().toISOString(),
+    data.appliedHash,
+    +data.active,
+    data.error,
+    id,
+  );
+  return relayDesired(db, host);
+}
+// These are subscription-only views of the existing destination, never polling nodes.
+export function appendRelaySubscriptionNodes(
+  db: Database.Database,
+  nodes: Hy2NodeRow[],
+) {
+  const hosts = db
+    .prepare("SELECT * FROM relay_hosts WHERE enabled=1")
+    .all() as RelayHost[];
+  const routes = db
+    .prepare("SELECT * FROM relay_routes WHERE enabled=1 ORDER BY created_at")
+    .all() as RelayRoute[];
+  const byId = new Map(nodes.map((n) => [n.id, n]));
+  const ready = new Map(
+    hosts
+      .filter(
+        (h) =>
+          h.service_active &&
+          !h.error &&
+          h.last_seen_at &&
+          Date.now() - Date.parse(h.last_seen_at) < 90000 &&
+          h.applied_hash === relayDesired(db, h).hash,
+      )
+      .map((h) => [h.id, h]),
+  );
+  return [
+    ...nodes,
+    ...routes.flatMap((r) => {
+      const host = ready.get(r.host_id),
+        node = byId.get(r.node_id);
+      if (!host || !node || !["hy2", "anytls"].includes(node.node_type))
+        return [];
+      return [
+        {
+          ...node,
+          slug: `relay/${r.id}`,
+          name: r.name,
+          client_name: r.name,
+          server_host: host.server_host,
+          server_port: r.public_port,
+          sni: node.sni || node.server_host,
+        },
+      ];
+    }),
+  ];
+}
@@ -1,3 +1,4 @@
+import { appendRelaySubscriptionNodes } from "@/lib/relays";
 import {
  parseCustomProxyYaml,
  extractProxyNameFromYaml,
@@ -1249,7 +1250,7 @@ export async function saveMainProxyGroupMemberSlugs(memberSlugs: string[]) {

 export async function listSubscriptionNodes() {
  const db = await ensureDb();
-  return db
+  const nodes = db
    .prepare(
      `
        SELECT *
@@ -1259,6 +1260,7 @@ export async function listSubscriptionNodes() {
      `,
    )
    .all() as Hy2NodeRow[];
+  return appendRelaySubscriptionNodes(db, nodes);
 }

 export async function writeAuthAudit(input: AuditRecordInput) {
@@ -0,0 +1,70 @@
+#!/usr/bin/env bash
+[ -n "${BASH_VERSION:-}" ] || exec bash "$0" "$@"
+set -Eeuo pipefail
+umask 077
+ID=__ID__
+ROOT=/etc/hy2-panel-relay
+[ "$(id -u)" -eq 0 ] && [ -d /run/systemd/system ] || { echo '需要 root 和 systemd Linux。'; exit 1; }
+if [ -d "$ROOT" ] || [ -e /etc/systemd/system/hy2-panel-relay.service ]; then
+  [ "$(cat "$ROOT/host-id" 2>/dev/null || true)" = "$ID" ] || { echo '已有其他中转安装，停止以避免覆盖。'; exit 1; }
+fi
+case "$(uname -m)" in
+  x86_64) ARCH=x86_64; SHA=b1cc335547bea8bb2a88178bef12ec7f2363e36200e7ea1d4e1e67627929bf65;;
+  aarch64|arm64) ARCH=aarch64; SHA=f4c0318dd86854da483dcb7645b4f39cae2cc3f91c688fef969d53220b949488;;
+  *) echo '支持 x86_64 / ARM64。'; exit 1;;
+esac
+if command -v apt-get >/dev/null; then apt-get update; DEBIAN_FRONTEND=noninteractive apt-get install -y curl ca-certificates python3 tar
+elif command -v dnf >/dev/null; then dnf install -y curl ca-certificates python3 tar
+else echo '需要 Debian/Ubuntu 或 dnf 发行版。'; exit 1; fi
+WORK=$(mktemp -d)
+trap 'rm -rf "$WORK"' EXIT
+curl -fsSL --retry 3 --connect-timeout 20 --max-time 300 "https://github.com/zhboner/realm/releases/download/v2.9.6/realm-$ARCH-unknown-linux-musl.tar.gz" -o "$WORK/realm.tar.gz"
+printf '%s  %s\n' "$SHA" "$WORK/realm.tar.gz" | sha256sum -c -
+tar -xzf "$WORK/realm.tar.gz" -C "$WORK" realm
+systemctl stop hy2-panel-relay-agent.service hy2-panel-relay.service 2>/dev/null || true
+install -d -m 700 "$ROOT" /usr/local/lib/hy2-panel-relay
+printf '%s' "$ID" > "$ROOT/host-id"
+printf '%s' __PAYLOAD__ | base64 -d > "$WORK/payload.json"
+python3 - "$WORK/payload.json" <<'PY'
+import json,sys
+from pathlib import Path
+p=json.load(open(sys.argv[1]))
+Path('/etc/hy2-panel-relay/agent.json').write_text(json.dumps(p['credentials']))
+Path('/usr/local/lib/hy2-panel-relay/agent.py').write_text(p['source'])
+PY
+install -m 755 "$WORK/realm" /usr/local/lib/hy2-panel-relay/realm
+cat > /etc/systemd/system/hy2-panel-relay.service <<'UNIT'
+[Unit]
+Description=HY2 Panel Realm relay
+After=network-online.target
+[Service]
+ExecStart=/usr/local/lib/hy2-panel-relay/realm -c /etc/hy2-panel-relay/config.json
+Restart=on-failure
+RestartSec=3
+LimitNOFILE=1048576
+NoNewPrivileges=true
+ProtectHome=true
+ProtectSystem=strict
+PrivateTmp=true
+UNIT
+cat > /etc/systemd/system/hy2-panel-relay-agent.service <<'UNIT'
+[Unit]
+Description=HY2 Panel relay management agent
+After=network-online.target
+Wants=network-online.target
+[Service]
+ExecStart=/usr/bin/python3 /usr/local/lib/hy2-panel-relay/agent.py
+ExecStopPost=/bin/systemctl stop hy2-panel-relay.service
+Restart=always
+RestartSec=5
+NoNewPrivileges=true
+ProtectHome=true
+ProtectSystem=strict
+ReadWritePaths=/etc/hy2-panel-relay
+PrivateTmp=true
+[Install]
+WantedBy=multi-user.target
+UNIT
+systemctl daemon-reload
+systemctl enable --now hy2-panel-relay-agent.service
+echo '中转管理程序已启动，请在面板配置线路并确认状态。无需开放公网管理端口。'
@@ -0,0 +1,121 @@
+#!/usr/bin/env python3
+"""Outbound-only management for a dedicated Realm instance."""
+import fcntl
+import json
+import os
+from pathlib import Path
+import socket
+import subprocess
+import time
+import urllib.request
+import urllib.error
+ROOT = Path('/etc/hy2-panel-relay')
+UNIT = 'hy2-panel-relay.service'
+
+
+def core(command):
+    subprocess.run(['systemctl', command, UNIT], check=True, stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL)
+
+
+def pid():
+    return int(subprocess.check_output(['systemctl', 'show', UNIT, '-p', 'MainPID', '--value'], text=True).strip())
+
+
+def active():
+    return subprocess.run(['systemctl', 'is-active', '--quiet', UNIT]).returncode == 0
+
+
+def verify_listeners(config):
+    process = pid()
+    if not process:
+        return False
+    inodes = set()
+    for fd in Path('/proc/' + str(process) + '/fd').iterdir():
+        try:
+            link = os.readlink(fd)
+            if link.startswith('socket:['):
+                inodes.add(link[8:-1])
+        except OSError:
+            pass
+    ports = set()
+    for network in ('tcp', 'udp'):
+        for line in Path('/proc/net/' + network).read_text().splitlines()[1:]:
+            fields = line.split()
+            if fields[9] in inodes and (network != 'tcp' or fields[3] == '0A'):
+                ports.add((network, int(fields[1].split(':')[1], 16)))
+    return all(('udp' if endpoint['network']['use_udp'] else 'tcp', int(endpoint['listen'].rsplit(':', 1)[1])) in ports for endpoint in config['endpoints'])
+
+
+def write_config(config):
+    staged = ROOT / 'next.json'
+    with open(staged, 'w') as handle:
+        json.dump(config, handle)
+        handle.flush()
+        os.fsync(handle.fileno())
+    os.replace(staged, ROOT / 'config.json')
+
+
+def apply(config):
+    core('stop')
+    sockets = []
+    try:
+        for endpoint in config['endpoints']:
+            sock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM if endpoint['network']['use_udp'] else socket.SOCK_STREAM)
+            sockets.append(sock)
+            # SO_REUSEADDR allows a stopped TCP listener's TIME_WAIT connections.
+            if not endpoint['network']['use_udp']:
+                sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
+            sock.bind(('0.0.0.0', int(endpoint['listen'].rsplit(':', 1)[1])))
+    except OSError:
+        return 'port_conflict'
+    finally:
+        for sock in sockets:
+            sock.close()
+    write_config(config)
+    if config['endpoints']:
+        core('start')
+        for _ in range(20):
+            time.sleep(0.2)
+            if active() and verify_listeners(config):
+                return ''
+        core('stop')
+        return 'apply_failed'
+    return ''
+
+
+def run():
+    lock = open(ROOT / 'agent.lock', 'w')
+    fcntl.flock(lock, fcntl.LOCK_EX | fcntl.LOCK_NB)
+    credentials = json.loads((ROOT / 'agent.json').read_text())
+    applied = ''
+    error = ''
+    deadline = 0
+    core('stop')
+    while True:
+        try:
+            request = urllib.request.Request(credentials['panel'] + '/api/relay/' + credentials['id'] + '/sync',
+                data=json.dumps({'appliedHash': applied, 'active': active(), 'error': error}).encode(),
+                headers={'Content-Type': 'application/json', 'Authorization': 'Bearer ' + credentials['token']})
+            with urllib.request.urlopen(request, timeout=10) as response:
+                desired = json.load(response)
+            if desired['hash'] != applied or (desired['config']['endpoints'] and not active()):
+                error = apply(desired['config'])
+                applied = desired['hash'] if not error else ''
+            deadline = time.monotonic() + min(90, desired['leaseSeconds'])
+            print('同步成功' if not error else error, flush=True)
+        except Exception as exc:
+            print('同步失败：' + type(exc).__name__, flush=True)
+            if isinstance(exc, urllib.error.HTTPError) and exc.code in (400, 401, 403, 404):
+                deadline = 0
+            if time.monotonic() >= deadline:
+                core('stop')
+                applied = ''
+        time.sleep(max(1, min(15, deadline - time.monotonic())) if deadline else 5)
+
+
+if __name__ == '__main__':
+    os.umask(0o077)
+    try:
+        run()
+    finally:
+        core('stop')