From 2674ddb772c32ad237b85e574906e0db7e1dc521 Mon Sep 17 00:00:00 2001 From: Hein Meling Date: Sun, 11 Oct 2026 16:24:26 +0200 Subject: [PATCH 1/9] stream: check the dialed peer with an optional verifier An outbound channel trusted whichever server answered at the dialed address. Add PeerVerifier and OutboundOptions.Verify: when set, the channel checks the remote end of each new stream against its node ID, using the stream's context, before it sends requests on the stream. A failed check closes the stream and fails the waiting request with codes.Unauthenticated; the channel then retries as for any failed stream. --- internal/stream/channel_test.go | 102 ++++++++++++++++++++++++++++++-- internal/stream/errors.go | 7 +++ internal/stream/outbound.go | 28 ++++++++- 3 files changed, 131 insertions(+), 6 deletions(-) diff --git a/internal/stream/channel_test.go b/internal/stream/channel_test.go index e70f7bd8..b16c7115 100644 --- a/internal/stream/channel_test.go +++ b/internal/stream/channel_test.go @@ -17,9 +17,12 @@ import ( "github.com/relab/gorums/internal/testutils/mock" "google.golang.org/grpc" + "google.golang.org/grpc/codes" "google.golang.org/grpc/connectivity" "google.golang.org/grpc/credentials/insecure" "google.golang.org/grpc/keepalive" + "google.golang.org/grpc/peer" + "google.golang.org/grpc/status" ) const ( @@ -169,6 +172,13 @@ func setupChannel(t testing.TB, serverFn func(Gorums_NodeStreamServer) error, op // stream reconnection (see [NewOutboundChannel]). func setupChannelEager(t testing.TB, eagerReconnect bool, serverFn func(Gorums_NodeStreamServer) error, opts ...grpc.ServerOption) *testChannel { t.Helper() + return setupChannelOptions(t, OutboundOptions{EagerReconnect: eagerReconnect}, serverFn, opts...) +} + +// setupChannelOptions is [setupChannel] with the given channel options. A +// zero SendBufferSize becomes 10, and a nil Latency becomes a new estimate. +func setupChannelOptions(t testing.TB, chOpts OutboundOptions, serverFn func(Gorums_NodeStreamServer) error, opts ...grpc.ServerOption) *testChannel { + t.Helper() // Start listener lis, err := net.Listen("tcp", "127.0.0.1:0") @@ -194,11 +204,13 @@ func setupChannelEager(t testing.TB, eagerReconnect bool, serverFn func(Gorums_N t.Fatalf("failed to dial: %v", err) } - c := NewOutboundChannel(t.Context(), 1, conn, OutboundOptions{ - SendBufferSize: 10, - Latency: newLatency(), - EagerReconnect: eagerReconnect, - }) + if chOpts.SendBufferSize == 0 { + chOpts.SendBufferSize = 10 + } + if chOpts.Latency == nil { + chOpts.Latency = newLatency() + } + c := NewOutboundChannel(t.Context(), 1, conn, chOpts) tc := &testChannel{ OutboundChannel: c, srv: srv, @@ -457,6 +469,86 @@ func TestChannelErrors(t *testing.T) { } } +// TestChannelVerify verifies that a channel with a verifier checks the remote +// end of each new stream against the channel's ID, with the stream's +// context, before it sends requests on the stream. +func TestChannelVerify(t *testing.T) { + errRefused := errors.New("certificate names another node") + tests := []struct { + name string + noVerifier bool + verifyErr error + wantCode codes.Code + }{ + {name: "NoVerifier", noVerifier: true, wantCode: codes.OK}, + {name: "Accepted", wantCode: codes.OK}, + {name: "Refused", verifyErr: errRefused, wantCode: codes.Unauthenticated}, + } + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + var received atomic.Int32 + countingEcho := func(stream Gorums_NodeStreamServer) error { + for { + in, err := stream.Recv() + if err != nil { + return err + } + received.Add(1) + if err := stream.Send(in); err != nil { + return err + } + } + } + var mu sync.Mutex + var checkedIDs []ID + missingPeer := false + var opts OutboundOptions + if !tt.noVerifier { + opts.Verify = func(ctx context.Context, id ID) error { + _, ok := peer.FromContext(ctx) + mu.Lock() + defer mu.Unlock() + checkedIDs = append(checkedIDs, id) + missingPeer = missingPeer || !ok + return tt.verifyErr + } + } + tc := setupChannelOptions(t, opts, countingEcho) + + resp := sendRequest(t, tc.OutboundChannel, Request{}, 1) + if got := status.Code(resp.Err); got != tt.wantCode { + t.Fatalf("request error = %v; want code %v", resp.Err, tt.wantCode) + } + if tt.verifyErr != nil { + want := "gorums: node 1 failed verification: " + tt.verifyErr.Error() + if !strings.Contains(resp.Err.Error(), want) { + t.Errorf("request error = %q; want it to contain %q", resp.Err, want) + } + if got := received.Load(); got != 0 { + t.Errorf("server received %d messages on refused streams; want 0", got) + } + } + + mu.Lock() + defer mu.Unlock() + if tt.noVerifier { + return + } + if len(checkedIDs) == 0 { + t.Fatal("verifier was not called") + } + for _, id := range checkedIDs { + if id != 1 { + t.Errorf("verifier checked node %d; want 1, the channel's ID", id) + } + } + if missingPeer { + t.Error("verifier context carries no peer information; want the stream's context") + } + }) + } +} + // TestChannelStreamFailureRecordsLastErr verifies that a request the channel // cannot deliver because no stream could be established leaves the reason in // LastErr, even when the request has no response channel to report it on. diff --git a/internal/stream/errors.go b/internal/stream/errors.go index f868f1e9..0fd05c5e 100644 --- a/internal/stream/errors.go +++ b/internal/stream/errors.go @@ -26,3 +26,10 @@ var ( // stallReportDelay is how long a send must be blocked before [Channel.LastErr] // reports [ErrSendStalled]. const stallReportDelay = time.Second + +// VerificationError returns the error for a stream whose remote end failed a +// [PeerVerifier] check as node id. Its code is codes.Unauthenticated, and its +// message includes the verifier's error. +func VerificationError(id ID, err error) error { + return status.Errorf(codes.Unauthenticated, "gorums: node %d failed verification: %v", id, err) +} diff --git a/internal/stream/outbound.go b/internal/stream/outbound.go index 63aa404c..50751a76 100644 --- a/internal/stream/outbound.go +++ b/internal/stream/outbound.go @@ -18,6 +18,14 @@ const ( eagerReconnectMaxDelay = 2 * time.Second ) +// PeerVerifier reports whether the remote end of the stream with context ctx +// is the node with ID id. It returns nil to accept the stream, and an error +// that explains the refusal otherwise. ctx is the stream's context, which +// carries the connection's [google.golang.org/grpc/peer.Peer] and, on a +// server, the stream's incoming metadata. A PeerVerifier must be safe for +// concurrent use. +type PeerVerifier func(ctx context.Context, id ID) error + // OutboundOptions configures an [OutboundChannel]. type OutboundOptions struct { // SendBufferSize is the send queue capacity. @@ -33,6 +41,9 @@ type OutboundOptions struct { // time the stream comes up or goes down; [OutboundChannel.StreamUp] // reports the new state. OnStreamChange func() + // Verify, if non-nil, checks the remote end of each new stream against + // the channel's ID before the stream carries requests. + Verify PeerVerifier } // OutboundChannel is a [Channel] over a stream that this node opens to a @@ -49,11 +60,16 @@ type OutboundOptions struct { // the connection (a GOAWAY). The channel then sends new calls on a new // stream. Calls already sent on the old stream complete there, and the old // stream ends once it has no calls left. +// +// With [OutboundOptions.Verify], a stream whose remote end fails the check is +// closed before any request is sent on it, and the waiting request fails with +// the error from [VerificationError]. type OutboundChannel struct { endpoint conn *grpc.ClientConn eagerReconnect bool onStreamChange func() + verify PeerVerifier streamUp atomic.Bool mu sync.Mutex @@ -72,6 +88,7 @@ func NewOutboundChannel(ctx context.Context, id ID, conn *grpc.ClientConn, opts conn: conn, eagerReconnect: opts.EagerReconnect, onStreamChange: opts.OnStreamChange, + verify: opts.Verify, sessions: make(map[*session]struct{}), } c.closeOnce = sync.OnceValue(func() error { @@ -152,7 +169,8 @@ func (c *OutboundChannel) pace(delay *time.Duration) <-chan time.Time { return timer } -// open opens a stream and starts the session's receive loop and GOAWAY watcher. +// open opens a stream, checks its remote end if the channel has a verifier, +// and starts the session's receive loop and GOAWAY watcher. func (c *OutboundChannel) open() (*session, error) { ctx, cancel := context.WithCancel(c.ctx) st, err := NewGorumsClient(c.conn).NodeStream(ctx) @@ -160,6 +178,14 @@ func (c *OutboundChannel) open() (*session, error) { cancel() return nil, err } + if c.verify != nil { + // The stream's context carries the server's peer information, + // including its TLS state, once NodeStream returns. + if err := c.verify(st.Context(), c.id); err != nil { + cancel() + return nil, VerificationError(c.id, err) + } + } s := newSession(ctx, cancel, &c.endpoint, st, true, true) c.mu.Lock() c.sessions[s] = struct{}{} From 6200530d965804cd7a77962a21fabb13c6e26f26 Mon Sep 17 00:00:00 2001 From: Hein Meling Date: Sun, 11 Oct 2026 16:36:22 +0200 Subject: [PATCH 2/9] conn: verify peer claims and the peers a server dials A server attached an inbound stream to peer X whenever the stream's metadata claimed X, so any client could pose as a configured peer. NewInboundManager now takes an optional PeerVerifier. With it, AcceptPeer attaches a stream that claims a peer's node ID only after the verifier accepts the claim, and refuses claims for unknown node IDs, both with codes.Unauthenticated. The verifier runs without the manager's lock. The outbound manager hands the same verifier to the channels of the configured peers that the server dials. --- internal/conn/inbound_manager.go | 22 ++- internal/conn/inbound_manager_test.go | 186 ++++++++++++++++++++++--- internal/conn/node.go | 6 + internal/conn/node_test.go | 14 +- internal/conn/outbound_manager.go | 4 +- internal/conn/outbound_manager_test.go | 106 ++++++++++++++ server.go | 1 + 7 files changed, 308 insertions(+), 31 deletions(-) create mode 100644 internal/conn/outbound_manager_test.go diff --git a/internal/conn/inbound_manager.go b/internal/conn/inbound_manager.go index 0fdbd85f..56d24e85 100644 --- a/internal/conn/inbound_manager.go +++ b/internal/conn/inbound_manager.go @@ -85,6 +85,7 @@ type InboundManager struct { sendBufferSize uint // send buffer size for inbound channels dispatchSize uint // request dispatch queue size for inbound channels handler stream.RequestHandler // handler for dispatching incoming requests on all inbound nodes + verify PeerVerifier // optional; checks peer claims in AcceptPeer and the peers this server dials onConfigChange func(Config) // optional; called after each connected-peer config change nextClientID uint64 // next candidate ID for a client peer; uint64 represents exhaustion configCh chan struct{} // closed and replaced on each connectedPeers/connectedClients change; protected by mu @@ -109,7 +110,10 @@ const ClientIDStart = 1 << 20 // their send queue and request dispatch capacities; a dispatchSize of 0 // selects the default. It returns an error if peerNodes is invalid, for // example because of an invalid address or a duplicate node. -func NewInboundManager(myID ID, peerNodes NodeSource, sendBufferSize, dispatchSize uint, onConfigChange func(Config), handler stream.RequestHandler) (*InboundManager, error) { +// If verify is non-nil, [InboundManager.AcceptPeer] attaches a stream that +// claims a peer's node ID only after verify accepts the claim, and the +// server's outbound peer nodes check the peers they dial with verify. +func NewInboundManager(myID ID, peerNodes NodeSource, sendBufferSize, dispatchSize uint, onConfigChange func(Config), handler stream.RequestHandler, verify PeerVerifier) (*InboundManager, error) { im := &InboundManager{ myID: myID, knownNodes: make(map[ID]*Node), @@ -117,6 +121,7 @@ func NewInboundManager(myID ID, peerNodes NodeSource, sendBufferSize, dispatchSi sendBufferSize: sendBufferSize, dispatchSize: dispatchSize, handler: handler, + verify: verify, onConfigChange: onConfigChange, nextClientID: ClientIDStart, configCh: make(chan struct{}), @@ -249,7 +254,9 @@ func (im *InboundManager) isKnown(id ID) bool { // AcceptPeer accepts an inbound stream and returns its channel and a cleanup // function that closes the channel. It returns [errSelfNodeIDStream] for a -// stream that presents this server's own node ID. +// stream that presents this server's own node ID. With a peer verifier, it +// returns a codes.Unauthenticated error for a stream that claims an unknown +// node ID or fails the verifier's check. // // If the stream identifies a known peer, AcceptPeer attaches the channel to // that peer's node. If the stream includes peer metadata with node ID 0, @@ -270,6 +277,17 @@ func (im *InboundManager) AcceptPeer(streamCtx context.Context, inboundStream st // in the configuration. return nil, noop, errSelfNodeIDStream } + if id != 0 && im.verify != nil { + // A claim for a node outside the peer configuration has no identity + // to check. A known peer's claim is attached only after the verifier + // accepts it; the verifier runs without holding im.mu. + if !im.isKnown(id) { + return nil, noop, status.Errorf(codes.Unauthenticated, "gorums: stream claims unknown node %d", id) + } + if err := im.verify(streamCtx, id); err != nil { + return nil, noop, stream.VerificationError(id, err) + } + } if im.isKnown(id) { // Known peer — register on pre-created node. return im.registerPeer(streamCtx, inboundStream, id) diff --git a/internal/conn/inbound_manager_test.go b/internal/conn/inbound_manager_test.go index 3426693a..571044ca 100644 --- a/internal/conn/inbound_manager_test.go +++ b/internal/conn/inbound_manager_test.go @@ -2,6 +2,7 @@ package conn import ( "context" + "errors" "fmt" "io" "slices" @@ -27,7 +28,7 @@ func (f requestHandlerFunc) HandleRequest(ctx context.Context, _ ID, msg *stream // [NewInboundManager] rejects peerNodes. func newInboundManager(t *testing.T, myID uint32, peerNodes NodeSource, onConfigChange func(Config), handler stream.RequestHandler) *InboundManager { t.Helper() - im, err := NewInboundManager(myID, peerNodes, 0, 0, onConfigChange, handler) + im, err := NewInboundManager(myID, peerNodes, 0, 0, onConfigChange, handler, nil) if err != nil { t.Fatalf("NewInboundManager: %v", err) } @@ -109,7 +110,7 @@ func TestNewInboundManager(t *testing.T) { } for _, tc := range tests { t.Run(tc.name, func(t *testing.T) { - im, err := NewInboundManager(1, tc.opt, 0, 0, nil, nil) + im, err := NewInboundManager(1, tc.opt, 0, 0, nil, nil, nil) if tc.wantErr != "" { if err == nil || !strings.Contains(err.Error(), tc.wantErr) { t.Fatalf("NewInboundManager() error = %v; want it to contain %q", err, tc.wantErr) @@ -339,22 +340,22 @@ func TestInboundManagerAcceptPeerUpdatesConfig(t *testing.T) { } } -func TestInboundManagerAcceptPeer(t *testing.T) { - im := newTestInboundManager(t, 1) - - // tracked reports whether ch is the active channel of a node in im. - tracked := func(ch *stream.InboundChannel) bool { - im.mu.RLock() - defer im.mu.RUnlock() - for _, nodes := range []map[uint32]*Node{im.knownNodes, im.clientNodes} { - for _, n := range nodes { - if n.activeChannel() == stream.Channel(ch) { - return true - } +// isTracked reports whether ch is the active channel of a node in im. +func isTracked(im *InboundManager, ch *stream.InboundChannel) bool { + im.mu.RLock() + defer im.mu.RUnlock() + for _, nodes := range []map[ID]*Node{im.knownNodes, im.clientNodes} { + for _, n := range nodes { + if n.activeChannel() == stream.Channel(ch) { + return true } } - return false } + return false +} + +func TestInboundManagerAcceptPeer(t *testing.T) { + im := newTestInboundManager(t, 1) tests := []struct { name string @@ -403,7 +404,7 @@ func TestInboundManagerAcceptPeer(t *testing.T) { if err != nil { t.Fatalf("AcceptPeer() unexpected error: %v", err) } - if got := tracked(ch); got != tc.wantTracked { + if got := isTracked(im, ch); got != tc.wantTracked { t.Errorf("AcceptPeer() channel tracked = %t; want %t", got, tc.wantTracked) } cleanup() @@ -411,6 +412,157 @@ func TestInboundManagerAcceptPeer(t *testing.T) { } } +// newVerifyingInboundManager returns an InboundManager with node ID 1, known +// peers 1, 2, and 3, and the given peer verifier. +func newVerifyingInboundManager(t *testing.T, verify PeerVerifier) *InboundManager { + t.Helper() + im, err := NewInboundManager(1, WithNodes(map[ID]mock.NodeAddr{ + 1: "127.0.0.1:9081", + 2: "127.0.0.1:9082", + 3: "127.0.0.1:9083", + }), 0, 0, nil, nil, verify) + if err != nil { + t.Fatalf("NewInboundManager: %v", err) + } + return im +} + +// TestInboundManagerAcceptPeerVerifier verifies that, with a peer verifier, +// AcceptPeer attaches a stream that claims a peer's node ID only after the +// verifier accepts the claim, refuses claims for unknown node IDs without +// calling the verifier, and leaves streams that claim no peer node ID +// unchecked. TestInboundManagerAcceptPeer pins the behavior without a +// verifier. +func TestInboundManagerAcceptPeerVerifier(t *testing.T) { + errRefused := errors.New("certificate names another node") + tests := []struct { + name string + ctx context.Context + verifyErr error + wantCode codes.Code + wantTracked bool + wantChecked []ID + }{ + {name: "RegularClient", ctx: t.Context(), wantCode: codes.OK}, + {name: "BackChannelClient", ctx: inboundCtx(t.Context(), 0), wantCode: codes.OK, wantTracked: true}, + {name: "SelfNodeID", ctx: inboundCtx(t.Context(), 1), wantCode: codes.InvalidArgument}, + {name: "UnknownPeerID", ctx: inboundCtx(t.Context(), 99), wantCode: codes.Unauthenticated}, + {name: "KnownPeerRefused", ctx: inboundCtx(t.Context(), 2), verifyErr: errRefused, wantCode: codes.Unauthenticated, wantChecked: []ID{2}}, + {name: "KnownPeerAccepted", ctx: inboundCtx(t.Context(), 2), wantCode: codes.OK, wantTracked: true, wantChecked: []ID{2}}, + } + for _, tc := range tests { + t.Run(tc.name, func(t *testing.T) { + var checked []ID + im := newVerifyingInboundManager(t, func(_ context.Context, id ID) error { + checked = append(checked, id) + return tc.verifyErr + }) + inStream := mock.NewBidiStream[*stream.Message]() + defer inStream.Close() + + ch, cleanup, err := im.AcceptPeer(tc.ctx, inStream) + if got := status.Code(err); got != tc.wantCode { + t.Fatalf("AcceptPeer() error = %v; want code %v", err, tc.wantCode) + } + if tc.verifyErr != nil && !strings.Contains(err.Error(), "gorums: node 2 failed verification: "+tc.verifyErr.Error()) { + t.Errorf("AcceptPeer() error = %q; want it to name node 2 and the verifier's error", err) + } + if err == nil { + if got := isTracked(im, ch); got != tc.wantTracked { + t.Errorf("AcceptPeer() channel tracked = %t; want %t", got, tc.wantTracked) + } + cleanup() + } else if im.knownPeer(2).activeChannel() != nil { + t.Errorf("peer 2 has an active channel after a refused stream") + } + if !slices.Equal(checked, tc.wantChecked) { + t.Errorf("verifier checked %v; want %v", checked, tc.wantChecked) + } + }) + } +} + +// TestInboundManagerAcceptPeerVerifierChecksEveryStream verifies that every +// stream claiming a peer's node ID is checked, including a reconnect of a +// peer whose earlier stream was accepted, so a revoked certificate takes +// effect on the peer's next stream. +func TestInboundManagerAcceptPeerVerifierChecksEveryStream(t *testing.T) { + refuse := false + im := newVerifyingInboundManager(t, func(context.Context, ID) error { + if refuse { + return errors.New("certificate revoked") + } + return nil + }) + first := mock.NewBidiStream[*stream.Message]() + defer first.Close() + _, cleanup, err := im.AcceptPeer(inboundCtx(t.Context(), 2), first) + if err != nil { + t.Fatalf("AcceptPeer(first stream) error = %v; want nil", err) + } + defer cleanup() + + refuse = true + second := mock.NewBidiStream[*stream.Message]() + defer second.Close() + if _, _, err := im.AcceptPeer(inboundCtx(t.Context(), 2), second); status.Code(err) != codes.Unauthenticated { + t.Errorf("AcceptPeer(reconnect after revocation) error = %v; want code %v", err, codes.Unauthenticated) + } +} + +// TestInboundManagerAcceptPeerVerifierRunsWithoutLock verifies that a slow +// verifier for one peer does not stall the admission of another peer or +// reads of the manager's configurations. +func TestInboundManagerAcceptPeerVerifierRunsWithoutLock(t *testing.T) { + entered := make(chan struct{}) + release := make(chan struct{}) + im := newVerifyingInboundManager(t, func(_ context.Context, id ID) error { + if id == 2 { + close(entered) + <-release + } + return nil + }) + + slowDone := make(chan error, 1) + go func() { + slow := mock.NewBidiStream[*stream.Message]() + defer slow.Close() + _, cleanup, err := im.AcceptPeer(inboundCtx(t.Context(), 2), slow) + if err == nil { + cleanup() + } + slowDone <- err + }() + <-entered + + fastTracked := make(chan bool, 1) + go func() { + fast := mock.NewBidiStream[*stream.Message]() + defer fast.Close() + ch, cleanup, err := im.AcceptPeer(inboundCtx(t.Context(), 3), fast) + if err != nil { + fastTracked <- false + return + } + fastTracked <- isTracked(im, ch) && slices.Contains(im.InboundPeers().NodeIDs(), 3) + cleanup() + }() + select { + case ok := <-fastTracked: + if !ok { + t.Error("peer 3 was not attached while peer 2's check was in progress") + } + case <-time.After(3 * time.Second): + t.Fatal("peer 3's admission waited for peer 2's verifier") + } + + close(release) + if err := <-slowDone; err != nil { + t.Fatalf("AcceptPeer(peer 2) error = %v; want nil", err) + } +} + // TestInboundManagerAcceptPeerInstallsNewActiveOnReconnect verifies that calling AcceptPeer // for a peer that already has a live stream installs the new channel as the // node's active channel. The prior channel stays live until its own stream ends @@ -541,7 +693,7 @@ func TestInboundManagerAcceptPeerReplyRidesReceivingStream(t *testing.T) { im, err := NewInboundManager(1, WithNodes(map[uint32]mock.NodeAddr{ 1: "127.0.0.1:9081", 2: "127.0.0.1:9082", - }), 4, 0, nil, echo) + }), 4, 0, nil, echo, nil) if err != nil { t.Fatalf("NewInboundManager: %v", err) } diff --git a/internal/conn/node.go b/internal/conn/node.go index bd7a0622..c719c087 100644 --- a/internal/conn/node.go +++ b/internal/conn/node.go @@ -48,6 +48,10 @@ func sharedNodeTransport(peer *Node) *stream.Transport { // deduplication, the peer with the lower ID dials. type ID = stream.ID +// PeerVerifier reports whether the remote end of a stream is the node with a +// given ID; see [stream.PeerVerifier]. +type PeerVerifier = stream.PeerVerifier + // Node encapsulates the state of a node on which a remote procedure call // can be performed. type Node struct { @@ -109,6 +113,7 @@ type nodeOptions struct { Handler stream.RequestHandler EagerReconnect bool // re-establish a lost stream proactively; see [stream.OutboundOptions] OnStreamChange func() // optional; invoked on outbound stream transitions + Verify PeerVerifier // optional; checks the remote end of each dialed stream Manager *outboundManager // owning manager } @@ -135,6 +140,7 @@ func newOutboundNode(addr string, opts nodeOptions) (*Node, error) { Latency: transport.Latency(), EagerReconnect: opts.EagerReconnect, OnStreamChange: opts.OnStreamChange, + Verify: opts.Verify, })) return n, nil } diff --git a/internal/conn/node_test.go b/internal/conn/node_test.go index 1e79388b..c368e1d6 100644 --- a/internal/conn/node_test.go +++ b/internal/conn/node_test.go @@ -3,7 +3,6 @@ package conn import ( "context" "errors" - "net" "testing" "time" @@ -187,16 +186,9 @@ func BenchmarkNodeEnqueue(b *testing.B) { // // go test -run=^$ -bench='BenchmarkChannelSend$|BenchmarkNodeEnqueueSend' -benchmem -count=10 ./internal/stream . func BenchmarkNodeEnqueueSend(b *testing.B) { - lis, err := net.Listen("tcp", "127.0.0.1:0") - if err != nil { - b.Fatalf("failed to listen: %v", err) - } - grpcSrv := grpc.NewServer() // skipcq: GO-S0902 - stream.RegisterGorumsServer(grpcSrv, benchEchoServer{}) - go func() { _ = grpcSrv.Serve(lis) }() - b.Cleanup(grpcSrv.Stop) + addr := startEchoServer(b) - conn, err := grpc.NewClient(lis.Addr().String(), grpc.WithTransportCredentials(insecure.NewCredentials())) + conn, err := grpc.NewClient(addr, grpc.WithTransportCredentials(insecure.NewCredentials())) if err != nil { b.Fatalf("failed to dial: %v", err) } @@ -204,7 +196,7 @@ func BenchmarkNodeEnqueueSend(b *testing.B) { // Wrap the outbound channel in a Node, adding the transport lookup that // Node.Enqueue performs on every dispatch. - n := newInboundNode(1, lis.Addr().String(), func() uint64 { return 0 }) + n := newInboundNode(1, addr, func() uint64 { return 0 }) ch := stream.NewOutboundChannel(context.Background(), 1, conn, stream.OutboundOptions{SendBufferSize: 10, Latency: NodeTransport(n).Latency()}) b.Cleanup(func() { _ = ch.Close() }) NodeTransport(n).StoreChannel(ch) diff --git a/internal/conn/outbound_manager.go b/internal/conn/outbound_manager.go index d2184c24..c72cb843 100644 --- a/internal/conn/outbound_manager.go +++ b/internal/conn/outbound_manager.go @@ -167,8 +167,10 @@ func (m *outboundManager) newNode(id ID, addr string) (*Node, error) { } if im := m.opts.InboundManager; im != nil && im.isKnown(id) { // Stream-state changes on a dialed peer feed the server's - // connected-peer view. + // connected-peer view, and the server's peer verifier checks the + // peer at the other end of each stream. opts.OnStreamChange = im.onStreamChange + opts.Verify = im.verify } n, err := newOutboundNode(addr, opts) if err != nil { diff --git a/internal/conn/outbound_manager_test.go b/internal/conn/outbound_manager_test.go new file mode 100644 index 00000000..3e46b8ab --- /dev/null +++ b/internal/conn/outbound_manager_test.go @@ -0,0 +1,106 @@ +package conn + +import ( + "context" + "errors" + "net" + "slices" + "sync" + "testing" + "time" + + "github.com/relab/gorums/internal/stream" + "github.com/relab/gorums/internal/testutils/mock" + "google.golang.org/grpc" + "google.golang.org/grpc/codes" + "google.golang.org/grpc/credentials/insecure" + "google.golang.org/grpc/status" +) + +// startEchoServer starts a gRPC server whose node streams echo every message, +// and returns its address. The server stops when the test ends. +func startEchoServer(tb testing.TB) string { + tb.Helper() + lis, err := net.Listen("tcp", "127.0.0.1:0") + if err != nil { + tb.Fatalf("failed to listen: %v", err) + } + grpcSrv := grpc.NewServer() // skipcq: GO-S0902 + stream.RegisterGorumsServer(grpcSrv, benchEchoServer{}) + go func() { _ = grpcSrv.Serve(lis) }() + tb.Cleanup(grpcSrv.Stop) + return lis.Addr().String() +} + +// TestOutboundManagerPeerVerifier verifies that a configuration carrying a +// server checks each configured peer it dials with the server's peer +// verifier, and leaves other nodes unchecked. +func TestOutboundManagerPeerVerifier(t *testing.T) { + addr := startEchoServer(t) + errRefused := errors.New("certificate names another node") + tests := []struct { + name string + nodeID ID // 2 is a configured peer; 5 is not + noVerifier bool + verifyErr error + wantCode codes.Code + wantChecked []ID + }{ + {name: "PeerAccepted", nodeID: 2, wantCode: codes.OK, wantChecked: []ID{2}}, + {name: "PeerRefused", nodeID: 2, verifyErr: errRefused, wantCode: codes.Unauthenticated, wantChecked: []ID{2}}, + {name: "NotAPeer", nodeID: 5, wantCode: codes.OK}, + {name: "NoVerifier", nodeID: 2, noVerifier: true, wantCode: codes.OK}, + } + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + var mu sync.Mutex + var checked []ID + var verify PeerVerifier + if !tt.noVerifier { + verify = func(_ context.Context, id ID) error { + mu.Lock() + defer mu.Unlock() + checked = append(checked, id) + return tt.verifyErr + } + } + im, err := NewInboundManager(1, WithNodes(map[ID]mock.NodeAddr{ + 1: "127.0.0.1:9081", + 2: mock.NodeAddr(addr), + }), 0, 0, nil, nil, verify) + if err != nil { + t.Fatalf("NewInboundManager: %v", err) + } + m := newOutboundManager(func(o *DialOptions) { + o.InboundManager = im + o.LocalNodeID = 1 + o.GRPCDialOptions = []grpc.DialOption{grpc.WithTransportCredentials(insecure.NewCredentials())} + }) + t.Cleanup(m.Close) + n, err := m.newNode(tt.nodeID, addr) + if err != nil { + t.Fatalf("newNode(%d): %v", tt.nodeID, err) + } + + responseChan := make(chan stream.NodeResponse[*stream.Message], 1) + msg := stream.Message_builder{MessageSeqNo: 1, Method: mock.TestMethod}.Build() + NodeTransport(n).Enqueue(stream.Request{Ctx: t.Context(), Msg: msg, Oneway: true, ResponseChan: responseChan}) + select { + case r := <-responseChan: + if got := status.Code(r.Err); got != tt.wantCode { + t.Errorf("send error = %v; want code %v", r.Err, tt.wantCode) + } + case <-time.After(3 * time.Second): + t.Fatal("no send result within 3s") + } + + mu.Lock() + defer mu.Unlock() + // The channel checks each stream it opens, and it may open more + // than one, so compare the distinct IDs. + if got := slices.Compact(checked); !slices.Equal(got, tt.wantChecked) { + t.Errorf("verifier checked %v; want %v", got, tt.wantChecked) + } + }) + } +} diff --git a/server.go b/server.go index b292a7b1..ab976782 100644 --- a/server.go +++ b/server.go @@ -132,6 +132,7 @@ func newServer(opts ...ServerOption) (*Server, error) { serverOpts.recvBufferSize, serverOpts.onConfigChange, s, + nil, ) if err != nil { s.grpcServer.Stop() From 61cd27b6039bb3574981f457550d5b198e440315 Mon Sep 17 00:00:00 2001 From: Hein Meling Date: Sun, 11 Oct 2026 16:51:50 +0200 Subject: [PATCH 3/9] gorums: add WithPeerVerifier to check peer identities Servers trusted the node ID that a peer claimed when it connected and whoever answered at a peer's address, so ServerContext.SenderID was not authenticated. WithPeerVerifier installs a PeerVerifier that checks each configured peer on every stream between the two servers, in both directions, before the stream carries calls. NewServer panics for a nil verifier and for a verifier without WithPeers. The user guide gains an Authenticating Peers section. --- doc/dev-guide.md | 3 + doc/user-guide.md | 49 +++++++++++++- peer_verifier.go | 11 +++ server.go | 16 ++++- server_options.go | 29 ++++++++ server_test.go | 169 ++++++++++++++++++++++++++++++++++++++++++++++ 6 files changed, 272 insertions(+), 5 deletions(-) create mode 100644 peer_verifier.go diff --git a/doc/dev-guide.md b/doc/dev-guide.md index ce3e33f1..fd302f87 100644 --- a/doc/dev-guide.md +++ b/doc/dev-guide.md @@ -173,6 +173,7 @@ A `session` sends queued requests on its `BidiStream`, records two-way calls in The dispatcher runs handlers one at a time, in arrival order. A handler that returns without calling `Release` hands its goroutine to the next queued handler, so a busy stream keeps one goroutine and the stack it has grown; an early `Release` starts the next handler on a new goroutine. On the server side, `Server.NodeStream` asks a `PeerAcceptor` for the `InboundChannel` of each accepted stream. +An `OutboundChannel` with a `PeerVerifier` checks the remote end of each new stream against the channel's node ID, with the stream's context, before it sends requests on the stream; a stream that fails the check is closed, and the waiting request fails with `codes.Unauthenticated`. ```mermaid classDiagram @@ -251,6 +252,7 @@ The manager's `Close` closes all its nodes and logs any node's close error, sinc A closed manager also rejects new nodes under the same lock that guards its node list, so a node that races with `Close` cannot outlive it. A server's `InboundManager` owns the nodes for its known peers and for the peer-capable clients that connect to it, and maintains the connected-peer and connected-client configurations. It implements `stream.PeerAcceptor`, attaching each accepted stream to the matching node. +With a `PeerVerifier`, it attaches a stream that claims a peer's node ID only after the verifier accepts the claim, and refuses a claim for an unknown node ID; the outbound manager hands the same verifier to the channels of the configured peers the server dials. Both managers implement `nodeRegistry`, which a `NodeSource` uses to build a configuration. ```mermaid @@ -513,6 +515,7 @@ sequenceDiagram P->>SS: open NodeStream SS->>IM: AcceptPeer(ctx, stream) + IM->>IM: verify(ctx, claimed ID) if a PeerVerifier is set IM->>N: attachStream: NewInboundChannel N->>N: Transport.StoreChannel(ch) IM-->>SS: InboundChannel, cleanup diff --git a/doc/user-guide.md b/doc/user-guide.md index 28b521f4..801170ad 100644 --- a/doc/user-guide.md +++ b/doc/user-guide.md @@ -1694,8 +1694,10 @@ The sender is the last hop. When a message travels through several nodes, for example down a tree, `SenderID` identifies the node that forwarded it, not the node where the message started. Put the origin in the message if the handler needs it. -The sender asserts its ID when it connects, and the server does not authenticate it. -A Byzantine fault-tolerant protocol must still sign its messages and check the signer. +The sender asserts its ID when it connects. +A server configured with `WithPeerVerifier` checks the ID of every configured peer; see [Authenticating Peers](#authenticating-peers). +Otherwise, the server does not authenticate it. +Even with `WithPeerVerifier`, `SenderID` names only the last hop, so a Byzantine fault-tolerant protocol that forwards messages must still sign them and check the signer. ## Latency-Based Node Selection @@ -2373,6 +2375,49 @@ Calls already in flight while a stream is being replaced may still fail because Each `*Node` reports its stream topology: `Node.IsOutbound` is true for a node reached over a stream this process opened, `Node.IsInbound` is true for a node reached over a stream the peer opened, and `Node.IsShared` is true for a node that borrows the stream its lower-ID peer dialed, which holds on the higher-ID side of each deduplicated pair. These let a symmetric deployment inspect the shared-stream topology. +## Authenticating Peers + +Servers configured with `WithPeers` learn each other's node IDs from claims that nobody checks by default. +A server attaches an inbound stream to peer X when the stream announces node ID X. +A server that dials peer X trusts whoever answers at X's address. +So any client that can reach a server can claim to be one of its peers. +Under `WithStreamDedup`, a forged claim for a lower ID also redirects the server's own calls to that peer, because the server sends them over the peer's inbound stream. + +`WithPeerVerifier` makes a server check the identity of every configured peer, on every stream between them, in both directions: + +```go +srv := gorums.NewServer( + gorums.WithPeers(myID, gorums.WithNodeList(addrs), dialOpts...), + gorums.WithPeerVerifier(verify), +) +``` + +A `gorums.PeerVerifier` is a function `func(ctx context.Context, id gorums.ID) error`. +It receives the stream's context, which carries the connection's `peer.Peer`, and the node ID to check. +It returns nil to accept the stream, and an error that explains the refusal otherwise. +When the server refuses a stream that claims a peer's node ID, the refusal sent to the remote end includes the error's text. +When a stream that the server opened fails the check, the text is in the error of the waiting call. +The server calls it once per stream, outside its internal locks, so it must be safe for concurrent use. + +| Stream | Check | When the check fails | +| --- | --- | --- | +| Claims the node ID of a configured peer | `verify(ctx, claimed ID)` before the stream is attached to the peer | The stream ends with `codes.Unauthenticated`; no connect callback or handler runs | +| Claims a node ID outside the peer configuration | None; there is no identity to check | The stream ends with `codes.Unauthenticated` | +| Opened by this server to a configured peer | `verify(ctx, dialed ID)` before any request is sent | The waiting call fails with `codes.Unauthenticated`; the server tries again later | +| Claims no peer node ID (a client) | None | Served as without the option | + +With `WithPeerVerifier`, `ServerContext.SenderID` is authenticated for every configured peer, with and without stream deduplication. +`NewServer` panics if the verifier is nil or the server has no `WithPeers`. + +### Authenticating Clients + +`WithPeerVerifier` does not check clients; the transport and interceptors do: + +* To require a certificate from every client, set `ClientAuth: tls.RequireAndVerifyClientCert` in the server's `tls.Config`. + The handshake then fails before Gorums sees a stream. +* To decide which certificates may connect, use `tls.Config.VerifyConnection`, which runs on every handshake. +* To decide per request, for example to let only peers call a method, use a server interceptor that checks `ctx.SenderID()` or `peer.FromContext(ctx)`. + ## Send Queue Capacity and Backpressure The per-node send queue defaults to 4096 entries. diff --git a/peer_verifier.go b/peer_verifier.go new file mode 100644 index 00000000..0e35c48f --- /dev/null +++ b/peer_verifier.go @@ -0,0 +1,11 @@ +package gorums + +import "github.com/relab/gorums/internal/conn" + +// PeerVerifier reports whether the remote end of a stream is the node with a +// given ID. It returns nil to accept the stream, and an error that explains +// the refusal otherwise. Its context is the stream's context, which carries +// the connection's [google.golang.org/grpc/peer.Peer] and, on a server, the +// stream's incoming metadata. A PeerVerifier must be safe for concurrent use. +// See [WithPeerVerifier]. +type PeerVerifier = conn.PeerVerifier diff --git a/server.go b/server.go index ab976782..4326ea7e 100644 --- a/server.go +++ b/server.go @@ -2,6 +2,7 @@ package gorums import ( "context" + "errors" "fmt" "net" "slices" @@ -116,6 +117,14 @@ func newServer(opts ...ServerOption) (*Server, error) { if serverOpts.sendBufferSize == 0 { serverOpts.sendBufferSize = conn.DefaultSendBufferSize } + if serverOpts.hasPeerVerifier { + if serverOpts.peerVerifier == nil { + return nil, errors.New("gorums: WithPeerVerifier requires a non-nil verifier") + } + if serverOpts.peerNodes == nil { + return nil, errors.New("gorums: WithPeerVerifier requires WithPeers") + } + } // Allocate s first so it can serve as the self-node handler for the [conn.InboundManager]. // HandleRequest only accesses s.handlers and s.interceptors, both of which are // set below before NewInboundManager is called, so the reference is safe to pass. @@ -132,7 +141,7 @@ func newServer(opts ...ServerOption) (*Server, error) { serverOpts.recvBufferSize, serverOpts.onConfigChange, s, - nil, + serverOpts.peerVerifier, ) if err != nil { s.grpcServer.Stop() @@ -384,8 +393,9 @@ type ServerContext struct { // SenderID returns 0 for a client that announces no ID, or an ID that this // server does not know. The sender is the last hop: when a message travels // through several nodes, SenderID identifies the node that forwarded it. The -// sender asserts its ID when it connects, and the server does not -// authenticate it. +// sender asserts its ID when it connects. A server configured with +// [WithPeerVerifier] checks the ID of every configured peer, on the streams in +// both directions; otherwise the server does not authenticate it. func (ctx *ServerContext) SenderID() ID { return ctx.senderID } diff --git a/server_options.go b/server_options.go index af47625e..c9ea7811 100644 --- a/server_options.go +++ b/server_options.go @@ -20,6 +20,8 @@ type serverOptions struct { onConfigChange func(Config) // Callback registered via [WithPeerChange]; invoked after each connected-peer config change. listenAddr string // Listener address recorded by WithAddr; bound by ListenAndServe. outboundDialOpts []DialOption + peerVerifier PeerVerifier // Checks peer identities on peer streams; set by WithPeerVerifier. + hasPeerVerifier bool // Records that WithPeerVerifier was applied, so a nil verifier is reported. } // ServerOption configures a [Server]. @@ -132,3 +134,30 @@ func WithStreamDedup() ServerOption { o.outboundDialOpts = append(o.outboundDialOpts, conn.WithStreamDedup()) } } + +// WithPeerVerifier makes the server check, with verify, the identity of the +// peers configured by [WithPeers] on every stream between them, in both +// directions: +// - An inbound stream that claims a peer's node ID is checked against that +// ID. A stream that fails the check, or that claims a node ID outside the +// peer configuration, ends with codes.Unauthenticated before any connect +// callback or handler runs. Streams that claim no peer node ID are not +// checked. +// - A stream that the server opens to a peer is checked against the dialed +// node's ID before any request is sent on it. If the check fails, the +// stream is closed, the waiting call fails with codes.Unauthenticated, +// and the server tries again later, as for a failed connection. +// +// When the server refuses a stream that a peer opened, the refusal's message +// includes the text of verify's error, so the peer learns why. When a stream +// that the server opened fails the check, the text is in the error of the +// waiting call. The server runs verify once per stream, outside its internal +// locks. With WithPeerVerifier, [ServerContext.SenderID] is authenticated for +// every configured peer. [NewServer] panics if verify is nil or if the server +// has no [WithPeers]. +func WithPeerVerifier(verify PeerVerifier) ServerOption { + return func(o *serverOptions) { + o.peerVerifier = verify + o.hasPeerVerifier = true + } +} diff --git a/server_test.go b/server_test.go index cbeafdbe..1793fbdb 100644 --- a/server_test.go +++ b/server_test.go @@ -4,6 +4,7 @@ import ( "context" "errors" "fmt" + "maps" "net" "runtime" "strings" @@ -18,7 +19,9 @@ import ( "github.com/relab/gorums/internal/testutils/mock" gorumsimpl "github.com/relab/gorums/runtime/gorumsimpl" "go.uber.org/goleak" + "google.golang.org/grpc/codes" "google.golang.org/grpc/metadata" + "google.golang.org/grpc/status" "google.golang.org/protobuf/proto" pb "google.golang.org/protobuf/types/known/wrapperspb" ) @@ -1300,3 +1303,169 @@ func TestServerSenderID(t *testing.T) { } }) } + +// TestServerPeerVerifier verifies that WithPeerVerifier checks every +// configured peer on the streams the peer opens to the server and on the +// streams the server opens to the peer, with and without stream +// deduplication. +func TestServerPeerVerifier(t *testing.T) { + const n = 3 + all := func(_, _ gorums.ID) bool { return true } + tests := []struct { + name string + opts []gorumstest.Option + wantInbound func(self, peer gorums.ID) bool // peer opens a stream to self + wantDialed func(self, peer gorums.ID) bool // self opens a stream to peer + }{ + {name: "Dual", wantInbound: all, wantDialed: all}, + { + name: "Dedup", + opts: []gorumstest.Option{gorums.WithStreamDedup()}, + wantInbound: func(self, peer gorums.ID) bool { return peer < self }, + wantDialed: func(self, peer gorums.ID) bool { return peer > self }, + }, + } + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + var mu sync.Mutex + // checked[self] holds "inbound N" or "dialed N" for each peer N + // that self's verifier checked. A server's stream context carries + // incoming metadata; the context of a stream it dialed does not. + checked := make(map[gorums.ID]map[string]bool) + verifierFor := func(self gorums.ID) []gorums.ServerOption { + mu.Lock() + checked[self] = make(map[string]bool) + mu.Unlock() + return []gorums.ServerOption{gorums.WithPeerVerifier(func(ctx context.Context, peer gorums.ID) error { + direction := "dialed" + if _, ok := metadata.FromIncomingContext(ctx); ok { + direction = "inbound" + } + mu.Lock() + defer mu.Unlock() + checked[self][fmt.Sprintf("%s %d", direction, peer)] = true + return nil + })} + } + opts := append([]gorumstest.Option{gorums.WithLocalServerOptionsFunc(verifierFor)}, tt.opts...) + servers := gorumstest.LocalServers(t, n, opts...) + gorumstest.WaitForPeers(t, servers) + + want := make(map[gorums.ID]map[string]bool) + for self := gorums.ID(1); self <= n; self++ { + want[self] = make(map[string]bool) + for peer := gorums.ID(1); peer <= n; peer++ { + if peer == self { + continue + } + if tt.wantInbound(self, peer) { + want[self][fmt.Sprintf("inbound %d", peer)] = true + } + if tt.wantDialed(self, peer) { + want[self][fmt.Sprintf("dialed %d", peer)] = true + } + } + } + match := func() bool { + mu.Lock() + defer mu.Unlock() + return maps.EqualFunc(checked, want, func(a, b map[string]bool) bool { return maps.Equal(a, b) }) + } + if !gorumstest.WaitUntil(t, 5*time.Second, match) { + mu.Lock() + defer mu.Unlock() + t.Errorf("verifier checks = %v; want %v", checked, want) + } + }) + } +} + +// TestServerPeerVerifierRefusals verifies that a server with a peer verifier +// fails its calls to a peer that does not pass the check, refuses a client +// that claims a node ID outside the peer configuration, and serves a client +// that claims no node ID. +func TestServerPeerVerifierRefusals(t *testing.T) { + refuseAll := gorums.WithPeerVerifier(func(context.Context, gorums.ID) error { return errors.New("not trusted") }) + acceptAll := gorums.WithPeerVerifier(func(context.Context, gorums.ID) error { return nil }) + start := func(t *testing.T, opt gorums.ServerOption) []*gorums.Server { + t.Helper() + servers := gorumstest.LocalServers(t, 2, opt) + for _, srv := range servers { + srv.RegisterHandler(mock.TestMethod, gorumstest.EchoHandler("echo")) + } + return servers + } + call := func(t *testing.T, node *gorums.Node, timeout time.Duration) error { + t.Helper() + ctx := gorumstest.Context(t, timeout) + _, err := gorumsimpl.RemoteCall[*pb.StringValue, *pb.StringValue](node.Context(ctx), pb.String("hello"), mock.TestMethod) + return err + } + client := func(t *testing.T, srv *gorums.Server, opts ...gorums.DialOption) *gorums.Node { + t.Helper() + cfg, closeFn, err := gorums.NewConfig(gorums.WithNodeList([]string{srv.Addr()}), + append([]gorums.DialOption{gorumstest.DialOptions(t)}, opts...)...) + if err != nil { + t.Fatalf("NewConfig: %v", err) + } + t.Cleanup(closeFn) + return cfg[0] + } + + t.Run("PeerRefused", func(t *testing.T) { + servers := start(t, refuseAll) + node := servers[0].PeerConfig().Node(2) + var err error + // Until the connection is ready, a call can fail for that reason + // instead; the verifier's refusal follows once it is ready. + if !gorumstest.WaitUntil(t, 5*time.Second, func() bool { + err = call(t, node, time.Second) + return status.Code(err) == codes.Unauthenticated + }) { + t.Fatalf("call to a refused peer: err = %v; want code %v", err, codes.Unauthenticated) + } + if want := "gorums: node 2 failed verification: not trusted"; !strings.Contains(err.Error(), want) { + t.Errorf("call to a refused peer: err = %q; want it to contain %q", err, want) + } + }) + t.Run("UnknownPeerIDRefused", func(t *testing.T) { + servers := start(t, acceptAll) + node := client(t, servers[0], gorums.WithMetadata(metadata.Pairs("gorums-node-id", "99"))) + if err := call(t, node, 500*time.Millisecond); err == nil { + t.Fatal("call that claims an unknown node ID succeeded; want failure") + } + // The server refuses the stream, which ends with its status; the + // client's node records it as its last error. + if !gorumstest.WaitUntil(t, 3*time.Second, func() bool { + return status.Code(node.LastErr()) == codes.Unauthenticated + }) { + t.Errorf("client node LastErr = %v; want code %v", node.LastErr(), codes.Unauthenticated) + } + }) + t.Run("ClientServed", func(t *testing.T) { + servers := start(t, acceptAll) + if err := call(t, client(t, servers[0]), 3*time.Second); err != nil { + t.Errorf("call from a client that claims no node ID: %v", err) + } + }) +} + +// TestServerPeerVerifierPanics verifies that NewServer panics for a nil peer +// verifier and for a peer verifier without WithPeers. +func TestServerPeerVerifierPanics(t *testing.T) { + peers := gorums.WithPeers(1, gorums.WithNodeList([]string{"127.0.0.1:1", "127.0.0.1:2"}), gorumstest.InsecureDialOptions(t)) + acceptAll := gorums.WithPeerVerifier(func(context.Context, gorums.ID) error { return nil }) + tests := []struct { + name string + opts []gorums.ServerOption + want string + }{ + {name: "NilVerifier", opts: []gorums.ServerOption{peers, gorums.WithPeerVerifier(nil)}, want: "gorums: invalid peer configuration: gorums: WithPeerVerifier requires a non-nil verifier"}, + {name: "NoPeers", opts: []gorums.ServerOption{acceptAll}, want: "gorums: invalid peer configuration: gorums: WithPeerVerifier requires WithPeers"}, + } + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + assertPanic(t, tt.want, func() { gorums.NewServer(tt.opts...) }) + }) + } +} From fa78161358cc1dd0f6f09e7c6ae66b68e4cbe748 Mon Sep 17 00:00:00 2001 From: Hein Meling Date: Sun, 11 Oct 2026 17:11:09 +0200 Subject: [PATCH 4/9] gorums: add TLSPeerVerifier to bind node IDs to certificates Add a PeerVerifier that binds each node ID to a SAN of the peer's verified mTLS certificate. It reads the certificate only from the verified chains, so a certificate that no trusted CA signed never passes. An identity with a URI scheme matches only a URI SAN with the same canonical form; any other identity matches only a DNS SAN, without regard to case. Wildcard DNS SANs, IP SANs, the Common Name, and empty identities never match. The user guide shows the certificate and TLS configuration that replicas need. --- doc/user-guide.md | 49 ++++++++++++++++++- peer_verifier.go | 89 +++++++++++++++++++++++++++++++++- peer_verifier_test.go | 109 ++++++++++++++++++++++++++++++++++++++++++ server_options.go | 3 +- 4 files changed, 247 insertions(+), 3 deletions(-) create mode 100644 peer_verifier_test.go diff --git a/doc/user-guide.md b/doc/user-guide.md index 801170ad..412ab7e8 100644 --- a/doc/user-guide.md +++ b/doc/user-guide.md @@ -2397,7 +2397,8 @@ It receives the stream's context, which carries the connection's `peer.Peer`, an It returns nil to accept the stream, and an error that explains the refusal otherwise. When the server refuses a stream that claims a peer's node ID, the refusal sent to the remote end includes the error's text. When a stream that the server opened fails the check, the text is in the error of the waiting call. -The server calls it once per stream, outside its internal locks, so it must be safe for concurrent use. +The server calls the verifier once per stream, outside its internal locks, so it must be safe for concurrent use. +Use `TLSPeerVerifier` to bind node IDs to mTLS certificates; see [Binding Node IDs to TLS Certificates](#binding-node-ids-to-tls-certificates). | Stream | Check | When the check fails | | --- | --- | --- | @@ -2409,6 +2410,52 @@ The server calls it once per stream, outside its internal locks, so it must be s With `WithPeerVerifier`, `ServerContext.SenderID` is authenticated for every configured peer, with and without stream deduplication. `NewServer` panics if the verifier is nil or the server has no `WithPeers`. +### Binding Node IDs to TLS Certificates + +`TLSPeerVerifier` binds each node ID to a name in the peer's mTLS certificate: + +```go +verify := gorums.TLSPeerVerifier(map[gorums.ID]string{ + 1: "spiffe://relab/node/1", + 2: "spiffe://relab/node/2", + 3: "spiffe://relab/node/3", +}) +``` + +It accepts a stream for a node ID only if the TLS handshake verified the remote end's certificate against a trusted CA, and the certificate has a SAN that matches the map's entry for that ID. +An identity with a URI scheme, such as `spiffe://relab/node/2`, matches only a URI SAN with the same canonical form. +Any other identity matches only a DNS SAN, without regard to case. +A wildcard DNS SAN never matches, IP SANs and the certificate's Common Name are never used, and a node ID with an empty or missing identity is refused. + +Each replica needs its own certificate with: + +* a URI SAN or DNS SAN that names the replica, such as `spiffe://relab/node/2`; +* an IP SAN or DNS SAN for the address its peers dial, which standard TLS checks; +* extended key usage for both server and client authentication, because the same certificate serves both directions. + +The server must verify client certificates, and its peer connections must present its certificate: + +```go +srv := gorums.NewServer( + gorums.WithGRPCServerOptions(grpc.Creds(credentials.NewTLS(&tls.Config{ + Certificates: []tls.Certificate{cert}, + ClientCAs: caPool, + ClientAuth: tls.VerifyClientCertIfGiven, // clients without a certificate are still served + }))), + gorums.WithPeers(2, gorums.WithNodeList(addrs), + gorums.WithGRPCDialOptions(grpc.WithTransportCredentials(credentials.NewTLS(&tls.Config{ + Certificates: []tls.Certificate{cert}, // presented to the peers this server dials + RootCAs: caPool, + }))), + ), + gorums.WithPeerVerifier(verify), +) +``` + +With `ClientAuth: tls.RequireAndVerifyClientCert`, clients without a certificate are refused as well. + +For a local cluster, `NewLocalServers` takes options that differ per server: `WithLocalServerOptionsFunc` and `WithLocalDialOptionsFunc` return the options for a given node ID, such as that server's certificate. + ### Authenticating Clients `WithPeerVerifier` does not check clients; the transport and interceptors do: diff --git a/peer_verifier.go b/peer_verifier.go index 0e35c48f..804bec2e 100644 --- a/peer_verifier.go +++ b/peer_verifier.go @@ -1,6 +1,16 @@ package gorums -import "github.com/relab/gorums/internal/conn" +import ( + "context" + "errors" + "fmt" + "net/url" + "strings" + + "github.com/relab/gorums/internal/conn" + "google.golang.org/grpc/credentials" + "google.golang.org/grpc/peer" +) // PeerVerifier reports whether the remote end of a stream is the node with a // given ID. It returns nil to accept the stream, and an error that explains @@ -9,3 +19,80 @@ import "github.com/relab/gorums/internal/conn" // stream's incoming metadata. A PeerVerifier must be safe for concurrent use. // See [WithPeerVerifier]. type PeerVerifier = conn.PeerVerifier + +// peerIdentity is the name that a node's certificate must carry: a URI SAN +// if uri is set, and a DNS SAN otherwise. +type peerIdentity struct { + name string + uri bool +} + +// kind returns the type of SAN that the identity names. +func (p peerIdentity) kind() string { + if p.uri { + return "URI" + } + return "DNS" +} + +// TLSPeerVerifier returns a [PeerVerifier] that accepts a stream for node ID +// id only if the TLS handshake verified the remote end's certificate against +// a trusted CA, and that certificate has a SAN that matches identities[id]. +// An identity with a URI scheme, such as spiffe://relab/node/2, matches only +// a URI SAN with the same canonical form. Any other identity matches only a +// DNS SAN, without regard to case. A wildcard DNS SAN never matches, IP SANs +// and the Common Name are never used, and a node ID with an empty or missing +// identity is refused. +// +// The server's TLS configuration must verify client certificates, with +// ClientAuth set to tls.VerifyClientCertIfGiven or +// tls.RequireAndVerifyClientCert, and each replica's certificate must be +// valid for both client and server authentication. TLSPeerVerifier copies +// identities. +func TLSPeerVerifier(identities map[ID]string) PeerVerifier { + want := make(map[ID]peerIdentity, len(identities)) + for id, name := range identities { + if name == "" { + continue // An empty identity is treated as missing. + } + if u, err := url.Parse(name); err == nil && u.Scheme != "" { + want[id] = peerIdentity{name: u.String(), uri: true} + continue + } + want[id] = peerIdentity{name: name} + } + return func(ctx context.Context, id ID) error { + p, ok := peer.FromContext(ctx) + if !ok { + return errors.New("no peer information in the stream context") + } + tlsInfo, ok := p.AuthInfo.(credentials.TLSInfo) + if !ok { + return errors.New("connection does not use TLS") + } + chains := tlsInfo.State.VerifiedChains + if len(chains) == 0 || len(chains[0]) == 0 { + return errors.New("no verified certificate") + } + identity, ok := want[id] + if !ok { + return fmt.Errorf("no identity configured for node %d", id) + } + // Every verified chain starts with the certificate the remote end sent. + cert := chains[0][0] + if identity.uri { + for _, u := range cert.URIs { + if u.String() == identity.name { + return nil + } + } + } else { + for _, name := range cert.DNSNames { + if !strings.HasPrefix(name, "*") && strings.EqualFold(name, identity.name) { + return nil + } + } + } + return fmt.Errorf("certificate has no %s SAN equal to %q", identity.kind(), identity.name) + } +} diff --git a/peer_verifier_test.go b/peer_verifier_test.go new file mode 100644 index 00000000..8f5681ac --- /dev/null +++ b/peer_verifier_test.go @@ -0,0 +1,109 @@ +package gorums_test + +import ( + "context" + "crypto/tls" + "crypto/x509" + "crypto/x509/pkix" + "net" + "net/url" + "strings" + "testing" + + "github.com/relab/gorums" + "google.golang.org/grpc/credentials" + "google.golang.org/grpc/peer" +) + +// tlsPeerContext returns a context whose peer presented cert over TLS. If +// verified is true, cert also starts a verified chain, as after a handshake +// that checked it against a trusted CA. +func tlsPeerContext(cert *x509.Certificate, verified bool) context.Context { + state := tls.ConnectionState{PeerCertificates: []*x509.Certificate{cert}} + if verified { + state.VerifiedChains = [][]*x509.Certificate{{cert}} + } + return peer.NewContext(context.Background(), &peer.Peer{AuthInfo: credentials.TLSInfo{State: state}}) +} + +// otherAuthInfo is the AuthInfo of a connection that does not use TLS. +type otherAuthInfo struct{} + +func (otherAuthInfo) AuthType() string { return "other" } + +func mustParseURL(t *testing.T, s string) *url.URL { + t.Helper() + u, err := url.Parse(s) + if err != nil { + t.Fatalf("url.Parse(%q): %v", s, err) + } + return u +} + +// TestPeerVerifierTLS verifies that TLSPeerVerifier accepts a stream only if +// the verified certificate has a SAN of the type that the node's configured +// identity names, equal to that identity, and that it ignores wildcard DNS +// SANs, IP SANs, the Common Name, and unverified certificates. +func TestPeerVerifierTLS(t *testing.T) { + node2 := mustParseURL(t, "spiffe://relab/node/2") + node3 := mustParseURL(t, "spiffe://relab/node/3") + node4 := mustParseURL(t, "node4.relab.example") + node9 := mustParseURL(t, "spiffe://relab/node/A") + node10 := mustParseURL(t, "spiffe://relab/node/10") + verify := gorums.TLSPeerVerifier(map[gorums.ID]string{ + 2: "spiffe://relab/node/2", + 4: "node4.relab.example", + 5: "*.relab.example", + 6: "10.0.0.6", + 7: "node7.relab.example", + 8: "", + 9: "spiffe://relab/node/a", + 10: "SPIFFE://relab/node/10", + }) + tests := []struct { + name string + ctx context.Context + id gorums.ID + wantErr string // empty means the stream is accepted + }{ + {name: "URIMatch", ctx: tlsPeerContext(&x509.Certificate{URIs: []*url.URL{node2}}, true), id: 2}, + {name: "OtherNodesCertificate", ctx: tlsPeerContext(&x509.Certificate{URIs: []*url.URL{node3}}, true), id: 2, wantErr: `no URI SAN equal to "spiffe://relab/node/2"`}, + {name: "DNSMatchIgnoresCase", ctx: tlsPeerContext(&x509.Certificate{DNSNames: []string{"Node4.Relab.Example"}}, true), id: 4}, + {name: "WildcardDNSNeverMatches", ctx: tlsPeerContext(&x509.Certificate{DNSNames: []string{"*.relab.example"}}, true), id: 5, wantErr: `no DNS SAN equal to "*.relab.example"`}, + {name: "IPSANIgnored", ctx: tlsPeerContext(&x509.Certificate{IPAddresses: []net.IP{net.ParseIP("10.0.0.6")}}, true), id: 6, wantErr: `no DNS SAN equal to "10.0.0.6"`}, + {name: "CommonNameIgnored", ctx: tlsPeerContext(&x509.Certificate{Subject: pkix.Name{CommonName: "node7.relab.example"}}, true), id: 7, wantErr: `no DNS SAN equal to "node7.relab.example"`}, + {name: "UnverifiedCertificate", ctx: tlsPeerContext(&x509.Certificate{URIs: []*url.URL{node2}}, false), id: 2, wantErr: "no verified certificate"}, + {name: "NotTLS", ctx: peer.NewContext(context.Background(), &peer.Peer{AuthInfo: otherAuthInfo{}}), id: 2, wantErr: "connection does not use TLS"}, + {name: "NoPeer", ctx: context.Background(), id: 2, wantErr: "no peer information in the stream context"}, + {name: "NoIdentity", ctx: tlsPeerContext(&x509.Certificate{URIs: []*url.URL{node3}}, true), id: 3, wantErr: "no identity configured for node 3"}, + {name: "EmptyIdentity", ctx: tlsPeerContext(&x509.Certificate{DNSNames: []string{""}}, true), id: 8, wantErr: "no identity configured for node 8"}, + {name: "DNSSANDoesNotMatchURIIdentity", ctx: tlsPeerContext(&x509.Certificate{DNSNames: []string{"SPIFFE://RELAB/NODE/2"}}, true), id: 2, wantErr: `no URI SAN equal to "spiffe://relab/node/2"`}, + {name: "URISANDoesNotMatchDNSIdentity", ctx: tlsPeerContext(&x509.Certificate{URIs: []*url.URL{node4}}, true), id: 4, wantErr: `no DNS SAN equal to "node4.relab.example"`}, + {name: "URIMatchIsCaseSensitive", ctx: tlsPeerContext(&x509.Certificate{URIs: []*url.URL{node9}}, true), id: 9, wantErr: `no URI SAN equal to "spiffe://relab/node/a"`}, + {name: "LaterSANMatches", ctx: tlsPeerContext(&x509.Certificate{URIs: []*url.URL{node3, node2}}, true), id: 2}, + {name: "URIIdentityCanonicalized", ctx: tlsPeerContext(&x509.Certificate{URIs: []*url.URL{node10}}, true), id: 10}, + } + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + err := verify(tt.ctx, tt.id) + if tt.wantErr == "" { + if err != nil { + t.Errorf("verify(node %d) = %v; want nil", tt.id, err) + } + return + } + if err == nil || !strings.Contains(err.Error(), tt.wantErr) { + t.Errorf("verify(node %d) = %v; want an error containing %q", tt.id, err, tt.wantErr) + } + }) + } + + t.Run("MapChangedAfterConstruction", func(t *testing.T) { + identities := map[gorums.ID]string{2: "spiffe://relab/node/2"} + verify := gorums.TLSPeerVerifier(identities) + identities[2] = "spiffe://relab/node/3" + if err := verify(tlsPeerContext(&x509.Certificate{URIs: []*url.URL{node2}}, true), 2); err != nil { + t.Errorf("verify(node 2) after the caller changed its map = %v; want nil", err) + } + }) +} diff --git a/server_options.go b/server_options.go index c9ea7811..267d2127 100644 --- a/server_options.go +++ b/server_options.go @@ -154,7 +154,8 @@ func WithStreamDedup() ServerOption { // waiting call. The server runs verify once per stream, outside its internal // locks. With WithPeerVerifier, [ServerContext.SenderID] is authenticated for // every configured peer. [NewServer] panics if verify is nil or if the server -// has no [WithPeers]. +// has no [WithPeers]. Use [TLSPeerVerifier] to bind node IDs to mTLS +// certificates. func WithPeerVerifier(verify PeerVerifier) ServerOption { return func(o *serverOptions) { o.peerVerifier = verify From 203d3713627a6292eb2defe7bc56c2ee4dd43000 Mon Sep 17 00:00:00 2001 From: Hein Meling Date: Sun, 11 Oct 2026 17:15:55 +0200 Subject: [PATCH 5/9] internal/tests/tls: test per-replica peer certificates Run peer groups whose servers check each other with TLSPeerVerifier and certificates from a test CA, with and without stream deduplication. The tests cover peers that connect, a server that presents another node's certificate (refused by the dial-side and the inbound check), an impostor that claims a lower peer's ID and must not take over its stream, and a client without a certificate that is still served. --- internal/tests/tls/peer_identity_test.go | 219 +++++++++++++++++++++++ internal/tests/tls/testca_test.go | 130 ++++++++++++++ 2 files changed, 349 insertions(+) create mode 100644 internal/tests/tls/peer_identity_test.go create mode 100644 internal/tests/tls/testca_test.go diff --git a/internal/tests/tls/peer_identity_test.go b/internal/tests/tls/peer_identity_test.go new file mode 100644 index 00000000..a6dbfe0f --- /dev/null +++ b/internal/tests/tls/peer_identity_test.go @@ -0,0 +1,219 @@ +package tls + +import ( + "crypto/tls" + "fmt" + "strings" + "sync/atomic" + "testing" + "time" + + "github.com/relab/gorums" + "github.com/relab/gorums/gorumstest" + "github.com/relab/gorums/internal/testutils/mock" + gorumsimpl "github.com/relab/gorums/runtime/gorumsimpl" + "google.golang.org/grpc" + "google.golang.org/grpc/codes" + "google.golang.org/grpc/metadata" + "google.golang.org/grpc/status" + pb "google.golang.org/protobuf/types/known/wrapperspb" +) + +// peerGroup starts n servers that check each other's identities with +// per-replica certificates from ca. Server id presents the certificate of +// node certIDs[id-1], or its own if certIDs is nil, both to clients that dial +// it and to the peers it dials. Each server answers a mock.TestMethod +// request with "server from : ", so a reply also +// shows the [gorums.ServerContext.SenderID] the handler saw, and counts the +// requests in calls[id-1]. +func peerGroup(t *testing.T, ca *testCA, n int, certIDs []gorums.ID, opts ...gorumstest.Option) (servers []*gorums.Server, calls []*atomic.Int32) { + t.Helper() + certs := make(map[gorums.ID]tls.Certificate, n) + for id := gorums.ID(1); id <= gorums.ID(n); id++ { + certID := id + if certIDs != nil { + certID = certIDs[id-1] + } + certs[id] = ca.issue(t, certID) + } + verify := peerVerifier(n) + opts = append([]gorumstest.Option{ + gorums.WithLocalServerOptionsFunc(func(id gorums.ID) []gorums.ServerOption { + return []gorums.ServerOption{ + gorums.WithGRPCServerOptions(grpc.Creds(ca.serverCreds(certs[id]))), + gorums.WithPeerVerifier(verify), + } + }), + gorums.WithLocalDialOptionsFunc(func(id gorums.ID) []gorums.DialOption { + return []gorums.DialOption{gorums.WithGRPCDialOptions(grpc.WithTransportCredentials(ca.dialCreds(certs[id])))} + }), + }, opts...) + servers = gorumstest.LocalServers(t, n, opts...) + for _, srv := range servers { + count := new(atomic.Int32) + calls = append(calls, count) + self := srv.NodeID() + srv.RegisterHandler(mock.TestMethod, func(ctx gorums.ServerContext, in *gorums.Message) (*gorums.Message, error) { + count.Add(1) + req := gorums.AsProto[*pb.StringValue](in) + reply := fmt.Sprintf("server %d from %d: %s", self, ctx.SenderID(), req.GetValue()) + return gorums.NewResponseMessage(in, pb.String(reply)), nil + }) + } + return servers, calls +} + +// remoteCall calls mock.TestMethod on node and returns the reply's value. +func remoteCall(t *testing.T, node *gorums.Node, timeout time.Duration) (string, error) { + t.Helper() + ctx := gorumstest.Context(t, timeout) + resp, err := gorumsimpl.RemoteCall[*pb.StringValue, *pb.StringValue](node.Context(ctx), pb.String("hello"), mock.TestMethod) + return resp.GetValue(), err +} + +// awaitCode calls node until the call fails with code, and fails the test if +// that does not happen within 5 seconds. Until the connection is ready, a +// call can fail for that reason instead. +func awaitCode(t *testing.T, node *gorums.Node, code codes.Code) error { + t.Helper() + var err error + if !gorumstest.WaitUntil(t, 5*time.Second, func() bool { + _, err = remoteCall(t, node, time.Second) + return status.Code(err) == code + }) { + t.Fatalf("call to node %d: err = %v; want code %v", node.ID(), err, code) + } + return err +} + +// awaitLastErrCode fails the test unless node's last error has code within 3 seconds. +func awaitLastErrCode(t *testing.T, node *gorums.Node, code codes.Code) { + t.Helper() + if !gorumstest.WaitUntil(t, 3*time.Second, func() bool { + return status.Code(node.LastErr()) == code + }) { + t.Errorf("node %d LastErr = %v; want code %v", node.ID(), node.LastErr(), code) + } +} + +var peerIdentityModes = []struct { + name string + opts []gorumstest.Option +}{ + {name: "Dual"}, + {name: "Dedup", opts: []gorumstest.Option{gorums.WithStreamDedup()}}, +} + +// TestPeerIdentityConnects verifies that peers with their own certificates +// connect, and that each handler sees the authenticated ID of the server that +// called it. +func TestPeerIdentityConnects(t *testing.T) { + for _, mode := range peerIdentityModes { + t.Run(mode.name, func(t *testing.T) { + servers, _ := peerGroup(t, newTestCA(t), 3, nil, mode.opts...) + gorumstest.WaitForPeers(t, servers) + for _, srv := range servers { + for _, node := range srv.PeerConfig() { + got, err := remoteCall(t, node, 3*time.Second) + if err != nil { + t.Fatalf("server %d: call to node %d: %v", srv.NodeID(), node.ID(), err) + } + // The handler on node.ID() sees the calling server as the sender. + if want := fmt.Sprintf("server %d from %d: hello", node.ID(), srv.NodeID()); got != want { + t.Errorf("server %d: call to node %d = %q; want %q", srv.NodeID(), node.ID(), got, want) + } + } + } + }) + } +} + +// TestPeerIdentityWrongCertificate verifies both checks with a server that +// presents another node's certificate: a peer that dials it refuses it, and +// a peer it dials refuses its claim. +func TestPeerIdentityWrongCertificate(t *testing.T) { + for _, mode := range peerIdentityModes { + t.Run(mode.name, func(t *testing.T) { + // Server 2 presents node 3's certificate. + servers, _ := peerGroup(t, newTestCA(t), 3, []gorums.ID{1, 3, 3}, mode.opts...) + + // Dial-side check: server 1 dials server 2 and finds node 3's certificate. + err := awaitCode(t, servers[0].PeerConfig().Node(2), codes.Unauthenticated) + if want := "gorums: node 2 failed verification"; !strings.Contains(err.Error(), want) { + t.Errorf("call from server 1 to node 2: err = %q; want it to contain %q", err, want) + } + // Inbound check: server 2 dials server 3 and claims ID 2 with node 3's certificate. + awaitLastErrCode(t, servers[1].PeerConfig().Node(3), codes.Unauthenticated) + // The other pair is unaffected. + if got, err := remoteCall(t, servers[0].PeerConfig().Node(3), 3*time.Second); err != nil || got != "server 3 from 1: hello" { + t.Errorf("call from server 1 to node 3 = %q, %v; want \"server 3 from 1: hello\", nil", got, err) + } + }) + } +} + +// TestPeerIdentityClients verifies how a server with peer identity checks +// treats clients: a client that claims a peer's ID without that peer's +// certificate is refused and cannot take over the peer's stream, and a +// client that claims no ID is served without a certificate. +func TestPeerIdentityClients(t *testing.T) { + ca := newTestCA(t) + // Under stream deduplication, server 2 sends its calls to node 1 over + // the stream that server 1 opened, which a forged claim would replace. + servers, calls := peerGroup(t, ca, 3, nil, gorums.WithStreamDedup()) + gorumstest.WaitForPeers(t, servers) + server2 := servers[1] + node3Cert := ca.issue(t, 3) + claim1 := gorums.WithMetadata(metadata.Pairs("gorums-node-id", "1")) + // The in-memory dialer belongs to the test that started the servers, so + // the subtests dial with this test's dial options. + dialOpt := gorumstest.DialOptions(t) + + tests := []struct { + name string + certs []tls.Certificate + opts []gorums.DialOption + wantServe bool + }{ + {name: "ImpostorClaimsLowerPeer", certs: []tls.Certificate{node3Cert}, opts: []gorums.DialOption{claim1}}, + {name: "ClaimWithoutCertificate", opts: []gorums.DialOption{claim1}}, + {name: "ClientWithoutCertificate", wantServe: true}, + } + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + before := calls[1].Load() + opts := append([]gorums.DialOption{ + dialOpt, + gorums.WithGRPCDialOptions(grpc.WithTransportCredentials(ca.dialCreds(tt.certs...))), + }, tt.opts...) + cfg, closeFn, err := gorums.NewConfig(gorums.WithNodeList([]string{server2.Addr()}), opts...) + if err != nil { + t.Fatalf("NewConfig: %v", err) + } + t.Cleanup(closeFn) + node := cfg[0] + + if tt.wantServe { + // A client that announces no ID has sender ID 0. + if got, err := remoteCall(t, node, 3*time.Second); err != nil || got != "server 2 from 0: hello" { + t.Errorf("call from the client = %q, %v; want \"server 2 from 0: hello\", nil", got, err) + } + return + } + // The client channel resends the call on each new stream until + // the call's context ends, so the call fails by its deadline; the + // refusal's status is the node's last error. + if _, err := remoteCall(t, node, time.Second); err == nil { + t.Fatal("call from a refused client succeeded; want failure") + } + awaitLastErrCode(t, node, codes.Unauthenticated) + if n := calls[1].Load() - before; n != 0 { + t.Errorf("server 2 handled %d requests from a refused client; want 0", n) + } + // Server 2's calls to node 1 still reach server 1. + if got, err := remoteCall(t, server2.PeerConfig().Node(1), 3*time.Second); err != nil || got != "server 1 from 2: hello" { + t.Errorf("call from server 2 to node 1 = %q, %v; want \"server 1 from 2: hello\", nil", got, err) + } + }) + } +} diff --git a/internal/tests/tls/testca_test.go b/internal/tests/tls/testca_test.go new file mode 100644 index 00000000..c06c4359 --- /dev/null +++ b/internal/tests/tls/testca_test.go @@ -0,0 +1,130 @@ +package tls + +import ( + "crypto/ecdsa" + "crypto/elliptic" + "crypto/rand" + "crypto/tls" + "crypto/x509" + "crypto/x509/pkix" + "fmt" + "math/big" + "net" + "net/url" + "testing" + "time" + + "github.com/relab/gorums" + "google.golang.org/grpc/credentials" +) + +// testCA is a throwaway certificate authority that issues per-replica +// certificates for the peer identity tests. +type testCA struct { + cert *x509.Certificate + key *ecdsa.PrivateKey + pool *x509.CertPool +} + +func newTestCA(t testing.TB) *testCA { + t.Helper() + key := newKey(t) + tmpl := &x509.Certificate{ + SerialNumber: newSerialNumber(t), + Subject: pkix.Name{CommonName: "Gorums Test CA"}, + NotBefore: time.Now().Add(-time.Minute), + NotAfter: time.Now().Add(time.Hour), + IsCA: true, + KeyUsage: x509.KeyUsageCertSign | x509.KeyUsageDigitalSignature, + BasicConstraintsValid: true, + } + der, err := x509.CreateCertificate(rand.Reader, tmpl, tmpl, key.Public(), key) + if err != nil { + t.Fatalf("create CA certificate: %v", err) + } + cert, err := x509.ParseCertificate(der) + if err != nil { + t.Fatalf("parse CA certificate: %v", err) + } + pool := x509.NewCertPool() + pool.AddCert(cert) + return &testCA{cert: cert, key: key, pool: pool} +} + +// nodeIdentity returns the URI SAN that names node id in these tests. +func nodeIdentity(id gorums.ID) string { + return fmt.Sprintf("spiffe://gorums.test/node/%d", id) +} + +// issue returns a certificate for node id with nodeIdentity(id) as its URI +// SAN, valid on the loopback addresses for both client and server +// authentication. +func (ca *testCA) issue(t testing.TB, id gorums.ID) tls.Certificate { + t.Helper() + key := newKey(t) + uri, err := url.Parse(nodeIdentity(id)) + if err != nil { + t.Fatalf("parse node identity: %v", err) + } + tmpl := &x509.Certificate{ + SerialNumber: newSerialNumber(t), + Subject: pkix.Name{CommonName: fmt.Sprintf("node %d", id)}, + URIs: []*url.URL{uri}, + IPAddresses: []net.IP{net.IPv4(127, 0, 0, 1), net.IPv6loopback}, + NotBefore: time.Now().Add(-time.Minute), + NotAfter: time.Now().Add(time.Hour), + KeyUsage: x509.KeyUsageDigitalSignature, + ExtKeyUsage: []x509.ExtKeyUsage{x509.ExtKeyUsageServerAuth, x509.ExtKeyUsageClientAuth}, + } + der, err := x509.CreateCertificate(rand.Reader, tmpl, ca.cert, key.Public(), ca.key) + if err != nil { + t.Fatalf("create certificate for node %d: %v", id, err) + } + return tls.Certificate{Certificate: [][]byte{der}, PrivateKey: key} +} + +// serverCreds returns server credentials that present cert and verify a +// client certificate against the CA when the client sends one. +func (ca *testCA) serverCreds(cert tls.Certificate) credentials.TransportCredentials { + return credentials.NewTLS(&tls.Config{ + Certificates: []tls.Certificate{cert}, + ClientCAs: ca.pool, + ClientAuth: tls.VerifyClientCertIfGiven, + }) +} + +// dialCreds returns dial credentials that verify servers against the CA and +// present certs, if any, as the client certificate. +func (ca *testCA) dialCreds(certs ...tls.Certificate) credentials.TransportCredentials { + return credentials.NewTLS(&tls.Config{ + Certificates: certs, + RootCAs: ca.pool, + }) +} + +// peerVerifier returns a TLSPeerVerifier for nodes 1 to n. +func peerVerifier(n int) gorums.PeerVerifier { + identities := make(map[gorums.ID]string, n) + for id := gorums.ID(1); id <= gorums.ID(n); id++ { + identities[id] = nodeIdentity(id) + } + return gorums.TLSPeerVerifier(identities) +} + +func newKey(t testing.TB) *ecdsa.PrivateKey { + t.Helper() + key, err := ecdsa.GenerateKey(elliptic.P256(), rand.Reader) + if err != nil { + t.Fatalf("generate key: %v", err) + } + return key +} + +func newSerialNumber(t testing.TB) *big.Int { + t.Helper() + sn, err := rand.Int(rand.Reader, new(big.Int).Lsh(big.NewInt(1), 128)) + if err != nil { + t.Fatalf("generate serial number: %v", err) + } + return sn +} From c33e0b595f802bfafb281eb49230211b41212a15 Mon Sep 17 00:00:00 2001 From: Hein Meling Date: Sun, 11 Oct 2026 17:49:21 +0200 Subject: [PATCH 6/9] stream: document the verifier contract and check every dialed stream A verifier on a dialing node runs on the channel's goroutine, so one that blocks delays the channel's sends and its Close. State that it should return promptly and when its context ends, in the stream package and in the root PeerVerifier alias. State the invariant behind calling the stream's Context before the check: it commits the stream to its transport, so requests reach the checked peer, and turns off gRPC's transparent retry for the stream. Add a test that pins re-verification of the replacement stream after the server kills the first one. --- internal/stream/channel_test.go | 48 +++++++++++++++++++++++++++++++++ internal/stream/outbound.go | 10 +++++-- peer_verifier.go | 5 +++- 3 files changed, 60 insertions(+), 3 deletions(-) diff --git a/internal/stream/channel_test.go b/internal/stream/channel_test.go index b16c7115..faa72f65 100644 --- a/internal/stream/channel_test.go +++ b/internal/stream/channel_test.go @@ -549,6 +549,54 @@ func TestChannelVerify(t *testing.T) { } } +// TestChannelVerifyChecksEveryStream verifies that a channel with a verifier +// checks the remote end of every stream it opens, not only the first: after +// the server kills the first stream, the eager reconnect's replacement stream +// is checked against the channel's ID before it carries a request. +func TestChannelVerifyChecksEveryStream(t *testing.T) { + var mu sync.Mutex + var checkedIDs []ID + opts := OutboundOptions{ + EagerReconnect: true, + Verify: func(_ context.Context, id ID) error { + mu.Lock() + defer mu.Unlock() + checkedIDs = append(checkedIDs, id) + return nil + }, + } + serverFn, conns := killFirstStreamServer() + tc := setupChannelOptions(t, opts, serverFn) + + // The server kills the first stream; the channel redials on its own. + for want := int32(1); want <= 2; want++ { + select { + case n := <-conns: + if n != want { + t.Fatalf("accepted stream ordinal = %d; want %d", n, want) + } + case <-time.After(defaultTestTimeout): + t.Fatalf("stream %d never reached the server", want) + } + } + + // A request is sent only after the second stream's check returns. + if resp := sendRequest(t, tc.OutboundChannel, Request{}, 1); resp.Err != nil { + t.Fatalf("request on the second stream: %v", resp.Err) + } + + mu.Lock() + defer mu.Unlock() + if len(checkedIDs) < 2 { + t.Fatalf("verifier called %d times; want at least 2, once per stream", len(checkedIDs)) + } + for _, id := range checkedIDs { + if id != 1 { + t.Errorf("verifier checked node %d; want 1, the channel's ID", id) + } + } +} + // TestChannelStreamFailureRecordsLastErr verifies that a request the channel // cannot deliver because no stream could be established leaves the reason in // LastErr, even when the request has no response channel to report it on. diff --git a/internal/stream/outbound.go b/internal/stream/outbound.go index 50751a76..759c04e6 100644 --- a/internal/stream/outbound.go +++ b/internal/stream/outbound.go @@ -23,7 +23,9 @@ const ( // that explains the refusal otherwise. ctx is the stream's context, which // carries the connection's [google.golang.org/grpc/peer.Peer] and, on a // server, the stream's incoming metadata. A PeerVerifier must be safe for -// concurrent use. +// concurrent use. It should return promptly, and return when ctx ends: on a +// dialing node it runs on the channel's goroutine, so a verifier that blocks +// delays the channel's sends and its Close. type PeerVerifier func(ctx context.Context, id ID) error // OutboundOptions configures an [OutboundChannel]. @@ -180,7 +182,11 @@ func (c *OutboundChannel) open() (*session, error) { } if c.verify != nil { // The stream's context carries the server's peer information, - // including its TLS state, once NodeStream returns. + // including its TLS state, once NodeStream returns. Calling Context + // commits the stream to its transport, so the requests sent on this + // stream reach the peer that was checked. It also turns off gRPC's + // transparent retry for this stream; the run loop's reconnect + // replaces it. if err := c.verify(st.Context(), c.id); err != nil { cancel() return nil, VerificationError(c.id, err) diff --git a/peer_verifier.go b/peer_verifier.go index 804bec2e..4ab27f95 100644 --- a/peer_verifier.go +++ b/peer_verifier.go @@ -17,7 +17,10 @@ import ( // the refusal otherwise. Its context is the stream's context, which carries // the connection's [google.golang.org/grpc/peer.Peer] and, on a server, the // stream's incoming metadata. A PeerVerifier must be safe for concurrent use. -// See [WithPeerVerifier]. +// It should return promptly, and return when its context ends: on a server +// that dials a peer, it runs on the goroutine that sends to that peer, so a +// verifier that blocks delays the sends to that peer and [Server.Stop]. See +// [WithPeerVerifier]. type PeerVerifier = conn.PeerVerifier // peerIdentity is the name that a node's certificate must carry: a URI SAN From 528334dfc2f27a3437f224c823dbbe24397ae263 Mon Sep 17 00:00:00 2001 From: Hein Meling Date: Sun, 11 Oct 2026 17:51:41 +0200 Subject: [PATCH 7/9] gorums: reject duplicate identities in TLSPeerVerifier If two node IDs have the same identity, one certificate matches both, so either replica could claim the other's ID. TLSPeerVerifier now panics at construction in that case. Two identities are the same when one SAN matches both: equal canonical URIs, or DNS names that are equal without regard to case. IDs are compared in increasing order so the panic message names the same pair on every run. --- peer_verifier.go | 27 +++++++++++++++++++++- peer_verifier_test.go | 52 +++++++++++++++++++++++++++++++++++++++++++ 2 files changed, 78 insertions(+), 1 deletion(-) diff --git a/peer_verifier.go b/peer_verifier.go index 4ab27f95..bb8ad6fa 100644 --- a/peer_verifier.go +++ b/peer_verifier.go @@ -4,7 +4,9 @@ import ( "context" "errors" "fmt" + "maps" "net/url" + "slices" "strings" "github.com/relab/gorums/internal/conn" @@ -30,6 +32,18 @@ type peerIdentity struct { uri bool } +// same reports whether p and q name the same SAN, so that one certificate +// matches both. +func (p peerIdentity) same(q peerIdentity) bool { + if p.uri != q.uri { + return false + } + if p.uri { + return p.name == q.name + } + return strings.EqualFold(p.name, q.name) +} + // kind returns the type of SAN that the identity names. func (p peerIdentity) kind() string { if p.uri { @@ -51,7 +65,8 @@ func (p peerIdentity) kind() string { // ClientAuth set to tls.VerifyClientCertIfGiven or // tls.RequireAndVerifyClientCert, and each replica's certificate must be // valid for both client and server authentication. TLSPeerVerifier copies -// identities. +// identities. TLSPeerVerifier panics if two node IDs have the same identity, +// since either replica could then claim the other's ID. func TLSPeerVerifier(identities map[ID]string) PeerVerifier { want := make(map[ID]peerIdentity, len(identities)) for id, name := range identities { @@ -64,6 +79,16 @@ func TLSPeerVerifier(identities map[ID]string) PeerVerifier { } want[id] = peerIdentity{name: name} } + // IDs are compared in increasing order, so the panic names the same pair + // on every run. + ids := slices.Sorted(maps.Keys(want)) + for i, id := range ids { + for _, lower := range ids[:i] { + if want[lower].same(want[id]) { + panic(fmt.Sprintf("gorums: TLSPeerVerifier: nodes %d and %d have the same identity %q", lower, id, want[lower].name)) + } + } + } return func(ctx context.Context, id ID) error { p, ok := peer.FromContext(ctx) if !ok { diff --git a/peer_verifier_test.go b/peer_verifier_test.go index 8f5681ac..d646762a 100644 --- a/peer_verifier_test.go +++ b/peer_verifier_test.go @@ -107,3 +107,55 @@ func TestPeerVerifierTLS(t *testing.T) { } }) } + +// TestPeerVerifierTLSDuplicateIdentity verifies that TLSPeerVerifier panics +// when one SAN would match two node IDs, naming the lowest such pair, and +// accepts identities that no single SAN matches together. +func TestPeerVerifierTLSDuplicateIdentity(t *testing.T) { + tests := []struct { + name string + identities map[gorums.ID]string + wantPanic string // empty means no panic + }{ + { + name: "DuplicateURI", + identities: map[gorums.ID]string{1: "spiffe://relab/node/1", 2: "spiffe://relab/node/2", 3: "spiffe://relab/node/1"}, + wantPanic: `gorums: TLSPeerVerifier: nodes 1 and 3 have the same identity "spiffe://relab/node/1"`, + }, + { + name: "DuplicateURICanonicalForm", + identities: map[gorums.ID]string{4: "SPIFFE://relab/node/4", 5: "spiffe://relab/node/4"}, + wantPanic: `gorums: TLSPeerVerifier: nodes 4 and 5 have the same identity "spiffe://relab/node/4"`, + }, + { + name: "DuplicateDNSIgnoringCase", + identities: map[gorums.ID]string{2: "node2.relab.example", 7: "Node2.Relab.Example"}, + wantPanic: `gorums: TLSPeerVerifier: nodes 2 and 7 have the same identity "node2.relab.example"`, + }, + { + // A URI SAN matches only URI identities and a DNS SAN only DNS + // identities, so no SAN matches both nodes. + name: "URIAndDNSIdentity", + identities: map[gorums.ID]string{1: "spiffe://node1.relab.example", 2: "node1.relab.example"}, + }, + { + // URI SANs match exactly, so a SAN matches at most one of them. + name: "URIsDifferingInCase", + identities: map[gorums.ID]string{1: "spiffe://relab/node/a", 2: "SPIFFE://RELAB/NODE/A"}, + }, + } + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + if tt.wantPanic != "" { + assertPanic(t, tt.wantPanic, func() { gorums.TLSPeerVerifier(tt.identities) }) + return + } + defer func() { + if r := recover(); r != nil { + t.Fatalf("TLSPeerVerifier(%v) panicked: %v; want no panic", tt.identities, r) + } + }() + gorums.TLSPeerVerifier(tt.identities) + }) + } +} From 284a67fba21732f0b92cccc1bd86dc4a95a0bcc0 Mon Sep 17 00:00:00 2001 From: Hein Meling Date: Sun, 11 Oct 2026 17:53:32 +0200 Subject: [PATCH 8/9] gorums: document refused peers, trust anchors, and sender IDs A refused peer does not see codes.Unauthenticated on its calls: it resends them on new streams until their contexts end, and the refusal with the verifier's text appears in Node.LastErr. State this in the WithPeerVerifier docs and the user guide, with the need for deadlines and the brief effect on ConnectedPeers and WaitForAll. Add a subtest that pins the refusal text in the refused peer's Node.LastErr. State that TLSPeerVerifier is only as strong as the CA pools, and which TLS settings verify no chain. Tabulate which SenderID values are authenticated, state that an unknown claimed ID is refused under a verifier, and show an interceptor check with Config.Contains, since a nonzero SenderID also covers unauthenticated back-channel clients. --- doc/user-guide.md | 27 +++++++++++++++++++++++---- peer_verifier.go | 19 +++++++++++++------ server.go | 9 +++++---- server_options.go | 18 ++++++++++-------- server_test.go | 30 +++++++++++++++++++++++++++--- 5 files changed, 78 insertions(+), 25 deletions(-) diff --git a/doc/user-guide.md b/doc/user-guide.md index 412ab7e8..f6884089 100644 --- a/doc/user-guide.md +++ b/doc/user-guide.md @@ -1688,7 +1688,7 @@ func (srv *storageSrv) WriteQC(ctx gorums.ServerContext, req *WriteRequest) (*Wr * a request on a connection that this server dialed has the dialed node's ID; * a request that this server sends to itself has this server's own ID. -`SenderID` returns 0 for a client that announces no ID, or an ID that this server does not know. +`SenderID` returns 0 for a client that announces no ID, or, without `WithPeerVerifier`, for an ID that this server does not know; with it, a stream that claims an unknown ID is refused. The sender is the last hop. When a message travels through several nodes, for example down a tree, `SenderID` identifies the node that forwarded it, not the node where the message started. @@ -2407,7 +2407,21 @@ Use `TLSPeerVerifier` to bind node IDs to mTLS certificates; see [Binding Node I | Opened by this server to a configured peer | `verify(ctx, dialed ID)` before any request is sent | The waiting call fails with `codes.Unauthenticated`; the server tries again later | | Claims no peer node ID (a client) | None | Served as without the option | -With `WithPeerVerifier`, `ServerContext.SenderID` is authenticated for every configured peer, with and without stream deduplication. +A peer that this server refuses sees the refusal, with the verifier's text, in `Node.LastErr` of its node for this server. +Its pending calls to this server are resent on new streams until their contexts end, so they fail with the context's error, not with `codes.Unauthenticated`; give calls between peers a deadline. +While the refused peer keeps retrying, its `ConnectedPeers` view can briefly include this server, and under `WithStreamDedup` its `WaitForAll` can return before this server has accepted any of its streams. + +With `WithPeerVerifier`, `ServerContext.SenderID` is authenticated for every configured peer, with and without stream deduplication: + +| Request arrives on | `SenderID` | Checked by | +| --- | --- | --- | +| A stream that a configured peer opened to this server | The peer's ID | The inbound check | +| A stream that this server opened to a configured peer | The dialed peer's ID | The dial-side check | +| A back-channel client's stream | The ID this server assigned | Not authenticated | +| A regular client's stream | 0 | Not authenticated | +| A call this server sends to itself | This server's own ID | No network involved | + +`SenderID` names only the last hop. `NewServer` panics if the verifier is nil or the server has no `WithPeers`. ### Binding Node IDs to TLS Certificates @@ -2453,8 +2467,12 @@ srv := gorums.NewServer( ``` With `ClientAuth: tls.RequireAndVerifyClientCert`, clients without a certificate are refused as well. +The check is only as strong as the CA pools. +Set `ClientCAs` on the server and `RootCAs` on the dialer to a pool that holds only the cluster's CA; a nil pool means the system roots. +That CA must issue each identity only to its replica. +With `InsecureSkipVerify` on the dialer, or `ClientAuth` set to `tls.RequestClientCert` or `tls.RequireAnyClientCert` on the server, no chain is verified and every stream is refused. -For a local cluster, `NewLocalServers` takes options that differ per server: `WithLocalServerOptionsFunc` and `WithLocalDialOptionsFunc` return the options for a given node ID, such as that server's certificate. +For a local cluster, `NewLocalServers` takes options that differ per server: `WithLocalServerOptionsFunc` and `WithLocalDialOptionsFunc` take a function that returns the options for a given node ID, such as that server's certificate. ### Authenticating Clients @@ -2463,7 +2481,8 @@ For a local cluster, `NewLocalServers` takes options that differ per server: `Wi * To require a certificate from every client, set `ClientAuth: tls.RequireAndVerifyClientCert` in the server's `tls.Config`. The handshake then fails before Gorums sees a stream. * To decide which certificates may connect, use `tls.Config.VerifyConnection`, which runs on every handshake. -* To decide per request, for example to let only peers call a method, use a server interceptor that checks `ctx.SenderID()` or `peer.FromContext(ctx)`. +* To decide per request, for example to let only peers call a method, use a server interceptor that checks `ctx.PeerConfig().Contains(ctx.SenderID())`. + A nonzero `SenderID` alone is not enough, because back-channel clients have nonzero IDs that are not authenticated. ## Send Queue Capacity and Backpressure diff --git a/peer_verifier.go b/peer_verifier.go index bb8ad6fa..4afb5eec 100644 --- a/peer_verifier.go +++ b/peer_verifier.go @@ -61,12 +61,19 @@ func (p peerIdentity) kind() string { // and the Common Name are never used, and a node ID with an empty or missing // identity is refused. // -// The server's TLS configuration must verify client certificates, with -// ClientAuth set to tls.VerifyClientCertIfGiven or -// tls.RequireAndVerifyClientCert, and each replica's certificate must be -// valid for both client and server authentication. TLSPeerVerifier copies -// identities. TLSPeerVerifier panics if two node IDs have the same identity, -// since either replica could then claim the other's ID. +// The server's [crypto/tls.Config] must verify client certificates, with +// ClientAuth set to [crypto/tls.VerifyClientCertIfGiven] or +// [crypto/tls.RequireAndVerifyClientCert], and each replica's certificate +// must be valid for both client and server authentication. TLSPeerVerifier +// copies identities. TLSPeerVerifier panics if two node IDs have the same +// identity, since either replica could then claim the other's ID. +// +// The check is only as strong as the CA pools: set ClientCAs on the server +// and RootCAs on the dialer to a pool that holds only the cluster's CA, since +// a nil pool means the system roots, and that CA must issue each identity +// only to its replica. With InsecureSkipVerify on the dialer, or ClientAuth +// set to [crypto/tls.RequestClientCert] or [crypto/tls.RequireAnyClientCert] +// on the server, no chain is verified and every stream is refused. func TLSPeerVerifier(identities map[ID]string) PeerVerifier { want := make(map[ID]peerIdentity, len(identities)) for id, name := range identities { diff --git a/server.go b/server.go index 4326ea7e..2c9b1cad 100644 --- a/server.go +++ b/server.go @@ -390,10 +390,11 @@ type ServerContext struct { // node's ID; // - a request that this server sends to itself has this server's own ID. // -// SenderID returns 0 for a client that announces no ID, or an ID that this -// server does not know. The sender is the last hop: when a message travels -// through several nodes, SenderID identifies the node that forwarded it. The -// sender asserts its ID when it connects. A server configured with +// SenderID returns 0 for a client that announces no ID, or, without +// [WithPeerVerifier], for an ID that this server does not know; with it, a +// stream that claims an unknown ID is refused. The sender is the last hop: +// when a message travels through several nodes, SenderID identifies the node +// that forwarded it. The sender asserts its ID when it connects. A server configured with // [WithPeerVerifier] checks the ID of every configured peer, on the streams in // both directions; otherwise the server does not authenticate it. func (ctx *ServerContext) SenderID() ID { diff --git a/server_options.go b/server_options.go index 267d2127..e9b5edfb 100644 --- a/server_options.go +++ b/server_options.go @@ -148,14 +148,16 @@ func WithStreamDedup() ServerOption { // stream is closed, the waiting call fails with codes.Unauthenticated, // and the server tries again later, as for a failed connection. // -// When the server refuses a stream that a peer opened, the refusal's message -// includes the text of verify's error, so the peer learns why. When a stream -// that the server opened fails the check, the text is in the error of the -// waiting call. The server runs verify once per stream, outside its internal -// locks. With WithPeerVerifier, [ServerContext.SenderID] is authenticated for -// every configured peer. [NewServer] panics if verify is nil or if the server -// has no [WithPeers]. Use [TLSPeerVerifier] to bind node IDs to mTLS -// certificates. +// When the server refuses a stream that a peer opened, the refusal's status +// carries the text of verify's error, and the refused peer sees it in +// [Node.LastErr]. The refused peer resends its pending calls on new streams +// until their contexts end, so calls between peers need deadlines. When a +// stream that the server opened fails the check, the waiting call fails with +// the check's error. The server runs verify once per stream, outside its +// internal locks. With WithPeerVerifier, [ServerContext.SenderID] is +// authenticated for every configured peer. [NewServer] panics if verify is nil +// or if the server has no [WithPeers]. Use [TLSPeerVerifier] to bind node IDs +// to mTLS certificates. func WithPeerVerifier(verify PeerVerifier) ServerOption { return func(o *serverOptions) { o.peerVerifier = verify diff --git a/server_test.go b/server_test.go index 1793fbdb..11be2478 100644 --- a/server_test.go +++ b/server_test.go @@ -1381,9 +1381,10 @@ func TestServerPeerVerifier(t *testing.T) { } // TestServerPeerVerifierRefusals verifies that a server with a peer verifier -// fails its calls to a peer that does not pass the check, refuses a client -// that claims a node ID outside the peer configuration, and serves a client -// that claims no node ID. +// fails its calls to a peer that does not pass the check, that a peer it +// refuses sees the verifier's text in its node's last error, that it refuses +// a client that claims a node ID outside the peer configuration, and that it +// serves a client that claims no node ID. func TestServerPeerVerifierRefusals(t *testing.T) { refuseAll := gorums.WithPeerVerifier(func(context.Context, gorums.ID) error { return errors.New("not trusted") }) acceptAll := gorums.WithPeerVerifier(func(context.Context, gorums.ID) error { return nil }) @@ -1428,6 +1429,29 @@ func TestServerPeerVerifierRefusals(t *testing.T) { t.Errorf("call to a refused peer: err = %q; want it to contain %q", err, want) } }) + t.Run("RefusedPeerSeesReason", func(t *testing.T) { + // Server 1's dial-side check of server 2 passes, and server 2 + // refuses server 1's claim; server 1's node for server 2 records + // the refusal, with server 2's verifier text, as its last error. + verifierFor := func(self gorums.ID) []gorums.ServerOption { + if self == 2 { + return []gorums.ServerOption{gorums.WithPeerVerifier(func(context.Context, gorums.ID) error { + return errors.New("certificate names another node") + })} + } + return []gorums.ServerOption{acceptAll} + } + servers := gorumstest.LocalServers(t, 2, gorums.WithLocalServerOptionsFunc(verifierFor)) + node := servers[0].PeerConfig().Node(2) + const want = "gorums: node 1 failed verification: certificate names another node" + var err error + if !gorumstest.WaitUntil(t, 3*time.Second, func() bool { + err = node.LastErr() + return status.Code(err) == codes.Unauthenticated && strings.Contains(err.Error(), want) + }) { + t.Errorf("server 1's node 2 LastErr = %v; want code %v containing %q", err, codes.Unauthenticated, want) + } + }) t.Run("UnknownPeerIDRefused", func(t *testing.T) { servers := start(t, acceptAll) node := client(t, servers[0], gorums.WithMetadata(metadata.Pairs("gorums-node-id", "99"))) From 6e771699ece6a13395885d2cbd40ad83743f11c2 Mon Sep 17 00:00:00 2001 From: Hein Meling Date: Sun, 11 Oct 2026 21:01:53 +0200 Subject: [PATCH 9/9] gorums: tighten peer verifier tests and docs * Make the cross-type duplicate test case able to fail by using a DNS identity that strings.EqualFold equates with a URI identity. * Document that TLSPeerVerifier panics on duplicate identities. * Fix wording: rewrap the SenderID comment, say that only two-way calls are resent, say "every peer stream is refused", and split a paragraph in the user guide. --- doc/user-guide.md | 6 ++++-- peer_verifier.go | 2 +- peer_verifier_test.go | 9 ++++++--- server.go | 7 ++++--- server_options.go | 5 +++-- 5 files changed, 18 insertions(+), 11 deletions(-) diff --git a/doc/user-guide.md b/doc/user-guide.md index f6884089..6dc813af 100644 --- a/doc/user-guide.md +++ b/doc/user-guide.md @@ -2408,7 +2408,7 @@ Use `TLSPeerVerifier` to bind node IDs to mTLS certificates; see [Binding Node I | Claims no peer node ID (a client) | None | Served as without the option | A peer that this server refuses sees the refusal, with the verifier's text, in `Node.LastErr` of its node for this server. -Its pending calls to this server are resent on new streams until their contexts end, so they fail with the context's error, not with `codes.Unauthenticated`; give calls between peers a deadline. +Its pending two-way calls to this server are resent on new streams until their contexts end, so they fail with the context's error, not with `codes.Unauthenticated`; give calls between peers a deadline. While the refused peer keeps retrying, its `ConnectedPeers` view can briefly include this server, and under `WithStreamDedup` its `WaitForAll` can return before this server has accepted any of its streams. With `WithPeerVerifier`, `ServerContext.SenderID` is authenticated for every configured peer, with and without stream deduplication: @@ -2422,6 +2422,7 @@ With `WithPeerVerifier`, `ServerContext.SenderID` is authenticated for every con | A call this server sends to itself | This server's own ID | No network involved | `SenderID` names only the last hop. + `NewServer` panics if the verifier is nil or the server has no `WithPeers`. ### Binding Node IDs to TLS Certificates @@ -2440,6 +2441,7 @@ It accepts a stream for a node ID only if the TLS handshake verified the remote An identity with a URI scheme, such as `spiffe://relab/node/2`, matches only a URI SAN with the same canonical form. Any other identity matches only a DNS SAN, without regard to case. A wildcard DNS SAN never matches, IP SANs and the certificate's Common Name are never used, and a node ID with an empty or missing identity is refused. +`TLSPeerVerifier` panics if two node IDs have the same identity, since either replica could then claim the other's ID. Each replica needs its own certificate with: @@ -2470,7 +2472,7 @@ With `ClientAuth: tls.RequireAndVerifyClientCert`, clients without a certificate The check is only as strong as the CA pools. Set `ClientCAs` on the server and `RootCAs` on the dialer to a pool that holds only the cluster's CA; a nil pool means the system roots. That CA must issue each identity only to its replica. -With `InsecureSkipVerify` on the dialer, or `ClientAuth` set to `tls.RequestClientCert` or `tls.RequireAnyClientCert` on the server, no chain is verified and every stream is refused. +With `InsecureSkipVerify` on the dialer, or `ClientAuth` set to `tls.RequestClientCert` or `tls.RequireAnyClientCert` on the server, no chain is verified and every peer stream is refused. For a local cluster, `NewLocalServers` takes options that differ per server: `WithLocalServerOptionsFunc` and `WithLocalDialOptionsFunc` take a function that returns the options for a given node ID, such as that server's certificate. diff --git a/peer_verifier.go b/peer_verifier.go index 4afb5eec..16ec03b8 100644 --- a/peer_verifier.go +++ b/peer_verifier.go @@ -73,7 +73,7 @@ func (p peerIdentity) kind() string { // a nil pool means the system roots, and that CA must issue each identity // only to its replica. With InsecureSkipVerify on the dialer, or ClientAuth // set to [crypto/tls.RequestClientCert] or [crypto/tls.RequireAnyClientCert] -// on the server, no chain is verified and every stream is refused. +// on the server, no chain is verified and every peer stream is refused. func TLSPeerVerifier(identities map[ID]string) PeerVerifier { want := make(map[ID]peerIdentity, len(identities)) for id, name := range identities { diff --git a/peer_verifier_test.go b/peer_verifier_test.go index d646762a..c8582ba7 100644 --- a/peer_verifier_test.go +++ b/peer_verifier_test.go @@ -134,9 +134,12 @@ func TestPeerVerifierTLSDuplicateIdentity(t *testing.T) { }, { // A URI SAN matches only URI identities and a DNS SAN only DNS - // identities, so no SAN matches both nodes. - name: "URIAndDNSIdentity", - identities: map[gorums.ID]string{1: "spiffe://node1.relab.example", 2: "node1.relab.example"}, + // identities, so no SAN matches both nodes. The first string + // starts with U+017F (LATIN SMALL LETTER LONG S), so url.Parse + // rejects it and it is a DNS identity, yet strings.EqualFold + // treats it as equal to the URI identity. + name: "URIAndDNSIdentityEqualFold", + identities: map[gorums.ID]string{1: "ſpiffe://relab/node/1", 2: "spiffe://relab/node/1"}, }, { // URI SANs match exactly, so a SAN matches at most one of them. diff --git a/server.go b/server.go index 2c9b1cad..fe50b83d 100644 --- a/server.go +++ b/server.go @@ -394,9 +394,10 @@ type ServerContext struct { // [WithPeerVerifier], for an ID that this server does not know; with it, a // stream that claims an unknown ID is refused. The sender is the last hop: // when a message travels through several nodes, SenderID identifies the node -// that forwarded it. The sender asserts its ID when it connects. A server configured with -// [WithPeerVerifier] checks the ID of every configured peer, on the streams in -// both directions; otherwise the server does not authenticate it. +// that forwarded it. The sender asserts its ID when it connects. A server +// configured with [WithPeerVerifier] checks the ID of every configured peer, +// on the streams in both directions; otherwise the server does not +// authenticate it. func (ctx *ServerContext) SenderID() ID { return ctx.senderID } diff --git a/server_options.go b/server_options.go index e9b5edfb..41b38650 100644 --- a/server_options.go +++ b/server_options.go @@ -150,8 +150,9 @@ func WithStreamDedup() ServerOption { // // When the server refuses a stream that a peer opened, the refusal's status // carries the text of verify's error, and the refused peer sees it in -// [Node.LastErr]. The refused peer resends its pending calls on new streams -// until their contexts end, so calls between peers need deadlines. When a +// [Node.LastErr]. The refused peer resends its pending two-way calls on new +// streams until their contexts end, so calls between peers need deadlines; its +// streaming calls fail instead. When a // stream that the server opened fails the check, the waiting call fails with // the check's error. The server runs verify once per stream, outside its // internal locks. With WithPeerVerifier, [ServerContext.SenderID] is