wangjia
f03d2dc8a6
与控制面同仓同 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 <noreply@anthropic.com>
67 lines
2.0 KiB
Go
67 lines
2.0 KiB
Go
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")
|
|
}
|
|
}
|