From e87619f3368a18d169cf60f33244bf105349c2a8 Mon Sep 17 00:00:00 2001 From: Niutopia <120653980+Niutopia@users.noreply.github.com> Date: Mon, 19 Oct 2026 19:47:48 +0800 Subject: [PATCH] =?UTF-8?q?feat(cluster):=20=E6=89=A9=E5=B1=95=E8=8A=82?= =?UTF-8?q?=E7=82=B9=E7=9B=B4=E8=BF=9E=E9=80=9A=E9=81=93=EF=BC=88kin-slot-?= =?UTF-8?q?gate=EF=BC=8CTLS=EF=BC=89=EF=BC=8C=E6=A7=BD=E4=BD=8D=E8=AF=B7?= =?UTF-8?q?=E6=B1=82=E4=B8=8D=E5=86=8D=E6=8C=A4=E5=9C=A8=20SSH=20=E9=95=BF?= =?UTF-8?q?=E8=BF=9E=E6=8E=A5=E9=87=8C?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .gitattributes | 1 + .github/workflows/release.yml | 2 + CHANGELOG.md | 1 + Dockerfile | 1 + deploy/Dockerfile.arm64-control | 2 + docker-compose.arm64.yml | 1 + docs/BUILD.md | 3 + docs/PANEL_API.md | 2 + package.json | 1 + scripts/docker-entrypoint.sh | 1 + src/lib/cluster/cluster-manager.mjs | 152 ++++++++- src/lib/cluster/cluster-routes.mjs | 5 + src/lib/cluster/direct-link.mjs | 272 ++++++++++++++++ src/lib/cluster/socket-relay.mjs | 37 ++- .../039_cluster_node_direct_links.sql | 15 + src/lib/db/repos/cluster-nodes-repo.mjs | 43 +++ src/server.mjs | 1 + test/unit/cluster-direct-link.test.mjs | 298 ++++++++++++++++++ test/unit/cluster-node-edit.test.mjs | 4 +- .../features/cluster/direct-link-config.tsx | 118 +++++++ web/src/features/cluster/model.ts | 10 + web/src/features/cluster/remote-list.tsx | 16 + web/src/features/cluster/ssh-panel.tsx | 3 + web/src/types/panel-cluster.ts | 13 + worker/cmd/kin-slot-gate/main.go | 36 +++ worker/internal/slotgate/cert.go | 78 +++++ worker/internal/slotgate/dial_linux.go | 31 ++ worker/internal/slotgate/dial_other.go | 21 ++ worker/internal/slotgate/slotgate.go | 203 ++++++++++++ worker/internal/slotgate/slotgate_test.go | 225 +++++++++++++ 30 files changed, 1589 insertions(+), 7 deletions(-) create mode 100644 src/lib/cluster/direct-link.mjs create mode 100644 src/lib/db/migrations/039_cluster_node_direct_links.sql create mode 100644 test/unit/cluster-direct-link.test.mjs create mode 100644 web/src/features/cluster/direct-link-config.tsx create mode 100644 worker/cmd/kin-slot-gate/main.go create mode 100644 worker/internal/slotgate/cert.go create mode 100644 worker/internal/slotgate/dial_linux.go create mode 100644 worker/internal/slotgate/dial_other.go create mode 100644 worker/internal/slotgate/slotgate.go create mode 100644 worker/internal/slotgate/slotgate_test.go diff --git a/.gitattributes b/.gitattributes index a006a3b08..7e793b29f 100644 --- a/.gitattributes +++ b/.gitattributes @@ -6,6 +6,7 @@ docker-compose.yml text eol=lf .dockerignore text eol=lf bin/kin-kernel binary bin/kin-egress binary +bin/kin-slot-gate binary bin/kin-worker binary bin/kin-codex-kernel binary bin/kin-oauth-auth binary diff --git a/.github/workflows/release.yml b/.github/workflows/release.yml index 7fd31bd92..1eb0770f0 100644 --- a/.github/workflows/release.yml +++ b/.github/workflows/release.yml @@ -47,6 +47,7 @@ jobs: mkdir -p dist (cd worker && CGO_ENABLED=0 go build -trimpath -o ../dist/kin-worker ./cmd/kin-worker) (cd worker && CGO_ENABLED=0 go build -trimpath -o ../dist/kin-egress ./cmd/kin-egress) + (cd worker && CGO_ENABLED=0 go build -trimpath -o ../dist/kin-slot-gate ./cmd/kin-slot-gate) node scripts/fetch-payload.mjs node scripts/fetch-payload.mjs --stage dist - uses: actions/upload-artifact@v7 @@ -73,6 +74,7 @@ jobs: dist/kin-codex-kernel dist/kin-worker dist/kin-egress + dist/kin-slot-gate dist/kin-oauth-auth dist/kin-computer dist/cli-node diff --git a/CHANGELOG.md b/CHANGELOG.md index 1b4ae9510..02cbe677f 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -5,6 +5,7 @@ - 槽位种子页：`settings.json` 的 env 行与 `theme` 可直接编辑并随播种写入；`autoUpdates`、`grove_enabled` 只读且播种时固定为 `false`；TZ/LANG/LC_ALL、遥测 kill-switch、Bedrock/Vertex、gateway hint 等同理只读（播种由契约重写）。缺 `CLAUDE_CODE_ATTRIBUTION_HEADER` 时默认 `0`。调试上传默认关，独立 query 拉最近 40 条上传，切换开关不重置未保存编辑；Go sidecar 写 `vms//run/telemetry-upload.log`，面板 `GET seed-settings?debug=1` 返回 `telemetry_log`。 - 槽位种子页按当前槽 `settings.json` 列出真实配置；`claude.setting` 可编辑播种；缺失文件返回 `null`、空文件返回空串，不用策略默认冒充槽内文件；遥测开关与 sidecar 进程状态分开展示，保存不因遥测进程刷新失败而 502。 - fleet 遥测刷新在 kin-egress 二进制被替换（含 `(deleted)` 旧进程与脏 pidfile）时，会按 egress 配置扫停全部 helper、等待 `/proc/` 消失后再绑定新二进制；同一 SOCKS 出口的 stop/start 与 `ensureProxyEgress` 共用 per-proxy 锁，避免并发 telemetry 互相 SIGTERM。（#604） +- 扩展节点直连通道：`PUT /api/panel/cluster/nodes/:id/direct-link` 在节点上建 `kin-slot-gate`（TLS 1.3，首次启动自签证书，控制面经 SSH 读取并固定指纹；每节点随机令牌），远程槽 kernel / worker 请求改为每次一条直连 TLS 连接，不再全部挤在一条 SSH 长连接里（长连接在零碎流量下拥塞窗口会卡在约 17KB/s，突发请求正文排队数秒）。请求正文随握手后的第一包发出、不等应答。gate 只连槽的两个 socket 且不跟随符号链接；连不上、证书不符或令牌被拒时该节点退回 SSH 30 秒。面板节点详情可启用 / 关闭，节点行显示「直连」/「直连→SSH」。新增 Go 命令 `kin-slot-gate`（`npm run build:gate`，Release 附件同名）。 - 面板：侧栏「GitHub / 文档」两枚链接不再折行，字体偏宽时「文档」不会逐字竖排，空间不够先缩箭头再省略文字；VM 列表搜索框关闭浏览器自动填充，不再被回填成登录邮箱。 - 集群页「扩展节点」新增 CPU / 内存 / 容器三列：健康检查（每 30 秒）经同一条 SSH 连接读节点 `/proc` 和 cgroup，口径同本机宿主列；容器 = 节点上全部 Docker 容器内存合计（扣除 inactive_file，同 `docker stats`）。修复本机宿主列 CPU 最高只显示 10%（`Math.min(100, …×1000)/10` 写错上限）。链路列不再折行。 - 集群槽无需重建 VM 即可更新 Go 槽位服务（kin-worker / kin-egress）：持久化 `slot-bin` 挂载推送二进制，fleet `action=telemetry` 热更新遥测与 egress。（#600） diff --git a/Dockerfile b/Dockerfile index 50d02ea9b..984611584 100644 --- a/Dockerfile +++ b/Dockerfile @@ -20,6 +20,7 @@ RUN cd api-kernel && go mod download COPY api-kernel/ ./api-kernel/ RUN cd worker && go build -buildvcs=false -trimpath -o /out/kin-worker ./cmd/kin-worker \ && go build -buildvcs=false -trimpath -o /out/kin-egress ./cmd/kin-egress \ + && CGO_ENABLED=0 go build -buildvcs=false -trimpath -o /out/kin-slot-gate ./cmd/kin-slot-gate \ && cd ../api-kernel && go build -buildvcs=false -trimpath -o /out/kin-api-kernel ./cmd/kin-api-kernel FROM --platform=$BUILDPLATFORM node:22-bookworm-slim AS payload diff --git a/deploy/Dockerfile.arm64-control b/deploy/Dockerfile.arm64-control index d4bba5339..6e92f83f0 100644 --- a/deploy/Dockerfile.arm64-control +++ b/deploy/Dockerfile.arm64-control @@ -18,6 +18,7 @@ RUN cd worker && go build -buildvcs=false -trimpath -ldflags="-s -w" -o /out/kin && go build -buildvcs=false -trimpath -ldflags="-s -w" -o /out/kin-worker ./cmd/kin-worker \ && GOARCH=amd64 go build -buildvcs=false -trimpath -o /out/amd64/kin-egress ./cmd/kin-egress \ && GOARCH=amd64 go build -buildvcs=false -trimpath -o /out/amd64/kin-worker ./cmd/kin-worker \ + && CGO_ENABLED=0 GOARCH=amd64 go build -buildvcs=false -trimpath -o /out/amd64/kin-slot-gate ./cmd/kin-slot-gate \ && cd ../api-kernel && go build -buildvcs=false -trimpath -ldflags="-s -w" -o /out/kin-api-kernel ./cmd/kin-api-kernel FROM --platform=linux/arm64 ${NODE_IMAGE} AS web @@ -79,6 +80,7 @@ ENV NODE_ENV=production \ KIN_API_KERNEL_BIN=/opt/vm2api/bin/kin-api-kernel \ KIN_CLUSTER_WORKER_BIN=/opt/vm2api/cluster-bin-amd64/kin-worker \ KIN_CLUSTER_EGRESS_BIN=/opt/vm2api/cluster-bin-amd64/kin-egress \ + KIN_CLUSTER_GATE_BIN=/opt/vm2api/cluster-bin-amd64/kin-slot-gate \ DOCKER_DEFAULT_PLATFORM=linux/amd64 EXPOSE 8787 ENTRYPOINT ["/usr/local/bin/vm2api-entrypoint"] diff --git a/docker-compose.arm64.yml b/docker-compose.arm64.yml index c566ddec4..557068c4d 100644 --- a/docker-compose.arm64.yml +++ b/docker-compose.arm64.yml @@ -12,6 +12,7 @@ services: DOCKER_DEFAULT_PLATFORM: linux/amd64 KIN_CLUSTER_WORKER_BIN: /opt/vm2api/cluster-bin-amd64/kin-worker KIN_CLUSTER_EGRESS_BIN: /opt/vm2api/cluster-bin-amd64/kin-egress + KIN_CLUSTER_GATE_BIN: /opt/vm2api/cluster-bin-amd64/kin-slot-gate volumes: # Keep native runtime helpers out of the repository's amd64 bin/ assets. - ./.local/arm64/vms:/opt/vm2api/vms diff --git a/docs/BUILD.md b/docs/BUILD.md index 6c657c201..6aca24467 100644 --- a/docs/BUILD.md +++ b/docs/BUILD.md @@ -41,6 +41,7 @@ git push origin v1.2.22 |---|---| | `kin-kernel` | Claude Code 槽内核（payload，推理必带） | | `kin-egress` | 远程 SOCKS5 透明网关 | +| `kin-slot-gate` | 扩展节点直连通道入口（控制面每请求一条 TLS 连接直达节点槽位） | | `kin-worker` | **只** `telemetry`，不是 hop | | `kin-codex-kernel` | Codex 槽（payload） | | `kin-oauth-auth` | OAuth / Setup Token 换票服务二进制；源码不随部署上传 | @@ -78,6 +79,7 @@ pnpm -C web install --frozen-lockfile npm run fetch:payload # bin/kin-{kernel,codex-kernel,oauth-auth}、share/wrap-cli、share/crag npm run build:egress # bin/kin-egress +npm run build:gate # bin/kin-slot-gate（扩展节点直连通道） npm run build:web # web/dist npm run build:worker # bin/kin-worker（telemetry） ``` @@ -93,6 +95,7 @@ npm run format:check ```bash CGO_ENABLED=0 go build -trimpath -o bin/kin-egress ./worker/cmd/kin-egress +CGO_ENABLED=0 go build -trimpath -o bin/kin-slot-gate ./worker/cmd/kin-slot-gate CGO_ENABLED=0 go build -trimpath -o bin/kin-worker ./worker/cmd/kin-worker pnpm -C web build ``` diff --git a/docs/PANEL_API.md b/docs/PANEL_API.md index dc8d8ed35..36e40ad09 100644 --- a/docs/PANEL_API.md +++ b/docs/PANEL_API.md @@ -508,6 +508,7 @@ sessionKey / 授权码导入必须走该槽 SOCKS5。 | PATCH | `/cluster/nodes/:id` | 编辑名称或 SSH 配置，必须带 `expected_updated_at`（取列表中的 `updated_at`）；返回更新后的公开节点信息 | | POST | `/cluster/nodes/:id/probe` | 与 PATCH 同体；使用已保存的未更换凭证及原主机指纹测试候选配置，不保存、不切断原连接；返回 `{ host_key, docker, vm_count, dependent_count }` | | POST | `/cluster/nodes/:id/reconnect` | 清掉 `error` 与退避，立刻重连 | +| PUT / DELETE | `/cluster/nodes/:id/direct-link` | 直连通道：PUT `{ address, port? }`（节点 IP 字面量，端口默认 7790）启用或修改，DELETE 关闭并删除节点上的 `kin-slot-gate`；返回节点（含 `direct_link`） | | POST | `/cluster/nodes/:id/shell-ticket` | 30 秒一次性终端票据 | | WS | `/cluster/nodes/:id/shell?ticket=&cols=&rows=` | 客户端发 `{"t":"d","d":"…"}` 输入、`{"t":"r","c":列,"r":行}` 改尺寸；服务端二进制帧为终端输出，文本帧 `{"t":"exit"\|"error"}` | | GET | `/cluster/nodes/:id/docker/info` | 版本 / 系统 / 容器数 | @@ -523,6 +524,7 @@ sessionKey / 授权码导入必须走该槽 SOCKS5。 - 仅改名无需 SSH 连接，也不重连。SSH 配置保存时服务端会重新验证，不能以先前 probe 成功代替：先使用原指纹建立候选连接，验证成功再保存并接管原链路；失败保留原配置和连接。已有终端和转发连接可能在切换时断开，远端 VM 容器不会被重建。这里只修改控制面保存的连接配置，不修改远端 sshd、密码或 authorized_keys。 - 省略凭证字段表示保留。切换认证方式必须提供新密码或新私钥，并清理旧方式的凭证；新私钥不继承旧私钥的口令，`passphrase: ""` 表示清空。空密码/空私钥不是“保持不变”，会校验失败；响应不返回凭证。 - 已有 VM 的节点不能修改登录用户（`node_username_in_use`）。候选连接必须仍能访问现有 VM 容器、挂载目录及匹配的 UID/GID（否则 `node_runtime_mismatch`）；不用于将已有节点迁移到另一台 VPS。被其他节点用作跳板、正在安装 Docker、构建镜像、处理调度请求或有 `withVmLock` 操作（如 VM 重置）尚未完成时拒绝修改连接配置，仍可改名。主机密钥轮换、跳板拓扑修改和 VM 迁移不在此接口范围。 +- 直连通道：远程槽的 kernel / worker 请求默认在节点那一条 SSH 长连接里各开一个通道，长时间零碎流量会让这条连接的拥塞窗口卡在很低的值（实测约 17KB/s），突发的请求正文全部排队。`PUT /direct-link` 后控制面经 SSH 在节点上建 `kin-slot-gate`（宿主网络、只读挂载 `/vms`、监听 `0.0.0.0:`），之后每个槽连接是一条独立的 TLS 1.3 连接。gate 首次启动自签证书（`/gate/tls/`），控制面经已固定主机密钥的 SSH 读取并固定其 SHA-256，指纹不符即不发令牌；每节点随机令牌加密存库，令牌与指纹任何响应都不回传。请求正文紧跟前言发出，不等 gate 应答。gate 只连 `/vms//run/{kernel,worker}.sock`，以 `O_PATH|O_NOFOLLOW` 打开，不跟随槽内植入的符号链接。连不上、证书不符或令牌被拒时该节点退回 SSH 30 秒，健康检查 ping 成功即恢复；Docker、终端、健康检查始终走 SSH。**节点防火墙应只对控制面出口 IP 放行该端口**（令牌之外的第二道门）。`direct_link.state`：`ok` 直连、`down` 已退回 SSH（`error` 为原因）、`pending` 尚未检查或尚未固定证书；`gate_error` 为节点端创建失败原因（如节点上还没有槽位镜像）。 - 本机 Docker 桥：每个节点在 `/cluster//docker.sock`（0600）监听，`docker -H unix://…/docker.sock ps` 直接操作远端 daemon。数据目录所在文件系统不支持 unix socket（如 WSL `/mnt/*`）时设 `VM2API_CLUSTER_SOCKET_DIR`。 ## 管理口（master） diff --git a/package.json b/package.json index e7d591d0e..e226872f2 100644 --- a/package.json +++ b/package.json @@ -14,6 +14,7 @@ "test:all": "npm run test:unit && npm run test:go", "build:worker": "mkdir -p bin && cd worker && CGO_ENABLED=0 go build -trimpath -o ../bin/kin-worker ./cmd/kin-worker", "build:egress": "mkdir -p bin && cd worker && CGO_ENABLED=0 go build -trimpath -o ../bin/kin-egress ./cmd/kin-egress", + "build:gate": "mkdir -p bin && cd worker && CGO_ENABLED=0 go build -trimpath -o ../bin/kin-slot-gate ./cmd/kin-slot-gate", "build:web": "pnpm -C web build", "fetch:payload": "node scripts/fetch-payload.mjs", "test:web": "pnpm -C web test", diff --git a/scripts/docker-entrypoint.sh b/scripts/docker-entrypoint.sh index 1b3e41e41..0b936259c 100644 --- a/scripts/docker-entrypoint.sh +++ b/scripts/docker-entrypoint.sh @@ -36,6 +36,7 @@ ensure_bin() { } ensure_bin kin-kernel ensure_bin kin-egress +ensure_bin kin-slot-gate ensure_bin kin-worker ensure_bin kin-api-kernel ensure_bin kin-codex-kernel diff --git a/src/lib/cluster/cluster-manager.mjs b/src/lib/cluster/cluster-manager.mjs index 51484aa94..d566e1838 100644 --- a/src/lib/cluster/cluster-manager.mjs +++ b/src/lib/cluster/cluster-manager.mjs @@ -19,6 +19,14 @@ import { verifyNodeRuntimes, } from './node-edit.mjs' import { NODE_HOST_COMMAND, NODE_HOST_TIMEOUT_MS, parseNodeHost } from './node-host.mjs' +import { + dialGate, + ensureNodeGate, + newGateToken, + parseDirectLinkInput, + pingGate, + removeNodeGate, +} from './direct-link.mjs' import { shellQuote } from './remote-fs.mjs' import { SocketRelay } from './socket-relay.mjs' import { configuredIpv6Enabled } from '../vm/proxy-policy.mjs' @@ -36,6 +44,8 @@ import { export const HEALTH_INTERVAL_MS = 30_000 export const SHELL_TICKET_TTL_MS = 30_000 export const LOCAL_STATUS_TTL_MS = 30_000 +// A failed gate dial sends that node's slot traffic over SSH until the next good health ping or this long. +export const DIRECT_LINK_BACKOFF_MS = 30_000 const INSTALL_LOG_MAX = 64 * 1024 const HOSTNAME_RE = /^(?=.{1,253}$)(?:[a-z0-9](?:[a-z0-9-]{0,61}[a-z0-9])?\.)*[a-z0-9](?:[a-z0-9-]{0,61}[a-z0-9])?$/i @@ -110,7 +120,10 @@ export class ClusterManager { onNodesChanged = null, onLinkState = null, assertNodeEditable = () => {}, + gateBin = process.env.KIN_CLUSTER_GATE_BIN || path.join(process.cwd(), 'bin', 'kin-slot-gate'), }) { + this.gateBin = gateBin + this.directState = new Map() this.listen = listen // Node add / remove: per-node resources (each VPS's local egress row) follow the list. this.onNodesChanged = onNodesChanged @@ -167,6 +180,7 @@ export class ClusterManager { Promise.resolve(this._checkHealth(node.id)) .catch(() => {}) .finally(() => this._linkState(node.id, state)) + this._ensureGate(node.id).catch(() => {}) } else { this.health.delete(node.id) this._linkState(node.id, state) @@ -226,6 +240,7 @@ export class ClusterManager { if (err.code === 'streamlocal_failed') latencyMs = performance.now() - t0 } const hostRead = execCollect(client, NODE_HOST_COMMAND, { timeoutMs: NODE_HOST_TIMEOUT_MS }).catch(() => null) + const directRead = this._probeDirect(id) let docker let containers = null try { @@ -235,6 +250,7 @@ export class ClusterManager { } catch (err) { docker = { ok: false, error: err.message } } + await directRead const hostOut = await hostRead const host = hostOut?.code === 0 ? parseNodeHost(hostOut.stdout) : null if (link.client !== client) return @@ -253,7 +269,7 @@ export class ClusterManager { }) } - _view(node) { + _view(node, directLinks = this.repo.directLinks()) { const link = this.links.get(node.id) return { ...node, @@ -261,11 +277,13 @@ export class ClusterManager { health: this.health.get(node.id) || null, bridge: this.bridges.get(node.id)?.snapshot() || null, install: this.installs.get(node.id) || null, + direct_link: this._directView(node.id, directLinks.get(node.id)), } } list() { - return this.repo.list().map((node) => this._view(node)) + const directLinks = this.repo.directLinks() + return this.repo.list().map((node) => this._view(node, directLinks)) } get(id) { @@ -274,6 +292,121 @@ export class ClusterManager { return this._view(node) } + _directView(id, cfg) { + if (!cfg) return null + const st = this.directState.get(id) || {} + const fallback = (st.downUntil || 0) > Date.now() + return { + address: cfg.address, + port: cfg.port, + state: st.checked_at ? (fallback || st.error ? 'down' : 'ok') : 'pending', + latency_ms: st.latency_ms ?? null, + error: st.error || null, + gate_error: st.gate_error || null, + checked_at: st.checked_at || null, + } + } + + _directPatch(id, patch) { + this.directState.set(id, { ...(this.directState.get(id) || {}), ...patch }) + } + + /** Ping the gate; a good answer lifts the SSH fallback early, a bad one starts it. */ + async _probeDirect(id) { + const cfg = this.repo.getDirectLink(id) + if (!cfg?.certSha256) return + const checked_at = new Date().toISOString() + try { + const ms = await pingGate(cfg) + this._directPatch(id, { latency_ms: Math.round(ms), error: null, downUntil: 0, checked_at }) + } catch (err) { + this._directPatch(id, { + latency_ms: null, + error: err.code || 'gate_unreachable', + downUntil: Date.now() + DIRECT_LINK_BACKOFF_MS, + checked_at, + }) + } + } + + /** + * One slot connection over the direct link (`{ sock, ready }`), or null for + * the SSH path: no link, no pinned certificate yet, in backoff, or the dial + * failed before anything was sent. A refusal after that (bad token) starts + * the backoff too; a missing slot socket does not, it is not a link fault. + */ + async _dialDirect(nodeId, vmId, kind) { + const cfg = this.repo.getDirectLink(nodeId) + if (!cfg?.certSha256) return null + if ((this.directState.get(nodeId)?.downUntil || 0) > Date.now()) return null + const down = (err) => { + this._directPatch(nodeId, { error: err.code, downUntil: Date.now() + DIRECT_LINK_BACKOFF_MS }) + this.logger.warn(`[cluster] direct link ${nodeId} unavailable (${err.code}), using ssh`) + } + try { + const conn = await dialGate({ ...cfg, vmId, kind }) + conn.ready.catch((err) => { + if (err.code !== 'gate_no_socket') down(err) + }) + return conn + } catch (err) { + down(err) + return null + } + } + + async _ensureGate(id, { strict = false } = {}) { + const cfg = this.repo.getDirectLink(id) + const link = this.links.get(id) + if (!cfg || link?.state !== 'ready' || !link.client) return + const client = link.client + try { + const host = await this.remoteHost(id) + const result = await ensureNodeGate({ client, host, docker: sshDocker(client) }, cfg, this.gateBin) + if (result.certSha256 !== cfg.certSha256) { + this.repo.setDirectLinkCert(id, result.certSha256) + this.logger.log(`[cluster] slot gate on ${id}: pinned certificate ${result.certSha256.slice(0, 16)}…`) + } + this._directPatch(id, { gate_error: null }) + if (result.changed) this.logger.log(`[cluster] slot gate on ${id} (re)created from ${result.image}`) + } catch (err) { + this._directPatch(id, { gate_error: err.message }) + this.logger.warn(`[cluster] slot gate on ${id}: ${err.message}`) + if (strict) throw err + } + } + + /** Enable or move the direct link. Token and certificate pin are kept across edits and never returned. */ + async setDirectLink(id, body) { + if (!this.repo.get(id)) throw new ClusterError(404, 'node_not_found', '节点不存在') + const { address, port } = parseDirectLinkInput(body) + const token = this.repo.getDirectLink(id)?.token || newGateToken() + this.repo.setDirectLink(id, { address, port, token }) + this.directState.delete(id) + await this._ensureGate(id).catch(() => {}) + // The gate needs a moment to bind after a (re)create. + for (let i = 0; i < 5; i += 1) { + await this._probeDirect(id) + if (!this.directState.get(id)?.error) break + await new Promise((resolve) => setTimeout(resolve, 400)) + } + return this.get(id) + } + + async clearDirectLink(id) { + if (!this.repo.get(id)) throw new ClusterError(404, 'node_not_found', '节点不存在') + this.repo.clearDirectLink(id) + this.directState.delete(id) + const link = this.links.get(id) + if (link?.state === 'ready' && link.client) { + const client = link.client + await removeNodeGate({ client, host: await this.remoteHost(id), docker: sshDocker(client) }).catch((err) => + this.logger.warn(`[cluster] remove slot gate on ${id}: ${err.message}`), + ) + } + return this.get(id) + } + /** Connect once with the form's credentials, report host key + docker reachability, disconnect. */ async probe(input) { const spec = parseNodeInput(input) @@ -524,6 +657,7 @@ export class ClusterManager { this.bridges.delete(id) this.health.delete(id) this.installs.delete(id) + this.directState.delete(id) this.repo.remove(id) this._nodesChanged() } @@ -588,8 +722,18 @@ export class ClusterManager { const entry = { nodeId, paths, - kernel: new SocketRelay({ socketPath: paths.kernel, remotePath: remote('kernel.sock'), getClient }), - worker: new SocketRelay({ socketPath: paths.worker, remotePath: remote('worker.sock'), getClient }), + kernel: new SocketRelay({ + socketPath: paths.kernel, + remotePath: remote('kernel.sock'), + getClient, + dialDirect: () => this._dialDirect(nodeId, vmId, 'kernel'), + }), + worker: new SocketRelay({ + socketPath: paths.worker, + remotePath: remote('worker.sock'), + getClient, + dialDirect: () => this._dialDirect(nodeId, vmId, 'worker'), + }), } entry.ready = Promise.all([entry.kernel.start(), entry.worker.start()]) this.slotRelays.set(vmId, entry) diff --git a/src/lib/cluster/cluster-routes.mjs b/src/lib/cluster/cluster-routes.mjs index 9bdf734c0..4542ff2cd 100644 --- a/src/lib/cluster/cluster-routes.mjs +++ b/src/lib/cluster/cluster-routes.mjs @@ -77,6 +77,11 @@ export function createClusterRoutes({ manager, json, readBody, ok }) { return json(res, 200, ok({ id })) } if (sub === '/reconnect' && m === 'POST') return json(res, 200, ok(manager.reconnect(id))) + if (sub === '/direct-link' && m === 'PUT') { + const body = await readBody(req, MAX_BODY) + return json(res, 200, ok(await manager.setDirectLink(id, body))) + } + if (sub === '/direct-link' && m === 'DELETE') return json(res, 200, ok(await manager.clearDirectLink(id))) if (sub === '/shell-ticket' && m === 'POST') return json(res, 200, ok(manager.issueShellTicket(id))) if (sub === '/docker/info' && m === 'GET') return json(res, 200, ok(await dockerInfo(manager.docker(id)))) diff --git a/src/lib/cluster/direct-link.mjs b/src/lib/cluster/direct-link.mjs new file mode 100644 index 000000000..7840cb351 --- /dev/null +++ b/src/lib/cluster/direct-link.mjs @@ -0,0 +1,272 @@ +/** + * Cluster direct link: remote slot traffic over one TLS connection per request + * to the node's kin-slot-gate, instead of a channel on the node's long-lived + * SSH connection. + * + * That one SSH flow carries every remote slot request. Bursty request bodies + * behind chatty idle traffic leave its congestion state stuck low (BBR measured + * ~17KB/s on a path that moves 350KB/s on a fresh connection), and the whole + * node queues behind it. Through the gate each request gets a fresh connection + * with its own ramp-up. Docker, terminal and health stay on SSH; a gate that + * cannot be reached falls back to SSH. + * + * Trust: the gate self-signs its certificate on first start; the control plane + * reads it over the authenticated SSH link and pins its SHA-256, so no CA is + * involved. A per-node token, sent only after the pin matched, authorizes the + * connection. The gate reaches nothing but /vms//run/{kernel,worker}.sock. + */ + +import crypto from 'node:crypto' +import fs from 'node:fs' +import net from 'node:net' +import tls from 'node:tls' +import { + containerAction, + createContainerRaw, + dockerJson, + inspectContainerOrNull, + removeContainer, +} from './docker-remote.mjs' +import { openSftp, readRemoteFileNoFollow, runRemote, shellQuote, writeRemoteFile } from './remote-fs.mjs' +import { ClusterError } from './ssh-link.mjs' + +export const GATE_CONTAINER = 'kin-slot-gate' +export const GATE_DEFAULT_PORT = 7790 +export const GATE_DIAL_TIMEOUT_MS = 3_000 +const GATE_MOUNT = '/etc/kin-slot-gate' +const PREFACE = 'KSG1' +const REPLY_MAX = 64 +const SLOT_IMAGE_PREFIX = 'vm2api/kin-slot-' +const CERT_WAIT_TRIES = 20 +const CERT_WAIT_MS = 250 + +export function newGateToken() { + return crypto.randomBytes(32).toString('hex') +} + +/** `{ address, port }` from the panel. An IP literal only: no DNS on the request path. */ +export function parseDirectLinkInput(body = {}) { + const address = String(body.address ?? '').trim() + if (!net.isIP(address)) throw new ClusterError(400, 'invalid_direct_address', '直连地址必须是 IP') + const port = body.port == null || body.port === '' ? GATE_DEFAULT_PORT : Number(body.port) + if (!Number.isInteger(port) || port < 1024 || port > 65535) { + throw new ClusterError(400, 'invalid_direct_port', '直连端口须为 1024–65535') + } + return { address, port } +} + +/** Hex SHA-256 of a certificate's DER, the form stored as the pin. */ +export function certSha256(pem) { + return new crypto.X509Certificate(pem).fingerprint256.replaceAll(':', '').toLowerCase() +} + +function gateError(code, message) { + return Object.assign(new Error(message), { code }) +} + +/** + * Open one gate connection for `vmId`'s `kind` socket ('kernel' | 'worker'), or + * 'ping'. Resolves `{ sock, ready }` as soon as TLS is up, the pinned + * certificate matched and the preface is written, so the caller can send the + * request body without waiting a round trip (the gate buffers it). Rejects with + * gate_unreachable or gate_cert_mismatch: nothing was sent, the caller may use SSH. + * + * `ready` settles when the gate answers: resolves with the socket paused and the + * "OK" line consumed (pipe or resume it), or rejects with gate_no_socket (slot + * socket absent, the meaning of a refused streamlocal), gate_auth, + * gate_bad_request or gate_unreachable. + */ +export function dialGate({ + address, + port, + token, + certSha256: pin, + vmId = '-', + kind, + timeoutMs = GATE_DIAL_TIMEOUT_MS, +}) { + return new Promise((resolveConn, rejectConn) => { + if (!pin) { + rejectConn(gateError('gate_cert_mismatch', 'no pinned gate certificate')) + return + } + const sock = tls.connect({ + host: address, + port, + rejectUnauthorized: false, + minVersion: 'TLSv1.3', + allowHalfOpen: true, + noDelay: true, + keepAlive: true, + keepAliveInitialDelay: 30_000, + }) + let connected = false + let resolveReady + let rejectReady + const ready = new Promise((resolve, reject) => { + resolveReady = resolve + rejectReady = reject + }) + ready.catch(() => {}) + let buf = Buffer.alloc(0) + let settled = false + const fail = (err) => { + if (settled) return + settled = true + clearTimeout(timer) + sock.off('data', onData) + sock.destroy() + if (connected) rejectReady(err) + else rejectConn(err) + } + const timer = setTimeout(() => fail(gateError('gate_unreachable', 'gate timeout')), timeoutMs) + const onData = (chunk) => { + buf = Buffer.concat([buf, chunk]) + const nl = buf.indexOf(10) + if (nl < 0) { + if (buf.length > REPLY_MAX) fail(gateError('gate_bad_reply', 'gate reply too long')) + return + } + const reply = buf.subarray(0, nl).toString('utf8') + if (reply !== 'OK') { + const code = /^ERR ([a-z_]+)$/.exec(reply)?.[1] || 'bad_reply' + fail(gateError(`gate_${code}`, `gate refused: ${code}`)) + return + } + settled = true + clearTimeout(timer) + sock.off('data', onData) + sock.pause() + const rest = buf.subarray(nl + 1) + if (rest.length) sock.unshift(rest) + resolveReady(sock) + } + sock.once('secureConnect', () => { + const presented = sock.getPeerCertificate()?.fingerprint256?.replaceAll(':', '').toLowerCase() + if (presented !== pin) { + fail(gateError('gate_cert_mismatch', 'gate certificate does not match the pinned one')) + return + } + connected = true + sock.on('data', onData) + sock.write(`${PREFACE} ${token} ${vmId} ${kind}\n`) + resolveConn({ sock, ready }) + }) + sock.once('error', (err) => fail(gateError('gate_unreachable', err.message))) + sock.once('close', () => fail(gateError('gate_unreachable', 'gate closed'))) + }) +} + +/** Round trip of one authenticated ping (TCP + TLS + preface), in ms. */ +export async function pingGate(link, timeoutMs = GATE_DIAL_TIMEOUT_MS) { + const t0 = performance.now() + const { sock, ready } = await dialGate({ ...link, kind: 'ping', timeoutMs }) + try { + await ready + } finally { + sock.destroy() + } + return performance.now() - t0 +} + +/** Newest slot image on the node: the gate is a static binary and only needs a userland to start in. */ +async function nodeSlotImage(docker) { + const images = (await dockerJson(docker, { path: '/images/json', timeoutMs: 15_000 })) || [] + const slot = images + .filter((img) => (img.RepoTags || []).some((tag) => tag.startsWith(SLOT_IMAGE_PREFIX))) + .sort((a, b) => (b.Created || 0) - (a.Created || 0))[0] + return slot ? slot.RepoTags.find((tag) => tag.startsWith(SLOT_IMAGE_PREFIX)) : null +} + +/** + * Container body. Root SSH logins run slots as per-slot uids with 0600 sockets, + * so a root gate keeps only DAC_OVERRIDE (what sshd streamlocal had as root); + * a non-root login runs slots as itself and needs no capability. Only tls/ is + * writable, for the certificate the gate creates on first start. + */ +export function gateContainer({ host, imageRef, body, binaryDigest }) { + const user = `${host.uid}:${host.gid}` + const digest = crypto + .createHash('sha256') + .update(body) + .update(imageRef) + .update(user) + .update(binaryDigest) + .digest('hex') + .slice(0, 16) + const vms = `${host.root}/vms` + return { + digest, + spec: { + Image: imageRef, + User: user, + Entrypoint: [`${GATE_MOUNT}/kin-slot-gate`], + Cmd: ['-config', `${GATE_MOUNT}/gate.json`], + Labels: { 'kin.gate': '1', 'kin.gate.cfg': digest, 'vm2api.cluster': '1' }, + HostConfig: { + NetworkMode: 'host', + Binds: [`${host.root}/gate:${GATE_MOUNT}:ro`, `${host.root}/gate/tls:${GATE_MOUNT}/tls`, `${vms}:${vms}:ro`], + RestartPolicy: { Name: 'unless-stopped' }, + ReadonlyRootfs: true, + CapDrop: ['ALL'], + CapAdd: host.uid === 0 ? ['DAC_OVERRIDE'] : [], + SecurityOpt: ['no-new-privileges'], + }, + }, + } +} + +export function gateConfigBody({ host, link }) { + return `${JSON.stringify({ listen: `0.0.0.0:${link.port}`, root: host.root, token: link.token, tls_dir: `${GATE_MOUNT}/tls` }, null, 2)}\n` +} + +async function readGateCert(client, host) { + for (let i = 0; i < CERT_WAIT_TRIES; i += 1) { + const pem = await readRemoteFileNoFollow(client, `${host.root}/gate/tls/cert.pem`, { maxBytes: 16 * 1024 }) + if (pem?.length) return certSha256(pem) + await new Promise((resolve) => setTimeout(resolve, CERT_WAIT_MS)) + } + throw new ClusterError(502, 'gate_cert_missing', '节点上的 kin-slot-gate 没有生成证书，查看该容器日志') +} + +/** + * Create or refresh the node's gate, then read its certificate over SSH for + * pinning. Idempotent: a running gate with the same spec is kept. + */ +export async function ensureNodeGate({ client, host, docker }, link, gateBin) { + let binary + try { + binary = fs.readFileSync(gateBin) + } catch { + throw new ClusterError(500, 'gate_binary_missing', `控制面缺少 ${gateBin}`) + } + const imageRef = await nodeSlotImage(docker) + if (!imageRef) throw new ClusterError(409, 'gate_image_missing', '节点上还没有槽位镜像，先在该节点建一个槽') + const body = gateConfigBody({ host, link }) + const binaryDigest = crypto.createHash('sha256').update(binary).digest('hex') + const { digest, spec } = gateContainer({ host, imageRef, body, binaryDigest }) + const existing = await inspectContainerOrNull(docker, GATE_CONTAINER) + let changed = false + if ( + !(existing?.State?.Running && !existing.State.Restarting && existing.Config?.Labels?.['kin.gate.cfg'] === digest) + ) { + const dir = `${host.root}/gate` + await runRemote( + client, + `umask 077 && mkdir -p ${shellQuote(`${dir}/tls`)} && chmod 700 ${shellQuote(dir)} ${shellQuote(`${dir}/tls`)}`, + ) + const sftp = await openSftp(client) + await writeRemoteFile(sftp, `${dir}/kin-slot-gate`, binary, { mode: 0o755 }) + await writeRemoteFile(sftp, `${dir}/gate.json`, body, { mode: 0o600 }) + if (existing) await removeContainer(docker, GATE_CONTAINER) + await createContainerRaw(docker, GATE_CONTAINER, spec) + await containerAction(docker, GATE_CONTAINER, 'start') + changed = true + } + return { changed, image: imageRef, certSha256: await readGateCert(client, host) } +} + +export async function removeNodeGate({ client, host, docker }) { + if (await inspectContainerOrNull(docker, GATE_CONTAINER)) await removeContainer(docker, GATE_CONTAINER) + await runRemote(client, `rm -rf ${shellQuote(`${host.root}/gate`)}`) +} diff --git a/src/lib/cluster/socket-relay.mjs b/src/lib/cluster/socket-relay.mjs index 3430fbfa3..19264ba2e 100644 --- a/src/lib/cluster/socket-relay.mjs +++ b/src/lib/cluster/socket-relay.mjs @@ -21,12 +21,15 @@ export class SocketRelay { /** * `remotePath` may be an async function: slot relays resolve the node's HOME * on first use, so they can listen at boot before the SSH link is up. - * @param {{ socketPath: string, remotePath: string|(() => Promise), getClient: () => import('ssh2').Client|null, logger?: Pick }} opts + * `dialDirect` (slot relays on a node with a direct link) returns `{ sock, ready }` + * from dialGate, or null to use the SSH link. + * @param {{ socketPath: string, remotePath: string|(() => Promise), getClient: () => import('ssh2').Client|null, dialDirect?: () => Promise<{ sock: import('node:net').Socket, ready: Promise }|null>, logger?: Pick }} opts */ - constructor({ socketPath, remotePath, getClient, logger = console }) { + constructor({ socketPath, remotePath, getClient, dialDirect = null, logger = console }) { this.socketPath = socketPath this.remotePath = remotePath this.getClient = getClient + this.dialDirect = dialDirect this.logger = logger this.server = null this.listening = false @@ -92,6 +95,13 @@ export class SocketRelay { this.active.add(local) local.once('close', () => this.active.delete(local)) local.on('error', () => {}) + if (this.dialDirect) { + const direct = await this.dialDirect().catch(() => null) + if (direct) { + this._pipeDirect(local, direct) + return + } + } const client = this.getClient() if (!client) { local.destroy() @@ -119,6 +129,29 @@ export class SocketRelay { remote.pipe(local) } + /** + * The request body goes out at once (the gate buffers it behind the preface); + * the reply is piped back only after the gate's OK. A refusal at that point + * (missing slot socket) drops the local connection, like a refused streamlocal. + */ + _pipeDirect(local, { sock, ready }) { + if (local.destroyed) { + sock.destroy() + return + } + sock.on('error', () => local.destroy()) + local.on('close', () => sock.destroy()) + // pipe() forwards each half-close as end(); see the half-open note in _listen. + local.pipe(sock) + ready.then( + () => { + sock.pipe(local) + sock.resume() + }, + () => local.destroy(), + ) + } + async stop() { const server = this.server this.server = null diff --git a/src/lib/db/migrations/039_cluster_node_direct_links.sql b/src/lib/db/migrations/039_cluster_node_direct_links.sql new file mode 100644 index 000000000..0dcc20cd7 --- /dev/null +++ b/src/lib/db/migrations/039_cluster_node_direct_links.sql @@ -0,0 +1,15 @@ +-- 039_cluster_node_direct_links — optional direct TLS path to a node's slot gate. +-- +-- When set, remote slot traffic dials kin-slot-gate at address:port with one +-- TLS connection per request instead of a channel on the long-lived SSH link. +-- cert_sha256 pins the gate's self-signed certificate (read over SSH); token +-- authorizes the connection and goes through maybeEncrypt. + +CREATE TABLE IF NOT EXISTS cluster_node_direct_links ( + node_id TEXT PRIMARY KEY REFERENCES cluster_nodes(id) ON DELETE CASCADE, + address TEXT NOT NULL, + port INTEGER NOT NULL, + token TEXT NOT NULL, + cert_sha256 TEXT, + updated_at TEXT NOT NULL +); diff --git a/src/lib/db/repos/cluster-nodes-repo.mjs b/src/lib/db/repos/cluster-nodes-repo.mjs index 77ff1cb12..9167e3b45 100644 --- a/src/lib/db/repos/cluster-nodes-repo.mjs +++ b/src/lib/db/repos/cluster-nodes-repo.mjs @@ -52,6 +52,7 @@ export class ClusterNodesRepo { RETURNING ${PUBLIC_COLUMNS.join(', ')} `) this._remove = db.prepare('DELETE FROM cluster_nodes WHERE id = ?') + this._stmts = new Map() } list() { @@ -132,4 +133,46 @@ export class ClusterNodesRepo { remove(id) { return this._remove.run(id).changes > 0 } + + // Prepared on first use: callers that only load 026 (older fixtures) never touch this table. + _stmt(sql) { + let stmt = this._stmts.get(sql) + if (!stmt) { + stmt = this.db.prepare(sql) + this._stmts.set(sql, stmt) + } + return stmt + } + + /** node_id → { address, port }; the gate token stays out of listings. */ + directLinks() { + const rows = this._stmt('SELECT node_id, address, port FROM cluster_node_direct_links').all() + return new Map(rows.map((row) => [row.node_id, { address: row.address, port: row.port }])) + } + + getDirectLink(id) { + const row = this._stmt( + 'SELECT address, port, token, cert_sha256 FROM cluster_node_direct_links WHERE node_id = ?', + ).get(id) + return row + ? { address: row.address, port: row.port, token: maybeDecrypt(row.token), certSha256: row.cert_sha256 || null } + : null + } + + setDirectLink(id, { address, port, token }, now = new Date().toISOString()) { + this._stmt(` + INSERT INTO cluster_node_direct_links (node_id, address, port, token, updated_at) VALUES (?, ?, ?, ?, ?) + ON CONFLICT(node_id) DO UPDATE SET address = excluded.address, port = excluded.port, + token = excluded.token, updated_at = excluded.updated_at + `).run(id, address, port, maybeEncrypt(token), now) + } + + /** The pin follows the node's gate, not the address: an address edit keeps it. */ + setDirectLinkCert(id, certSha256) { + this._stmt('UPDATE cluster_node_direct_links SET cert_sha256 = ? WHERE node_id = ?').run(certSha256, id) + } + + clearDirectLink(id) { + return this._stmt('DELETE FROM cluster_node_direct_links WHERE node_id = ?').run(id).changes > 0 + } } diff --git a/src/server.mjs b/src/server.mjs index 2eab606bc..dff624191 100644 --- a/src/server.mjs +++ b/src/server.mjs @@ -897,6 +897,7 @@ function syncNodeLocalEgress() { const clusterManager = new ClusterManager({ repo: new ClusterNodesRepo(), dataDir, + gateBin: process.env.KIN_CLUSTER_GATE_BIN || path.join(cfg.paths.project, 'bin', 'kin-slot-gate'), listen: { host: cfg.host, port: cfg.port }, vmsOnNode: (nodeId) => listVms(cfg.paths.project).filter((vm) => vm.node_id === nodeId), assertNodeEditable: (nodeId) => { diff --git a/test/unit/cluster-direct-link.test.mjs b/test/unit/cluster-direct-link.test.mjs new file mode 100644 index 000000000..3ff849c12 --- /dev/null +++ b/test/unit/cluster-direct-link.test.mjs @@ -0,0 +1,298 @@ +import test from 'node:test' +import assert from 'node:assert/strict' +import { execFileSync } from 'node:child_process' +import fs from 'node:fs' +import net from 'node:net' +import os from 'node:os' +import path from 'node:path' +import tls from 'node:tls' +import { DatabaseSync } from 'node:sqlite' +import { ClusterManager, DIRECT_LINK_BACKOFF_MS } from '../../src/lib/cluster/cluster-manager.mjs' +import { + certSha256, + dialGate, + gateConfigBody, + gateContainer, + parseDirectLinkInput, +} from '../../src/lib/cluster/direct-link.mjs' +import { SocketRelay } from '../../src/lib/cluster/socket-relay.mjs' +import { ClusterNodesRepo } from '../../src/lib/db/repos/cluster-nodes-repo.mjs' + +const TOKEN = 'a'.repeat(64) + +function shortDir(t) { + const dir = fs.mkdtempSync(path.join('/tmp', 'kdl-')) + t.after(() => fs.rmSync(dir, { recursive: true, force: true })) + return dir +} + +function selfSigned(t) { + const dir = shortDir(t) + const key = path.join(dir, 'key.pem') + const cert = path.join(dir, 'cert.pem') + execFileSync( + 'openssl', + [ + 'req', + '-x509', + '-newkey', + 'rsa:2048', + '-nodes', + '-keyout', + key, + '-out', + cert, + '-days', + '1', + '-subj', + '/CN=kin-slot-gate', + ], + { stdio: 'ignore' }, + ) + return { key: fs.readFileSync(key), cert: fs.readFileSync(cert) } +} + +/** + * In-process double of kin-slot-gate: TLS, same preface and replies. Bytes that + * follow the preface are the request body (sent before OK), and a slot answers + * with the upper-cased body once the client half-closes. + */ +async function fakeGate(t, { token = TOKEN, slots = new Set(['vm-1']), extra = '' } = {}) { + const pair = selfSigned(t) + const seen = [] + const held = new Set() + const server = tls.createServer({ ...pair, minVersion: 'TLSv1.3', allowHalfOpen: true }, (sock) => { + held.add(sock) + sock.on('close', () => held.delete(sock)) + let buf = Buffer.alloc(0) + const onData = (chunk) => { + buf = Buffer.concat([buf, chunk]) + const nl = buf.indexOf(10) + if (nl < 0) return + sock.off('data', onData) + const [magic, tok, vm, kind] = buf.subarray(0, nl).toString().split(' ') + const parts = [buf.subarray(nl + 1)] + seen.push({ vm, kind, early: parts[0].toString() }) + if (magic !== 'KSG1' || tok !== token) return sock.end('ERR auth\n') + if (kind === 'ping') return sock.end('OK\n') + if (!slots.has(vm)) return sock.end('ERR no_socket\n') + sock.write(`OK\n${extra}`) + sock.on('data', (c) => parts.push(c)) + sock.on('end', () => sock.end(Buffer.concat(parts).toString().toUpperCase())) + } + sock.on('data', onData) + sock.on('error', () => {}) + }) + await new Promise((resolve) => server.listen(0, '127.0.0.1', resolve)) + t.after(() => { + for (const s of held) s.destroy() + server.close() + }) + return { address: '127.0.0.1', port: server.address().port, token, certSha256: certSha256(pair.cert), seen } +} + +function readAll(sock) { + return new Promise((resolve, reject) => { + const parts = [] + sock.on('data', (c) => parts.push(c)) + sock.once('end', () => resolve(Buffer.concat(parts).toString('utf8'))) + sock.once('error', reject) + sock.resume() + }) +} + +test('parseDirectLinkInput takes an IP literal and a non-privileged port', () => { + assert.deepEqual(parseDirectLinkInput({ address: ' 203.0.113.7 ' }), { address: '203.0.113.7', port: 7790 }) + assert.deepEqual(parseDirectLinkInput({ address: '2001:db8::2', port: '7800' }), { + address: '2001:db8::2', + port: 7800, + }) + for (const body of [ + { address: 'node.example' }, + { address: '10.0.0.1', port: 22 }, + { address: '10.0.0.1', port: 'x' }, + ]) { + assert.throws(() => parseDirectLinkInput(body), /直连/) + } +}) + +test('dialGate sends the body before OK and strips OK from the reply', async (t) => { + const gate = await fakeGate(t, { extra: 'early:' }) + const { sock, ready } = await dialGate({ ...gate, vmId: 'vm-1', kind: 'kernel' }) + sock.end('hello') + await ready + assert.equal(await readAll(sock), 'early:HELLO') + assert.equal(gate.seen[0].vm, 'vm-1') +}) + +test('dialGate refuses a gate whose certificate is not the pinned one, before sending anything', async (t) => { + const gate = await fakeGate(t) + await assert.rejects(dialGate({ ...gate, certSha256: '0'.repeat(64), vmId: 'vm-1', kind: 'kernel' }), { + code: 'gate_cert_mismatch', + }) + await assert.rejects(dialGate({ ...gate, certSha256: null, kind: 'ping' }), { code: 'gate_cert_mismatch' }) + await new Promise((resolve) => setTimeout(resolve, 50)) + assert.deepEqual(gate.seen, [], 'no preface (token) reached a mismatched gate') +}) + +test('dialGate reports refusals through ready and silence as unreachable', async (t) => { + const gate = await fakeGate(t) + const missing = await dialGate({ ...gate, vmId: 'vm-9', kind: 'worker' }) + await assert.rejects(missing.ready, { code: 'gate_no_socket' }) + const badToken = await dialGate({ ...gate, token: 'b'.repeat(64), vmId: 'vm-1', kind: 'worker' }) + await assert.rejects(badToken.ready, { code: 'gate_auth' }) + const held = [] + const silent = net.createServer((sock) => held.push(sock)) + await new Promise((resolve) => silent.listen(0, '127.0.0.1', resolve)) + t.after(() => { + for (const sock of held) sock.destroy() + silent.close() + }) + await assert.rejects(dialGate({ ...gate, port: silent.address().port, kind: 'ping', timeoutMs: 150 }), { + code: 'gate_unreachable', + }) +}) + +test('gate container: root keeps only DAC_OVERRIDE, only tls/ is writable, spec follows the config', () => { + const root = { uid: 0, gid: 0, root: '/root/.vm2api' } + const body = gateConfigBody({ host: root, link: { port: 7790, token: 't' } }) + assert.deepEqual(JSON.parse(body), { + listen: '0.0.0.0:7790', + root: '/root/.vm2api', + token: 't', + tls_dir: '/etc/kin-slot-gate/tls', + }) + const a = gateContainer({ host: root, imageRef: 'vm2api/kin-slot-x:1', body, binaryDigest: 'd' }) + assert.deepEqual(a.spec.HostConfig.CapDrop, ['ALL']) + assert.deepEqual(a.spec.HostConfig.CapAdd, ['DAC_OVERRIDE']) + assert.equal(a.spec.HostConfig.ReadonlyRootfs, true) + assert.deepEqual(a.spec.HostConfig.Binds, [ + '/root/.vm2api/gate:/etc/kin-slot-gate:ro', + '/root/.vm2api/gate/tls:/etc/kin-slot-gate/tls', + '/root/.vm2api/vms:/root/.vm2api/vms:ro', + ]) + const user = gateContainer({ + host: { uid: 1000, gid: 1000, root: '/home/u/.vm2api' }, + imageRef: 'i', + body, + binaryDigest: 'd', + }) + assert.deepEqual(user.spec.HostConfig.CapAdd, []) + assert.notEqual(gateContainer({ host: root, imageRef: 'i', body: 'other', binaryDigest: 'd' }).digest, a.digest) +}) + +test('SocketRelay: direct dial, SSH fallback on null, local drop when the slot socket is missing', async (t) => { + const gate = await fakeGate(t) + const dir = shortDir(t) + const sshAsked = [] + const mk = (name, dialDirect) => { + const relay = new SocketRelay({ + socketPath: path.join(dir, name), + remotePath: '/r', + getClient: () => { + sshAsked.push(name) + return null + }, + dialDirect, + }) + t.after(() => relay.stop()) + return relay + } + const direct = mk('d.sock', () => dialGate({ ...gate, vmId: 'vm-1', kind: 'worker' })) + const fallback = mk('f.sock', async () => null) + const missing = mk('m.sock', () => dialGate({ ...gate, vmId: 'vm-2', kind: 'worker' })) + await Promise.all([direct.start(), fallback.start(), missing.start()]) + + const c = net.connect(direct.socketPath) + const reply = readAll(c) + c.end('over the gate') + assert.equal(await reply, 'OVER THE GATE') + + for (const relay of [fallback, missing]) { + const s = net.connect(relay.socketPath) + s.on('error', () => {}) + s.end('x') + await new Promise((resolve) => s.once('close', resolve)) + } + assert.deepEqual(sshAsked, ['f.sock']) +}) + +function managerWithRepo(t) { + const db = new DatabaseSync(':memory:') + for (const name of ['026_cluster_nodes.sql', '039_cluster_node_direct_links.sql']) { + db.exec(fs.readFileSync(new URL(`../../src/lib/db/migrations/${name}`, import.meta.url), 'utf8')) + } + db.exec('PRAGMA foreign_keys = ON') + t.after(() => db.close()) + const repo = new ClusterNodesRepo(db) + repo.insert({ + id: 'node-a', + label: 'A', + host: '203.0.113.5', + port: 22, + username: 'root', + auth_type: 'password', + password: 'pw', + host_key_alg: 'ssh-ed25519', + host_key_sha256: 'SHA256:x', + }) + const warnings = [] + const manager = new ClusterManager({ + repo, + dataDir: os.tmpdir(), + logger: { log() {}, warn: (m) => warnings.push(m) }, + }) + return { db, repo, manager, warnings } +} + +test('direct link: token and pin survive edits and never leave; failures fall back to SSH with backoff', async (t) => { + const { db, repo, manager, warnings } = managerWithRepo(t) + const gate = await fakeGate(t) + + await manager.setDirectLink('node-a', { address: '127.0.0.1', port: 1025 }) + const token = repo.getDirectLink('node-a').token + assert.equal(await manager._dialDirect('node-a', 'vm-1', 'kernel'), null, 'no pin yet: SSH') + assert.equal(manager.get('node-a').direct_link.state, 'pending') + + repo.setDirectLink('node-a', { address: gate.address, port: gate.port, token: TOKEN }) + repo.setDirectLinkCert('node-a', gate.certSha256) + await manager.setDirectLink('node-a', { address: gate.address, port: gate.port }) + assert.equal(repo.getDirectLink('node-a').token, TOKEN, 'edit keeps the token') + assert.notEqual(token, TOKEN) + assert.equal(repo.getDirectLink('node-a').certSha256, gate.certSha256, 'edit keeps the pin') + assert.equal(manager.get('node-a').direct_link.state, 'ok') + const listed = JSON.stringify(manager.list()) + assert.equal(listed.includes(TOKEN), false) + assert.equal(listed.includes(gate.certSha256), false) + + const conn = await manager._dialDirect('node-a', 'vm-1', 'kernel') + conn.sock.end('ok') + await conn.ready + conn.sock.destroy() + const missing = await manager._dialDirect('node-a', 'vm-2', 'kernel') + await assert.rejects(missing.ready, { code: 'gate_no_socket' }) + assert.equal(manager.get('node-a').direct_link.state, 'ok', 'a missing slot socket is not a link fault') + + repo.setDirectLink('node-a', { address: gate.address, port: gate.port, token: 'c'.repeat(64) }) + const refused = await manager._dialDirect('node-a', 'vm-1', 'kernel') + await assert.rejects(refused.ready, { code: 'gate_auth' }) + assert.match(warnings.at(-1), /gate_auth/) + repo.setDirectLink('node-a', { address: gate.address, port: gate.port, token: TOKEN }) + const before = gate.seen.length + assert.equal(await manager._dialDirect('node-a', 'vm-1', 'kernel'), null, 'still in backoff') + assert.equal(gate.seen.length, before) + assert.equal(manager.get('node-a').direct_link.state, 'down') + assert.ok(manager.directState.get('node-a').downUntil <= Date.now() + DIRECT_LINK_BACKOFF_MS) + await manager._probeDirect('node-a') + assert.equal(manager.get('node-a').direct_link.state, 'ok', 'a good ping lifts the backoff') + + repo.setDirectLink('node-a', { address: '127.0.0.1', port: 1, token: TOKEN }) + assert.equal(await manager._dialDirect('node-a', 'vm-1', 'kernel'), null, 'unreachable: SSH') + assert.match(warnings.at(-1), /gate_unreachable/) + + await manager.clearDirectLink('node-a') + assert.equal(manager.get('node-a').direct_link, null) + repo.setDirectLink('node-a', { address: gate.address, port: gate.port, token: TOKEN }) + db.prepare('DELETE FROM cluster_nodes WHERE id = ?').run('node-a') + assert.equal(repo.getDirectLink('node-a'), null, 'removing the node drops its link') +}) diff --git a/test/unit/cluster-node-edit.test.mjs b/test/unit/cluster-node-edit.test.mjs index 008e05cad..3a1af006a 100644 --- a/test/unit/cluster-node-edit.test.mjs +++ b/test/unit/cluster-node-edit.test.mjs @@ -11,7 +11,9 @@ import { dockerPing, sshDocker } from '../../src/lib/cluster/docker-remote.mjs' import { ClusterNodesRepo } from '../../src/lib/db/repos/cluster-nodes-repo.mjs' const { Server, utils } = ssh2 -const schema = fs.readFileSync(new URL('../../src/lib/db/migrations/026_cluster_nodes.sql', import.meta.url), 'utf8') +const schema = ['026_cluster_nodes.sql', '039_cluster_node_direct_links.sql'] + .map((name) => fs.readFileSync(new URL(`../../src/lib/db/migrations/${name}`, import.meta.url), 'utf8')) + .join('\n') const key = () => utils.generateKeyPairSync('ecdsa', { bits: 256 }) const defer = () => { let resolve diff --git a/web/src/features/cluster/direct-link-config.tsx b/web/src/features/cluster/direct-link-config.tsx new file mode 100644 index 000000000..b8032b867 --- /dev/null +++ b/web/src/features/cluster/direct-link-config.tsx @@ -0,0 +1,118 @@ +import { useState } from 'react' +import { useMutation, useQueryClient } from '@tanstack/react-query' +import type { ClusterApiNode } from '@/types/panel-cluster' +import { toast } from 'sonner' +import { api } from '@/lib/api' +import { Button } from '@/components/ui/button' +import { Input } from '@/components/ui/input' +import { Label } from '@/components/ui/label' +import { directLinkText } from './model' +import { CLUSTER_NODES_KEY } from './queries' + +const DEFAULT_PORT = 7790 +// 只用来预填：SSH 地址本身是 IP 时直接沿用，主机名留空让人填（服务端只收 IP）。 +const looksLikeIp = (s: string) => + /^\d{1,3}(\.\d{1,3}){3}$/.test(s) || s.includes(':') + +/** 槽位流量经 TLS 直连节点上的 kin-slot-gate，每个请求一条 TCP；不通时自动走 SSH。 */ +export function DirectLinkConfig({ node }: { node: ClusterApiNode }) { + const qc = useQueryClient() + const link = node.direct_link || null + const [address, setAddress] = useState( + link?.address || (looksLikeIp(node.host) ? node.host : '') + ) + const [port, setPort] = useState(link ? String(link.port) : '') + const path = `/api/panel/cluster/nodes/${encodeURIComponent(node.id)}/direct-link` + const done = () => qc.invalidateQueries({ queryKey: CLUSTER_NODES_KEY }) + const save = useMutation({ + mutationFn: () => + api(path, { + method: 'PUT', + body: JSON.stringify({ + address: address.trim(), + port: port.trim() || DEFAULT_PORT, + }), + }), + onSuccess: (saved) => { + const next = saved.direct_link + if (next?.state === 'ok') toast.success('直连通道已生效') + else toast.warning(`已保存，${directLinkText(next)}`) + done() + }, + onError: (err) => toast.error(err.message), + }) + const clear = useMutation({ + mutationFn: () => api(path, { method: 'DELETE' }), + onSuccess: () => { + toast.success('已关闭直连通道，槽位流量改走 SSH') + setAddress('') + setPort('') + done() + }, + onError: (err) => toast.error(err.message), + }) + const busy = save.isPending || clear.isPending + return ( +
+
+

直连通道

+ + {directLinkText(link)} + +
+

+ 槽位请求改为每次一条 TLS 直连，不再挤在 SSH 长连接里；直连不通时自动退回 + SSH。节点上会起 kin-slot-gate 监听该端口，请在节点防火墙只对控制面的出口 + IP 放行。 +

+ {link?.gate_error ? ( +

节点端：{link.gate_error}

+ ) : null} +
+
+ + setAddress(e.target.value)} + /> +
+
+ + setPort(e.target.value)} + /> +
+
+
+ + {link ? ( + + ) : null} +
+
+ ) +} diff --git a/web/src/features/cluster/model.ts b/web/src/features/cluster/model.ts index 1f3851631..995666032 100644 --- a/web/src/features/cluster/model.ts +++ b/web/src/features/cluster/model.ts @@ -179,3 +179,13 @@ export function remoteNodeFromApi( remote: node, } } + +/** 直连通道状态一句话；面板与节点行共用。 */ +export function directLinkText(link: ClusterApiNode['direct_link']): string { + if (!link) return '未启用' + if (link.state === 'pending') return '等待检查' + if (link.state === 'ok') { + return `直连中${link.latency_ms != null ? ` · ${link.latency_ms}ms` : ''}` + } + return `已退回 SSH${link.error ? `（${link.error}）` : ''}` +} diff --git a/web/src/features/cluster/remote-list.tsx b/web/src/features/cluster/remote-list.tsx index 4c4bd0ac9..e25da644e 100644 --- a/web/src/features/cluster/remote-list.tsx +++ b/web/src/features/cluster/remote-list.tsx @@ -9,6 +9,7 @@ import { LINK_LATENCY_WARN_MS, LINK_STATE_TEXT, LINK_TONE, + directLinkText, type ClusterNode, } from '@/features/cluster/model' import { PanelCard } from '@/features/overview/panel-card' @@ -118,6 +119,21 @@ export function RemoteList({ {LINK_STATE_TEXT[node.remote.link.state]} ) : null} + {live && node.remote?.direct_link ? ( + + {node.remote.direct_link.state === 'ok' + ? '直连' + : '直连→SSH'} + + ) : null} {lat ? ( n.jump_node_id === node.id).length} /> + +
本机 Docker 桥
{bridge?.listening ? ( diff --git a/web/src/types/panel-cluster.ts b/web/src/types/panel-cluster.ts index b496b17dd..1c9386687 100644 --- a/web/src/types/panel-cluster.ts +++ b/web/src/types/panel-cluster.ts @@ -39,6 +39,19 @@ export type ClusterApiNode = { active: number } | null install: DockerInstallJob | null + /** 直连通道（kin-slot-gate，TLS）；未配置为 null。令牌与证书指纹从不返回。 */ + direct_link?: ClusterDirectLink | null +} + +export type ClusterDirectLink = { + address: string + port: number + /** ok 直连；down 已退回 SSH；pending 尚未检查。 */ + state: 'ok' | 'down' | 'pending' + latency_ms: number | null + error: string | null + gate_error: string | null + checked_at: string | null } export type NodeHostStats = { diff --git a/worker/cmd/kin-slot-gate/main.go b/worker/cmd/kin-slot-gate/main.go new file mode 100644 index 000000000..7cbfcdcbb --- /dev/null +++ b/worker/cmd/kin-slot-gate/main.go @@ -0,0 +1,36 @@ +package main + +import ( + "context" + "encoding/json" + "flag" + "log" + "os" + "os/signal" + "syscall" + + "github.com/dofastted/kin-gateway/worker/internal/slotgate" +) + +func main() { + configPath := flag.String("config", "", "path to gate JSON config") + flag.Parse() + data, err := os.ReadFile(*configPath) + if err != nil { + log.Fatalf("read gate config: %v", err) + } + var cfg slotgate.Config + if err = json.Unmarshal(data, &cfg); err != nil { + log.Fatalf("decode gate config: %v", err) + } + srv, err := slotgate.New(cfg) + if err != nil { + log.Fatalf("kin-slot-gate: %v", err) + } + ctx, stop := signal.NotifyContext(context.Background(), syscall.SIGINT, syscall.SIGTERM) + defer stop() + log.Printf("kin-slot-gate listening on %s", cfg.Listen) + if err = srv.Serve(ctx); err != nil { + log.Fatalf("kin-slot-gate serve: %v", err) + } +} diff --git a/worker/internal/slotgate/cert.go b/worker/internal/slotgate/cert.go new file mode 100644 index 000000000..ab1c4fe28 --- /dev/null +++ b/worker/internal/slotgate/cert.go @@ -0,0 +1,78 @@ +package slotgate + +import ( + "crypto/ecdsa" + "crypto/elliptic" + "crypto/rand" + "crypto/tls" + "crypto/x509" + "crypto/x509/pkix" + "encoding/pem" + "math/big" + "os" + "path/filepath" + "time" +) + +// loadOrCreateCert keeps one self-signed certificate per node across restarts, +// so the control plane's pinned fingerprint stays valid. A missing or broken +// pair is replaced; the control plane re-reads and re-pins it over SSH. +func loadOrCreateCert(dir string) (tls.Certificate, error) { + keyPath, certPath := filepath.Join(dir, "key.pem"), filepath.Join(dir, "cert.pem") + if cert, err := tls.LoadX509KeyPair(certPath, keyPath); err == nil { + return cert, nil + } + key, err := ecdsa.GenerateKey(elliptic.P256(), rand.Reader) + if err != nil { + return tls.Certificate{}, err + } + serial, err := rand.Int(rand.Reader, new(big.Int).Lsh(big.NewInt(1), 128)) + if err != nil { + return tls.Certificate{}, err + } + now := time.Now() + tmpl := &x509.Certificate{ + SerialNumber: serial, + Subject: pkix.Name{CommonName: "kin-slot-gate"}, + NotBefore: now.Add(-time.Hour), + NotAfter: now.AddDate(20, 0, 0), + KeyUsage: x509.KeyUsageDigitalSignature, + ExtKeyUsage: []x509.ExtKeyUsage{x509.ExtKeyUsageServerAuth}, + } + der, err := x509.CreateCertificate(rand.Reader, tmpl, tmpl, &key.PublicKey, key) + if err != nil { + return tls.Certificate{}, err + } + keyDER, err := x509.MarshalPKCS8PrivateKey(key) + if err != nil { + return tls.Certificate{}, err + } + if err = os.MkdirAll(dir, 0o700); err != nil { + return tls.Certificate{}, err + } + if err = writeAtomic(keyPath, pem.EncodeToMemory(&pem.Block{Type: "PRIVATE KEY", Bytes: keyDER}), 0o600); err != nil { + return tls.Certificate{}, err + } + if err = writeAtomic(certPath, pem.EncodeToMemory(&pem.Block{Type: "CERTIFICATE", Bytes: der}), 0o644); err != nil { + return tls.Certificate{}, err + } + return tls.LoadX509KeyPair(certPath, keyPath) +} + +func writeAtomic(path string, data []byte, mode os.FileMode) error { + tmp, err := os.CreateTemp(filepath.Dir(path), ".tmp-*") + if err != nil { + return err + } + defer os.Remove(tmp.Name()) + if err = tmp.Chmod(mode); err == nil { + _, err = tmp.Write(data) + } + if cerr := tmp.Close(); err == nil { + err = cerr + } + if err != nil { + return err + } + return os.Rename(tmp.Name(), path) +} diff --git a/worker/internal/slotgate/dial_linux.go b/worker/internal/slotgate/dial_linux.go new file mode 100644 index 000000000..ef46c0922 --- /dev/null +++ b/worker/internal/slotgate/dial_linux.go @@ -0,0 +1,31 @@ +//go:build linux + +package slotgate + +import ( + "fmt" + "net" + "syscall" +) + +// oPath is O_PATH on every Linux arch Go supports; package syscall does not export it. +const oPath = 0x200000 + +// dialSlotSocket never follows a planted symlink: the guest can create entries +// in run/, so the final component is opened O_PATH|O_NOFOLLOW, checked to be a +// socket, and connected through its /proc fd with no second path lookup. +func dialSlotSocket(path string) (net.Conn, error) { + fd, err := syscall.Open(path, oPath|syscall.O_NOFOLLOW|syscall.O_CLOEXEC, 0) + if err != nil { + return nil, err + } + defer syscall.Close(fd) + var st syscall.Stat_t + if err = syscall.Fstat(fd, &st); err != nil { + return nil, err + } + if st.Mode&syscall.S_IFMT != syscall.S_IFSOCK { + return nil, fmt.Errorf("%s is not a socket", path) + } + return net.Dial("unix", fmt.Sprintf("/proc/self/fd/%d", fd)) +} diff --git a/worker/internal/slotgate/dial_other.go b/worker/internal/slotgate/dial_other.go new file mode 100644 index 000000000..8f3fe9734 --- /dev/null +++ b/worker/internal/slotgate/dial_other.go @@ -0,0 +1,21 @@ +//go:build !linux + +package slotgate + +import ( + "fmt" + "net" + "os" +) + +// Nodes are Linux; this keeps the package testable elsewhere (lstat, then dial). +func dialSlotSocket(path string) (net.Conn, error) { + st, err := os.Lstat(path) + if err != nil { + return nil, err + } + if st.Mode()&os.ModeSocket == 0 { + return nil, fmt.Errorf("%s is not a socket", path) + } + return net.Dial("unix", path) +} diff --git a/worker/internal/slotgate/slotgate.go b/worker/internal/slotgate/slotgate.go new file mode 100644 index 000000000..0f11bae23 --- /dev/null +++ b/worker/internal/slotgate/slotgate.go @@ -0,0 +1,203 @@ +// Package slotgate is the node side of the cluster direct link: one TLS +// listener that the control plane dials once per request instead of opening a +// channel on its long-lived SSH link. Each connection names a slot and one of +// its two sockets; the gate splices it to /vms//run/.sock and +// nothing else. The certificate is self-signed on first start; the control +// plane pins its fingerprint after reading it over SSH. +package slotgate + +import ( + "bufio" + "context" + "crypto/sha256" + "crypto/subtle" + "crypto/tls" + "errors" + "fmt" + "io" + "log" + "net" + "path/filepath" + "regexp" + "strings" + "sync" + "time" +) + +type Config struct { + Listen string `json:"listen"` + Root string `json:"root"` + Token string `json:"token"` + // TLSDir holds key.pem / cert.pem, created on first start. + TLSDir string `json:"tls_dir"` +} + +// Preface: "KSG1 \n"; reply "OK\n" or "ERR \n". +const ( + Magic = "KSG1" + prefaceMax = 512 + prefaceTimeout = 5 * time.Second + keepAlive = 30 * time.Second + MaxConns = 512 +) + +var ( + vmIDRe = regexp.MustCompile(`^[A-Za-z0-9_-]{1,128}$`) + kinds = map[string]string{"kernel": "kernel.sock", "worker": "worker.sock"} +) + +type Server struct { + cfg Config + token [sha256.Size]byte + tls *tls.Config + slots chan struct{} + wg sync.WaitGroup +} + +func New(cfg Config) (*Server, error) { + if strings.TrimSpace(cfg.Listen) == "" { + return nil, errors.New("listen required") + } + if !filepath.IsAbs(cfg.Root) { + return nil, errors.New("root must be absolute") + } + if len(cfg.Token) < 32 || strings.ContainsAny(cfg.Token, " \t\r\n") { + return nil, errors.New("token must be at least 32 non-space characters") + } + if !filepath.IsAbs(cfg.TLSDir) { + return nil, errors.New("tls_dir must be absolute") + } + cert, err := loadOrCreateCert(cfg.TLSDir) + if err != nil { + return nil, fmt.Errorf("tls: %w", err) + } + return &Server{ + cfg: cfg, + token: sha256.Sum256([]byte(cfg.Token)), + tls: &tls.Config{Certificates: []tls.Certificate{cert}, MinVersion: tls.VersionTLS13}, + slots: make(chan struct{}, MaxConns), + }, nil +} + +// TLSConfig is the server side config, for callers that bring their own listener. +func (s *Server) TLSConfig() *tls.Config { return s.tls } + +func (s *Server) Serve(ctx context.Context) error { + lc := net.ListenConfig{KeepAlive: keepAlive} + ln, err := lc.Listen(ctx, "tcp", s.cfg.Listen) + if err != nil { + return err + } + return s.ServeListener(ctx, tls.NewListener(ln, s.tls)) +} + +func (s *Server) ServeListener(ctx context.Context, ln net.Listener) error { + stop := context.AfterFunc(ctx, func() { ln.Close() }) + defer stop() + defer s.wg.Wait() + for { + conn, err := ln.Accept() + if err != nil { + if ctx.Err() != nil { + return nil + } + var ne net.Error + if errors.As(err, &ne) && ne.Timeout() { + continue + } + return err + } + select { + case s.slots <- struct{}{}: + default: + conn.Close() + continue + } + s.wg.Add(1) + go func() { + defer s.wg.Done() + defer func() { <-s.slots }() + s.handle(conn) + }() + } +} + +// handle runs the TLS handshake and preface under one deadline, so a client +// that stalls in either is dropped after prefaceTimeout. +func (s *Server) handle(conn net.Conn) { + defer conn.Close() + conn.SetDeadline(time.Now().Add(prefaceTimeout)) + br := bufio.NewReaderSize(conn, prefaceMax) + line, err := br.ReadSlice('\n') + if err != nil { + return + } + vm, kind, code := s.parse(string(line)) + if code != "" { + fmt.Fprintf(conn, "ERR %s\n", code) + return + } + if kind == "ping" { + io.WriteString(conn, "OK\n") + return + } + target, err := dialSlotSocket(filepath.Join(s.cfg.Root, "vms", vm, "run", kinds[kind])) + if err != nil { + fmt.Fprintf(conn, "ERR no_socket\n") + return + } + defer target.Close() + if _, err = io.WriteString(conn, "OK\n"); err != nil { + return + } + conn.SetDeadline(time.Time{}) + splice(conn, br, target) +} + +func (s *Server) parse(line string) (vm, kind, code string) { + f := strings.Fields(line) + if len(f) != 4 || f[0] != Magic { + return "", "", "bad_request" + } + sum := sha256.Sum256([]byte(f[1])) + if subtle.ConstantTimeCompare(sum[:], s.token[:]) != 1 { + return "", "", "auth" + } + kind = f[3] + if kind == "ping" { + return "", kind, "" + } + if _, ok := kinds[kind]; !ok || !vmIDRe.MatchString(f[2]) { + return "", "", "bad_request" + } + return f[2], kind, "" +} + +type closeWriter interface{ CloseWrite() error } + +// splice copies both ways and forwards each half-close, so a client that +// finishes sending still receives the rest of a streamed reply. An error on +// either side (reset, dead peer found by keepalive) tears down both. +func splice(client net.Conn, clientReader io.Reader, target net.Conn) { + done := make(chan struct{}, 2) + pipe := func(dst net.Conn, src io.Reader) { + defer func() { done <- struct{}{} }() + if _, err := io.Copy(dst, src); err != nil { + if !errors.Is(err, net.ErrClosed) { + log.Printf("slotgate splice: %v", err) + } + client.Close() + target.Close() + return + } + if cw, ok := dst.(closeWriter); ok { + cw.CloseWrite() + } else { + dst.Close() + } + } + go pipe(target, clientReader) + go pipe(client, target) + <-done + <-done +} diff --git a/worker/internal/slotgate/slotgate_test.go b/worker/internal/slotgate/slotgate_test.go new file mode 100644 index 000000000..54a6ad924 --- /dev/null +++ b/worker/internal/slotgate/slotgate_test.go @@ -0,0 +1,225 @@ +package slotgate + +import ( + "bufio" + "bytes" + "context" + "crypto/sha256" + "crypto/tls" + "encoding/pem" + "io" + "net" + "os" + "path/filepath" + "strings" + "testing" + "time" +) + +const testToken = "0123456789abcdef0123456789abcdef" + +// Unix socket paths are capped near 104 bytes, so the tree lives in a short /tmp dir. +func shortRoot(t *testing.T) string { + t.Helper() + dir, err := os.MkdirTemp("/tmp", "ksg") + if err != nil { + t.Fatal(err) + } + t.Cleanup(func() { os.RemoveAll(dir) }) + return dir +} + +// slotSocket serves an upper-casing echo that half-closes after the client does. +func slotSocket(t *testing.T, root, vm, name string) { + t.Helper() + dir := filepath.Join(root, "vms", vm, "run") + if err := os.MkdirAll(dir, 0o700); err != nil { + t.Fatal(err) + } + ln, err := net.Listen("unix", filepath.Join(dir, name)) + if err != nil { + t.Fatal(err) + } + t.Cleanup(func() { ln.Close() }) + go func() { + for { + c, err := ln.Accept() + if err != nil { + return + } + go func() { + defer c.Close() + data, _ := io.ReadAll(c) + c.Write([]byte(strings.ToUpper(string(data)))) + }() + } + }() +} + +func certFingerprint(t *testing.T, root string) [32]byte { + t.Helper() + data, err := os.ReadFile(filepath.Join(root, "tls", "cert.pem")) + if err != nil { + t.Fatal(err) + } + block, _ := pem.Decode(data) + return sha256.Sum256(block.Bytes) +} + +func startGate(t *testing.T, root string) string { + t.Helper() + srv, err := New(Config{Listen: "127.0.0.1:0", Root: root, Token: testToken, TLSDir: filepath.Join(root, "tls")}) + if err != nil { + t.Fatal(err) + } + ln, err := net.Listen("tcp", "127.0.0.1:0") + if err != nil { + t.Fatal(err) + } + ctx, cancel := context.WithCancel(context.Background()) + done := make(chan struct{}) + go func() { + srv.ServeListener(ctx, tls.NewListener(ln, srv.TLSConfig())) + close(done) + }() + t.Cleanup(func() { + cancel() + <-done + }) + testRoot[ln.Addr().String()] = root + return ln.Addr().String() +} + +var testRoot = map[string]string{} + +// dialGate pins the gate's certificate the way the control plane does. +func dialGate(t *testing.T, addr, preface string) (net.Conn, *bufio.Reader, string) { + t.Helper() + want := certFingerprint(t, testRoot[addr]) + c, err := tls.DialWithDialer(&net.Dialer{Timeout: 2 * time.Second}, "tcp", addr, &tls.Config{ + InsecureSkipVerify: true, + MinVersion: tls.VersionTLS13, + VerifyConnection: func(cs tls.ConnectionState) error { + if got := sha256.Sum256(cs.PeerCertificates[0].Raw); !bytes.Equal(got[:], want[:]) { + t.Errorf("gate certificate %x, pinned %x", got, want) + } + return nil + }, + }) + if err != nil { + t.Fatal(err) + } + t.Cleanup(func() { c.Close() }) + c.SetDeadline(time.Now().Add(5 * time.Second)) + if _, err = io.WriteString(c, preface); err != nil { + t.Fatal(err) + } + br := bufio.NewReader(c) + reply, _ := br.ReadString('\n') + return c, br, reply +} + +func TestSplicesToSlotSocketWithHalfClose(t *testing.T) { + root := shortRoot(t) + slotSocket(t, root, "vm-1", "kernel.sock") + addr := startGate(t, root) + c, br, reply := dialGate(t, addr, "KSG1 "+testToken+" vm-1 kernel\n") + if reply != "OK\n" { + t.Fatalf("reply = %q", reply) + } + io.WriteString(c, "hello slot") + c.(*tls.Conn).CloseWrite() + got, err := io.ReadAll(br) + if err != nil || string(got) != "HELLO SLOT" { + t.Fatalf("got %q err %v", got, err) + } +} + +func TestRejectsBadPrefaces(t *testing.T) { + root := shortRoot(t) + slotSocket(t, root, "vm-1", "worker.sock") + addr := startGate(t, root) + for _, tc := range []struct{ preface, want string }{ + {"KSG1 wrong-token-wrong-token-wrong-token vm-1 worker\n", "ERR auth\n"}, + {"KSG1 " + testToken + " ../vm-1 worker\n", "ERR bad_request\n"}, + {"KSG1 " + testToken + " vm-1 docker\n", "ERR bad_request\n"}, + {"KSG0 " + testToken + " vm-1 worker\n", "ERR bad_request\n"}, + {"KSG1 " + testToken + " vm-2 worker\n", "ERR no_socket\n"}, + {"KSG1 " + testToken + " - ping\n", "OK\n"}, + } { + if _, _, reply := dialGate(t, addr, tc.preface); reply != tc.want { + t.Errorf("%q: reply %q, want %q", tc.preface, reply, tc.want) + } + } +} + +func TestDoesNotFollowPlantedSymlink(t *testing.T) { + root := shortRoot(t) + slotSocket(t, root, "victim", "worker.sock") + run := filepath.Join(root, "vms", "guest", "run") + if err := os.MkdirAll(run, 0o700); err != nil { + t.Fatal(err) + } + if err := os.Symlink(filepath.Join(root, "vms", "victim", "run", "worker.sock"), filepath.Join(run, "worker.sock")); err != nil { + t.Fatal(err) + } + addr := startGate(t, root) + if _, _, reply := dialGate(t, addr, "KSG1 "+testToken+" guest worker\n"); reply != "ERR no_socket\n" { + t.Fatalf("symlinked socket reply = %q", reply) + } +} + +func TestCertificateSurvivesRestartAndIsRecreatedWhenLost(t *testing.T) { + root := shortRoot(t) + cfg := Config{Listen: "127.0.0.1:1", Root: root, Token: testToken, TLSDir: filepath.Join(root, "tls")} + if _, err := New(cfg); err != nil { + t.Fatal(err) + } + first := certFingerprint(t, root) + if st, _ := os.Stat(filepath.Join(root, "tls", "key.pem")); st.Mode().Perm() != 0o600 { + t.Fatalf("key mode %v", st.Mode().Perm()) + } + if _, err := New(cfg); err != nil { + t.Fatal(err) + } + if certFingerprint(t, root) != first { + t.Fatal("certificate changed across restart") + } + os.Remove(filepath.Join(root, "tls", "key.pem")) + if _, err := New(cfg); err != nil { + t.Fatal(err) + } + if certFingerprint(t, root) == first { + t.Fatal("lost key did not produce a new certificate") + } +} + +func TestPlainTCPClientGetsNothing(t *testing.T) { + root := shortRoot(t) + slotSocket(t, root, "vm-1", "kernel.sock") + addr := startGate(t, root) + c, err := net.DialTimeout("tcp", addr, 2*time.Second) + if err != nil { + t.Fatal(err) + } + defer c.Close() + c.SetDeadline(time.Now().Add(3 * time.Second)) + io.WriteString(c, "KSG1 "+testToken+" vm-1 kernel\n") + reply, _ := bufio.NewReader(c).ReadString('\n') + if strings.HasPrefix(reply, "OK") { + t.Fatalf("plaintext preface accepted: %q", reply) + } +} + +func TestNewValidatesConfig(t *testing.T) { + for _, cfg := range []Config{ + {Listen: "", Root: "/r", Token: testToken, TLSDir: "/t"}, + {Listen: "127.0.0.1:1", Root: "rel", Token: testToken, TLSDir: "/t"}, + {Listen: "127.0.0.1:1", Root: "/r", Token: "short", TLSDir: "/t"}, + {Listen: "127.0.0.1:1", Root: "/r", Token: testToken, TLSDir: "rel"}, + } { + if _, err := New(cfg); err == nil { + t.Errorf("New(%+v) accepted", cfg) + } + } +}