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..6dc813af 100644 --- a/doc/user-guide.md +++ b/doc/user-guide.md @@ -1688,14 +1688,16 @@ 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. 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,117 @@ 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 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 | +| --- | --- | --- | +| 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 | + +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 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: + +| 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 + +`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. +`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: + +* 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. +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 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. + +### 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.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 The per-node send queue defaults to 4096 entries. 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/internal/stream/channel_test.go b/internal/stream/channel_test.go index e70f7bd8..faa72f65 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,134 @@ 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") + } + }) + } +} + +// 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/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..759c04e6 100644 --- a/internal/stream/outbound.go +++ b/internal/stream/outbound.go @@ -18,6 +18,16 @@ 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. 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]. type OutboundOptions struct { // SendBufferSize is the send queue capacity. @@ -33,6 +43,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 +62,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 +90,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 +171,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 +180,18 @@ 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. 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) + } + } s := newSession(ctx, cancel, &c.endpoint, st, true, true) c.mu.Lock() c.sessions[s] = struct{}{} 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 +} diff --git a/peer_verifier.go b/peer_verifier.go new file mode 100644 index 00000000..16ec03b8 --- /dev/null +++ b/peer_verifier.go @@ -0,0 +1,133 @@ +package gorums + +import ( + "context" + "errors" + "fmt" + "maps" + "net/url" + "slices" + "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 +// 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. +// 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 +// if uri is set, and a DNS SAN otherwise. +type peerIdentity struct { + name string + 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 { + 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 [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 peer stream is refused. +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} + } + // 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 { + 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..c8582ba7 --- /dev/null +++ b/peer_verifier_test.go @@ -0,0 +1,164 @@ +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) + } + }) +} + +// 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. 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. + 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) + }) + } +} diff --git a/server.go b/server.go index b292a7b1..fe50b83d 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,6 +141,7 @@ func newServer(opts ...ServerOption) (*Server, error) { serverOpts.recvBufferSize, serverOpts.onConfigChange, s, + serverOpts.peerVerifier, ) if err != nil { s.grpcServer.Stop() @@ -380,10 +390,13 @@ 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, and the server does not +// 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 { return ctx.senderID diff --git a/server_options.go b/server_options.go index af47625e..41b38650 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,34 @@ 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 status +// carries the text of verify's error, and the refused peer sees it in +// [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 +// 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 + o.hasPeerVerifier = true + } +} diff --git a/server_test.go b/server_test.go index cbeafdbe..11be2478 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,193 @@ 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, 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 }) + 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("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"))) + 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...) }) + }) + } +}