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..aff6888e 100644 --- a/internal/logging/logging.go +++ b/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() diff --git a/internal/logging/logging_test.go b/internal/logging/logging_test.go index 51061a1e..bffd6d2c 100644 --- a/internal/logging/logging_test.go +++ b/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) } } 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 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.