From dfd1ce436df67ec62a14223d7029f1be0840f227 Mon Sep 17 00:00:00 2001 From: Saurav Date: Sat, 17 Oct 2026 12:38:58 +0530 Subject: [PATCH 1/3] logging: the node's /debug/logs is always JSON, whatever --log-format says --log-format switched stderr and the in-memory ring buffer together, so a tool reading /debug/logs had to parse the console format or make the operator give up a readable stderr. The ring buffer is now its own sink with its own JSON encoder, attached through go-log's pipe reader, and --log-format only shapes stderr (and the log file under --daemonize). The ringbuffer:// zap sink and the Sync and Close methods it needed are gone. Setup attaches the sink once, so calling it again does not retain every line twice. Lines reach the ring just after the log call returns, so the ring tests wait on RecentLogs instead of reading it at once. They pass 300 runs under -race and 200 pinned to two CPUs. Refs #332 --- cmd/agentmesh-node/main.go | 2 +- internal/logging/logging.go | 58 ++++++---- internal/logging/logging_test.go | 108 ++++++++++++++++-- site/content/docs/reference/agentmesh-node.md | 2 +- site/content/docs/reference/node-api.md | 5 +- 5 files changed, 141 insertions(+), 34 deletions(-) diff --git a/cmd/agentmesh-node/main.go b/cmd/agentmesh-node/main.go index 9609dfcc..9028c139 100644 --- a/cmd/agentmesh-node/main.go +++ b/cmd/agentmesh-node/main.go @@ -866,7 +866,7 @@ func main() { runCmd.Flags().BoolVar(&enableRelayFlag, "enable-relay", false, "Allow this node to serve as a relay for others") runCmd.Flags().BoolVar(&daemonizeFlag, "daemonize", false, "Run the node in the background and return once its local API answers (writes agentmesh-node.log and agentmesh-node.pid to the data directory, and generates an API token there if none is configured)") runCmd.Flags().StringVar(&logLevelFlag, "log-level", "info", "Log level (debug, info, warn, error)") - runCmd.Flags().StringVar(&logFormatFlag, "log-format", "text", "Log format (text, json)") + runCmd.Flags().StringVar(&logFormatFlag, "log-format", "text", "Log format on stderr and in the log file (text, json); /debug/logs is always JSON") runCmd.Flags().DurationVar(&keyGracePeriodFlag, "key-grace-period", 24*time.Hour, "Key grace period for old keys (e.g. 24h)") runCmd.Flags().BoolVar(&allowLoopbackFlag, "allow-loopback", false, "Allow publishing and connecting to loopback/link-local addresses") runCmd.Flags().BoolVar(&announcePrivateFlag, "announce-private", true, "Publish this host's private (RFC1918/ULA) addresses to the mesh; keep enabled for LAN or on-premises meshes, disable when peers are only reachable through routers or public addresses") diff --git a/internal/logging/logging.go b/internal/logging/logging.go index 99360bad..51a3704e 100644 --- a/internal/logging/logging.go +++ b/internal/logging/logging.go @@ -16,20 +16,18 @@ package logging import ( + "bufio" "container/ring" - "net/url" "strings" "sync" golog "github.com/ipfs/go-log/v2" - "go.uber.org/zap" ) var logger = golog.Logger("logging") const ( - ringBufferScheme = "ringbuffer" - logBufferLines = 500 + logBufferLines = 500 // maxLogLineBytes bounds each retained line so the ring's memory is // bounded too (logBufferLines * maxLogLineBytes), whatever a remote peer // manages to get logged. @@ -44,9 +42,11 @@ var ( // Options selects how a binary logs. Zero values keep the defaults: info // level, text format and no in-memory ring buffer. type Options struct { - Level string // log level: debug, info, warn or error (default: info) - Format string // output format: text or json (default: text) - RingBuffer bool // also retain the most recent lines in memory (see RecentLogs) + Level string // log level: debug, info, warn or error (default: info) + // Format is the output format on stderr: text or json (default: text). The + // ring buffer is always JSON, so tools never parse the human format. + Format string + RingBuffer bool // also retain the most recent lines in memory as JSON (see RecentLogs) } // Setup applies opts process-wide. @@ -62,17 +62,7 @@ func Setup(opts Options) { cfg := golog.GetConfig() cfg.Level = lvl - if opts.RingBuffer { - ringBufferOnce.Do(func() { - ringBuffer = newRingBufferSink(logBufferLines) - _ = zap.RegisterSink(ringBufferScheme, func(*url.URL) (zap.Sink, error) { - return ringBuffer, nil - }) - }) - cfg.URL = ringBufferScheme + "://" - } else { - cfg.URL = "" - } + cfg.URL = "" badFormat := false switch opts.Format { case "json": @@ -84,6 +74,9 @@ func Setup(opts Options) { cfg.Format = golog.PlaintextOutput } golog.SetupLogging(cfg) + if opts.RingBuffer { + ringBufferOnce.Do(startRingBuffer) + } // go-log starts at error level, so warnings only show after SetAllLoggers. if levelErr != nil { @@ -101,8 +94,30 @@ func Setup(opts Options) { } } -// RecentLogs returns the most recent log lines captured when Options.RingBuffer -// was enabled in Setup. +// startRingBuffer attaches the ring buffer to every logger as a second sink +// with its own JSON encoder. go-log registers the sink once, outside the +// primary output, so later Setup calls reconfigure stderr and leave it alone. +func startRingBuffer() { + ringBuffer = newRingBufferSink(logBufferLines) + pipe := golog.NewPipeReader(golog.PipeFormat(golog.JSONOutput)) + go func() { + // The pipe is synchronous: a reader that stops would block every log + // call, so this loop must never return before the process does. + r := bufio.NewReader(pipe) + for { + line, err := r.ReadBytes('\n') + if len(line) > 0 { + _, _ = ringBuffer.Write(line) + } + if err != nil { + return + } + } + }() +} + +// RecentLogs returns the most recent log lines, as JSON, captured when +// Options.RingBuffer was enabled in Setup. func RecentLogs() []string { if ringBuffer == nil { return []string{} @@ -133,9 +148,6 @@ func (s *ringBufferSink) Write(p []byte) (int, error) { return len(p), nil } -func (s *ringBufferSink) Sync() error { return nil } -func (s *ringBufferSink) Close() error { return nil } - func (s *ringBufferSink) lines() []string { s.mu.Lock() defer s.mu.Unlock() diff --git a/internal/logging/logging_test.go b/internal/logging/logging_test.go index 51061a1e..f3ecf6eb 100644 --- a/internal/logging/logging_test.go +++ b/internal/logging/logging_test.go @@ -15,8 +15,12 @@ package logging import ( + "encoding/json" + "fmt" + "os" "strings" "testing" + "time" golog "github.com/ipfs/go-log/v2" "go.uber.org/zap/zapcore" @@ -81,16 +85,104 @@ func TestSetupInvalidLevelFallsBackToInfo(t *testing.T) { func TestSetupRingBuffer(t *testing.T) { Setup(Options{Level: "info", RingBuffer: true}) golog.Logger("logging-test-ring").Infof("hello ring buffer") - lines := RecentLogs() - found := false - for _, line := range lines { - if strings.Contains(line, "hello ring buffer") { - found = true - break + ringLine(t, "hello ring buffer") +} + +// captureStderr points os.Stderr at a file for the rest of the test, so a +// Setup that opens the "stderr" sink writes there. +func captureStderr(t *testing.T) *os.File { + t.Helper() + f, err := os.CreateTemp(t.TempDir(), "stderr") + if err != nil { + t.Fatal(err) + } + orig := os.Stderr + os.Stderr = f + t.Cleanup(func() { + os.Stderr = orig + _ = f.Close() + }) + return f +} + +// ringLine waits for the ring buffer to hold a line containing want: the +// lines reach it through a pipe, so they land just after the log call returns. +func ringLine(t *testing.T, want string) string { + t.Helper() + deadline := time.Now().Add(5 * time.Second) + for { + for _, line := range RecentLogs() { + if strings.Contains(line, want) { + return line + } + } + if time.Now().After(deadline) { + t.Fatalf("RecentLogs() = %v, want a line containing %q", RecentLogs(), want) + } + time.Sleep(5 * time.Millisecond) + } +} + +// The ring buffer is what tools read through /debug/logs, so it holds JSON +// whatever --log-format says; the format only shapes what humans read on stderr. +func TestRingBufferIsJSONWhateverTheFormat(t *testing.T) { + for _, format := range []string{"text", "json"} { + t.Run(format, func(t *testing.T) { + stderr := captureStderr(t) + Setup(Options{Level: "info", Format: format, RingBuffer: true}) + msg := fmt.Sprintf("ring-json-%s-%d", format, time.Now().UnixNano()) + golog.Logger("logging-test-sinks").Infof("%s", msg) + + var entry struct { + Level string `json:"level"` + Logger string `json:"logger"` + Msg string `json:"msg"` + } + line := ringLine(t, msg) + if err := json.Unmarshal([]byte(line), &entry); err != nil { + t.Fatalf("ring line %q is not JSON: %v", line, err) + } + if entry.Msg != msg || entry.Level != "info" || entry.Logger != "logging-test-sinks" { + t.Errorf("ring entry = %+v, want msg %q, level info, logger logging-test-sinks", entry, msg) + } + + out, err := os.ReadFile(stderr.Name()) + if err != nil { + t.Fatal(err) + } + stderrLine := strings.TrimSpace(string(out)) + if !strings.Contains(stderrLine, msg) { + t.Fatalf("stderr = %q, want a line containing %q", stderrLine, msg) + } + if gotJSON := json.Valid([]byte(stderrLine)); gotJSON != (format == "json") { + t.Errorf("stderr line %q is JSON = %v, want %v for --log-format=%s", stderrLine, gotJSON, format == "json", format) + } + }) + } +} + +// A second Setup must not add a second ring sink, or every line would be +// retained twice. +func TestSetupRingBufferTwiceKeepsOneCopy(t *testing.T) { + Setup(Options{Level: "info", RingBuffer: true}) + Setup(Options{Level: "info", RingBuffer: true}) + id := time.Now().UnixNano() + msg := fmt.Sprintf("only once please %d", id) + log := golog.Logger("logging-test-twice") + log.Infof("%s", msg) + // Lines reach the ring in order, so once the later line is there, every + // copy of the earlier one is too. + log.Infof("after %d", id) + ringLine(t, fmt.Sprintf("after %d", id)) + + n := 0 + for _, line := range RecentLogs() { + if strings.Contains(line, msg) { + n++ } } - if !found { - t.Errorf("RecentLogs() = %v, want line containing %q", lines, "hello ring buffer") + if n != 1 { + t.Errorf("ring holds %d copies of %q, want 1", n, msg) } } diff --git a/site/content/docs/reference/agentmesh-node.md b/site/content/docs/reference/agentmesh-node.md index 84b50be8..88a1741f 100644 --- a/site/content/docs/reference/agentmesh-node.md +++ b/site/content/docs/reference/agentmesh-node.md @@ -126,7 +126,7 @@ explains how to enroll. | `--monitor-interval` | `1m` | How often the monitor checks that a router is connected. | | `--autorelay-min-interval`, `--autorelay-boot-delay`, `--autorelay-backoff` | `30s`, `0s`, `3s` | libp2p AutoRelay tuning. | | `--log-level` | `info` | `debug`, `info`, `warn` or `error`. | -| `--log-format` | `text` | `text` or `json`. | +| `--log-format` | `text` | `text` or `json`: the format of the node's log output on stderr, and of the log file under `--daemonize`. `agentmesh-node debug logs` is always JSON, one object per line, whatever this is set to. | ## `reset` diff --git a/site/content/docs/reference/node-api.md b/site/content/docs/reference/node-api.md index 898bfd86..918bcc31 100644 --- a/site/content/docs/reference/node-api.md +++ b/site/content/docs/reference/node-api.md @@ -385,6 +385,9 @@ every peer it is connected to; `network-info` lists the addresses the node listens on, the ones it announces, and the ones in the records it last announced to the directory (`directory_addresses`), which follow the announced ones at the next announcement pass; `peer-addrs` is where the -directory says a peer is reached, looked up and not dialed. `agentmesh-node +directory says a peer is reached, looked up and not dialed. `logs` returns the +node's most recent log lines as `logs`, a list of strings; each is one JSON +object (`ts`, `level`, `logger`, `caller`, `msg`), whatever `--log-format` +sets for stderr, so a tool needs no console-format parsing. `agentmesh-node debug` on the command line wraps them. Their output is not stable across releases. From 231145bca04d1d0c2a5b042a2d5e0e7c3b30c38f Mon Sep 17 00:00:00 2001 From: Saurav Date: Sat, 17 Oct 2026 13:04:25 +0530 Subject: [PATCH 2/3] logging: say why the log ring buffer stopped The goroutine that feeds the ring left on a read error without a trace. Its pipe is synchronous, so a registered pipe with no reader blocks every log call, and the line that would explain it blocks too. The copy loop is now copyLines, which returns the read error and treats end of input as a clean stop. The goroutine closes the pipe, which unregisters it, and only then logs the error. --- internal/logging/logging.go | 38 +++++++++++++++++++++++--------- internal/logging/logging_test.go | 34 ++++++++++++++++++++++++++++ 2 files changed, 61 insertions(+), 11 deletions(-) diff --git a/internal/logging/logging.go b/internal/logging/logging.go index 51a3704e..aff6888e 100644 --- a/internal/logging/logging.go +++ b/internal/logging/logging.go @@ -18,6 +18,8 @@ package logging import ( "bufio" "container/ring" + "errors" + "io" "strings" "sync" @@ -101,21 +103,35 @@ func startRingBuffer() { ringBuffer = newRingBufferSink(logBufferLines) pipe := golog.NewPipeReader(golog.PipeFormat(golog.JSONOutput)) go func() { - // The pipe is synchronous: a reader that stops would block every log - // call, so this loop must never return before the process does. - r := bufio.NewReader(pipe) - for { - line, err := r.ReadBytes('\n') - if len(line) > 0 { - _, _ = ringBuffer.Write(line) - } - if err != nil { - return - } + err := copyLines(pipe, ringBuffer) + // The pipe is synchronous: while it stays registered with no reader, + // every log call blocks, the one below included. Closing it + // unregisters it, so only then is it safe to say why the ring stopped. + _ = pipe.Close() + if err != nil { + logger.Errorf("log ring buffer stopped, /debug/logs gets no new lines: %v", err) } }() } +// copyLines writes each line r yields to sink until r ends. A clean end of +// input is not an error; a read error is returned. +func copyLines(r io.Reader, sink io.Writer) error { + br := bufio.NewReader(r) + for { + line, err := br.ReadBytes('\n') + if len(line) > 0 { + _, _ = sink.Write(line) + } + if errors.Is(err, io.EOF) { + return nil + } + if err != nil { + return err + } + } +} + // RecentLogs returns the most recent log lines, as JSON, captured when // Options.RingBuffer was enabled in Setup. func RecentLogs() []string { diff --git a/internal/logging/logging_test.go b/internal/logging/logging_test.go index f3ecf6eb..bffd6d2c 100644 --- a/internal/logging/logging_test.go +++ b/internal/logging/logging_test.go @@ -16,10 +16,13 @@ package logging import ( "encoding/json" + "errors" "fmt" + "io" "os" "strings" "testing" + "testing/iotest" "time" golog "github.com/ipfs/go-log/v2" @@ -186,6 +189,37 @@ func TestSetupRingBufferTwiceKeepsOneCopy(t *testing.T) { } } +func TestCopyLinesEndOfInputIsNotAnError(t *testing.T) { + sink := newRingBufferSink(4) + if err := copyLines(strings.NewReader("one\ntwo\nthree"), sink); err != nil { + t.Fatalf("copyLines returned %v at end of input, want nil", err) + } + got := sink.lines() + want := []string{"one", "two", "three"} + if len(got) != len(want) { + t.Fatalf("lines = %q, want %q (a last line without a newline is kept)", got, want) + } + for i := range want { + if got[i] != want[i] { + t.Errorf("lines[%d] = %q, want %q", i, got[i], want[i]) + } + } +} + +// The ring's reader reports why it stopped, so the caller can say so in the +// log instead of leaving /debug/logs silently stale. +func TestCopyLinesReturnsTheReadError(t *testing.T) { + boom := errors.New("boom") + sink := newRingBufferSink(4) + err := copyLines(io.MultiReader(strings.NewReader("kept\n"), iotest.ErrReader(boom)), sink) + if !errors.Is(err, boom) { + t.Errorf("copyLines returned %v, want %v", err, boom) + } + if got := sink.lines(); len(got) != 1 || got[0] != "kept" { + t.Errorf("lines = %q, want [kept]: lines read before the error stay in the ring", got) + } +} + // The ring keeps whole lines; without a per-line cap a remote peer that gets // an 8 MiB string logged pins it in memory for 500 rounds. func TestRingBufferSinkCapsLineBytes(t *testing.T) { From 306db6cc3ed97199970db7b94b0b74154c2ef9f9 Mon Sep 17 00:00:00 2001 From: Saurav Date: Sat, 17 Oct 2026 13:24:14 +0530 Subject: [PATCH 3/3] node: report the verdict of the key that signed the token, not the last key tried With two trusted control-plane keys, the node tried each in turn and returned the last error. A token that verified under the first key and was denied by policy then came back as "biscuit: invalid signature" from the second key, so the caller never saw the denial. Which error won depended on key order, which made the policy e2e Negative and Attenuation paths flaky. Only a signature mismatch now moves on to the next key; once a key verifies the token, its verdict is the answer. A trusted key of the wrong size is skipped with a warning instead of ending the search. --- internal/node/middleware.go | 33 ++++++++------ internal/node/middleware_test.go | 78 ++++++++++++++++++++++++++++++++ 2 files changed, 96 insertions(+), 15 deletions(-) diff --git a/internal/node/middleware.go b/internal/node/middleware.go index 1ac4d060..890f309d 100644 --- a/internal/node/middleware.go +++ b/internal/node/middleware.go @@ -16,6 +16,7 @@ package node import ( "crypto/ed25519" + "errors" "fmt" "runtime/debug" "sync/atomic" @@ -223,28 +224,30 @@ func (n *AgentMeshNode) verifyBiscuitTokenWithRules(biscuitBytes []byte, reqCtx keys := n.trustedKeys n.keysMu.RUnlock() - var authorized bool - var taskRules []*api.TaskAuthorizationRule + // Only a signature mismatch moves on to the next trusted key. Once a key + // verifies the token, that key's verdict is the answer: carrying on would + // replace a policy denial with the next key's "invalid signature". var lastErr error for _, pubKey := range keys { + if len(pubKey.Key) != ed25519.PublicKeySize { + logger.Warnf("[Auth] Skipping trusted key of invalid size %d", len(pubKey.Key)) + continue + } logger.Infof("[Auth] Trying key: %x", pubKey.Key) - if rules, err := n.authorizeWithRules(biscuitBytes, reqCtx, pubKey.Key); err == nil { - authorized = true - taskRules = rules - break - } else { - lastErr = err + rules, err := n.authorizeWithRules(biscuitBytes, reqCtx, pubKey.Key) + if err == nil { + return rules, nil } - } - - if !authorized { - if lastErr != nil { - return nil, lastErr + if !errors.Is(err, biscuit.ErrInvalidSignature) { + return nil, err } - return nil, fmt.Errorf("authorization failed") + lastErr = err } - return taskRules, nil + if lastErr != nil { + return nil, lastErr + } + return nil, fmt.Errorf("authorization failed") } func (n *AgentMeshNode) Authorize(rawToken []byte, req RequestContext, pubKey ed25519.PublicKey) error { diff --git a/internal/node/middleware_test.go b/internal/node/middleware_test.go index 839c5667..0c0f6939 100644 --- a/internal/node/middleware_test.go +++ b/internal/node/middleware_test.go @@ -21,6 +21,7 @@ import ( cryptorand "crypto/rand" "encoding/base64" "encoding/json" + "errors" "fmt" "io" @@ -253,6 +254,83 @@ func TestAuthorizeRejectsExpiredBiscuit(t *testing.T) { } } +// TestVerifyBiscuitTokenReportsVerdictOfSigningKey reproduces the flaky +// policy e2e: with two trusted control-plane keys, a token whose signature +// verifies under one key and is denied by policy was reported as +// "invalid signature" when the other key happened to be tried last, hiding +// the denial from the caller. +func TestVerifyBiscuitTokenReportsVerdictOfSigningKey(t *testing.T) { + store, err := NewStore(t.TempDir()) + if err != nil { + t.Fatal(err) + } + t.Cleanup(func() { _ = store.Close() }) + + signingPub, signingPriv, err := ed25519.GenerateKey(nil) + if err != nil { + t.Fatal(err) + } + otherPub, _, err := ed25519.GenerateKey(nil) + if err != nil { + t.Fatal(err) + } + + caller := peer.ID("caller-peer") + builder := biscuit.NewBuilder(signingPriv) + _ = builder.AddAuthorityFact(api.MarkerFact(api.FactTargetUnrestricted)) + _ = builder.AddAuthorityFact(biscuit.Fact{Predicate: biscuit.Predicate{ + Name: "node", + IDs: []biscuit.Term{biscuit.String(caller.String())}, + }}) + _ = builder.AddAuthorityFact(biscuit.Fact{Predicate: biscuit.Predicate{ + Name: api.FactClientPeerID, + IDs: []biscuit.Term{biscuit.String(caller.String())}, + }}) + _ = builder.AddAuthorityFact(biscuit.Fact{Predicate: biscuit.Predicate{ + Name: api.FactGrantedServiceExact, + IDs: []biscuit.Term{biscuit.String("mcp"), biscuit.String("calculator")}, + }}) + b, err := builder.Build() + if err != nil { + t.Fatal(err) + } + tokenBytes, err := b.Serialize() + if err != nil { + t.Fatal(err) + } + + // The token grants mcp://calculator only, so this request is denied by + // policy after its signature verified under signingPub. + req := RequestContext{ + PeerID: caller, + Protocol: "/mesh/mcp/1.0.0", + Target: "mcp://unauthorized-service", + } + + for _, tt := range []struct { + name string + keys []TrustedKey + }{ + {"signing key first", []TrustedKey{{Key: signingPub}, {Key: otherPub}}}, + {"signing key last", []TrustedKey{{Key: otherPub}, {Key: signingPub}}}, + } { + t.Run(tt.name, func(t *testing.T) { + node := &AgentMeshNode{ + Store: store, + trustedKeys: tt.keys, + BiscuitTimeout: 500 * time.Millisecond, + } + err := node.VerifyBiscuitToken(tokenBytes, req) + if err == nil { + t.Fatal("VerifyBiscuitToken allowed a target the token does not grant") + } + if errors.Is(err, biscuit.ErrInvalidSignature) { + t.Fatalf("VerifyBiscuitToken = %v, want the policy denial from the key that signed the token", err) + } + }) + } +} + // TestAuthorizeRejectsAppendedPeerBinding covers the dataplane half of the // binding bypass. Appending a block needs no root key, so a peer holding any // valid token could append node() and connect under its own