logging: the node's /debug/logs is always JSON, whatever --log-format says

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")

internal/logging/logging.go

@@ -16,20 +16,20 @@
 package logging
 
 import (
+	"bufio"
 	"container/ring"
-	"net/url"
+	"errors"
+	"io"
 	"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 +44,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 +64,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 +76,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 +96,44 @@ 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() {
+		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 {
 	if ringBuffer == nil {
 		return []string{}
@@ -133,9 +164,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()

internal/logging/logging_test.go

@@ -15,8 +15,15 @@
 package logging
 
 import (
+	"encoding/json"
+	"errors"
+	"fmt"
+	"io"
+	"os"
 	"strings"
 	"testing"
+	"testing/iotest"
+	"time"
 
 	golog "github.com/ipfs/go-log/v2"
 	"go.uber.org/zap/zapcore"
@@ -81,16 +88,135 @@ 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 n != 1 {
+		t.Errorf("ring holds %d copies of %q, want 1", n, msg)
+	}
+}
+
+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 !found {
-		t.Errorf("RecentLogs() = %v, want line containing %q", lines, "hello ring buffer")
+	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)
 	}
 }
 

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 {