From f03d2dc8a6d58172c55c7846ff845ab0bbfcd48c Mon Sep 17 00:00:00 2001 From: wangjia <809946525@qq.com> Date: Sat, 13 Jun 2026 12:26:58 +0800 Subject: [PATCH] =?UTF-8?q?feat(agent):=20node=20agent=20=E2=80=94=20enrol?= =?UTF-8?q?l/mTLS,=20heartbeat,=20command=20stream,=20sing-box=20=E7=94=A8?= =?UTF-8?q?=E6=88=B7=E8=A1=A8=20(tsk=5F=5FR8M4jEw43JR)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 与控制面同仓同 go.mod，新增节点 agent 实现： - proto/agent/v1/agent.proto + internal/pb/agentv1：冻结的控制面↔agent gRPC 契约 （Enroll/Register/Heartbeat/Subscribe/Ack/ReportUsage）。仓库尚无 protoc 流水线， 暂以手写 Go 类型 + JSON gRPC codec 实现，与 proto 1:1 对应，待 protoc 接入即可替换。 - internal/agentd： - enroll.go：首启生成 EC 密钥+CSR，持 bootstrap token 调 Enroll 换 90d 节点证书 （CN=node_uuid），落 /etc/pangolin-agent/，此后 mTLS。 - conn.go(agent.go)+creds.go：mTLS 主动拨号 + 指数退避重连；重连携带 last_command_id； Register 取 ConfigSnapshot 全量配置覆盖本地。 - heartbeat.go：30s 上报 peer/带宽/CPU + config_version；need_full_resync→全量同步。 - command.go：消费 Subscribe，Upsert/Revoke/Rotate/ApplyConfig/Lifecycle 幂等处理后 Ack（at-least-once，按 command_id 去重）。 - singbox.go+render.go：内存用户表 + 落盘 state.json（仅 dp_uuid+expires_at）；任何变更 渲染完整 sing-box 配置（REALITY users[uuid,flow] + Hy2 users[派生口令]）→ 500ms 去抖 合并 → systemd 重启。 - ttl.go：凭证 TTL 定时移除并上报。 - usage.go：按 dp_uuid 聚合上报，绝无 user_id/email/目的地址。 - derive.go：Hy2 口令 = HMAC-SHA256(key, dp_uuid)，与控制面同源派生。 - cmd/agent：入口（flag/env 配置）。 - infra/cloud-init/{node.yaml.tmpl,install-node.sh,README.md}：一段式安装，下载锁定版本 二进制并校验 SHA-256，systemd 拉管，首启即 Enroll/Register。shellcheck -S warning 通过。 测试（bufconn mock 控制面，无需 docker）：Enroll→Register→Heartbeat 全流转；Upsert/Revoke 渲染正确；Rotate 宽限期新旧并存到点移除；TTL 自动移除并上报；断流重连 last_command_id 续发不丢不重；need_full_resync 触发重注册；state.json 恢复；去抖合并；扫描确认无身份字段。 go test -race ./internal/agentd/... ./internal/pb/... 通过；go vet ./... 通过。 落实 doc/04 §2 节点无状态化与 doc/06 §3 数据面红线（节点仅见 dp_uuid）。 Co-Authored-By: Claude Opus 4.8 --- infra/cloud-init/README.md | 39 +++ infra/cloud-init/install-node.sh | 113 +++++++ infra/cloud-init/node.yaml.tmpl | 45 +++ server/Makefile | 6 + server/cmd/agent/main.go | 58 ++++ server/internal/agentd/agent.go | 202 +++++++++++ server/internal/agentd/command.go | 105 ++++++ server/internal/agentd/command_test.go | 66 ++++ server/internal/agentd/config.go | 112 +++++++ server/internal/agentd/creds.go | 101 ++++++ server/internal/agentd/derive.go | 34 ++ server/internal/agentd/derive_test.go | 30 ++ server/internal/agentd/enroll.go | 139 ++++++++ server/internal/agentd/heartbeat.go | 63 ++++ server/internal/agentd/integration_test.go | 370 +++++++++++++++++++++ server/internal/agentd/render.go | 95 ++++++ server/internal/agentd/singbox.go | 354 ++++++++++++++++++++ server/internal/agentd/singbox_test.go | 229 +++++++++++++ server/internal/agentd/ttl.go | 30 ++ server/internal/agentd/usage.go | 54 +++ server/internal/agentd/util.go | 62 ++++ server/internal/pb/agentv1/codec.go | 30 ++ server/internal/pb/agentv1/service.go | 238 +++++++++++++ server/internal/pb/agentv1/types.go | 176 ++++++++++ server/proto/agent/v1/agent.proto | 212 ++++++++++++ 25 files changed, 2963 insertions(+) create mode 100644 infra/cloud-init/README.md create mode 100644 infra/cloud-init/install-node.sh create mode 100644 infra/cloud-init/node.yaml.tmpl create mode 100644 server/cmd/agent/main.go create mode 100644 server/internal/agentd/agent.go create mode 100644 server/internal/agentd/command.go create mode 100644 server/internal/agentd/command_test.go create mode 100644 server/internal/agentd/config.go create mode 100644 server/internal/agentd/creds.go create mode 100644 server/internal/agentd/derive.go create mode 100644 server/internal/agentd/derive_test.go create mode 100644 server/internal/agentd/enroll.go create mode 100644 server/internal/agentd/heartbeat.go create mode 100644 server/internal/agentd/integration_test.go create mode 100644 server/internal/agentd/render.go create mode 100644 server/internal/agentd/singbox.go create mode 100644 server/internal/agentd/singbox_test.go create mode 100644 server/internal/agentd/ttl.go create mode 100644 server/internal/agentd/usage.go create mode 100644 server/internal/agentd/util.go create mode 100644 server/internal/pb/agentv1/codec.go create mode 100644 server/internal/pb/agentv1/service.go create mode 100644 server/internal/pb/agentv1/types.go create mode 100644 server/proto/agent/v1/agent.proto diff --git a/infra/cloud-init/README.md b/infra/cloud-init/README.md new file mode 100644 index 0000000..6539f91 --- /dev/null +++ b/infra/cloud-init/README.md @@ -0,0 +1,39 @@ +# infra/cloud-init —— 节点一段式安装 + +加速节点开机即用所需的全部物料。控制面（或 #14 弹性开机）在创建节点前，用 `sed` +把模板里的 `__PLACEHOLDER__` 替换成真实值（沿用 `deploy/` 渲染约定），再作为 +user-data 注入。 + +## 文件 + +- `node.yaml.tmpl` —— cloud-config 模板。写入 `/etc/pangolin-agent/agent.env` + （含一次性 bootstrap token，0600），下载并按 SHA-256 校验 `install-node.sh` 后执行。 +- `install-node.sh` —— 幂等安装脚本：下载锁定版本的 `sing-box` + `pangolin-agent` + 二进制并校验 SHA-256，装 systemd 单元（agent 拉管 sing-box），生成 Hy2 自签证书， + 启动 agent（首启即 Enroll/Register）。`shellcheck -S warning` 通过。 + +## 占位符 + +| 占位符 | 含义 | +| --- | --- | +| `__CONTROL_PLANE__` | 控制面 gRPC 地址 `host:port` | +| `__SERVER_NAME__` | 校验控制面证书的 TLS SNI（留空取 host） | +| `__BOOTSTRAP_TOKEN__` | 一次性引导 token（`mtls.BootstrapTokenManager`，15min TTL） | +| `__DERIVE_KEY__` | Hy2 口令派生密钥（与控制面同源，见 `agentd.DeriveHy2Password`） | +| `__AGENT_VERSION__` | agent 版本（与产物命名 + SHA 一致） | +| `__SINGBOX_VERSION__` / `__SINGBOX_SHA256__` | sing-box 版本与校验和 | +| `__AGENT_SHA256__` | agent 二进制校验和 | +| `__ARTIFACT_BASE__` | 二进制 + 安装脚本下载根 URL | +| `__INSTALL_SHA256__` | `install-node.sh` 自身的 SHA-256（防篡改） | + +## 可复现性 + +版本与 SHA-256 全部锁定：`install-node.sh` 顶部的版本/校验和占位符由发布流水线渲染 +为固定值并随产物一起发布；cloud-init 下载脚本后先校验脚本 SHA-256 再执行，脚本再逐个 +校验二进制 SHA-256。任何一环不匹配即 `set -e` 中止，保证整链可复现、不可被中间人替换。 + +## 无状态红线 + +节点除 `/etc/pangolin-agent/{node.key,node.crt,ca.crt,state.json}` 与 `/etc/sing-box` +外不落任何持久数据；`state.json` 仅含 `dp_uuid + expires_at`，无任何账号/设备/目的地址/ +DNS 信息，亦无访问日志（见 `doc/04 §2`、`doc/06 §3`）。 diff --git a/infra/cloud-init/install-node.sh b/infra/cloud-init/install-node.sh new file mode 100644 index 0000000..3b0dd89 --- /dev/null +++ b/infra/cloud-init/install-node.sh @@ -0,0 +1,113 @@ +#!/usr/bin/env bash +# install-node.sh —— Pangolin 加速节点一键安装（由 cloud-init runcmd 调用，幂等）。 +# +# 职责：下载锁定版本的 sing-box + pangolin-agent 二进制并校验 SHA-256 → 安装 systemd +# 单元（agent 拉管 sing-box）→ 生成 Hy2 自签证书 → 启动 agent（首启即 Enroll/Register）。 +# +# 版本与校验和通过 __PLACEHOLDER__ 由发布流水线/控制面 provisioner 用 sed 渲染（沿用 +# deploy/ 约定），保证「二进制版本 + SHA-256 锁定可复现」。也可用同名环境变量覆盖。 +# +# 数据面无状态红线：本脚本不写任何用户身份、不开访问日志；唯一持久数据是 +# /etc/pangolin-agent/state.json（仅 dp_uuid + expires_at，由 agent 维护）。 +set -euo pipefail + +SINGBOX_VERSION="${SINGBOX_VERSION:-__SINGBOX_VERSION__}" +SINGBOX_SHA256="${SINGBOX_SHA256:-__SINGBOX_SHA256__}" +AGENT_VERSION="${AGENT_VERSION:-__AGENT_VERSION__}" +AGENT_SHA256="${AGENT_SHA256:-__AGENT_SHA256__}" +ARTIFACT_BASE="${ARTIFACT_BASE:-__ARTIFACT_BASE__}" + +STATE_DIR=/etc/pangolin-agent +INSTALL_DIR=/usr/local/bin +SB_CFG_DIR=/etc/sing-box +MASQ_CN=www.bing.com + +log() { echo "[install-node] $*"; } + +# 架构映射（与发布产物命名对齐）。 +arch="$(uname -m)" +case "$arch" in + x86_64 | amd64) goarch=amd64 ;; + aarch64 | arm64) goarch=arm64 ;; + *) + echo "[install-node] unsupported arch: $arch" >&2 + exit 1 + ;; +esac + +# fetch_verify URL EXPECTED_SHA256 DEST —— 下载、校验、原子安装为可执行文件。 +fetch_verify() { + url="$1" + sha="$2" + dest="$3" + tmp="$(mktemp)" + log "downloading $url" + curl -fsSL --retry 3 "$url" -o "$tmp" + echo "${sha} ${tmp}" | sha256sum -c - + install -m 0755 "$tmp" "$dest" + rm -f "$tmp" + log "installed $dest" +} + +install -d -m 700 "$STATE_DIR" +install -d -m 755 "$SB_CFG_DIR" "$SB_CFG_DIR/cert" + +fetch_verify "${ARTIFACT_BASE}/sing-box-${SINGBOX_VERSION}-linux-${goarch}" \ + "$SINGBOX_SHA256" "${INSTALL_DIR}/sing-box" +fetch_verify "${ARTIFACT_BASE}/pangolin-agent-${AGENT_VERSION}-linux-${goarch}" \ + "$AGENT_SHA256" "${INSTALL_DIR}/pangolin-agent" + +# Hy2 入站需要 TLS 证书；自签即可（masquerade 用），幂等。 +if [[ ! -f "${SB_CFG_DIR}/cert/hy2.crt" || ! -f "${SB_CFG_DIR}/cert/hy2.key" ]]; then + log "generating self-signed Hy2 certificate" + openssl ecparam -genkey -name prime256v1 -out "${SB_CFG_DIR}/cert/hy2.key" + openssl req -new -x509 -days 3650 -key "${SB_CFG_DIR}/cert/hy2.key" \ + -out "${SB_CFG_DIR}/cert/hy2.crt" -subj "/CN=${MASQ_CN}" + chmod 600 "${SB_CFG_DIR}/cert/hy2.key" + chmod 644 "${SB_CFG_DIR}/cert/hy2.crt" +fi + +# sing-box 单元：由 agent 渲染配置后 `systemctl restart` 拉起（不预启，避免无配置失败循环）。 +cat > /etc/systemd/system/sing-box.service <<'UNIT' +[Unit] +Description=sing-box data plane (managed by pangolin-agent) +After=network-online.target +Wants=network-online.target + +[Service] +ExecStart=/usr/local/bin/sing-box run -c /etc/sing-box/config.json +Restart=on-failure +RestartSec=1 +LimitNOFILE=1048576 +# 无状态：禁止持久写盘（配置目录除外）。 +NoNewPrivileges=true +ProtectHome=true + +[Install] +WantedBy=multi-user.target +UNIT + +# pangolin-agent 单元：常驻、自动重连；EnvironmentFile 由 cloud-init 写入。 +cat > /etc/systemd/system/pangolin-agent.service <<'UNIT' +[Unit] +Description=Pangolin node agent +After=network-online.target +Wants=network-online.target + +[Service] +EnvironmentFile=/etc/pangolin-agent/agent.env +ExecStart=/usr/local/bin/pangolin-agent +Restart=always +RestartSec=2 +NoNewPrivileges=true + +[Install] +WantedBy=multi-user.target +UNIT + +systemctl daemon-reload +# sing-box 仅 enable 不 start：等 agent 首次渲染配置后由 agent restart 拉起。 +systemctl enable sing-box.service +systemctl enable --now pangolin-agent.service + +log "done; agent will Enroll/Register and render sing-box config on first connect" diff --git a/infra/cloud-init/node.yaml.tmpl b/infra/cloud-init/node.yaml.tmpl new file mode 100644 index 0000000..7236193 --- /dev/null +++ b/infra/cloud-init/node.yaml.tmpl @@ -0,0 +1,45 @@ +#cloud-config +# Pangolin 加速节点一段式安装模板（与 #14 弹性开机共用）。 +# +# 渲染：控制面 provisioner 用 sed 替换所有 __PLACEHOLDER__（沿用 deploy/ 约定）。 +# __CONTROL_PLANE__ 控制面 gRPC 地址 host:port +# __SERVER_NAME__ 校验控制面证书的 TLS SNI（留空则取 host 部分） +# __BOOTSTRAP_TOKEN__ 一次性引导 token（由 mtls.BootstrapTokenManager.IssueToken 预签发，15min TTL） +# __DERIVE_KEY__ Hy2 口令派生密钥（与控制面同源，见 agentd.DeriveHy2Password） +# __AGENT_VERSION__ pangolin-agent 版本（与产物命名/SHA 锁定一致） +# __ARTIFACT_BASE__ 二进制与安装脚本下载根 URL +# __INSTALL_SHA256__ install-node.sh 的 SHA-256（整链可复现校验） +# +# 安全：agent.env 内含 bootstrap token，权限 0600；token 一次性消费后即失效。 +# 节点无状态：除 /etc/pangolin-agent/{node.key,node.crt,ca.crt,state.json} 与 +# /etc/sing-box 外不落任何持久数据，且 state.json 仅含 dp_uuid + expires_at。 + +write_files: + - path: /etc/pangolin-agent/agent.env + permissions: '0600' + owner: root:root + content: | + PANGOLIN_AGENT_CONTROL_PLANE=__CONTROL_PLANE__ + PANGOLIN_AGENT_SERVER_NAME=__SERVER_NAME__ + PANGOLIN_AGENT_BOOTSTRAP_TOKEN=__BOOTSTRAP_TOKEN__ + PANGOLIN_AGENT_DERIVE_KEY=__DERIVE_KEY__ + PANGOLIN_AGENT_STATE_DIR=/etc/pangolin-agent + PANGOLIN_AGENT_SINGBOX_CONFIG=/etc/sing-box/config.json + PANGOLIN_AGENT_SINGBOX_UNIT=sing-box + PANGOLIN_AGENT_VERSION=__AGENT_VERSION__ + +runcmd: + - | + set -euo pipefail + mkdir -p /opt/pangolin + tmp="$(mktemp)" + # 下载安装脚本并按锁定 SHA-256 校验后再执行（拒绝被篡改的脚本）。 + curl -fsSL --retry 3 "__ARTIFACT_BASE__/install-node.sh" -o "$tmp" + echo "__INSTALL_SHA256__ $tmp" | sha256sum -c - + install -m 0755 "$tmp" /opt/pangolin/install-node.sh + rm -f "$tmp" + ARTIFACT_BASE="__ARTIFACT_BASE__" AGENT_VERSION="__AGENT_VERSION__" \ + bash /opt/pangolin/install-node.sh + +# 整机 3–5 分钟内可服务：cloud-init 完成 → agent 首启 Enroll → Register 取全量配置 → +# 渲染 sing-box config 并 restart sing-box → REALITY/Hy2 入站上线。 diff --git a/server/Makefile b/server/Makefile index 792a8ff..6723007 100644 --- a/server/Makefile +++ b/server/Makefile @@ -12,6 +12,9 @@ build: build-codegen: ## 激活码批次生成 CLI go build -o bin/codegen ./cmd/codegen +build-agent: ## 节点 agent 二进制（部署到加速节点） + go build -o bin/pangolin-agent ./cmd/agent + # ── test ─────────────────────────────────────────────────────────────────────── test: go test ./... @@ -19,6 +22,9 @@ test: test-unit: ## 单测（不依赖外部服务） go test -count=1 -race ./internal/codes/... -run 'Test[^I][^n][^t]' +test-agent: ## 节点 agent 单测 + 集成（bufconn mock 控制面，无需 docker） + go test -count=1 -race ./internal/agentd/... ./internal/pb/... + test-integration: ## 集成测试（需本机 docker，testcontainers） go test -count=1 ./internal/codes/... -run TestInt diff --git a/server/cmd/agent/main.go b/server/cmd/agent/main.go new file mode 100644 index 0000000..2871312 --- /dev/null +++ b/server/cmd/agent/main.go @@ -0,0 +1,58 @@ +// Command agent is the Pangolin node agent. It runs on every acceleration node, +// self-enrolls over mTLS, dials the control plane, applies the streamed command +// feed by managing the local sing-box user table, and reports heartbeat/usage. +// +// Configuration comes from flags or PANGOLIN_AGENT_* environment variables +// (cloud-init injects the latter). The node holds zero user identities on disk. +package main + +import ( + "context" + "flag" + "log" + "os" + "os/signal" + "syscall" + "time" + + "github.com/wangjia/pangolin/server/internal/agentd" +) + +func main() { + cfg := agentd.Config{} + var singboxUnit string + + flag.StringVar(&cfg.ControlPlaneAddr, "control-plane", env("PANGOLIN_AGENT_CONTROL_PLANE", ""), "control plane gRPC address host:port") + flag.StringVar(&cfg.ServerName, "server-name", env("PANGOLIN_AGENT_SERVER_NAME", ""), "TLS SNI to validate (defaults to control-plane host)") + flag.StringVar(&cfg.BootstrapToken, "bootstrap-token", env("PANGOLIN_AGENT_BOOTSTRAP_TOKEN", ""), "one-time enrollment token") + flag.StringVar(&cfg.StateDir, "state-dir", env("PANGOLIN_AGENT_STATE_DIR", agentd.DefaultStateDir), "directory for node key/cert/ca/state") + flag.StringVar(&cfg.SingboxConfigPath, "singbox-config", env("PANGOLIN_AGENT_SINGBOX_CONFIG", agentd.DefaultSingboxCfg), "path to rendered sing-box config") + flag.StringVar(&cfg.DeriveKey, "derive-key", env("PANGOLIN_AGENT_DERIVE_KEY", ""), "Hy2 password derivation key") + flag.StringVar(&cfg.AgentVersion, "agent-version", env("PANGOLIN_AGENT_VERSION", "dev"), "reported agent version") + flag.StringVar(&singboxUnit, "singbox-unit", env("PANGOLIN_AGENT_SINGBOX_UNIT", "sing-box"), "systemd unit to restart on config change") + flag.BoolVar(&cfg.Insecure, "insecure", env("PANGOLIN_AGENT_INSECURE", "") == "1", "dial without mTLS (dev only)") + flag.Parse() + + if cfg.ControlPlaneAddr == "" { + log.Fatal("agent: --control-plane (or PANGOLIN_AGENT_CONTROL_PLANE) is required") + } + + agent := agentd.New(cfg, agentd.WithRestarter(agentd.SystemdRestarter{Unit: singboxUnit})) + + ctx, stop := signal.NotifyContext(context.Background(), syscall.SIGINT, syscall.SIGTERM) + defer stop() + + log.Printf("[pangolin-agent] starting (control-plane=%s, version=%s)", cfg.ControlPlaneAddr, cfg.AgentVersion) + start := time.Now() + if err := agent.Run(ctx); err != nil { + log.Fatalf("[pangolin-agent] exited after %s: %v", time.Since(start), err) + } + log.Printf("[pangolin-agent] shut down cleanly after %s", time.Since(start)) +} + +func env(key, def string) string { + if v, ok := os.LookupEnv(key); ok { + return v + } + return def +} diff --git a/server/internal/agentd/agent.go b/server/internal/agentd/agent.go new file mode 100644 index 0000000..eb28922 --- /dev/null +++ b/server/internal/agentd/agent.go @@ -0,0 +1,202 @@ +package agentd + +import ( + "context" + "errors" + "math/rand" + "sync/atomic" + "time" + + agentv1 "github.com/wangjia/pangolin/server/internal/pb/agentv1" +) + +// errResync is returned within a session to force a reconnect + full Register. +var errResync = errors.New("agentd: full resync required") + +// Agent is the top-level node daemon. Construct with New and drive with Run. +type Agent struct { + cfg Config + sb *SingBox + nodeUUID atomic.Value // string; written once by Run, read by all loops + tests + + dial Dialer + enrollDial EnrollDialer + load LoadSource + usage UsageSource + clock func() time.Time + + lastCommandID atomic.Int64 + shutdown context.CancelFunc + + restarter Restarter +} + +// Option customises an Agent (primarily for tests / dependency injection). +type Option func(*Agent) + +func WithRestarter(r Restarter) Option { return func(a *Agent) { a.restarter = r } } +func WithDialer(d Dialer) Option { return func(a *Agent) { a.dial = d } } +func WithEnrollDialer(d EnrollDialer) Option { return func(a *Agent) { a.enrollDial = d } } +func WithLoadSource(l LoadSource) Option { return func(a *Agent) { a.load = l } } +func WithUsageSource(u UsageSource) Option { return func(a *Agent) { a.usage = u } } +func WithClock(f func() time.Time) Option { return func(a *Agent) { a.clock = f } } + +// New builds an Agent. Production callers pass nothing extra and get mTLS dialers; +// tests inject bufconn dialers, fake restarters, etc. +func New(cfg Config, opts ...Option) *Agent { + cfg = cfg.withDefaults() + a := &Agent{cfg: cfg, clock: time.Now} + for _, o := range opts { + o(a) + } + if a.dial == nil { + if cfg.Insecure { + a.dial = insecureDialer(cfg.ControlPlaneAddr) + } else { + a.dial = NewMTLSDialer(cfg) + } + } + if a.enrollDial == nil { + if cfg.Insecure { + a.enrollDial = EnrollDialer(insecureDialer(cfg.ControlPlaneAddr)) + } else { + a.enrollDial = NewEnrollDialer(cfg) + } + } + a.sb = NewSingBox(cfg, a.restarter) + a.sb.clock = a.clock + if a.load == nil { + a.load = defaultLoadSource{sb: a.sb} + } + if a.usage == nil { + a.usage = nopUsageSource{} + } + return a +} + +// SingBox exposes the underlying manager (tests / introspection). +func (a *Agent) SingBox() *SingBox { return a.sb } + +// NodeUUID returns the enrolled node UUID (empty until Run enrolls). +func (a *Agent) NodeUUID() string { + v, _ := a.nodeUUID.Load().(string) + return v +} + +// Run enrolls (if needed), restores state, then maintains the control-plane +// connection with exponential-backoff reconnect until ctx is cancelled. +func (a *Agent) Run(ctx context.Context) error { + ctx, cancel := context.WithCancel(ctx) + defer cancel() + a.shutdown = cancel + + uuid, err := EnsureEnrolled(ctx, a.cfg, a.enrollDial) + if err != nil { + return err + } + a.nodeUUID.Store(uuid) + + if err := a.sb.LoadState(); err != nil { + return err + } + + go a.sb.Run(ctx) + go a.runTTL(ctx) + a.sb.markDirty() // render whatever we restored before the first connect + + backoff := a.cfg.BackoffMin + for { + if ctx.Err() != nil { + return nil + } + registered, err := a.runSession(ctx) + if ctx.Err() != nil { + return nil + } + if registered { + backoff = a.cfg.BackoffMin // healthy session → reset backoff + } + if err != nil && !errors.Is(err, errResync) { + logf("session ended: %v; reconnecting in %s", err, backoff) + } + if !sleepCtx(ctx, jitter(backoff)) { + return nil + } + backoff = nextBackoff(backoff, a.cfg.BackoffMax) + } +} + +// runSession establishes one connection: Register (full snapshot), then runs +// heartbeat/subscribe/usage until the first of them fails. Returns whether +// Register succeeded (used to reset backoff). +func (a *Agent) runSession(ctx context.Context) (registered bool, err error) { + conn, err := a.dial(ctx) + if err != nil { + return false, err + } + defer conn.Close() + + client := agentv1.NewAgentServiceClient(conn) + + snap, err := client.Register(ctx, &agentv1.RegisterRequest{ + NodeUUID: a.NodeUUID(), + AgentVersion: a.cfg.AgentVersion, + LocalConfigVersion: a.sb.ConfigVersion(), + }) + if err != nil { + return false, err + } + // Full snapshot is authoritative: overwrite local credential table + inbounds. + a.sb.ApplyConfig(snap, true) + if snap.LastCommandID > a.lastCommandID.Load() { + a.lastCommandID.Store(snap.LastCommandID) + } + + sctx, cancel := context.WithCancel(ctx) + defer cancel() + + errc := make(chan error, 3) + go func() { errc <- a.runHeartbeat(sctx, client) }() + go func() { errc <- a.runSubscribe(sctx, client) }() + go func() { errc <- a.runUsage(sctx, client) }() + + first := <-errc + cancel() + <-errc + <-errc + return true, first +} + +// ─── backoff helpers ───────────────────────────────────────────────────────── + +func nextBackoff(cur, max time.Duration) time.Duration { + next := cur * 2 + if next > max { + next = max + } + return next +} + +// jitter applies ±20% randomisation to avoid thundering-herd reconnects. +func jitter(d time.Duration) time.Duration { + if d <= 0 { + return 0 + } + delta := float64(d) * 0.2 + return d + time.Duration((rand.Float64()*2-1)*delta) +} + +// sleepCtx sleeps for d unless ctx is cancelled first. Returns false if cancelled. +func sleepCtx(ctx context.Context, d time.Duration) bool { + if d <= 0 { + return ctx.Err() == nil + } + t := time.NewTimer(d) + defer t.Stop() + select { + case <-ctx.Done(): + return false + case <-t.C: + return true + } +} diff --git a/server/internal/agentd/command.go b/server/internal/agentd/command.go new file mode 100644 index 0000000..e4add11 --- /dev/null +++ b/server/internal/agentd/command.go @@ -0,0 +1,105 @@ +package agentd + +import ( + "context" + + agentv1 "github.com/wangjia/pangolin/server/internal/pb/agentv1" +) + +// runSubscribe opens the Command stream (resuming from the last durably processed +// command id) and applies each command idempotently, Ack'ing after apply +// (at-least-once). It returns when ctx is cancelled or the stream errors, so the +// session can reconnect and resume. +func (a *Agent) runSubscribe(ctx context.Context, client agentv1.AgentServiceClient) error { + nodeUUID := a.NodeUUID() + stream, err := client.Subscribe(ctx, &agentv1.SubscribeRequest{ + NodeUUID: nodeUUID, + LastCommandID: a.lastCommandID.Load(), + }) + if err != nil { + return err + } + for { + cmd, err := stream.Recv() + if err != nil { + return err + } + a.handleCommand(cmd) + // Ack after apply. A failed Ack just causes redelivery, which the + // dedup-by-command-id + idempotent apply below tolerates. + if _, err := client.Ack(ctx, &agentv1.AckRequest{ + NodeUUID: nodeUUID, + CommandID: cmd.CommandID, + }); err != nil { + return err + } + } +} + +// handleCommand applies one command. Commands at or below the high-water mark are +// skipped (already applied) — this makes redelivery after a dropped Ack safe. +func (a *Agent) handleCommand(cmd *agentv1.Command) { + if cmd.CommandID != 0 && cmd.CommandID <= a.lastCommandID.Load() { + return + } + switch cmd.Type { + case agentv1.CommandTypeUpsert: + if cmd.Upsert != nil && cmd.Upsert.Credential != nil { + a.sb.Upsert(credFromPB(cmd.Upsert.Credential)) + } + case agentv1.CommandTypeRevoke: + if cmd.Revoke != nil { + a.sb.Revoke(cmd.Revoke.DpUUID) + } + case agentv1.CommandTypeRotateCredential: + if cmd.Rotate != nil && cmd.Rotate.NewCredential != nil { + a.sb.Rotate(cmd.Rotate.OldDpUUID, credFromPB(cmd.Rotate.NewCredential), cmd.Rotate.GraceUntilUnix) + } + case agentv1.CommandTypeApplyConfig: + if cmd.ApplyConfig != nil { + a.sb.ApplyConfig(cmd.ApplyConfig, cmd.ApplyConfig.Credentials != nil) + } + case agentv1.CommandTypeLifecycle: + a.handleLifecycle(cmd.Lifecycle) + default: + logf("ignoring command id=%d with unknown type=%d", cmd.CommandID, cmd.Type) + } + + if cmd.CommandID > a.lastCommandID.Load() { + a.lastCommandID.Store(cmd.CommandID) + } +} + +func (a *Agent) handleLifecycle(p *agentv1.LifecyclePayload) { + if p == nil { + return + } + switch p.Action { + case agentv1.LifecycleActionDrain: + // Draining is steered by control-plane node weight; the data plane keeps + // serving existing sessions. Nothing to mutate locally. + logf("lifecycle: drain") + case agentv1.LifecycleActionResume: + logf("lifecycle: resume") + case agentv1.LifecycleActionShutdown: + logf("lifecycle: shutdown requested") + if a.shutdown != nil { + a.shutdown() + } + default: + logf("lifecycle: unknown action %d", p.Action) + } +} + +func credFromPB(c *agentv1.Credential) *Cred { + flow := c.Flow + if flow == "" { + flow = DefaultFlow + } + return &Cred{ + DpUUID: c.DpUUID, + Protocol: c.Protocol, + Flow: flow, + ExpiresAt: c.ExpiresAtUnix, + } +} diff --git a/server/internal/agentd/command_test.go b/server/internal/agentd/command_test.go new file mode 100644 index 0000000..8794a5a --- /dev/null +++ b/server/internal/agentd/command_test.go @@ -0,0 +1,66 @@ +package agentd + +import ( + "testing" + + agentv1 "github.com/wangjia/pangolin/server/internal/pb/agentv1" +) + +func TestHandleCommandIdempotentAndOrdered(t *testing.T) { + a := New(testConfig(t), WithRestarter(&fakeRestarter{})) + a.sb.ApplyConfig(sampleSnapshot(), true) // inbounds present + + upsert := func(id int64, dp string) *agentv1.Command { + return &agentv1.Command{ + CommandID: id, + Type: agentv1.CommandTypeUpsert, + Upsert: &agentv1.UpsertPayload{Credential: &agentv1.Credential{DpUUID: dp, Protocol: agentv1.ProtocolBoth}}, + } + } + + // Apply id=1, then redelivered id=1 (dropped Ack) — applied once, no regression. + a.handleCommand(upsert(1, "alpha")) + a.handleCommand(upsert(1, "alpha")) + if a.lastCommandID.Load() != 1 { + t.Fatalf("lastCommandID = %d, want 1", a.lastCommandID.Load()) + } + + // A stale command at or below the high-water mark must be skipped (idempotent + // at-least-once redelivery): revoke at id=1 should NOT remove alpha. + a.handleCommand(&agentv1.Command{ + CommandID: 1, + Type: agentv1.CommandTypeRevoke, + Revoke: &agentv1.RevokePayload{DpUUID: "alpha"}, + }) + if !a.sb.Has("alpha") { + t.Error("revoke with stale id=1 should have been skipped, but alpha was removed") + } + + // Forward progress: id=2 revokes alpha. + a.handleCommand(&agentv1.Command{ + CommandID: 2, + Type: agentv1.CommandTypeRevoke, + Revoke: &agentv1.RevokePayload{DpUUID: "alpha"}, + }) + if a.sb.Has("alpha") { + t.Error("alpha should be revoked by id=2") + } + if a.lastCommandID.Load() != 2 { + t.Fatalf("lastCommandID = %d, want 2", a.lastCommandID.Load()) + } +} + +func TestHandleApplyConfigUpdatesInbounds(t *testing.T) { + a := New(testConfig(t), WithRestarter(&fakeRestarter{})) + a.handleCommand(&agentv1.Command{ + CommandID: 1, + Type: agentv1.CommandTypeApplyConfig, + ApplyConfig: sampleSnapshot(&agentv1.Credential{DpUUID: "z", Protocol: agentv1.ProtocolReality}), + }) + if a.sb.ConfigVersion() != 7 { + t.Errorf("config version = %d, want 7", a.sb.ConfigVersion()) + } + if !a.sb.Has("z") { + t.Error("apply_config credential z not applied") + } +} diff --git a/server/internal/agentd/config.go b/server/internal/agentd/config.go new file mode 100644 index 0000000..2375096 --- /dev/null +++ b/server/internal/agentd/config.go @@ -0,0 +1,112 @@ +// Package agentd implements the Pangolin node agent: a lightweight daemon that +// runs on every acceleration node. It self-enrolls over mTLS, dials the control +// plane (the agent is always the dialer), keeps a long-lived gRPC connection with +// exponential-backoff reconnect, reports heartbeat/usage, and applies the streamed +// command feed by managing the local sing-box user table. +// +// No-state invariant (doc/04 §2, doc/06 §3): the agent persists ONLY the +// credential table (dp_uuid + expires_at) to disk. It keeps zero user identities, +// zero destination/DNS data and writes no access logs. A seized node leaks only +// opaque dp_uuids, never accounts. +package agentd + +import ( + "path/filepath" + "time" +) + +// Default tuning values. The intervals match doc/06 (30s heartbeat) and the +// 500ms render debounce that batches bursts of credential changes into a single +// sing-box restart (sing-box has no hot-reload; a restart is a ~1-2s blip). +const ( + DefaultHeartbeatInterval = 30 * time.Second + DefaultUsageInterval = 60 * time.Second + DefaultTTLScanInterval = 15 * time.Second + DefaultDebounceWindow = 500 * time.Millisecond + DefaultDialTimeout = 10 * time.Second + + DefaultBackoffMin = 1 * time.Second + DefaultBackoffMax = 60 * time.Second + + DefaultStateDir = "/etc/pangolin-agent" + DefaultSingboxCfg = "/etc/sing-box/config.json" + + // DefaultFlow is the REALITY VLESS flow (doc/02 §3.1). + DefaultFlow = "xtls-rprx-vision" +) + +// Config holds everything the agent needs. Most fields come from cloud-init +// (control-plane address + bootstrap token) or have safe defaults. +type Config struct { + // ControlPlaneAddr is the control plane gRPC endpoint (host:port). + ControlPlaneAddr string + // ServerName overrides the TLS SNI used when dialing (defaults to the host + // part of ControlPlaneAddr). The control plane cert must match it. + ServerName string + + // BootstrapToken is the one-time cloud-init token consumed during Enroll. + // Ignored once the node already holds a certificate. + BootstrapToken string + + // StateDir holds node.key/node.crt/ca.crt and state.json. + StateDir string + // SingboxConfigPath is where the rendered sing-box config is written. + SingboxConfigPath string + + // DeriveKey keys the Hy2 password derivation (see DeriveHy2Password). + DeriveKey string + + AgentVersion string + + HeartbeatInterval time.Duration + UsageInterval time.Duration + TTLScanInterval time.Duration + DebounceWindow time.Duration + DialTimeout time.Duration + BackoffMin time.Duration + BackoffMax time.Duration + + // Insecure dials without mTLS. ONLY for local dev / integration tests. + Insecure bool +} + +// withDefaults returns a copy of c with zero-valued tunables filled in. +func (c Config) withDefaults() Config { + if c.StateDir == "" { + c.StateDir = DefaultStateDir + } + if c.SingboxConfigPath == "" { + c.SingboxConfigPath = DefaultSingboxCfg + } + if c.HeartbeatInterval == 0 { + c.HeartbeatInterval = DefaultHeartbeatInterval + } + if c.UsageInterval == 0 { + c.UsageInterval = DefaultUsageInterval + } + if c.TTLScanInterval == 0 { + c.TTLScanInterval = DefaultTTLScanInterval + } + if c.DebounceWindow == 0 { + c.DebounceWindow = DefaultDebounceWindow + } + if c.DialTimeout == 0 { + c.DialTimeout = DefaultDialTimeout + } + if c.BackoffMin == 0 { + c.BackoffMin = DefaultBackoffMin + } + if c.BackoffMax == 0 { + c.BackoffMax = DefaultBackoffMax + } + if c.AgentVersion == "" { + c.AgentVersion = "dev" + } + return c +} + +// Paths to the on-disk PKI + state material. +func (c Config) KeyPath() string { return filepath.Join(c.StateDir, "node.key") } +func (c Config) CertPath() string { return filepath.Join(c.StateDir, "node.crt") } +func (c Config) CAPath() string { return filepath.Join(c.StateDir, "ca.crt") } +func (c Config) StatePath() string { return filepath.Join(c.StateDir, "state.json") } diff --git a/server/internal/agentd/creds.go b/server/internal/agentd/creds.go new file mode 100644 index 0000000..2490b5f --- /dev/null +++ b/server/internal/agentd/creds.go @@ -0,0 +1,101 @@ +package agentd + +import ( + "context" + "crypto/tls" + "crypto/x509" + "errors" + "fmt" + "net" + "os" + + "google.golang.org/grpc" + "google.golang.org/grpc/credentials" + "google.golang.org/grpc/credentials/insecure" +) + +// Dialer opens an authenticated long-lived connection to the control plane. +type Dialer func(ctx context.Context) (*grpc.ClientConn, error) + +// mtlsClientConfig builds the client *tls.Config from the persisted node key/cert +// and pinned CA. +func mtlsClientConfig(cfg Config) (*tls.Config, error) { + cert, err := tls.LoadX509KeyPair(cfg.CertPath(), cfg.KeyPath()) + if err != nil { + return nil, fmt.Errorf("agentd: load client keypair: %w", err) + } + caPEM, err := os.ReadFile(cfg.CAPath()) + if err != nil { + return nil, fmt.Errorf("agentd: read CA: %w", err) + } + pool := x509.NewCertPool() + if !pool.AppendCertsFromPEM(caPEM) { + return nil, errors.New("agentd: CA file contains no certificate") + } + return &tls.Config{ + Certificates: []tls.Certificate{cert}, + RootCAs: pool, + ServerName: cfg.tlsServerName(), + MinVersion: tls.VersionTLS13, + }, nil +} + +// tlsServerName returns the SNI to validate the control-plane cert against. +func (c Config) tlsServerName() string { + if c.ServerName != "" { + return c.ServerName + } + host, _, err := net.SplitHostPort(c.ControlPlaneAddr) + if err != nil { + return c.ControlPlaneAddr + } + return host +} + +// NewMTLSDialer returns a Dialer that connects to the control plane over mTLS, +// presenting the node's client certificate. +func NewMTLSDialer(cfg Config) Dialer { + return func(ctx context.Context) (*grpc.ClientConn, error) { + tlsCfg, err := mtlsClientConfig(cfg) + if err != nil { + return nil, err + } + return grpc.NewClient(cfg.ControlPlaneAddr, + grpc.WithTransportCredentials(credentials.NewTLS(tlsCfg)), + ) + } +} + +// NewEnrollDialer returns an EnrollDialer for the bootstrap (pre-certificate) +// Enroll call. It pins the control-plane CA if ca.crt is already present +// (cloud-init may inject it); otherwise it falls back to a server-unauthenticated +// TLS handshake — the bootstrap token is the trust anchor for that single call. +func NewEnrollDialer(cfg Config) EnrollDialer { + return func(ctx context.Context) (*grpc.ClientConn, error) { + var creds credentials.TransportCredentials + if caPEM, err := os.ReadFile(cfg.CAPath()); err == nil { + pool := x509.NewCertPool() + if !pool.AppendCertsFromPEM(caPEM) { + return nil, errors.New("agentd: pinned CA file invalid") + } + creds = credentials.NewTLS(&tls.Config{ + RootCAs: pool, + ServerName: cfg.tlsServerName(), + MinVersion: tls.VersionTLS13, + }) + } else { + creds = credentials.NewTLS(&tls.Config{ + InsecureSkipVerify: true, //nolint:gosec // bootstrap-token-authenticated enroll only + MinVersion: tls.VersionTLS13, + }) + } + return grpc.NewClient(cfg.ControlPlaneAddr, grpc.WithTransportCredentials(creds)) + } +} + +// insecureDialer is used only by tests/dev (cfg.Insecure). +func insecureDialer(addr string) Dialer { + return func(ctx context.Context) (*grpc.ClientConn, error) { + return grpc.NewClient(addr, grpc.WithTransportCredentials(insecure.NewCredentials())) + } +} diff --git a/server/internal/agentd/derive.go b/server/internal/agentd/derive.go new file mode 100644 index 0000000..5a1d144 --- /dev/null +++ b/server/internal/agentd/derive.go @@ -0,0 +1,34 @@ +package agentd + +import ( + "crypto/hmac" + "crypto/sha256" + "encoding/base64" +) + +// DeriveHy2Password derives a node-agnostic Hysteria2 password from the opaque +// data-plane credential id (dp_uuid). The REALITY inbound uses dp_uuid directly as +// the VLESS user uuid; the Hy2 inbound cannot reuse a UUID as a password verbatim +// (it must look like an opaque secret), so it is derived from the SAME source — +// "password = dp_uuid 同源派生" — via a keyed HMAC. +// +// The derivation is deterministic given (dp_uuid, key): the control plane runs the +// exact same function when it builds the client's connect config (doc/02 §3.1), so +// both sides agree without the password ever crossing the agent contract. +// +// Risk note (carried from task #5): the derivation key is shared material. If a +// node is seized the attacker still only sees dp_uuids and this node's key, which +// lets them recompute Hy2 passwords for dp_uuids THEY ALREADY HOLD — it does not +// reveal other subscribers' credentials or any account identity. Rotating the key +// rotates every Hy2 password. When key == "" the password falls back to the raw +// dp_uuid (acceptable for dev; production cloud-init always injects a key). +func DeriveHy2Password(dpUUID, key string) string { + if key == "" { + return dpUUID + } + mac := hmac.New(sha256.New, []byte(key)) + mac.Write([]byte(dpUUID)) + sum := mac.Sum(nil) + // base64url without padding → URL/JSON-safe, 43 chars. + return base64.RawURLEncoding.EncodeToString(sum) +} diff --git a/server/internal/agentd/derive_test.go b/server/internal/agentd/derive_test.go new file mode 100644 index 0000000..f0a754f --- /dev/null +++ b/server/internal/agentd/derive_test.go @@ -0,0 +1,30 @@ +package agentd + +import "testing" + +func TestDeriveHy2Password(t *testing.T) { + const dp = "11111111-1111-1111-1111-111111111111" + + // Deterministic for a given (dp, key). + a := DeriveHy2Password(dp, "secret") + b := DeriveHy2Password(dp, "secret") + if a != b { + t.Fatalf("derivation not deterministic: %q != %q", a, b) + } + // Different key → different password. + if DeriveHy2Password(dp, "other") == a { + t.Fatal("different key produced identical password") + } + // Different dp_uuid → different password. + if DeriveHy2Password("22222222-2222-2222-2222-222222222222", "secret") == a { + t.Fatal("different dp_uuid produced identical password") + } + // The derived value is not the raw dp_uuid (must look like an opaque secret). + if a == dp { + t.Fatal("derived password equals raw dp_uuid") + } + // Empty key → raw dp_uuid fallback (dev mode). + if got := DeriveHy2Password(dp, ""); got != dp { + t.Fatalf("empty key fallback = %q, want raw dp_uuid", got) + } +} diff --git a/server/internal/agentd/enroll.go b/server/internal/agentd/enroll.go new file mode 100644 index 0000000..5fb98d7 --- /dev/null +++ b/server/internal/agentd/enroll.go @@ -0,0 +1,139 @@ +package agentd + +import ( + "context" + "crypto/ecdsa" + "crypto/elliptic" + "crypto/rand" + "crypto/x509" + "crypto/x509/pkix" + "encoding/pem" + "errors" + "fmt" + "io/fs" + "os" + + agentv1 "github.com/wangjia/pangolin/server/internal/pb/agentv1" + "google.golang.org/grpc" +) + +// EnrollDialer opens a connection suitable for the (un-authenticated) Enroll RPC. +// The agent has no client certificate yet, so this connection presents none; the +// bootstrap token authenticates the call instead. +type EnrollDialer func(ctx context.Context) (*grpc.ClientConn, error) + +// EnsureEnrolled guarantees the node holds a valid client certificate. +// +// - If node.crt/node.key already exist, it loads and returns the certificate's +// CN (the node UUID) without contacting the control plane (idempotent first +// start vs. restart). +// - Otherwise it generates an EC P-256 key + CSR, calls Enroll with the +// bootstrap token, and persists node.key (0600), node.crt and ca.crt. +// +// The private key is generated on the node and never leaves it. +func EnsureEnrolled(ctx context.Context, cfg Config, dial EnrollDialer) (string, error) { + cfg = cfg.withDefaults() + + if uuid, ok, err := loadEnrolledUUID(cfg); err != nil { + return "", err + } else if ok { + return uuid, nil + } + + if cfg.BootstrapToken == "" { + return "", errors.New("agentd: not enrolled and no bootstrap token provided") + } + + keyPEM, csrPEM, err := generateKeyAndCSR() + if err != nil { + return "", err + } + + conn, err := dial(ctx) + if err != nil { + return "", fmt.Errorf("agentd: dial for enroll: %w", err) + } + defer conn.Close() + + client := agentv1.NewAgentServiceClient(conn) + resp, err := client.Enroll(ctx, &agentv1.EnrollRequest{ + BootstrapToken: cfg.BootstrapToken, + CSRPEM: csrPEM, + AgentVersion: cfg.AgentVersion, + }) + if err != nil { + return "", fmt.Errorf("agentd: enroll rpc: %w", err) + } + if resp.NodeUUID == "" || len(resp.CertPEM) == 0 { + return "", errors.New("agentd: enroll response missing node_uuid/cert") + } + + if err := os.MkdirAll(cfg.StateDir, 0o700); err != nil { + return "", fmt.Errorf("agentd: mkdir state dir: %w", err) + } + if err := atomicWrite(cfg.KeyPath(), keyPEM, 0o600); err != nil { + return "", err + } + if err := atomicWrite(cfg.CertPath(), resp.CertPEM, 0o644); err != nil { + return "", err + } + if len(resp.CAPEM) > 0 { + if err := atomicWrite(cfg.CAPath(), resp.CAPEM, 0o644); err != nil { + return "", err + } + } + logf("enrolled successfully (cert valid until unix=%d)", resp.NotAfterUnix) + return resp.NodeUUID, nil +} + +// loadEnrolledUUID returns the CN of the persisted node certificate, if present. +func loadEnrolledUUID(cfg Config) (string, bool, error) { + certPEM, err := os.ReadFile(cfg.CertPath()) + if errors.Is(err, fs.ErrNotExist) { + return "", false, nil + } + if err != nil { + return "", false, fmt.Errorf("agentd: read node cert: %w", err) + } + if _, err := os.Stat(cfg.KeyPath()); err != nil { + // Cert without key is unusable — treat as not enrolled. + return "", false, nil + } + block, _ := pem.Decode(certPEM) + if block == nil { + return "", false, errors.New("agentd: node cert has no PEM block") + } + cert, err := x509.ParseCertificate(block.Bytes) + if err != nil { + return "", false, fmt.Errorf("agentd: parse node cert: %w", err) + } + if cert.Subject.CommonName == "" { + return "", false, errors.New("agentd: node cert has empty CN") + } + return cert.Subject.CommonName, true, nil +} + +// generateKeyAndCSR creates an EC P-256 private key and a PKCS#10 CSR. The CSR's +// CN is a placeholder; the CA overrides it with the authoritative node UUID. +func generateKeyAndCSR() (keyPEM, csrPEM []byte, err error) { + key, err := ecdsa.GenerateKey(elliptic.P256(), rand.Reader) + if err != nil { + return nil, nil, fmt.Errorf("agentd: generate key: %w", err) + } + keyDER, err := x509.MarshalECPrivateKey(key) + if err != nil { + return nil, nil, fmt.Errorf("agentd: marshal key: %w", err) + } + keyPEM = pem.EncodeToMemory(&pem.Block{Type: "EC PRIVATE KEY", Bytes: keyDER}) + + tmpl := &x509.CertificateRequest{ + Subject: pkix.Name{CommonName: "pangolin-node-pending"}, + SignatureAlgorithm: x509.ECDSAWithSHA256, + } + csrDER, err := x509.CreateCertificateRequest(rand.Reader, tmpl, key) + if err != nil { + return nil, nil, fmt.Errorf("agentd: create CSR: %w", err) + } + csrPEM = pem.EncodeToMemory(&pem.Block{Type: "CERTIFICATE REQUEST", Bytes: csrDER}) + return keyPEM, csrPEM, nil +} diff --git a/server/internal/agentd/heartbeat.go b/server/internal/agentd/heartbeat.go new file mode 100644 index 0000000..2b76e03 --- /dev/null +++ b/server/internal/agentd/heartbeat.go @@ -0,0 +1,63 @@ +package agentd + +import ( + "context" + "time" + + agentv1 "github.com/wangjia/pangolin/server/internal/pb/agentv1" +) + +// LoadSource provides the runtime metrics reported in each heartbeat. The +// production implementation can read sing-box stats + host CPU; the default only +// reports the online peer count (number of provisioned credentials), which never +// reveals identities. +type LoadSource interface { + Load() (onlinePeers int32, upBps, downBps int64, cpuPercent float64) +} + +// defaultLoadSource reports peer count from the SingBox table and zeros for the +// rest. Bandwidth/CPU collection is a documented extension point. +type defaultLoadSource struct{ sb *SingBox } + +func (d defaultLoadSource) Load() (int32, int64, int64, float64) { + return int32(d.sb.OnlinePeers()), 0, 0, 0 +} + +// runHeartbeat sends a heartbeat every cfg.HeartbeatInterval until ctx is +// cancelled or an RPC error occurs. A need_full_resync response returns +// errResync so the session re-Registers and overwrites local state. +func (a *Agent) runHeartbeat(ctx context.Context, client agentv1.AgentServiceClient) error { + ticker := time.NewTicker(a.cfg.HeartbeatInterval) + defer ticker.Stop() + for { + if err := a.sendHeartbeat(ctx, client); err != nil { + return err + } + select { + case <-ctx.Done(): + return ctx.Err() + case <-ticker.C: + } + } +} + +func (a *Agent) sendHeartbeat(ctx context.Context, client agentv1.AgentServiceClient) error { + peers, up, down, cpu := a.load.Load() + resp, err := client.Heartbeat(ctx, &agentv1.HeartbeatRequest{ + NodeUUID: a.NodeUUID(), + ConfigVersion: a.sb.ConfigVersion(), + OnlinePeers: peers, + BandwidthUpBps: up, + BandwidthDownBps: down, + CPUPercent: cpu, + TimestampUnix: a.clock().Unix(), + }) + if err != nil { + return err + } + if resp.NeedFullResync { + logf("control plane requested full resync") + return errResync + } + return nil +} diff --git a/server/internal/agentd/integration_test.go b/server/internal/agentd/integration_test.go new file mode 100644 index 0000000..180159a --- /dev/null +++ b/server/internal/agentd/integration_test.go @@ -0,0 +1,370 @@ +package agentd + +import ( + "context" + "net" + "os" + "sort" + "strings" + "sync" + "testing" + "time" + + "github.com/wangjia/pangolin/server/internal/mtls" + agentv1 "github.com/wangjia/pangolin/server/internal/pb/agentv1" + "google.golang.org/grpc" + "google.golang.org/grpc/codes" + "google.golang.org/grpc/credentials/insecure" + "google.golang.org/grpc/status" + "google.golang.org/grpc/test/bufconn" +) + +const testNodeUUID = "node-test-uuid" + +// mockCP is an in-memory control plane implementing AgentServiceServer. +type mockCP struct { + ca *mtls.CA + + mu sync.Mutex + snapCreds map[string]*agentv1.Credential // converged credential set for Register + configVer int64 + lastCmdID int64 + issued []*agentv1.Command // backlog for resume-after-drop + acked map[int64]bool + registerN int + heartbeatN int + subReqs []int64 + usage []*agentv1.UsageReport + resyncOnce bool + enrollToken string + + liveCh chan *agentv1.Command + dropCh chan struct{} +} + +func newMockCP(t *testing.T) *mockCP { + t.Helper() + dir := t.TempDir() + ca, err := mtls.NewCA(mtls.CAConfig{ + KeyPath: dir + "/ca.key", + CertPath: dir + "/ca.crt", + }) + if err != nil { + t.Fatalf("new CA: %v", err) + } + return &mockCP{ + ca: ca, + snapCreds: map[string]*agentv1.Credential{}, + acked: map[int64]bool{}, + enrollToken: "test-token", + liveCh: make(chan *agentv1.Command), + dropCh: make(chan struct{}, 1), + } +} + +func (m *mockCP) Enroll(_ context.Context, req *agentv1.EnrollRequest) (*agentv1.EnrollResponse, error) { + if req.BootstrapToken != m.enrollToken { + return nil, status.Error(codes.Unauthenticated, "bad bootstrap token") + } + certPEM, err := m.ca.SignCSR(req.CSRPEM, testNodeUUID) + if err != nil { + return nil, status.Errorf(codes.InvalidArgument, "sign csr: %v", err) + } + return &agentv1.EnrollResponse{ + NodeUUID: testNodeUUID, + CertPEM: certPEM, + CAPEM: m.ca.CAPEM(), + NotAfterUnix: time.Now().Add(90 * 24 * time.Hour).Unix(), + }, nil +} + +func (m *mockCP) Register(_ context.Context, _ *agentv1.RegisterRequest) (*agentv1.ConfigSnapshot, error) { + m.mu.Lock() + defer m.mu.Unlock() + m.registerN++ + creds := make([]*agentv1.Credential, 0, len(m.snapCreds)) + for _, c := range m.snapCreds { + creds = append(creds, c) + } + sort.Slice(creds, func(i, j int) bool { return creds[i].DpUUID < creds[j].DpUUID }) + return &agentv1.ConfigSnapshot{ + ConfigVersion: m.configVer, + Credentials: creds, + Reality: &agentv1.RealityInbound{ + ListenPort: 11443, PrivateKey: "pk", ShortID: "deadbeef", + ServerName: "www.apple.com", HandshakeServer: "www.apple.com", HandshakePort: 443, + }, + Hy2: &agentv1.Hy2Inbound{ListenPort: 443, Masquerade: "https://www.bing.com", CertPath: "/c", KeyPath: "/k"}, + LastCommandID: m.lastCmdID, + }, nil +} + +func (m *mockCP) Heartbeat(_ context.Context, _ *agentv1.HeartbeatRequest) (*agentv1.HeartbeatResponse, error) { + m.mu.Lock() + defer m.mu.Unlock() + m.heartbeatN++ + resync := false + if m.resyncOnce { + m.resyncOnce = false + resync = true + } + return &agentv1.HeartbeatResponse{NeedFullResync: resync, ServerTimeUnix: time.Now().Unix()}, nil +} + +func (m *mockCP) Ack(_ context.Context, req *agentv1.AckRequest) (*agentv1.AckResponse, error) { + m.mu.Lock() + m.acked[req.CommandID] = true + m.mu.Unlock() + return &agentv1.AckResponse{}, nil +} + +func (m *mockCP) ReportUsage(_ context.Context, req *agentv1.UsageReport) (*agentv1.UsageAck, error) { + m.mu.Lock() + m.usage = append(m.usage, req) + m.mu.Unlock() + return &agentv1.UsageAck{}, nil +} + +func (m *mockCP) Subscribe(req *agentv1.SubscribeRequest, stream agentv1.AgentService_SubscribeServer) error { + m.mu.Lock() + m.subReqs = append(m.subReqs, req.LastCommandID) + backlog := make([]*agentv1.Command, 0) + for _, c := range m.issued { + if c.CommandID > req.LastCommandID { + backlog = append(backlog, c) + } + } + m.mu.Unlock() + + for _, c := range backlog { + if err := stream.Send(c); err != nil { + return err + } + } + for { + select { + case <-stream.Context().Done(): + return stream.Context().Err() + case <-m.dropCh: + return status.Error(codes.Unavailable, "simulated stream drop") + case c := <-m.liveCh: + if err := stream.Send(c); err != nil { + return err + } + } + } +} + +// ─── mock helpers (test goroutine) ─────────────────────────────────────────── + +// seedCred adds a credential to the converged Register snapshot before start. +func (m *mockCP) seedCred(c *agentv1.Credential) { + m.mu.Lock() + m.snapCreds[c.DpUUID] = c + if m.configVer == 0 { + m.configVer = 1 + } + m.mu.Unlock() +} + +// push issues a live command AND folds it into the Register snapshot, so a later +// reconnect's Register stays consistent with the command stream. +func (m *mockCP) push(cmd *agentv1.Command) { + m.applyToSnapshot(cmd, true) // updates snapCreds + lastCmdID + m.mu.Lock() + m.issued = append(m.issued, cmd) + m.mu.Unlock() + m.liveCh <- cmd +} + +// enqueueOffline appends a command to the resume backlog WITHOUT touching the +// Register snapshot — so the only way the agent learns it is via Subscribe resume. +func (m *mockCP) enqueueOffline(cmd *agentv1.Command) { + m.mu.Lock() + // Intentionally does NOT advance lastCmdID: this command is not yet reflected + // in the Register snapshot, so the agent must learn it via Subscribe resume. + m.issued = append(m.issued, cmd) + m.mu.Unlock() +} + +func (m *mockCP) applyToSnapshot(cmd *agentv1.Command, bumpVer bool) { + m.mu.Lock() + defer m.mu.Unlock() + switch cmd.Type { + case agentv1.CommandTypeUpsert: + m.snapCreds[cmd.Upsert.Credential.DpUUID] = cmd.Upsert.Credential + case agentv1.CommandTypeRevoke: + delete(m.snapCreds, cmd.Revoke.DpUUID) + } + if cmd.CommandID > m.lastCmdID { + m.lastCmdID = cmd.CommandID + } + if bumpVer { + m.configVer++ + } +} + +func (m *mockCP) drop() { m.dropCh <- struct{}{} } +func (m *mockCP) registerCount() int { m.mu.Lock(); defer m.mu.Unlock(); return m.registerN } +func (m *mockCP) heartbeatCount() int { m.mu.Lock(); defer m.mu.Unlock(); return m.heartbeatN } +func (m *mockCP) isAcked(id int64) bool { m.mu.Lock(); defer m.mu.Unlock(); return m.acked[id] } +func (m *mockCP) subReqList() []int64 { + m.mu.Lock() + defer m.mu.Unlock() + return append([]int64(nil), m.subReqs...) +} + +// ─── test harness ───────────────────────────────────────────────────────────── + +func startMock(t *testing.T, m *mockCP) func(ctx context.Context) (*grpc.ClientConn, error) { + t.Helper() + lis := bufconn.Listen(1024 * 1024) + srv := grpc.NewServer() + agentv1.RegisterAgentServiceServer(srv, m) + go func() { _ = srv.Serve(lis) }() + t.Cleanup(func() { srv.Stop() }) + return func(ctx context.Context) (*grpc.ClientConn, error) { + return grpc.NewClient("passthrough:///bufnet", + grpc.WithContextDialer(func(ctx context.Context, _ string) (net.Conn, error) { + return lis.DialContext(ctx) + }), + grpc.WithTransportCredentials(insecure.NewCredentials()), + ) + } +} + +func fastConfig(t *testing.T) Config { + cfg := testConfig(t) + cfg.BootstrapToken = "test-token" + cfg.HeartbeatInterval = 25 * time.Millisecond + cfg.UsageInterval = time.Hour + cfg.TTLScanInterval = time.Hour + cfg.DebounceWindow = 10 * time.Millisecond + cfg.BackoffMin = 10 * time.Millisecond + cfg.BackoffMax = 40 * time.Millisecond + return cfg +} + +func eventually(t *testing.T, timeout time.Duration, fn func() bool, msg string) { + t.Helper() + deadline := time.Now().Add(timeout) + for time.Now().Before(deadline) { + if fn() { + return + } + time.Sleep(5 * time.Millisecond) + } + t.Fatalf("timeout waiting for: %s", msg) +} + +func readRender(path string) string { + b, err := os.ReadFile(path) + if err != nil { + return "" + } + return string(b) +} + +// TestIntegrationFullFlow exercises Enroll → Register → Heartbeat → Subscribe/Ack, +// command application (upsert/revoke), render correctness, heartbeat-triggered +// resync, and reconnect resume via last_command_id. +func TestIntegrationFullFlow(t *testing.T) { + m := newMockCP(t) + m.seedCred(&agentv1.Credential{DpUUID: "boot", Protocol: agentv1.ProtocolBoth}) + dial := startMock(t, m) + + cfg := fastConfig(t) + a := New(cfg, + WithDialer(Dialer(dial)), + WithEnrollDialer(EnrollDialer(dial)), + WithRestarter(&fakeRestarter{}), + ) + + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + runErr := make(chan error, 1) + go func() { runErr <- a.Run(ctx) }() + + // 1) Enrollment persisted the cert/key/ca and the node UUID is set. + eventually(t, 2*time.Second, func() bool { + _, err1 := os.Stat(cfg.CertPath()) + _, err2 := os.Stat(cfg.KeyPath()) + _, err3 := os.Stat(cfg.CAPath()) + return err1 == nil && err2 == nil && err3 == nil && a.NodeUUID() == testNodeUUID + }, "enrollment to persist cert/key/ca and set node UUID") + + // 2) Initial Register snapshot rendered with the seeded credential. + eventually(t, 2*time.Second, func() bool { + return strings.Contains(readRender(cfg.SingboxConfigPath), `"uuid": "boot"`) + }, "initial snapshot credential 'boot' to be rendered") + + // 3) Heartbeats flowing. + eventually(t, 2*time.Second, func() bool { return m.heartbeatCount() > 0 }, "heartbeats") + + // 4) Stream a command: upsert 'extra'. Applied + Ack'd + rendered. + m.push(&agentv1.Command{ + CommandID: 1, Type: agentv1.CommandTypeUpsert, + Upsert: &agentv1.UpsertPayload{Credential: &agentv1.Credential{DpUUID: "extra", Protocol: agentv1.ProtocolBoth}}, + }) + eventually(t, 2*time.Second, func() bool { + return strings.Contains(readRender(cfg.SingboxConfigPath), `"uuid": "extra"`) && m.isAcked(1) + }, "upsert 'extra' applied and acked") + + // 5) Stream a revoke for 'boot'. Removed from render + Ack'd. + m.push(&agentv1.Command{ + CommandID: 2, Type: agentv1.CommandTypeRevoke, + Revoke: &agentv1.RevokePayload{DpUUID: "boot"}, + }) + eventually(t, 2*time.Second, func() bool { + r := readRender(cfg.SingboxConfigPath) + return !strings.Contains(r, `"uuid": "boot"`) && m.isAcked(2) + }, "revoke 'boot' applied and acked") + + // 6) Reconnect resume: enqueue an OFFLINE command (only reachable via backlog), + // then drop the stream. After reconnect the agent must resume from + // last_command_id and receive id=3 without redelivering 1/2. + m.enqueueOffline(&agentv1.Command{ + CommandID: 3, Type: agentv1.CommandTypeUpsert, + Upsert: &agentv1.UpsertPayload{Credential: &agentv1.Credential{DpUUID: "resumed", Protocol: agentv1.ProtocolBoth}}, + }) + m.drop() + eventually(t, 3*time.Second, func() bool { + return strings.Contains(readRender(cfg.SingboxConfigPath), `"uuid": "resumed"`) && m.isAcked(3) + }, "offline command 3 delivered after reconnect resume") + + // The second Subscribe must have resumed from a non-zero high-water mark. + reqs := m.subReqList() + if len(reqs) < 2 { + t.Fatalf("expected at least 2 Subscribe calls (reconnect), got %v", reqs) + } + if reqs[len(reqs)-1] < 2 { + t.Errorf("reconnect Subscribe last_command_id = %d, want >= 2 (resume, no redelivery)", reqs[len(reqs)-1]) + } + + cancel() + select { + case <-runErr: + case <-time.After(2 * time.Second): + t.Fatal("agent did not shut down after context cancel") + } +} + +// TestIntegrationHeartbeatResync verifies a need_full_resync heartbeat forces a +// reconnect + re-Register so the node converges on the authoritative snapshot. +func TestIntegrationHeartbeatResync(t *testing.T) { + m := newMockCP(t) + m.seedCred(&agentv1.Credential{DpUUID: "boot", Protocol: agentv1.ProtocolBoth}) + m.resyncOnce = true + dial := startMock(t, m) + + a := New(fastConfig(t), + WithDialer(Dialer(dial)), + WithEnrollDialer(EnrollDialer(dial)), + WithRestarter(&fakeRestarter{}), + ) + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + go func() { _ = a.Run(ctx) }() + + eventually(t, 3*time.Second, func() bool { return m.registerCount() >= 2 }, "re-Register after need_full_resync") +} diff --git a/server/internal/agentd/render.go b/server/internal/agentd/render.go new file mode 100644 index 0000000..2d5dd8d --- /dev/null +++ b/server/internal/agentd/render.go @@ -0,0 +1,95 @@ +package agentd + +import ( + "encoding/json" + + agentv1 "github.com/wangjia/pangolin/server/internal/pb/agentv1" +) + +// renderSingboxConfig produces a complete sing-box SERVER config JSON for the node: +// a VLESS+REALITY inbound and a Hysteria2 inbound, each carrying one user per +// provisioned credential. REALITY users key on the dp_uuid; Hy2 users key on the +// derived password (DeriveHy2Password) — both from the same dp_uuid source. +// +// Only the opaque dp_uuid is ever written; no account identity touches the node. +func renderSingboxConfig(creds []Cred, reality *agentv1.RealityInbound, hy2 *agentv1.Hy2Inbound, deriveKey string) ([]byte, error) { + cfg := map[string]any{ + "log": map[string]any{"level": "warn", "timestamp": true}, + "inbounds": buildInbounds(creds, reality, hy2, deriveKey), + "outbounds": []any{map[string]any{"type": "direct", "tag": "direct"}}, + } + return json.MarshalIndent(cfg, "", " ") +} + +func buildInbounds(creds []Cred, reality *agentv1.RealityInbound, hy2 *agentv1.Hy2Inbound, deriveKey string) []any { + inbounds := make([]any, 0, 2) + + if reality != nil { + users := make([]any, 0, len(creds)) + for _, c := range creds { + if c.Protocol == agentv1.ProtocolReality || c.Protocol == agentv1.ProtocolBoth { + flow := c.Flow + if flow == "" { + flow = DefaultFlow + } + users = append(users, map[string]any{ + "name": c.DpUUID, + "uuid": c.DpUUID, + "flow": flow, + }) + } + } + realityTLS := map[string]any{ + "enabled": true, + "server_name": reality.ServerName, + "reality": map[string]any{ + "enabled": true, + "private_key": reality.PrivateKey, + "short_id": []string{reality.ShortID}, + "handshake": map[string]any{ + "server": reality.HandshakeServer, + "server_port": reality.HandshakePort, + }, + }, + } + inbounds = append(inbounds, map[string]any{ + "type": "vless", + "tag": "reality-in", + "listen": "::", + "listen_port": reality.ListenPort, + "users": users, + "tls": realityTLS, + }) + } + + if hy2 != nil { + users := make([]any, 0, len(creds)) + for _, c := range creds { + if c.Protocol == agentv1.ProtocolHy2 || c.Protocol == agentv1.ProtocolBoth { + users = append(users, map[string]any{ + "name": c.DpUUID, + "password": DeriveHy2Password(c.DpUUID, deriveKey), + }) + } + } + hy2In := map[string]any{ + "type": "hysteria2", + "tag": "hy2-in", + "listen": "::", + "listen_port": hy2.ListenPort, + "users": users, + "tls": map[string]any{ + "enabled": true, + "alpn": []string{"h3"}, + "certificate_path": hy2.CertPath, + "key_path": hy2.KeyPath, + }, + } + if hy2.Masquerade != "" { + hy2In["masquerade"] = hy2.Masquerade + } + inbounds = append(inbounds, hy2In) + } + + return inbounds +} diff --git a/server/internal/agentd/singbox.go b/server/internal/agentd/singbox.go new file mode 100644 index 0000000..24f7fe2 --- /dev/null +++ b/server/internal/agentd/singbox.go @@ -0,0 +1,354 @@ +package agentd + +import ( + "context" + "encoding/json" + "errors" + "fmt" + "io/fs" + "os" + "path/filepath" + "sort" + "sync" + "time" + + agentv1 "github.com/wangjia/pangolin/server/internal/pb/agentv1" +) + +// Restarter applies a freshly-rendered sing-box config. sing-box has no hot +// reload, so the production implementation restarts the systemd unit (a ~1-2s +// data-plane blip). Tests inject a fake. +type Restarter interface { + Restart(ctx context.Context) error +} + +// Cred is the agent's in-memory view of one data-plane credential. It is the +// authoritative local state together with the rendered sing-box config. +type Cred struct { + DpUUID string + Protocol agentv1.Protocol + Flow string + ExpiresAt int64 // unix seconds; 0 = no expiry +} + +// persistedCred is the on-disk shape. Per the no-state invariant we persist ONLY +// dp_uuid + expires_at — protocol/flow are reconstructed (defaults) on load and +// then corrected by the next Register reconciliation. +type persistedCred struct { + DpUUID string `json:"dp_uuid"` + ExpiresAt int64 `json:"expires_at"` +} + +type persistedState struct { + Creds []persistedCred `json:"creds"` +} + +// SingBox owns the credential table, state.json persistence, sing-box config +// rendering and the debounced restart loop. +type SingBox struct { + cfg Config + restarter Restarter + clock func() time.Time + + // OnExpire, if set, is invoked (outside the lock) when the TTL sweep removes + // a credential, so the agent can surface the removal upstream. + OnExpire func(dpUUID string) + + mu sync.Mutex + creds map[string]*Cred + reality *agentv1.RealityInbound + hy2 *agentv1.Hy2Inbound + configVersion int64 + + // debounce machinery + dirty chan struct{} + done chan struct{} + closed bool + startWG sync.WaitGroup +} + +// NewSingBox builds a manager. restarter may be nil (renders only, no restart). +func NewSingBox(cfg Config, restarter Restarter) *SingBox { + if restarter == nil { + restarter = noopRestarter{} + } + return &SingBox{ + cfg: cfg.withDefaults(), + restarter: restarter, + clock: time.Now, + creds: make(map[string]*Cred), + dirty: make(chan struct{}, 1), + done: make(chan struct{}), + } +} + +// LoadState restores the credential table from state.json after a restart. Only +// dp_uuid + expires_at survive; protocol defaults to BOTH and flow to the +// configured default until the next Register overwrites them. Expired entries are +// dropped on load. +func (s *SingBox) LoadState() error { + data, err := os.ReadFile(s.cfg.StatePath()) + if errors.Is(err, fs.ErrNotExist) { + return nil + } + if err != nil { + return fmt.Errorf("agentd: read state: %w", err) + } + var ps persistedState + if err := json.Unmarshal(data, &ps); err != nil { + return fmt.Errorf("agentd: parse state: %w", err) + } + now := s.clock().Unix() + s.mu.Lock() + defer s.mu.Unlock() + for _, pc := range ps.Creds { + if pc.ExpiresAt != 0 && pc.ExpiresAt <= now { + continue + } + s.creds[pc.DpUUID] = &Cred{ + DpUUID: pc.DpUUID, + Protocol: agentv1.ProtocolBoth, + Flow: DefaultFlow, + ExpiresAt: pc.ExpiresAt, + } + } + return nil +} + +// saveStateLocked writes state.json atomically. Caller holds s.mu. +func (s *SingBox) saveStateLocked() error { + ps := persistedState{Creds: make([]persistedCred, 0, len(s.creds))} + for _, c := range s.creds { + ps.Creds = append(ps.Creds, persistedCred{DpUUID: c.DpUUID, ExpiresAt: c.ExpiresAt}) + } + sort.Slice(ps.Creds, func(i, j int) bool { return ps.Creds[i].DpUUID < ps.Creds[j].DpUUID }) + data, err := json.MarshalIndent(ps, "", " ") + if err != nil { + return err + } + return atomicWrite(s.cfg.StatePath(), data, 0o600) +} + +// ─── mutations (all idempotent, all schedule a render) ─────────────────────────── + +// Upsert adds or replaces a credential. +func (s *SingBox) Upsert(c *Cred) { + s.mu.Lock() + cp := *c + if cp.Flow == "" { + cp.Flow = DefaultFlow + } + s.creds[cp.DpUUID] = &cp + _ = s.saveStateLocked() + s.mu.Unlock() + s.markDirty() +} + +// Revoke removes a credential. No-op if absent (idempotent). +func (s *SingBox) Revoke(dpUUID string) { + s.mu.Lock() + _, existed := s.creds[dpUUID] + delete(s.creds, dpUUID) + if existed { + _ = s.saveStateLocked() + } + s.mu.Unlock() + if existed { + s.markDirty() + } +} + +// Rotate installs new alongside old, keeping old alive until graceUntil. The TTL +// sweep removes old when its grace expires, giving "宽限期内新旧并存". +func (s *SingBox) Rotate(oldDpUUID string, newCred *Cred, graceUntil int64) { + s.mu.Lock() + if old, ok := s.creds[oldDpUUID]; ok && oldDpUUID != newCred.DpUUID { + // Clamp old credential's expiry to the grace deadline. + if graceUntil > 0 && (old.ExpiresAt == 0 || graceUntil < old.ExpiresAt) { + old.ExpiresAt = graceUntil + } + } + nc := *newCred + if nc.Flow == "" { + nc.Flow = DefaultFlow + } + s.creds[nc.DpUUID] = &nc + _ = s.saveStateLocked() + s.mu.Unlock() + s.markDirty() +} + +// ApplyConfig replaces the node-wide inbound parameters and (if the snapshot +// carries credentials) the full credential table. Used by APPLY_CONFIG commands +// and full resync. +func (s *SingBox) ApplyConfig(snap *agentv1.ConfigSnapshot, replaceCreds bool) { + s.mu.Lock() + if snap.Reality != nil { + s.reality = snap.Reality + } + if snap.Hy2 != nil { + s.hy2 = snap.Hy2 + } + if snap.ConfigVersion != 0 { + s.configVersion = snap.ConfigVersion + } + if replaceCreds { + s.creds = make(map[string]*Cred, len(snap.Credentials)) + for _, c := range snap.Credentials { + flow := c.Flow + if flow == "" { + flow = DefaultFlow + } + s.creds[c.DpUUID] = &Cred{ + DpUUID: c.DpUUID, + Protocol: c.Protocol, + Flow: flow, + ExpiresAt: c.ExpiresAtUnix, + } + } + } + _ = s.saveStateLocked() + s.mu.Unlock() + s.markDirty() +} + +// SetConfigVersion records the authoritative config version (e.g. from Register). +func (s *SingBox) SetConfigVersion(v int64) { + s.mu.Lock() + s.configVersion = v + s.mu.Unlock() +} + +// ─── read accessors ────────────────────────────────────────────────────────── + +func (s *SingBox) ConfigVersion() int64 { + s.mu.Lock() + defer s.mu.Unlock() + return s.configVersion +} + +func (s *SingBox) OnlinePeers() int { + s.mu.Lock() + defer s.mu.Unlock() + return len(s.creds) +} + +// Has reports whether a dp_uuid is currently provisioned. +func (s *SingBox) Has(dpUUID string) bool { + s.mu.Lock() + defer s.mu.Unlock() + _, ok := s.creds[dpUUID] + return ok +} + +// Snapshot returns a copy of the current credential set (sorted by dp_uuid). +func (s *SingBox) Snapshot() []Cred { + s.mu.Lock() + defer s.mu.Unlock() + out := make([]Cred, 0, len(s.creds)) + for _, c := range s.creds { + out = append(out, *c) + } + sort.Slice(out, func(i, j int) bool { return out[i].DpUUID < out[j].DpUUID }) + return out +} + +// ─── TTL sweep ─────────────────────────────────────────────────────────────── + +// sweepExpired removes every credential whose expires_at has passed. Returns the +// removed dp_uuids and whether anything changed. +func (s *SingBox) sweepExpired() []string { + now := s.clock().Unix() + s.mu.Lock() + var removed []string + for id, c := range s.creds { + if c.ExpiresAt != 0 && c.ExpiresAt <= now { + delete(s.creds, id) + removed = append(removed, id) + } + } + if len(removed) > 0 { + _ = s.saveStateLocked() + } + s.mu.Unlock() + if len(removed) > 0 { + s.markDirty() + if s.OnExpire != nil { + for _, id := range removed { + s.OnExpire(id) + } + } + } + sort.Strings(removed) + return removed +} + +// ─── rendering ─────────────────────────────────────────────────────────────── + +// RenderConfig builds the sing-box server config JSON from the current state. +// Exported (and pure w.r.t. the snapshot) so tests can assert its contents. +func (s *SingBox) RenderConfig() ([]byte, error) { + s.mu.Lock() + creds := make([]Cred, 0, len(s.creds)) + for _, c := range s.creds { + creds = append(creds, *c) + } + reality := s.reality + hy2 := s.hy2 + s.mu.Unlock() + sort.Slice(creds, func(i, j int) bool { return creds[i].DpUUID < creds[j].DpUUID }) + return renderSingboxConfig(creds, reality, hy2, s.cfg.DeriveKey) +} + +// writeAndRestart renders, writes the config file and restarts sing-box. +func (s *SingBox) writeAndRestart(ctx context.Context) error { + data, err := s.RenderConfig() + if err != nil { + return err + } + if err := os.MkdirAll(filepath.Dir(s.cfg.SingboxConfigPath), 0o755); err != nil { + return fmt.Errorf("agentd: mkdir singbox cfg: %w", err) + } + if err := atomicWrite(s.cfg.SingboxConfigPath, data, 0o644); err != nil { + return err + } + return s.restarter.Restart(ctx) +} + +// markDirty signals the debounce loop that a render is pending (non-blocking). +func (s *SingBox) markDirty() { + select { + case s.dirty <- struct{}{}: + default: + } +} + +// Run drives the debounce loop: it coalesces bursts of changes within +// DebounceWindow into a single write+restart. It blocks until ctx is cancelled. +func (s *SingBox) Run(ctx context.Context) { + timer := time.NewTimer(time.Hour) + timer.Stop() + pending := false + for { + select { + case <-ctx.Done(): + timer.Stop() + return + case <-s.dirty: + if !pending { + pending = true + timer.Reset(s.cfg.DebounceWindow) + } + case <-timer.C: + pending = false + if err := s.writeAndRestart(ctx); err != nil { + logf("singbox render/restart failed: %v", err) + } + } + } +} + +// noopRestarter renders without restarting (dev / config-check only). +type noopRestarter struct{} + +func (noopRestarter) Restart(context.Context) error { return nil } diff --git a/server/internal/agentd/singbox_test.go b/server/internal/agentd/singbox_test.go new file mode 100644 index 0000000..209ee9c --- /dev/null +++ b/server/internal/agentd/singbox_test.go @@ -0,0 +1,229 @@ +package agentd + +import ( + "context" + "encoding/json" + "os" + "path/filepath" + "strings" + "sync" + "testing" + "time" + + agentv1 "github.com/wangjia/pangolin/server/internal/pb/agentv1" +) + +// fakeRestarter counts Restart calls. +type fakeRestarter struct { + mu sync.Mutex + n int +} + +func (f *fakeRestarter) Restart(context.Context) error { + f.mu.Lock() + f.n++ + f.mu.Unlock() + return nil +} + +func (f *fakeRestarter) count() int { + f.mu.Lock() + defer f.mu.Unlock() + return f.n +} + +func testConfig(t *testing.T) Config { + t.Helper() + dir := t.TempDir() + return Config{ + ControlPlaneAddr: "bufnet", + StateDir: dir, + SingboxConfigPath: filepath.Join(dir, "singbox", "config.json"), + DeriveKey: "test-derive-key", + DebounceWindow: 20 * time.Millisecond, + Insecure: true, + }.withDefaults() +} + +func sampleSnapshot(creds ...*agentv1.Credential) *agentv1.ConfigSnapshot { + return &agentv1.ConfigSnapshot{ + ConfigVersion: 7, + Credentials: creds, + Reality: &agentv1.RealityInbound{ + ListenPort: 11443, PrivateKey: "pk", ShortID: "deadbeef", + ServerName: "www.apple.com", HandshakeServer: "www.apple.com", HandshakePort: 443, + }, + Hy2: &agentv1.Hy2Inbound{ListenPort: 443, Masquerade: "https://www.bing.com", CertPath: "/c", KeyPath: "/k"}, + LastCommandID: 3, + } +} + +func TestRenderContainsCredentials(t *testing.T) { + sb := NewSingBox(testConfig(t), nil) + sb.ApplyConfig(sampleSnapshot( + &agentv1.Credential{DpUUID: "aaaa", Protocol: agentv1.ProtocolBoth}, + &agentv1.Credential{DpUUID: "bbbb", Protocol: agentv1.ProtocolReality}, + &agentv1.Credential{DpUUID: "cccc", Protocol: agentv1.ProtocolHy2}, + ), true) + + data, err := sb.RenderConfig() + if err != nil { + t.Fatal(err) + } + s := string(data) + + // REALITY inbound: aaaa + bbbb as VLESS uuids, with flow; cccc must NOT be there. + for _, want := range []string{`"uuid": "aaaa"`, `"uuid": "bbbb"`, `"flow": "xtls-rprx-vision"`} { + if !strings.Contains(s, want) { + t.Errorf("rendered config missing %q\n%s", want, s) + } + } + if strings.Contains(s, `"uuid": "cccc"`) { + t.Errorf("hy2-only credential cccc leaked into REALITY users") + } + + // Hy2 inbound: derived passwords for aaaa + cccc, never the raw dp_uuid as pw. + var cfg map[string]any + if err := json.Unmarshal(data, &cfg); err != nil { + t.Fatalf("rendered config is not valid JSON: %v", err) + } + if !strings.Contains(s, DeriveHy2Password("aaaa", "test-derive-key")) { + t.Error("hy2 password for aaaa not found") + } + if !strings.Contains(s, DeriveHy2Password("cccc", "test-derive-key")) { + t.Error("hy2 password for cccc not found") + } +} + +func TestRevokeRemovesFromRender(t *testing.T) { + sb := NewSingBox(testConfig(t), nil) + sb.ApplyConfig(sampleSnapshot(&agentv1.Credential{DpUUID: "aaaa", Protocol: agentv1.ProtocolBoth}), true) + sb.Revoke("aaaa") + data, _ := sb.RenderConfig() + if strings.Contains(string(data), "aaaa") { + t.Errorf("revoked credential still present:\n%s", data) + } + if sb.Has("aaaa") { + t.Error("Has reports revoked credential as present") + } +} + +func TestRotateGraceBothPresentThenOldExpires(t *testing.T) { + cfg := testConfig(t) + sb := NewSingBox(cfg, nil) + now := time.Unix(1000, 0) + sb.clock = func() time.Time { return now } + sb.ApplyConfig(sampleSnapshot(&agentv1.Credential{DpUUID: "old", Protocol: agentv1.ProtocolBoth}), true) + + grace := int64(1005) + sb.Rotate("old", &Cred{DpUUID: "new", Protocol: agentv1.ProtocolBoth}, grace) + + // During grace both coexist. + if !sb.Has("old") || !sb.Has("new") { + t.Fatalf("expected both old+new during grace; old=%v new=%v", sb.Has("old"), sb.Has("new")) + } + + // After grace, sweep removes old, keeps new. + now = time.Unix(1006, 0) + removed := sb.sweepExpired() + if len(removed) != 1 || removed[0] != "old" { + t.Fatalf("expected old removed by sweep, got %v", removed) + } + if sb.Has("old") || !sb.Has("new") { + t.Fatalf("after grace: old should be gone, new present; old=%v new=%v", sb.Has("old"), sb.Has("new")) + } +} + +func TestStateRecovery(t *testing.T) { + cfg := testConfig(t) + now := time.Unix(2000, 0) + + // First manager: add creds (one already expired), persist via mutation. + sb1 := NewSingBox(cfg, nil) + sb1.clock = func() time.Time { return now } + sb1.Upsert(&Cred{DpUUID: "live", Protocol: agentv1.ProtocolBoth, ExpiresAt: 3000}) + sb1.Upsert(&Cred{DpUUID: "dead", Protocol: agentv1.ProtocolBoth, ExpiresAt: 1000}) // already expired + + // state.json must contain ONLY dp_uuid + expires_at — no identity fields. + raw, err := os.ReadFile(cfg.StatePath()) + if err != nil { + t.Fatal(err) + } + assertNoIdentityFields(t, raw) + + // Second manager (simulated kill -9 + restart) restores from disk. + sb2 := NewSingBox(cfg, nil) + sb2.clock = func() time.Time { return now } + if err := sb2.LoadState(); err != nil { + t.Fatal(err) + } + if !sb2.Has("live") { + t.Error("live credential not recovered from state.json") + } + if sb2.Has("dead") { + t.Error("expired credential should be dropped on load") + } + // Recovered creds default to BOTH + default flow until Register reconciles. + for _, c := range sb2.Snapshot() { + if c.DpUUID == "live" && (c.Protocol != agentv1.ProtocolBoth || c.Flow != DefaultFlow) { + t.Errorf("recovered cred has unexpected defaults: %+v", c) + } + } +} + +func TestTTLSweepReportsExpiry(t *testing.T) { + cfg := testConfig(t) + sb := NewSingBox(cfg, nil) + now := time.Unix(5000, 0) + sb.clock = func() time.Time { return now } + + var mu sync.Mutex + var reported []string + sb.OnExpire = func(dp string) { mu.Lock(); reported = append(reported, dp); mu.Unlock() } + + sb.Upsert(&Cred{DpUUID: "x", Protocol: agentv1.ProtocolBoth, ExpiresAt: 4999}) + sb.Upsert(&Cred{DpUUID: "y", Protocol: agentv1.ProtocolBoth, ExpiresAt: 9999}) + + removed := sb.sweepExpired() + if len(removed) != 1 || removed[0] != "x" { + t.Fatalf("sweep removed = %v, want [x]", removed) + } + mu.Lock() + defer mu.Unlock() + if len(reported) != 1 || reported[0] != "x" { + t.Fatalf("OnExpire reported = %v, want [x]", reported) + } +} + +func TestDebounceCoalescesRestarts(t *testing.T) { + cfg := testConfig(t) + cfg.DebounceWindow = 40 * time.Millisecond + fr := &fakeRestarter{} + sb := NewSingBox(cfg, fr) + sb.ApplyConfig(sampleSnapshot(), true) // set inbounds so render succeeds + + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + go sb.Run(ctx) + + // Burst of changes within one debounce window. + for i := 0; i < 5; i++ { + sb.Upsert(&Cred{DpUUID: string(rune('a' + i)), Protocol: agentv1.ProtocolBoth}) + } + time.Sleep(150 * time.Millisecond) + + if n := fr.count(); n != 1 { + t.Fatalf("expected 1 coalesced restart, got %d", n) + } +} + +// assertNoIdentityFields fails if the blob contains any user-identity token. +func assertNoIdentityFields(t *testing.T, blob []byte) { + t.Helper() + lower := strings.ToLower(string(blob)) + for _, banned := range []string{"email", "user_id", "userid", "phone", "device_id", "deviceid", "account", "destination", "dns"} { + if strings.Contains(lower, banned) { + t.Errorf("forbidden identity field %q found in persisted/rendered data:\n%s", banned, blob) + } + } +} diff --git a/server/internal/agentd/ttl.go b/server/internal/agentd/ttl.go new file mode 100644 index 0000000..c1bcd9d --- /dev/null +++ b/server/internal/agentd/ttl.go @@ -0,0 +1,30 @@ +package agentd + +import ( + "context" + "time" +) + +// runTTL periodically sweeps expired credentials out of the local table. It runs +// for the whole agent lifetime (independent of the control-plane connection) so +// credentials expire on time even during a network partition. +// +// Per doc/06, sing-box has no native per-user TTL: free-plan credentials carry the +// current day's remaining minutes as expires_at, and paid (24h) credentials are +// refreshed by RotateCredential/Upsert commands from the control plane. When a +// credential expires the agent removes it locally, re-renders sing-box and (via +// SingBox.OnExpire) surfaces the removal upstream. +func (a *Agent) runTTL(ctx context.Context) { + ticker := time.NewTicker(a.cfg.TTLScanInterval) + defer ticker.Stop() + for { + select { + case <-ctx.Done(): + return + case <-ticker.C: + if removed := a.sb.sweepExpired(); len(removed) > 0 { + logf("ttl sweep removed %d expired credential(s)", len(removed)) + } + } + } +} diff --git a/server/internal/agentd/usage.go b/server/internal/agentd/usage.go new file mode 100644 index 0000000..b59f396 --- /dev/null +++ b/server/internal/agentd/usage.go @@ -0,0 +1,54 @@ +package agentd + +import ( + "context" + "time" + + agentv1 "github.com/wangjia/pangolin/server/internal/pb/agentv1" +) + +// UsageSource yields per-dp_uuid aggregate counters for the elapsed window. It +// MUST return only dp_uuids and byte/minute counts — never user ids, addresses or +// DNS. The default returns nothing (no stats backend wired); production plugs in a +// sing-box stats reader. +type UsageSource interface { + // Collect drains and returns usage accumulated since the previous call. + Collect() []*agentv1.UsageEntry +} + +// nopUsageSource reports nothing. +type nopUsageSource struct{} + +func (nopUsageSource) Collect() []*agentv1.UsageEntry { return nil } + +// runUsage periodically aggregates and uploads usage. Reporting failures are +// logged but do not tear down the session (usage is best-effort, at-least-once). +func (a *Agent) runUsage(ctx context.Context, client agentv1.AgentServiceClient) error { + ticker := time.NewTicker(a.cfg.UsageInterval) + defer ticker.Stop() + windowStart := a.clock().Unix() + for { + select { + case <-ctx.Done(): + return ctx.Err() + case <-ticker.C: + } + entries := a.usage.Collect() + if len(entries) == 0 { + windowStart = a.clock().Unix() + continue + } + now := a.clock().Unix() + _, err := client.ReportUsage(ctx, &agentv1.UsageReport{ + NodeUUID: a.NodeUUID(), + WindowStartUnix: windowStart, + WindowEndUnix: now, + Entries: entries, + }) + if err != nil { + // Connection-level errors should restart the session; surface them. + return err + } + windowStart = now + } +} diff --git a/server/internal/agentd/util.go b/server/internal/agentd/util.go new file mode 100644 index 0000000..d2019cf --- /dev/null +++ b/server/internal/agentd/util.go @@ -0,0 +1,62 @@ +package agentd + +import ( + "context" + "fmt" + "log" + "os" + "os/exec" + "path/filepath" +) + +// logf is the agent's structured-ish logger. It deliberately never logs +// dp_uuids, peers' addresses or any payload — only operational events — to honour +// the no-log policy. +func logf(format string, args ...any) { + log.Printf("[pangolin-agent] "+format, args...) +} + +// atomicWrite writes data to a temp file in the same directory and renames it into +// place, so a crash mid-write never leaves a truncated config/state file. +func atomicWrite(path string, data []byte, perm os.FileMode) error { + dir := filepath.Dir(path) + tmp, err := os.CreateTemp(dir, ".tmp-*") + if err != nil { + return fmt.Errorf("agentd: temp file: %w", err) + } + tmpName := tmp.Name() + defer os.Remove(tmpName) // no-op if the rename succeeded + if _, err := tmp.Write(data); err != nil { + tmp.Close() + return fmt.Errorf("agentd: write temp: %w", err) + } + if err := tmp.Chmod(perm); err != nil { + tmp.Close() + return fmt.Errorf("agentd: chmod temp: %w", err) + } + if err := tmp.Close(); err != nil { + return fmt.Errorf("agentd: close temp: %w", err) + } + if err := os.Rename(tmpName, path); err != nil { + return fmt.Errorf("agentd: rename temp: %w", err) + } + return nil +} + +// SystemdRestarter restarts the sing-box unit via `systemctl restart`. +type SystemdRestarter struct { + Unit string // e.g. "sing-box" +} + +// Restart runs `systemctl restart `. +func (r SystemdRestarter) Restart(ctx context.Context) error { + unit := r.Unit + if unit == "" { + unit = "sing-box" + } + cmd := exec.CommandContext(ctx, "systemctl", "restart", unit) + if out, err := cmd.CombinedOutput(); err != nil { + return fmt.Errorf("agentd: restart %s: %w: %s", unit, err, out) + } + return nil +} diff --git a/server/internal/pb/agentv1/codec.go b/server/internal/pb/agentv1/codec.go new file mode 100644 index 0000000..8767249 --- /dev/null +++ b/server/internal/pb/agentv1/codec.go @@ -0,0 +1,30 @@ +package agentv1 + +import ( + "encoding/json" + + "google.golang.org/grpc/encoding" +) + +// CodecName is the gRPC content-subtype used by the agent contract. Because this +// repository has no protoc pipeline yet, messages are encoded as JSON rather than +// protobuf wire format. Clients select it with grpc.CallContentSubtype(CodecName) +// (wired automatically by the client/server helpers in service.go); the server +// echoes the same subtype on responses. +const CodecName = "pangolinagentjson" + +func init() { encoding.RegisterCodec(jsonCodec{}) } + +// jsonCodec implements google.golang.org/grpc/encoding.Codec over encoding/json. +type jsonCodec struct{} + +func (jsonCodec) Marshal(v any) ([]byte, error) { return json.Marshal(v) } + +func (jsonCodec) Unmarshal(data []byte, v any) error { + if len(data) == 0 { + return nil // empty message (e.g. AckResponse / UsageAck) + } + return json.Unmarshal(data, v) +} + +func (jsonCodec) Name() string { return CodecName } diff --git a/server/internal/pb/agentv1/service.go b/server/internal/pb/agentv1/service.go new file mode 100644 index 0000000..89e1485 --- /dev/null +++ b/server/internal/pb/agentv1/service.go @@ -0,0 +1,238 @@ +package agentv1 + +import ( + "context" + + "google.golang.org/grpc" +) + +// ServiceName is the fully-qualified gRPC service name. It MUST match the value +// the mTLS interceptor whitelists for Enroll (server/internal/mtls/identity.go). +const ServiceName = "pangolin.agent.v1.AgentService" + +// Fully-qualified method names. +const ( + MethodEnroll = "/" + ServiceName + "/Enroll" + MethodRegister = "/" + ServiceName + "/Register" + MethodHeartbeat = "/" + ServiceName + "/Heartbeat" + MethodSubscribe = "/" + ServiceName + "/Subscribe" + MethodAck = "/" + ServiceName + "/Ack" + MethodReportUsage = "/" + ServiceName + "/ReportUsage" +) + +// withCodec forces the JSON content-subtype on every call so the contract does +// not depend on the caller remembering to set dial-level call options. +func withCodec(opts []grpc.CallOption) []grpc.CallOption { + return append([]grpc.CallOption{grpc.CallContentSubtype(CodecName)}, opts...) +} + +// ─── server interface ────────────────────────────────────────────────────────── + +// AgentServiceServer is implemented by the control plane. +type AgentServiceServer interface { + Enroll(context.Context, *EnrollRequest) (*EnrollResponse, error) + Register(context.Context, *RegisterRequest) (*ConfigSnapshot, error) + Heartbeat(context.Context, *HeartbeatRequest) (*HeartbeatResponse, error) + Subscribe(*SubscribeRequest, AgentService_SubscribeServer) error + Ack(context.Context, *AckRequest) (*AckResponse, error) + ReportUsage(context.Context, *UsageReport) (*UsageAck, error) +} + +// AgentService_SubscribeServer is the server side of the Command stream. +type AgentService_SubscribeServer interface { + Send(*Command) error + grpc.ServerStream +} + +type subscribeServer struct{ grpc.ServerStream } + +func (s *subscribeServer) Send(m *Command) error { return s.ServerStream.SendMsg(m) } + +// RegisterAgentServiceServer wires srv into a gRPC server (or any ServiceRegistrar). +func RegisterAgentServiceServer(s grpc.ServiceRegistrar, srv AgentServiceServer) { + s.RegisterService(&serviceDesc, srv) +} + +func handlerEnroll(srv any, ctx context.Context, dec func(any) error, interceptor grpc.UnaryServerInterceptor) (any, error) { + in := new(EnrollRequest) + if err := dec(in); err != nil { + return nil, err + } + if interceptor == nil { + return srv.(AgentServiceServer).Enroll(ctx, in) + } + info := &grpc.UnaryServerInfo{Server: srv, FullMethod: MethodEnroll} + return interceptor(ctx, in, info, func(ctx context.Context, req any) (any, error) { + return srv.(AgentServiceServer).Enroll(ctx, req.(*EnrollRequest)) + }) +} + +func handlerRegister(srv any, ctx context.Context, dec func(any) error, interceptor grpc.UnaryServerInterceptor) (any, error) { + in := new(RegisterRequest) + if err := dec(in); err != nil { + return nil, err + } + if interceptor == nil { + return srv.(AgentServiceServer).Register(ctx, in) + } + info := &grpc.UnaryServerInfo{Server: srv, FullMethod: MethodRegister} + return interceptor(ctx, in, info, func(ctx context.Context, req any) (any, error) { + return srv.(AgentServiceServer).Register(ctx, req.(*RegisterRequest)) + }) +} + +func handlerHeartbeat(srv any, ctx context.Context, dec func(any) error, interceptor grpc.UnaryServerInterceptor) (any, error) { + in := new(HeartbeatRequest) + if err := dec(in); err != nil { + return nil, err + } + if interceptor == nil { + return srv.(AgentServiceServer).Heartbeat(ctx, in) + } + info := &grpc.UnaryServerInfo{Server: srv, FullMethod: MethodHeartbeat} + return interceptor(ctx, in, info, func(ctx context.Context, req any) (any, error) { + return srv.(AgentServiceServer).Heartbeat(ctx, req.(*HeartbeatRequest)) + }) +} + +func handlerAck(srv any, ctx context.Context, dec func(any) error, interceptor grpc.UnaryServerInterceptor) (any, error) { + in := new(AckRequest) + if err := dec(in); err != nil { + return nil, err + } + if interceptor == nil { + return srv.(AgentServiceServer).Ack(ctx, in) + } + info := &grpc.UnaryServerInfo{Server: srv, FullMethod: MethodAck} + return interceptor(ctx, in, info, func(ctx context.Context, req any) (any, error) { + return srv.(AgentServiceServer).Ack(ctx, req.(*AckRequest)) + }) +} + +func handlerReportUsage(srv any, ctx context.Context, dec func(any) error, interceptor grpc.UnaryServerInterceptor) (any, error) { + in := new(UsageReport) + if err := dec(in); err != nil { + return nil, err + } + if interceptor == nil { + return srv.(AgentServiceServer).ReportUsage(ctx, in) + } + info := &grpc.UnaryServerInfo{Server: srv, FullMethod: MethodReportUsage} + return interceptor(ctx, in, info, func(ctx context.Context, req any) (any, error) { + return srv.(AgentServiceServer).ReportUsage(ctx, req.(*UsageReport)) + }) +} + +func handlerSubscribe(srv any, stream grpc.ServerStream) error { + m := new(SubscribeRequest) + if err := stream.RecvMsg(m); err != nil { + return err + } + return srv.(AgentServiceServer).Subscribe(m, &subscribeServer{stream}) +} + +var serviceDesc = grpc.ServiceDesc{ + ServiceName: ServiceName, + HandlerType: (*AgentServiceServer)(nil), + Methods: []grpc.MethodDesc{ + {MethodName: "Enroll", Handler: handlerEnroll}, + {MethodName: "Register", Handler: handlerRegister}, + {MethodName: "Heartbeat", Handler: handlerHeartbeat}, + {MethodName: "Ack", Handler: handlerAck}, + {MethodName: "ReportUsage", Handler: handlerReportUsage}, + }, + Streams: []grpc.StreamDesc{ + {StreamName: "Subscribe", Handler: handlerSubscribe, ServerStreams: true}, + }, + Metadata: "proto/agent/v1/agent.proto", +} + +// ─── client ───────────────────────────────────────────────────────────────────── + +// AgentServiceClient is consumed by the node agent. +type AgentServiceClient interface { + Enroll(ctx context.Context, in *EnrollRequest, opts ...grpc.CallOption) (*EnrollResponse, error) + Register(ctx context.Context, in *RegisterRequest, opts ...grpc.CallOption) (*ConfigSnapshot, error) + Heartbeat(ctx context.Context, in *HeartbeatRequest, opts ...grpc.CallOption) (*HeartbeatResponse, error) + Subscribe(ctx context.Context, in *SubscribeRequest, opts ...grpc.CallOption) (AgentService_SubscribeClient, error) + Ack(ctx context.Context, in *AckRequest, opts ...grpc.CallOption) (*AckResponse, error) + ReportUsage(ctx context.Context, in *UsageReport, opts ...grpc.CallOption) (*UsageAck, error) +} + +// AgentService_SubscribeClient is the client side of the Command stream. +type AgentService_SubscribeClient interface { + Recv() (*Command, error) + grpc.ClientStream +} + +type subscribeClient struct{ grpc.ClientStream } + +func (c *subscribeClient) Recv() (*Command, error) { + m := new(Command) + if err := c.ClientStream.RecvMsg(m); err != nil { + return nil, err + } + return m, nil +} + +type agentServiceClient struct{ cc grpc.ClientConnInterface } + +// NewAgentServiceClient returns a client bound to cc. All calls use the JSON codec. +func NewAgentServiceClient(cc grpc.ClientConnInterface) AgentServiceClient { + return &agentServiceClient{cc} +} + +func (c *agentServiceClient) Enroll(ctx context.Context, in *EnrollRequest, opts ...grpc.CallOption) (*EnrollResponse, error) { + out := new(EnrollResponse) + if err := c.cc.Invoke(ctx, MethodEnroll, in, out, withCodec(opts)...); err != nil { + return nil, err + } + return out, nil +} + +func (c *agentServiceClient) Register(ctx context.Context, in *RegisterRequest, opts ...grpc.CallOption) (*ConfigSnapshot, error) { + out := new(ConfigSnapshot) + if err := c.cc.Invoke(ctx, MethodRegister, in, out, withCodec(opts)...); err != nil { + return nil, err + } + return out, nil +} + +func (c *agentServiceClient) Heartbeat(ctx context.Context, in *HeartbeatRequest, opts ...grpc.CallOption) (*HeartbeatResponse, error) { + out := new(HeartbeatResponse) + if err := c.cc.Invoke(ctx, MethodHeartbeat, in, out, withCodec(opts)...); err != nil { + return nil, err + } + return out, nil +} + +func (c *agentServiceClient) Ack(ctx context.Context, in *AckRequest, opts ...grpc.CallOption) (*AckResponse, error) { + out := new(AckResponse) + if err := c.cc.Invoke(ctx, MethodAck, in, out, withCodec(opts)...); err != nil { + return nil, err + } + return out, nil +} + +func (c *agentServiceClient) ReportUsage(ctx context.Context, in *UsageReport, opts ...grpc.CallOption) (*UsageAck, error) { + out := new(UsageAck) + if err := c.cc.Invoke(ctx, MethodReportUsage, in, out, withCodec(opts)...); err != nil { + return nil, err + } + return out, nil +} + +func (c *agentServiceClient) Subscribe(ctx context.Context, in *SubscribeRequest, opts ...grpc.CallOption) (AgentService_SubscribeClient, error) { + stream, err := c.cc.NewStream(ctx, &serviceDesc.Streams[0], MethodSubscribe, withCodec(opts)...) + if err != nil { + return nil, err + } + x := &subscribeClient{stream} + if err := x.ClientStream.SendMsg(in); err != nil { + return nil, err + } + if err := x.ClientStream.CloseSend(); err != nil { + return nil, err + } + return x, nil +} diff --git a/server/internal/pb/agentv1/types.go b/server/internal/pb/agentv1/types.go new file mode 100644 index 0000000..90dfcbd --- /dev/null +++ b/server/internal/pb/agentv1/types.go @@ -0,0 +1,176 @@ +// Package agentv1 is the control-plane <-> node-agent gRPC contract. +// +// These types mirror server/proto/agent/v1/agent.proto 1:1. The repository has no +// protoc/buf pipeline wired yet (Makefile `generate` is a stub from task #5), so +// the messages are hand-written Go structs and the gRPC stubs (service.go) use a +// JSON codec (codec.go) instead of protobuf wire format. The JSON field names are +// the proto field names so a future protoc-generated package is drop-in +// compatible at the API surface. +// +// Data-plane red line: nothing in this package carries user_id, email, device id, +// destination address or DNS data — only the opaque dp_uuid and aggregate counters. +package agentv1 + +// Protocol selects which sing-box inbound(s) a credential is provisioned on. +type Protocol int32 + +const ( + ProtocolUnspecified Protocol = 0 + ProtocolReality Protocol = 1 + ProtocolHy2 Protocol = 2 + ProtocolBoth Protocol = 3 +) + +// CommandType discriminates the Command payload. +type CommandType int32 + +const ( + CommandTypeUnspecified CommandType = 0 + CommandTypeUpsert CommandType = 1 + CommandTypeRevoke CommandType = 2 + CommandTypeRotateCredential CommandType = 3 + CommandTypeApplyConfig CommandType = 4 + CommandTypeLifecycle CommandType = 5 +) + +// LifecycleAction is the verb of a LIFECYCLE command. +type LifecycleAction int32 + +const ( + LifecycleActionUnspecified LifecycleAction = 0 + LifecycleActionDrain LifecycleAction = 1 + LifecycleActionResume LifecycleAction = 2 + LifecycleActionShutdown LifecycleAction = 3 +) + +// ─── enroll ────────────────────────────────────────────────────────────────── + +type EnrollRequest struct { + BootstrapToken string `json:"bootstrap_token,omitempty"` + CSRPEM []byte `json:"csr_pem,omitempty"` + AgentVersion string `json:"agent_version,omitempty"` +} + +type EnrollResponse struct { + NodeUUID string `json:"node_uuid,omitempty"` + CertPEM []byte `json:"cert_pem,omitempty"` + CAPEM []byte `json:"ca_pem,omitempty"` + NotAfterUnix int64 `json:"not_after_unix,omitempty"` +} + +// ─── register / config ───────────────────────────────────────────────────────── + +type RegisterRequest struct { + NodeUUID string `json:"node_uuid,omitempty"` + AgentVersion string `json:"agent_version,omitempty"` + LocalConfigVersion int64 `json:"local_config_version,omitempty"` +} + +// Credential is the only per-subscriber object a node ever sees. +type Credential struct { + DpUUID string `json:"dp_uuid,omitempty"` + Protocol Protocol `json:"protocol,omitempty"` + Flow string `json:"flow,omitempty"` + ExpiresAtUnix int64 `json:"expires_at_unix,omitempty"` +} + +type ConfigSnapshot struct { + ConfigVersion int64 `json:"config_version,omitempty"` + Credentials []*Credential `json:"credentials,omitempty"` + Reality *RealityInbound `json:"reality,omitempty"` + Hy2 *Hy2Inbound `json:"hy2,omitempty"` + LastCommandID int64 `json:"last_command_id,omitempty"` +} + +type RealityInbound struct { + ListenPort int32 `json:"listen_port,omitempty"` + PrivateKey string `json:"private_key,omitempty"` + ShortID string `json:"short_id,omitempty"` + ServerName string `json:"server_name,omitempty"` + HandshakeServer string `json:"handshake_server,omitempty"` + HandshakePort int32 `json:"handshake_port,omitempty"` +} + +type Hy2Inbound struct { + ListenPort int32 `json:"listen_port,omitempty"` + Masquerade string `json:"masquerade,omitempty"` + CertPath string `json:"cert_path,omitempty"` + KeyPath string `json:"key_path,omitempty"` +} + +// ─── heartbeat ─────────────────────────────────────────────────────────────── + +type HeartbeatRequest struct { + NodeUUID string `json:"node_uuid,omitempty"` + ConfigVersion int64 `json:"config_version,omitempty"` + OnlinePeers int32 `json:"online_peers,omitempty"` + BandwidthUpBps int64 `json:"bandwidth_up_bps,omitempty"` + BandwidthDownBps int64 `json:"bandwidth_down_bps,omitempty"` + CPUPercent float64 `json:"cpu_percent,omitempty"` + TimestampUnix int64 `json:"timestamp_unix,omitempty"` +} + +type HeartbeatResponse struct { + NeedFullResync bool `json:"need_full_resync,omitempty"` + ServerTimeUnix int64 `json:"server_time_unix,omitempty"` +} + +// ─── command stream ────────────────────────────────────────────────────────── + +type Command struct { + CommandID int64 `json:"command_id,omitempty"` + Type CommandType `json:"type,omitempty"` + Upsert *UpsertPayload `json:"upsert,omitempty"` + Revoke *RevokePayload `json:"revoke,omitempty"` + Rotate *RotatePayload `json:"rotate,omitempty"` + ApplyConfig *ConfigSnapshot `json:"apply_config,omitempty"` + Lifecycle *LifecyclePayload `json:"lifecycle,omitempty"` +} + +type UpsertPayload struct { + Credential *Credential `json:"credential,omitempty"` +} + +type RevokePayload struct { + DpUUID string `json:"dp_uuid,omitempty"` +} + +type RotatePayload struct { + OldDpUUID string `json:"old_dp_uuid,omitempty"` + NewCredential *Credential `json:"new_credential,omitempty"` + GraceUntilUnix int64 `json:"grace_until_unix,omitempty"` +} + +type LifecyclePayload struct { + Action LifecycleAction `json:"action,omitempty"` +} + +type SubscribeRequest struct { + NodeUUID string `json:"node_uuid,omitempty"` + LastCommandID int64 `json:"last_command_id,omitempty"` +} + +type AckRequest struct { + NodeUUID string `json:"node_uuid,omitempty"` + CommandID int64 `json:"command_id,omitempty"` +} + +type AckResponse struct{} + +// ─── usage ─────────────────────────────────────────────────────────────────── + +type UsageEntry struct { + DpUUID string `json:"dp_uuid,omitempty"` + BytesUp int64 `json:"bytes_up,omitempty"` + BytesDown int64 `json:"bytes_down,omitempty"` + SessionMinutes int64 `json:"session_minutes,omitempty"` +} + +type UsageReport struct { + NodeUUID string `json:"node_uuid,omitempty"` + WindowStartUnix int64 `json:"window_start_unix,omitempty"` + WindowEndUnix int64 `json:"window_end_unix,omitempty"` + Entries []*UsageEntry `json:"entries,omitempty"` +} + +type UsageAck struct{} diff --git a/server/proto/agent/v1/agent.proto b/server/proto/agent/v1/agent.proto new file mode 100644 index 0000000..d8b6bd8 --- /dev/null +++ b/server/proto/agent/v1/agent.proto @@ -0,0 +1,212 @@ +// Package pangolin.agent.v1 is the frozen control-plane <-> node-agent contract. +// +// CONNECTION DIRECTION: the node agent ALWAYS dials the control plane (never the +// reverse). Every RPC except Enroll requires a verified mTLS client certificate +// whose Subject CN equals the node UUID (see server/internal/mtls). Enroll is the +// single bootstrap exception, authenticated by a one-time bootstrap token instead. +// +// DATA-PLANE RED LINE (doc/06 §3, doc/04 §2): messages on this contract carry ONLY +// the opaque data-plane credential id (dp_uuid) and aggregate counters. They MUST +// NOT carry user_id, email, device id, destination address, SNI or any DNS query. +// +// NOTE ON CODE GENERATION: this repository has no protoc/buf toolchain wired yet +// (Makefile `generate` is a stub). Until task #5's proto pipeline lands, the Go +// types and gRPC stubs in server/internal/pb/agentv1 are hand-written to mirror +// this file 1:1 and are wire-encoded with a JSON gRPC codec. When protoc is +// introduced this file is the source of truth; field numbers are reserved here so +// the generated code stays compatible. +syntax = "proto3"; + +package pangolin.agent.v1; + +option go_package = "github.com/wangjia/pangolin/server/internal/pb/agentv1;agentv1"; + +// AgentService is implemented by the control plane and consumed by node agents. +service AgentService { + // Enroll exchanges a one-time bootstrap token + CSR for a node client + // certificate (CN = node_uuid, 90d). Exempt from mTLS — this is the only RPC an + // un-enrolled agent may call. Idempotent only within the token TTL. + rpc Enroll(EnrollRequest) returns (EnrollResponse); + + // Register is called after (re)connecting over mTLS. It returns the authoritative + // full ConfigSnapshot so the agent can converge local state. + rpc Register(RegisterRequest) returns (ConfigSnapshot); + + // Heartbeat reports liveness + load every ~30s and the agent's local + // config_version. The control plane flags need_full_resync when it detects drift. + rpc Heartbeat(HeartbeatRequest) returns (HeartbeatResponse); + + // Subscribe is a server-streaming feed of Commands. The agent passes the last + // command_id it durably processed so the control plane can resume without gaps + // (at-least-once; commands are idempotent and Ack'd individually). + rpc Subscribe(SubscribeRequest) returns (stream Command); + + // Ack confirms a command was applied. Safe to retry. + rpc Ack(AckRequest) returns (AckResponse); + + // ReportUsage uploads per-dp_uuid aggregate byte/minute counters. No identities. + rpc ReportUsage(UsageReport) returns (UsageAck); +} + +// ─── enroll ────────────────────────────────────────────────────────────────── + +message EnrollRequest { + string bootstrap_token = 1; // one-time, injected via cloud-init + bytes csr_pem = 2; // PKCS#10 CSR, agent-generated key never leaves node + string agent_version = 3; +} + +message EnrollResponse { + string node_uuid = 1; // authoritative; becomes the cert CN + bytes cert_pem = 2; // signed leaf, 90d, EKU=clientAuth + bytes ca_pem = 3; // CA to pin for server validation + int64 not_after_unix = 4; +} + +// ─── register / config ───────────────────────────────────────────────────────── + +message RegisterRequest { + string node_uuid = 1; + string agent_version = 2; + int64 local_config_version = 3; // 0 on a fresh node +} + +// Protocol selects which inbound(s) a credential is provisioned on. +enum Protocol { + PROTOCOL_UNSPECIFIED = 0; + PROTOCOL_REALITY = 1; // VLESS+REALITY inbound, user = {uuid, flow} + PROTOCOL_HY2 = 2; // Hysteria2 inbound, user = {password} + PROTOCOL_BOTH = 3; // both inbounds share the same dp_uuid +} + +// Credential is the ONLY per-subscriber object a node ever sees. +message Credential { + string dp_uuid = 1; // opaque data-plane id; REALITY user uuid + Protocol protocol = 2; + string flow = 3; // e.g. "xtls-rprx-vision" + int64 expires_at_unix = 4; // agent-side TTL; 0 = no expiry +} + +// ConfigSnapshot is the full authoritative node state. +message ConfigSnapshot { + int64 config_version = 1; + repeated Credential credentials = 2; + + // Inbound parameters (node-wide, not per-user). + RealityInbound reality = 3; + Hy2Inbound hy2 = 4; + + // Baseline command id already reflected in this snapshot; the agent resumes + // Subscribe from here. + int64 last_command_id = 5; +} + +message RealityInbound { + int32 listen_port = 1; + string private_key = 2; // REALITY server private key (base64) + string short_id = 3; + string server_name = 4; // masquerade SNI handshake target + string handshake_server = 5; + int32 handshake_port = 6; +} + +message Hy2Inbound { + int32 listen_port = 1; + string masquerade = 2; + string cert_path = 3; + string key_path = 4; +} + +// ─── heartbeat ─────────────────────────────────────────────────────────────── + +message HeartbeatRequest { + string node_uuid = 1; + int64 config_version = 2; + int32 online_peers = 3; + int64 bandwidth_up_bps = 4; + int64 bandwidth_down_bps = 5; + double cpu_percent = 6; + int64 timestamp_unix = 7; +} + +message HeartbeatResponse { + bool need_full_resync = 1; // agent must Register + overwrite local state + int64 server_time_unix = 2; +} + +// ─── command stream ────────────────────────────────────────────────────────── + +enum CommandType { + COMMAND_TYPE_UNSPECIFIED = 0; + COMMAND_TYPE_UPSERT = 1; // add/update a credential + COMMAND_TYPE_REVOKE = 2; // remove a credential + COMMAND_TYPE_ROTATE_CREDENTIAL = 3; // new replaces old, old kept until grace + COMMAND_TYPE_APPLY_CONFIG = 4; // replace inbound params / full snapshot + COMMAND_TYPE_LIFECYCLE = 5; // drain / resume / shutdown +} + +enum LifecycleAction { + LIFECYCLE_ACTION_UNSPECIFIED = 0; + LIFECYCLE_ACTION_DRAIN = 1; + LIFECYCLE_ACTION_RESUME = 2; + LIFECYCLE_ACTION_SHUTDOWN = 3; +} + +message Command { + int64 command_id = 1; // monotonically increasing per node + CommandType type = 2; + + UpsertPayload upsert = 3; + RevokePayload revoke = 4; + RotatePayload rotate = 5; + ConfigSnapshot apply_config = 6; + LifecyclePayload lifecycle = 7; +} + +message UpsertPayload { + Credential credential = 1; +} + +message RevokePayload { + string dp_uuid = 1; +} + +message RotatePayload { + string old_dp_uuid = 1; + Credential new_credential = 2; + int64 grace_until_unix = 3; // old credential kept active until this time +} + +message LifecyclePayload { + LifecycleAction action = 1; +} + +message SubscribeRequest { + string node_uuid = 1; + int64 last_command_id = 2; // resume point; 0 = from the beginning +} + +message AckRequest { + string node_uuid = 1; + int64 command_id = 2; +} + +message AckResponse {} + +// ─── usage ─────────────────────────────────────────────────────────────────── + +message UsageEntry { + string dp_uuid = 1; // never a user id + int64 bytes_up = 2; + int64 bytes_down = 3; + int64 session_minutes = 4; +} + +message UsageReport { + string node_uuid = 1; + int64 window_start_unix = 2; + int64 window_end_unix = 3; + repeated UsageEntry entries = 4; +} + +message UsageAck {}