feat: 配对管理端点与中继侧设备注册上报
- agent --pair:临时监听 :3001 仅开放 /provision 与 /destroy,以节点密码为共享秘密; 窗口由三个独立条件关闭(首次成功配对 / 失败尝试 10 次 / 10 分钟超时),配对成功后 管理监听即关(设计 §7 第 4 步) - agent 为每台配对手机生成中继 bearer(config.Save 0600 原子写;启动时为手写条目补发) - transport.PushDevices:握手成功后全量上报本节点受信设备集;空集是吊销声明而非无操作 - relay /v1/agent/devices:agent 会话后鉴权的设备注册;全量替换本节点条目; 操作员 MSA_DEVICES 为 static 不可被覆盖;跨节点 id 不可抢占 - 中继设备注册替代了 §11 第 6 项的一部分:节点 destroy + 重握手即可在路由层吊销 - 配置:provisionSecret 字段(无密码保护节点需显式设置才可配对);Validate(pairMode) - 测试:pair 流程 8 项、设备注册 6 项、transport 1 项;文档 §7 实现修订注记与 README 状态
This commit is contained in:
+90
-6
@@ -14,13 +14,16 @@ import (
|
||||
"flag"
|
||||
"fmt"
|
||||
"log/slog"
|
||||
"net/http"
|
||||
"os"
|
||||
"os/signal"
|
||||
"strconv"
|
||||
"syscall"
|
||||
"time"
|
||||
|
||||
"fun.chick/msa-agent/internal/agent"
|
||||
"fun.chick/msa-agent/internal/config"
|
||||
"fun.chick/msa-agent/internal/pair"
|
||||
"fun.chick/msa-agent/internal/policy"
|
||||
"fun.chick/msa-agent/internal/runtime"
|
||||
"fun.chick/msa-agent/internal/transport"
|
||||
@@ -41,21 +44,86 @@ func main() {
|
||||
configPath = flag.String("config", "/etc/msa-agent/config.json", "path to the agent config file")
|
||||
check = flag.Bool("check", false, "validate the config and policy, then exit")
|
||||
once = flag.Bool("once", false, "run a single poll cycle and exit")
|
||||
pairMode = flag.Bool("pair", false, "open the temporary pairing window on the LAN and exit when it closes")
|
||||
pairPort = flag.Int("pair-port", pair.DefaultAdminPort, "port for the pairing window's management listener")
|
||||
)
|
||||
flag.Parse()
|
||||
|
||||
logger := slog.New(slog.NewJSONHandler(os.Stdout, &slog.HandlerOptions{Level: slog.LevelInfo}))
|
||||
slog.SetDefault(logger)
|
||||
|
||||
if err := run(*configPath, *check, *once); err != nil {
|
||||
var err error
|
||||
if *pairMode {
|
||||
err = runPair(*configPath, *pairPort)
|
||||
} else {
|
||||
err = run(*configPath, *check, *once)
|
||||
}
|
||||
if err != nil {
|
||||
// One line, then exit non-zero. systemd restarts it, and the journal keeps the reason.
|
||||
logger.Error("msa-agent exiting", "error", err)
|
||||
os.Exit(1)
|
||||
}
|
||||
}
|
||||
|
||||
// runPair serves design §7's management surface: a LAN listener offering /provision and /destroy
|
||||
// for as long as the window is open, then exit. It is a separate mode from the poll loop because
|
||||
// the invariant "the agent binds no socket" is a property of the daemon; the pairing window is a
|
||||
// deliberate, operator-started exception with three independent shutdowns (first pairing, attempt
|
||||
// ceiling, deadline). Pairing is run again by the operator for each additional phone.
|
||||
func runPair(configPath string, adminPort int) error {
|
||||
cfg, err := config.LoadForPairing(configPath)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
handler, err := pair.New(configPath, cfg, func(event string, fields map[string]any) {
|
||||
slog.Info(event, fieldsToArgs(fields)...)
|
||||
})
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
server := &http.Server{
|
||||
Addr: ":" + strconv.Itoa(adminPort),
|
||||
Handler: handler.Routes(),
|
||||
ReadHeaderTimeout: 5 * time.Second,
|
||||
}
|
||||
go func() {
|
||||
slog.Info("pairing window open", "port", adminPort, "node", cfg.NodeID)
|
||||
if err := server.ListenAndServe(); err != nil && err != http.ErrServerClosed {
|
||||
slog.Error("pairing listener failed", "error", err)
|
||||
handler.Close()
|
||||
}
|
||||
}()
|
||||
defer server.Close()
|
||||
|
||||
ctx, stop := signal.NotifyContext(context.Background(), os.Interrupt, syscall.SIGTERM)
|
||||
defer stop()
|
||||
deadline := time.NewTimer(pair.DefaultWindow)
|
||||
defer deadline.Stop()
|
||||
select {
|
||||
case <-handler.Done:
|
||||
slog.Info("pairing window closed")
|
||||
case <-deadline.C:
|
||||
slog.Warn("pairing window expired unused")
|
||||
case <-ctx.Done():
|
||||
slog.Info("pairing window interrupted")
|
||||
}
|
||||
// The listener is already closing via defer; Stop with a short grace so an in-flight request
|
||||
// finishes rather than being truncated mid-JSON.
|
||||
shutdownCtx, cancel := context.WithTimeout(context.Background(), 2*time.Second)
|
||||
defer cancel()
|
||||
return server.Shutdown(shutdownCtx)
|
||||
}
|
||||
|
||||
func fieldsToArgs(fields map[string]any) []any {
|
||||
args := make([]any, 0, len(fields)*2)
|
||||
for key, value := range fields {
|
||||
args = append(args, key, value)
|
||||
}
|
||||
return args
|
||||
}
|
||||
|
||||
func run(configPath string, checkOnly, once bool) error {
|
||||
cfg, err := config.Load(configPath)
|
||||
cfg, _, err := config.EnsureRelayTokens(configPath)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
@@ -126,7 +194,15 @@ func run(configPath string, checkOnly, once bool) error {
|
||||
if once {
|
||||
return cycle(ctx, node, client, pol)
|
||||
}
|
||||
return loop(ctx, node, client, pol)
|
||||
return loop(ctx, node, client, pol, registrationsFrom(cfg))
|
||||
}
|
||||
|
||||
func registrationsFrom(cfg config.File) []transport.DeviceRegistration {
|
||||
out := make([]transport.DeviceRegistration, 0, len(cfg.Devices))
|
||||
for _, device := range cfg.Devices {
|
||||
out = append(out, transport.DeviceRegistration{App: device.ID, Token: device.RelayToken})
|
||||
}
|
||||
return out
|
||||
}
|
||||
|
||||
// loop is the agent's whole life: handshake, then poll forever, re-handshaking whenever the relay
|
||||
@@ -135,13 +211,13 @@ func run(configPath string, checkOnly, once bool) error {
|
||||
// Errors are classified here rather than handled per call site, because the three situations look
|
||||
// identical from inside a function and need three different responses: retry now, re-authenticate,
|
||||
// or wait.
|
||||
func loop(ctx context.Context, node *agent.Agent, client *transport.Client, pol *policy.Policy) error {
|
||||
func loop(ctx context.Context, node *agent.Agent, client *transport.Client, pol *policy.Policy, devices []transport.DeviceRegistration) error {
|
||||
backoff := minBackoff
|
||||
for {
|
||||
if ctx.Err() != nil {
|
||||
return nil
|
||||
}
|
||||
err := handshakeAndPoll(ctx, node, client, pol)
|
||||
err := handshakeAndPoll(ctx, node, client, pol, devices)
|
||||
switch {
|
||||
case err == nil:
|
||||
backoff = minBackoff
|
||||
@@ -176,10 +252,18 @@ func next(current time.Duration) time.Duration {
|
||||
return doubled
|
||||
}
|
||||
|
||||
func handshakeAndPoll(ctx context.Context, node *agent.Agent, client *transport.Client, pol *policy.Policy) error {
|
||||
func handshakeAndPoll(ctx context.Context, node *agent.Agent, client *transport.Client, pol *policy.Policy, devices []transport.DeviceRegistration) error {
|
||||
if _, err := client.Hello(ctx, version, pol.SHA256()); err != nil {
|
||||
return err
|
||||
}
|
||||
// The relay holds no device state across restarts, and it must never trust a device claim it
|
||||
// cannot tie to an authenticated agent. Reporting the full set after every handshake is what
|
||||
// makes that true without a second channel: the node's paired-device set, including any
|
||||
// revocations from a destroy, is re-asserted each time the session is established. Empty is
|
||||
// pushed too, because "empty" is exactly the claim that revocation needs.
|
||||
if err := client.PushDevices(ctx, devices); err != nil {
|
||||
return err
|
||||
}
|
||||
return cycle(ctx, node, client, pol)
|
||||
}
|
||||
|
||||
|
||||
Reference in New Issue
Block a user