feat: 端到端测试(真 agent + 真中继 + TLS),并修三处只有它能发现的缺陷
新增 test/e2e.test.js:编译并运行 Go agent,经 TLS 接到真中继,由模拟手机发起 签名请求,再断言节点实际收到的内容(含 audit 日志不含 body/密码、危险写 body 逐字节一致)。缺 go/openssl 则跳过并说明原因。 它一次抓出三个单元测试各自都通过、合起来却坏了的问题: 1. transport.Client 把会话 token 写进 c.token,而 c.token 是 bearer——Poll/Result 因此用会话 token 当 bearer 发出,中继一律 401。现象是 agent 连上后无限重连, 日志说'凭据被拒'。现在两个凭据分成两个字段,并加测试断言握手带 bearer 不带会话、 轮询带会话。 2. main.loop 每个周期都重新握手。中继每个节点只保留一个会话,新握手会顶掉旧的并 丢弃其挂起的轮询,于是两次握手之间到达的请求被投进已不存在的会话、以'agent 离线' 超时——而 agent 全程在线。改为只在会话失效/传输失败时重新握手。 3. auth.registerDevices 把'已存在'一律跳过,随后又被'本轮未声明'的回收循环删掉, 所以节点的第二次握手会把自己注册的设备全部注销。改为本节点已持有即确认(幂等)。 另:transport 支持 relayCAFile(PEM 钉 CA,替代系统信任库,不是关闭校验), 测试用自签证书 + 钉 CA 走真实 TLS 校验路径。
This commit is contained in:
+64
-28
@@ -9,6 +9,7 @@ package main
|
||||
import (
|
||||
"context"
|
||||
"crypto/ed25519"
|
||||
"crypto/x509"
|
||||
"encoding/base64"
|
||||
"errors"
|
||||
"flag"
|
||||
@@ -177,12 +178,17 @@ func run(configPath string, checkOnly, once bool) error {
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
roots, err := relayRoots(cfg.RelayCAFile)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
client, err := transport.New(transport.Config{
|
||||
BaseURL: cfg.RelayURL,
|
||||
NodeID: cfg.NodeID,
|
||||
Token: cfg.AgentToken,
|
||||
Signer: sign,
|
||||
UserAgent: "msa-agent/" + version,
|
||||
RootCAs: roots,
|
||||
})
|
||||
if err != nil {
|
||||
return err
|
||||
@@ -205,35 +211,80 @@ func registrationsFrom(cfg config.File) []transport.DeviceRegistration {
|
||||
return out
|
||||
}
|
||||
|
||||
// loop is the agent's whole life: handshake, then poll forever, re-handshaking whenever the relay
|
||||
// says the session is gone and backing off when the relay is unreachable.
|
||||
// relayRoots loads a PEM bundle to trust instead of the system store, or returns nil to keep the
|
||||
// system store. An unreadable or empty bundle is a startup error rather than a silent fallback: an
|
||||
// operator who pinned a CA and got a system-store verification instead would be running a trust
|
||||
// decision they did not choose, and the failure would look like a network problem.
|
||||
func relayRoots(path string) (*x509.CertPool, error) {
|
||||
if path == "" {
|
||||
return nil, nil
|
||||
}
|
||||
pem, err := os.ReadFile(path)
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("relayCAFile: %w", err)
|
||||
}
|
||||
pool := x509.NewCertPool()
|
||||
if !pool.AppendCertsFromPEM(pem) {
|
||||
return nil, fmt.Errorf("relayCAFile %s contains no PEM certificates", path)
|
||||
}
|
||||
return pool, nil
|
||||
}
|
||||
|
||||
// loop is the agent's whole life: handshake once, then poll forever, re-handshaking only when the
|
||||
// relay says the session is gone.
|
||||
//
|
||||
// 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.
|
||||
// The handshake is *not* part of the per-cycle work, and that distinction is load-bearing rather
|
||||
// than an optimisation. The relay's registry holds one session per node: a second handshake replaces
|
||||
// the first, and replacing it drops whatever poll was parked. So an agent that re-handshook on every
|
||||
// cycle would tear down its own parked poll each time, and any request that arrived in the window
|
||||
// between the two handshakes would be delivered into a session that no longer exists and time out
|
||||
// with "agent offline" — while the agent was connected the whole time. An earlier version of this
|
||||
// loop did exactly that, and only the end-to-end test caught it, because each component was behaving
|
||||
// correctly on its own.
|
||||
//
|
||||
// Errors are classified here rather than handled per call site, because the situations look identical
|
||||
// from inside a function and need different responses: retry now, re-authenticate, or wait.
|
||||
func loop(ctx context.Context, node *agent.Agent, client *transport.Client, pol *policy.Policy, devices []transport.DeviceRegistration) error {
|
||||
backoff := minBackoff
|
||||
authenticated := false
|
||||
for {
|
||||
if ctx.Err() != nil {
|
||||
return nil
|
||||
}
|
||||
err := handshakeAndPoll(ctx, node, client, pol, devices)
|
||||
var err error
|
||||
if !authenticated {
|
||||
if _, err = client.Hello(ctx, version, pol.SHA256()); err == nil {
|
||||
// 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 right after
|
||||
// the handshake is what makes that true without a second channel: the node's paired
|
||||
// set, including any revocation, is re-asserted every time a session is established.
|
||||
// Empty is pushed too, because "nobody" is exactly the claim revocation needs.
|
||||
err = client.PushDevices(ctx, devices)
|
||||
authenticated = err == nil
|
||||
}
|
||||
}
|
||||
if err == nil {
|
||||
err = cycle(ctx, node, client, pol)
|
||||
}
|
||||
switch {
|
||||
case err == nil:
|
||||
backoff = minBackoff
|
||||
case errors.Is(err, transport.ErrNoWork):
|
||||
case err == nil, errors.Is(err, transport.ErrNoWork):
|
||||
backoff = minBackoff
|
||||
case errors.Is(err, transport.ErrUnauthorized), errors.Is(err, transport.ErrSessionExpired):
|
||||
// Re-handshake on the next iteration. The wait is what stops a bad token from becoming
|
||||
// a tight loop against the relay.
|
||||
// Re-handshake on the next iteration. The wait is what stops a bad token from becoming a
|
||||
// tight loop against the relay.
|
||||
slog.Warn("relay rejected the agent session, re-authenticating", "error", err)
|
||||
authenticated = false
|
||||
backoff = minBackoff * 5
|
||||
case errors.Is(err, transport.ErrRateLimited):
|
||||
// The relay said how long to wait; doubling its own backoff on top of that would be
|
||||
// the agent overriding an explicit instruction.
|
||||
// The relay said how long to wait; doubling its own backoff on top of that would be the
|
||||
// agent overriding an explicit instruction.
|
||||
backoff = client.RetryAfter()
|
||||
default:
|
||||
// A transport failure leaves the session's fate unknown — the relay may still hold the
|
||||
// parked poll, or may have dropped it. Re-handshaking is the safe reading, because a
|
||||
// stale session fails every subsequent request.
|
||||
slog.Warn("poll cycle failed", "error", err, "retryIn", backoff)
|
||||
authenticated = false
|
||||
backoff = next(backoff)
|
||||
}
|
||||
select {
|
||||
@@ -252,21 +303,6 @@ func next(current time.Duration) time.Duration {
|
||||
return doubled
|
||||
}
|
||||
|
||||
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)
|
||||
}
|
||||
|
||||
// cycle drains one batch. Each request is handled and answered before the next is fetched, so a
|
||||
// slow upstream request delays the queue rather than buffering it in memory.
|
||||
func cycle(ctx context.Context, node *agent.Agent, client *transport.Client, pol *policy.Policy) error {
|
||||
|
||||
@@ -50,6 +50,10 @@ type File struct {
|
||||
// handles or transmits it. Empty declares the node unprotected; a protected node then refuses
|
||||
// commands with node_auth_failed rather than looping against a lockout.
|
||||
NodePassword string `json:"nodePassword,omitempty"`
|
||||
// RelayCAFile is a PEM bundle to trust instead of the system store when verifying the relay's
|
||||
// certificate. It exists for a relay behind a private CA; the default is the system store, which
|
||||
// is what a public certificate needs. Setting it pins trust rather than disabling verification.
|
||||
RelayCAFile string `json:"relayCAFile,omitempty"`
|
||||
// AuditPath is where the audit log is appended.
|
||||
AuditPath string `json:"auditPath"`
|
||||
// Devices may be empty only in pair mode (--pair). In normal operation an agent with no
|
||||
|
||||
@@ -14,6 +14,8 @@ import (
|
||||
"bytes"
|
||||
"context"
|
||||
"crypto/ed25519"
|
||||
"crypto/tls"
|
||||
"crypto/x509"
|
||||
"encoding/json"
|
||||
"errors"
|
||||
"fmt"
|
||||
@@ -75,10 +77,17 @@ type Batch struct {
|
||||
}
|
||||
|
||||
// Client talks to one relay.
|
||||
//
|
||||
// The two credentials are kept apart on purpose, and keeping them apart is not cosmetic. The bearer
|
||||
// token identifies the node to the relay's authentication layer; the session token is what the relay
|
||||
// issued after the challenge signature proved possession of the node's key. The relay checks both on
|
||||
// every poll, and using one where the other belongs fails as "relay rejected the agent credential" —
|
||||
// which reads like a bad token rather than like a mixed-up header.
|
||||
type Client struct {
|
||||
baseURL string
|
||||
nodeID string
|
||||
token string
|
||||
session string
|
||||
signer ed25519.PrivateKey
|
||||
http *http.Client
|
||||
userAgent string
|
||||
@@ -95,6 +104,14 @@ type Config struct {
|
||||
Timeout time.Duration
|
||||
UserAgent string
|
||||
MaxBytes int64
|
||||
// RootCAs replaces the system trust store when non-nil. It exists for one deployment shape: a
|
||||
// relay behind a private CA, where installing the CA into the node's system store is either not
|
||||
// possible or not desirable. The zero value keeps the system store, which is what an operator
|
||||
// using a public certificate wants and what the agent does by default.
|
||||
//
|
||||
// It is a *CertPool and not an "insecure" flag on purpose: the way to trust a private relay is to
|
||||
// pin the CA that signed it, never to stop verifying.
|
||||
RootCAs *x509.CertPool
|
||||
}
|
||||
|
||||
// New validates the relay URL before any traffic is sent.
|
||||
@@ -127,6 +144,15 @@ func New(cfg Config) (*Client, error) {
|
||||
if maxBytes <= 0 {
|
||||
maxBytes = 1 << 20
|
||||
}
|
||||
client := &http.Client{Timeout: timeout}
|
||||
if cfg.RootCAs != nil {
|
||||
// A fresh transport rather than a mutated DefaultTransport: mutating the shared one would
|
||||
// change trust for every other client in the process, which is a surprising thing for a
|
||||
// constructor to do.
|
||||
client.Transport = &http.Transport{
|
||||
TLSClientConfig: &tls.Config{RootCAs: cfg.RootCAs, MinVersion: tls.VersionTLS12},
|
||||
}
|
||||
}
|
||||
return &Client{
|
||||
baseURL: strings.TrimRight(cfg.BaseURL, "/"),
|
||||
nodeID: cfg.NodeID,
|
||||
@@ -134,7 +160,7 @@ func New(cfg Config) (*Client, error) {
|
||||
signer: cfg.Signer,
|
||||
userAgent: cfg.UserAgent,
|
||||
maxBytes: maxBytes,
|
||||
http: &http.Client{Timeout: timeout},
|
||||
http: client,
|
||||
}, nil
|
||||
}
|
||||
|
||||
@@ -163,7 +189,7 @@ func (c *Client) Hello(ctx context.Context, build, policySHA string) (HelloRespo
|
||||
if err := c.post(ctx, PathHello, signed, &completed, "", false); err != nil {
|
||||
return completed, err
|
||||
}
|
||||
c.token = completed.Session.Token
|
||||
c.session = completed.Session.Token
|
||||
return completed, nil
|
||||
}
|
||||
|
||||
@@ -179,7 +205,7 @@ func (c *Client) Poll(ctx context.Context) (Batch, error) {
|
||||
var batch Batch
|
||||
// A long poll that times out answers with an empty body, and that is the normal case rather than
|
||||
// a failure, so an empty body means ErrNoWork here and only here.
|
||||
err := c.post(ctx, PathPoll, map[string]string{"node": c.nodeID}, &batch, c.session(), true)
|
||||
err := c.post(ctx, PathPoll, map[string]string{"node": c.nodeID}, &batch, c.sessionToken(), true)
|
||||
if err != nil {
|
||||
return Batch{}, err
|
||||
}
|
||||
@@ -192,10 +218,12 @@ func (c *Client) Poll(ctx context.Context) (Batch, error) {
|
||||
// must not share that reading of an empty body, or a confirmed delivery would be treated as a
|
||||
// reason to back off.
|
||||
func (c *Client) Result(ctx context.Context, envelope json.RawMessage) error {
|
||||
return c.post(ctx, PathResult, envelope, &json.RawMessage{}, c.session(), false)
|
||||
return c.post(ctx, PathResult, envelope, &json.RawMessage{}, c.sessionToken(), false)
|
||||
}
|
||||
|
||||
func (c *Client) session() string { return c.token }
|
||||
// session returns the token the relay issued for this session. It is empty before the handshake, and
|
||||
// an empty value makes the relay answer 401, which the loop treats as "re-authenticate".
|
||||
func (c *Client) sessionToken() string { return c.session }
|
||||
|
||||
// DeviceRegistration is one paired phone as the relay needs to know it: the bearer token the phone
|
||||
// presents, minted by the agent at pairing time and never seen by the relay before that moment.
|
||||
@@ -213,7 +241,7 @@ type DeviceRegistration struct {
|
||||
func (c *Client) PushDevices(ctx context.Context, devices []DeviceRegistration) error {
|
||||
return c.post(ctx, PathDevices, map[string]any{
|
||||
"node": c.nodeID, "devices": devices,
|
||||
}, &json.RawMessage{}, c.session(), false)
|
||||
}, &json.RawMessage{}, c.sessionToken(), false)
|
||||
}
|
||||
|
||||
// post is the single place a relay request is made, so authentication, size limits, status mapping
|
||||
|
||||
@@ -3,6 +3,7 @@ package transport
|
||||
import (
|
||||
"context"
|
||||
"crypto/ed25519"
|
||||
"crypto/x509"
|
||||
"encoding/json"
|
||||
"errors"
|
||||
"io"
|
||||
@@ -193,6 +194,10 @@ func TestResultPostsTheEnvelopeVerbatim(t *testing.T) {
|
||||
got, _ = io.ReadAll(r.Body)
|
||||
w.WriteHeader(http.StatusOK)
|
||||
}), "t")
|
||||
// The session the handshake would have issued. It is set directly rather than by running the
|
||||
// handshake because this test is about the result body, and the two tokens being distinct is
|
||||
// asserted on its own in TestTheBearerAndTheSessionAreDifferentCredentials.
|
||||
client.session = "session-token"
|
||||
envelope := json.RawMessage(`{"v":1,"node":"nd_test0001","p":"AAAA"}`)
|
||||
if err := client.Result(context.Background(), envelope); err != nil {
|
||||
t.Fatal(err)
|
||||
@@ -214,12 +219,16 @@ func TestPushDevicesPostsTheFullSetUnderTheSession(t *testing.T) {
|
||||
if r.URL.Path != PathDevices {
|
||||
t.Fatalf("posted to %q", r.URL.Path)
|
||||
}
|
||||
if r.Header.Get("X-MSA-Session") == "" {
|
||||
t.Error("the device push must carry the session")
|
||||
if r.Header.Get("X-MSA-Session") != "session-token" {
|
||||
t.Errorf("the device push carried session %q", r.Header.Get("X-MSA-Session"))
|
||||
}
|
||||
if r.Header.Get("Authorization") != "Bearer t" {
|
||||
t.Errorf("the device push carried authorization %q", r.Header.Get("Authorization"))
|
||||
}
|
||||
json.NewDecoder(r.Body).Decode(&got)
|
||||
w.WriteHeader(http.StatusOK)
|
||||
}), "t")
|
||||
client.session = "session-token"
|
||||
err := client.PushDevices(context.Background(), []DeviceRegistration{
|
||||
{App: "ap_5b1d0000", Token: "tok-a"},
|
||||
})
|
||||
@@ -237,6 +246,7 @@ func TestPushDevicesPostsTheFullSetUnderTheSession(t *testing.T) {
|
||||
client2, _ := newClient(t, http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
|
||||
w.WriteHeader(http.StatusOK)
|
||||
}), "t")
|
||||
client2.session = "session-token"
|
||||
if err := client2.PushDevices(context.Background(), []DeviceRegistration{}); err != nil {
|
||||
t.Fatalf("empty push: %v", err)
|
||||
}
|
||||
@@ -270,3 +280,109 @@ func TestRetryAfterHonoursTheRelayHeader(t *testing.T) {
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// The bearer token and the session token are different credentials with different jobs, and swapping
|
||||
// them is the failure this test exists to catch. The bearer identifies the node to the relay's
|
||||
// authentication layer; the session is what the relay issued once the challenge signature proved
|
||||
// possession of the node's key. Sending the session as the bearer (or vice versa) makes the relay
|
||||
// answer 401, which the loop reads as "re-authenticate" — so the symptom is an agent that connects
|
||||
// and immediately reconnects forever, with a log line that blames the token.
|
||||
func TestTheBearerAndTheSessionAreDifferentCredentials(t *testing.T) {
|
||||
var seen []struct {
|
||||
path string
|
||||
authorization string
|
||||
session string
|
||||
}
|
||||
client, _ := newClient(t, http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
|
||||
seen = append(seen, struct {
|
||||
path string
|
||||
authorization string
|
||||
session string
|
||||
}{r.URL.Path, r.Header.Get("Authorization"), r.Header.Get("X-MSA-Session")})
|
||||
w.Header().Set("Content-Type", "application/json")
|
||||
switch r.URL.Path {
|
||||
case PathHello:
|
||||
var body HelloRequest
|
||||
json.NewDecoder(r.Body).Decode(&body)
|
||||
if body.Challenge == "" {
|
||||
json.NewEncoder(w).Encode(HelloResponse{Challenge: "issued"})
|
||||
} else {
|
||||
json.NewEncoder(w).Encode(HelloResponse{Session: Session{Token: "session-abc"}})
|
||||
}
|
||||
case PathPoll:
|
||||
w.WriteHeader(http.StatusNoContent)
|
||||
default:
|
||||
w.WriteHeader(http.StatusOK)
|
||||
}
|
||||
}), "bearer-token")
|
||||
|
||||
if _, err := client.Hello(context.Background(), "build", "sha"); err != nil {
|
||||
t.Fatalf("Hello: %v", err)
|
||||
}
|
||||
if _, err := client.Poll(context.Background()); !errors.Is(err, ErrNoWork) {
|
||||
t.Fatalf("Poll: %v", err)
|
||||
}
|
||||
|
||||
for _, request := range seen {
|
||||
if request.authorization != "Bearer bearer-token" {
|
||||
t.Errorf("%s sent authorization %q", request.path, request.authorization)
|
||||
}
|
||||
switch request.path {
|
||||
case PathHello:
|
||||
// The handshake is authenticated by the bearer and the signature; there is no session yet.
|
||||
if request.session != "" {
|
||||
t.Errorf("the handshake carried a session %q", request.session)
|
||||
}
|
||||
case PathPoll:
|
||||
if request.session != "session-abc" {
|
||||
t.Errorf("poll carried session %q, want the one the handshake issued", request.session)
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// A pinned CA replaces the system store rather than adding to it, and a certificate that only the
|
||||
// system store would accept must then fail. That asymmetry is the point: `relayCAFile` is for a
|
||||
// relay behind a private CA, and an operator who set it expects the relay to be verified against
|
||||
// exactly that root.
|
||||
func TestRootCAsReplaceTheSystemStore(t *testing.T) {
|
||||
server := httptest.NewTLSServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
|
||||
w.WriteHeader(http.StatusOK)
|
||||
}))
|
||||
t.Cleanup(server.Close)
|
||||
|
||||
_, signer, err := ed25519.GenerateKey(nil)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
// The test server's own certificate is self-signed, so trusting it works...
|
||||
pool := x509.NewCertPool()
|
||||
pool.AddCert(server.Certificate())
|
||||
client, err := New(Config{
|
||||
BaseURL: server.URL, NodeID: "nd_test0001", Token: "t", Signer: signer,
|
||||
Timeout: 2 * time.Second, RootCAs: pool,
|
||||
})
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
// The transport is left as configured, so this exercises the pinned pool rather than the test
|
||||
// server's convenience client. The handler answers 200 with no body, so ErrNoWork is the success
|
||||
// case here: what is under test is whether the TLS handshake completed, and a verification
|
||||
// failure surfaces as a transport error instead.
|
||||
if _, err := client.Poll(context.Background()); !errors.Is(err, ErrNoWork) {
|
||||
t.Fatalf("a pinned CA did not verify the relay: %v", err)
|
||||
}
|
||||
|
||||
// ...and an empty pool verifies nothing, which is the failure mode an operator sees when they
|
||||
// point relayCAFile at the wrong file: a TLS error, never a silent fallback.
|
||||
empty, err := New(Config{
|
||||
BaseURL: server.URL, NodeID: "nd_test0001", Token: "t", Signer: signer,
|
||||
Timeout: 2 * time.Second, RootCAs: x509.NewCertPool(),
|
||||
})
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if _, err := empty.Poll(context.Background()); err == nil || errors.Is(err, ErrNoWork) {
|
||||
t.Fatalf("an empty root pool still verified the relay: %v", err)
|
||||
}
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user