GODRIVER-1584 Ensure cluster time is updated from handshakes (#390)

Author: Divjot Arora

.evergreen/config.yml

@@ -1235,6 +1235,20 @@ tasks:
           AUTH: "auth"
           SSL: "ssl"
 
+  - name: test-replicaset-auth-nossl
+    tags: ["test", "replicaset", "authssl"]
+    commands:
+      - func: bootstrap-mongo-orchestration
+        vars:
+          TOPOLOGY: "replica_set"
+          AUTH: "auth"
+          SSL: "nossl"
+      - func: run-tests
+        vars:
+          TOPOLOGY: "replica_set"
+          AUTH: "auth"
+          SSL: "nossl"
+
   - name: test-sharded-noauth-nossl
     tags: ["test", "sharded"]
     commands:

mongo/client.go

@@ -324,6 +324,9 @@ func (c *Client) configure(opts *options.ClientOptions) error {
 
 	// TODO(GODRIVER-814): Add tests for topology, server, and connection related options.
 
+	// ClusterClock
+	c.clock = new(session.ClusterClock)
+
 	// Pass down URI so topology can determine whether or not SRV polling is required
 	topologyOpts = append(topologyOpts, topology.WithURI(func(uri string) string {
 		return opts.GetURI()
@@ -368,7 +371,7 @@ func (c *Client) configure(opts *options.ClientOptions) error {
 	}
 	// Handshaker
 	var handshaker = func(driver.Handshaker) driver.Handshaker {
-		return operation.NewIsMaster().AppName(appName).Compressors(comps)
+		return operation.NewIsMaster().AppName(appName).Compressors(comps).ClusterClock(c.clock)
 	}
 	// Auth & Database & Password & Username
 	if opts.Auth != nil {
@@ -399,6 +402,7 @@ func (c *Client) configure(opts *options.ClientOptions) error {
 			AppName:       appName,
 			Authenticator: authenticator,
 			Compressors:   comps,
+			ClusterClock:  c.clock,
 		}
 		if mechanism == "" {
 			// Required for SASL mechanism negotiation during handshake
@@ -553,9 +557,6 @@ func (c *Client) configure(opts *options.ClientOptions) error {
 		}
 	}
 
-	// ClusterClock
-	c.clock = new(session.ClusterClock)
-
 	// OCSP cache
 	ocspCache := ocsp.NewCache()
 	connOpts = append(

mongo/integration/client_test.go

@@ -71,7 +71,7 @@ func TestClient(t *testing.T) {
 
 		assert.Equal(mt, int64(-10), got.ID, "expected ID -10, got %v", got.ID)
 	})
-	mt.RunOpts("tls connection", mtest.NewOptions().MinServerVersion("3.0").Auth(true), func(mt *mtest.T) {
+	mt.RunOpts("tls connection", mtest.NewOptions().MinServerVersion("3.0").SSL(true), func(mt *mtest.T) {
 		var result bson.Raw
 		err := mt.Coll.Database().RunCommand(mtest.Background, bson.D{
 			{"serverStatus", 1},
@@ -86,7 +86,7 @@ func TestClient(t *testing.T) {
 		_, found = security.Document().LookupErr("SSLServerHasCertificateAuthority")
 		assert.Nil(mt, found, "SSLServerHasCertificateAuthority not found in result")
 	})
-	mt.RunOpts("x509", mtest.NewOptions().Auth(true), func(mt *mtest.T) {
+	mt.RunOpts("x509", mtest.NewOptions().Auth(true).SSL(true), func(mt *mtest.T) {
 		const user = "C=US,ST=New York,L=New York City,O=MongoDB,OU=other,CN=external"
 		db := mt.Client.Database("$external")
 
@@ -396,13 +396,13 @@ func TestClient(t *testing.T) {
 		err := mt.Client.Ping(mtest.Background, mtest.PrimaryRp)
 		assert.Nil(mt, err, "Ping error: %v", err)
 
-		sent := appNameProxyDialer.sent
-		assert.True(mt, len(sent) >= 2, "expected at least 2 events sent, got %v", len(sent))
+		msgPairs := appNameProxyDialer.messages
+		assert.True(mt, len(msgPairs) >= 2, "expected at least 2 events sent, got %v", len(msgPairs))
 
 		// First two messages should be connection handshakes: one for the heartbeat connection and the other for the
 		// application connection.
-		for idx, wm := range sent[:2] {
-			cmd, err := drivertest.GetCommandFromQueryWireMessage(wm)
+		for idx, pair := range msgPairs[:2] {
+			cmd, err := drivertest.GetCommandFromQueryWireMessage(pair.sent)
 			assert.Nil(mt, err, "GetCommandFromQueryWireMessage error at index %d: %v", idx, err)
 			heartbeatCmdName := cmd.Index(0).Key()
 			assert.Equal(mt, "isMaster", heartbeatCmdName,
@@ -441,13 +441,19 @@ func TestClient(t *testing.T) {
 	})
 }
 
+type proxyMessage struct {
+	serverAddress string
+	sent          wiremessage.WireMessage
+	received      wiremessage.WireMessage
+}
+
 // proxyDialer is a ContextDialer implementation that wraps a net.Dialer and records the messages sent and received
 // using connections created through it.
 type proxyDialer struct {
 	*net.Dialer
 	sync.Mutex
-	sent     []wiremessage.WireMessage
-	received []wiremessage.WireMessage
+	messages []proxyMessage
+	sentMap  sync.Map
 }
 
 var _ options.ContextDialer = (*proxyDialer)(nil)
@@ -480,7 +486,9 @@ func (p *proxyDialer) storeSentMessage(msg []byte) {
 
 	msgCopy := make(wiremessage.WireMessage, len(msg))
 	copy(msgCopy, msg)
-	p.sent = append(p.sent, msgCopy)
+
+	_, requestID, _, _, _, _ := wiremessage.ReadHeader(msgCopy)
+	p.sentMap.Store(requestID, msgCopy)
 }
 
 // storeReceivedMessage stores a copy of the wire message being received from the server.
@@ -490,7 +498,16 @@ func (p *proxyDialer) storeReceivedMessage(msg []byte) {
 
 	msgCopy := make(wiremessage.WireMessage, len(msg))
 	copy(msgCopy, msg)
-	p.received = append(p.received, msgCopy)
+
+	_, _, responseTo, _, _, _ := wiremessage.ReadHeader(msgCopy)
+	sentMsg, _ := p.sentMap.Load(responseTo)
+	p.sentMap.Delete(responseTo)
+
+	proxyMsg := proxyMessage{
+		sent:     sentMsg.(wiremessage.WireMessage),
+		received: msgCopy,
+	}
+	p.messages = append(p.messages, proxyMsg)
 }
 
 // proxyConn is a net.Conn that wraps a network connection. All messages sent/received through a proxyConn are stored

mongo/integration/mtest/mongotest.go

@@ -92,6 +92,7 @@ type T struct {
 	maxServerVersion string
 	validTopologies  []TopologyKind
 	auth             *bool
+	ssl              *bool
 	enterprise       *bool
 	collCreateOpts   bson.D
 	connsCheckedOut  int // net number of connections checked out during test execution
@@ -493,6 +494,11 @@ func (t *T) AuthEnabled() bool {
 	return testContext.authEnabled
 }
 
+// SSLEnabled returns whether or not this test is running in an environment with SSL.
+func (t *T) SSLEnabled() bool {
+	return testContext.sslEnabled
+}
+
 // TopologyKind returns the topology kind of the environment
 func (t *T) TopologyKind() TopologyKind {
 	return testContext.topoKind
@@ -662,6 +668,9 @@ func (t *T) shouldSkip() bool {
 	if t.auth != nil && *t.auth != testContext.authEnabled {
 		return true
 	}
+	if t.ssl != nil && *t.ssl != testContext.sslEnabled {
+		return true
+	}
 	if t.enterprise != nil && *t.enterprise != testContext.enterpriseServer {
 		return true
 	}

mongo/integration/mtest/options.go

@@ -184,6 +184,15 @@ func (op *Options) Auth(auth bool) *Options {
 	return op
 }
 
+// SSL specifies whether or not SSL should be enabled for this test to run. By default, a test will run regardless
+// of whether or not SSL is enabled.
+func (op *Options) SSL(ssl bool) *Options {
+	op.optFuncs = append(op.optFuncs, func(t *T) {
+		t.ssl = &ssl
+	})
+	return op
+}
+
 // Enterprise specifies whether or not this test should only be run on enterprise server variants. Defaults to false.
 func (op *Options) Enterprise(ent bool) *Options {
 	op.optFuncs = append(op.optFuncs, func(t *T) {

mongo/integration/mtest/setup.go

@@ -42,6 +42,7 @@ var testContext struct {
 	client           *mongo.Client // client used for setup and teardown
 	serverVersion    string
 	authEnabled      bool
+	sslEnabled       bool
 	enterpriseServer bool
 }
 
@@ -121,7 +122,8 @@ func Setup() error {
 		}
 	}
 
-	testContext.authEnabled = len(os.Getenv("MONGO_GO_DRIVER_CA_FILE")) != 0
+	testContext.authEnabled = os.Getenv("AUTH") == "auth"
+	testContext.sslEnabled = os.Getenv("SSL") == "ssl"
 	biRes, err := testContext.client.Database("admin").RunCommand(Background, bson.D{{"buildInfo", 1}}).DecodeBytes()
 	if err != nil {
 		return fmt.Errorf("buildInfo error: %v", err)

mongo/integration/sessions_test.go

@@ -8,6 +8,7 @@ package integration
 
 import (
 	"bytes"
+	"os"
 	"reflect"
 	"testing"
 	"time"
@@ -18,6 +19,7 @@ import (
 	"go.mongodb.org/mongo-driver/mongo/integration/mtest"
 	"go.mongodb.org/mongo-driver/mongo/options"
 	"go.mongodb.org/mongo-driver/mongo/readpref"
+	"go.mongodb.org/mongo-driver/x/mongo/driver/drivertest"
 	"go.mongodb.org/mongo-driver/x/mongo/driver/session"
 )
 
@@ -113,6 +115,53 @@ func TestSessions(t *testing.T) {
 			})
 		}
 	})
+
+	clusterTimeDialer := newProxyDialer()
+	hosts := options.Client().ApplyURI(mt.ConnString()).Hosts
+	clusterTimeHandshakeOpts := options.Client().
+		SetDialer(clusterTimeDialer).
+		SetHosts(hosts[:1]).
+		SetDirect(true)
+	clusterTimeHandshakeMtOpts := mtest.NewOptions().
+		ClientOptions(clusterTimeHandshakeOpts).
+		CreateCollection(false).
+		SSL(false) // The proxy dialer doesn't work for SSL connections.
+	mt.RunOpts("cluster time is updated from handshakes", clusterTimeHandshakeMtOpts, func(mt *mtest.T) {
+		// Compression uses a different opcode that the existing drivertest helpers can't handle and doesn't affect
+		// the functionality tested here, so we can skip the test if compression is enabled.
+		if len(os.Getenv("MONGO_GO_DRIVER_COMPRESSOR")) > 0 {
+			mt.Skip("skipping for compression")
+		}
+
+		err := mt.Client.Ping(mtest.Background, mtest.PrimaryRp)
+		assert.Nil(mt, err, "Ping error: %v", err)
+
+		msgPairs := clusterTimeDialer.messages
+		for idx, pair := range msgPairs {
+			// Get the command sent to the server.
+			cmd, err := drivertest.GetCommandFromQueryWireMessage(pair.sent)
+			if err != nil {
+				cmd, err = drivertest.GetCommandFromMsgWireMessage(pair.sent)
+			}
+			if err != nil {
+				mt.Fatalf("error reading command document from wire message: %v", err)
+			}
+
+			// Get the $clusterTime value sent to the server. The first two messages are the handshakes for the
+			// heartbeat and application connections. These should not contain $clusterTime because they happen on
+			// connections that don't know the server's wire version and therefore don't know if the server supports
+			// $clusterTime.
+			_, err = cmd.LookupErr("$clusterTime")
+			if idx <= 1 {
+				assert.NotNil(mt, err, "expected no $clusterTime field in command %s", cmd)
+				continue
+			}
+
+			// All messages after the first two should contain $clusterTime.
+			assert.Nil(mt, err, "expected $clusterTime field in command %s", cmd)
+		}
+	})
+
 	mt.RunOpts("explicit implicit session arguments", noClientOpts, func(mt *mtest.T) {
 		// lsid is included in commands with explicit and implicit sessions
 

x/mongo/driver/auth/auth.go

@@ -15,6 +15,7 @@ import (
 	"go.mongodb.org/mongo-driver/x/mongo/driver/address"
 	"go.mongodb.org/mongo-driver/x/mongo/driver/description"
 	"go.mongodb.org/mongo-driver/x/mongo/driver/operation"
+	"go.mongodb.org/mongo-driver/x/mongo/driver/session"
 )
 
 // AuthenticatorFactory constructs an authenticator.
@@ -56,6 +57,7 @@ type HandshakeOptions struct {
 	Compressors           []string
 	DBUser                string
 	PerformAuthentication func(description.Server) bool
+	ClusterClock          *session.ClusterClock
 }
 
 type authHandshaker struct {
@@ -74,7 +76,8 @@ func (ah *authHandshaker) GetDescription(ctx context.Context, addr address.Addre
 	op := operation.NewIsMaster().
 		AppName(ah.options.AppName).
 		Compressors(ah.options.Compressors).
-		SASLSupportedMechs(ah.options.DBUser)
+		SASLSupportedMechs(ah.options.DBUser).
+		ClusterClock(ah.options.ClusterClock)
 
 	if ah.options.Authenticator != nil {
 		if speculativeAuth, ok := ah.options.Authenticator.(SpeculativeAuthenticator); ok {
@@ -112,7 +115,13 @@ func (ah *authHandshaker) FinishHandshake(ctx context.Context, conn driver.Conne
 
 	desc := conn.Description()
 	if performAuth(desc) && ah.options.Authenticator != nil {
-		if err := ah.authenticate(ctx, desc, conn); err != nil {
+		cfg := &Config{
+			Description:  desc,
+			Connection:   conn,
+			ClusterClock: ah.options.ClusterClock,
+		}
+
+		if err := ah.authenticate(ctx, cfg); err != nil {
 			return newAuthError("auth error", err)
 		}
 	}
@@ -123,20 +132,20 @@ func (ah *authHandshaker) FinishHandshake(ctx context.Context, conn driver.Conne
 	return ah.wrapped.FinishHandshake(ctx, conn)
 }
 
-func (ah *authHandshaker) authenticate(ctx context.Context, desc description.Server, conn driver.Connection) error {
+func (ah *authHandshaker) authenticate(ctx context.Context, cfg *Config) error {
 	// If the initial isMaster reply included a response to the speculative authentication attempt, we only need to
 	// conduct the remainder of the conversation.
-	if desc.SpeculativeAuthenticate != nil {
+	if speculativeResponse := cfg.Description.SpeculativeAuthenticate; speculativeResponse != nil {
 		// Defensively ensure that the server did not include a response if speculative auth was not attempted.
 		if ah.conversation == nil {
 			return errors.New("speculative auth was not attempted but the server included a response")
 		}
-		return ah.conversation.Finish(ctx, desc.SpeculativeAuthenticate, conn)
+		return ah.conversation.Finish(ctx, cfg, speculativeResponse)
 	}
 
 	// If the server does not support speculative authentication or the first attempt was not successful, we need to
 	// perform authentication from scratch.
-	return ah.options.Authenticator.Auth(ctx, desc, conn)
+	return ah.options.Authenticator.Auth(ctx, cfg)
 }
 
 // Handshaker creates a connection handshaker for the given authenticator.
@@ -147,10 +156,17 @@ func Handshaker(h driver.Handshaker, options *HandshakeOptions) driver.Handshake
 	}
 }
 
+// Config holds the information necessary to perform an authentication attempt.
+type Config struct {
+	Description  description.Server
+	Connection   driver.Connection
+	ClusterClock *session.ClusterClock
+}
+
 // Authenticator handles authenticating a connection.
 type Authenticator interface {
 	// Auth authenticates the connection.
-	Auth(context.Context, description.Server, driver.Connection) error
+	Auth(context.Context, *Config) error
 }
 
 func newAuthError(msg string, inner error) error {

x/mongo/driver/auth/conversation.go

@@ -10,7 +10,6 @@ import (
 	"context"
 
 	"go.mongodb.org/mongo-driver/x/bsonx/bsoncore"
-	"go.mongodb.org/mongo-driver/x/mongo/driver"
 )
 
 // SpeculativeConversation represents an authentication conversation that can be merged with the initial connection
@@ -23,7 +22,7 @@ import (
 // authenticate the provided connection.
 type SpeculativeConversation interface {
 	FirstMessage() (bsoncore.Document, error)
-	Finish(ctx context.Context, firstResponse bsoncore.Document, conn driver.Connection) error
+	Finish(ctx context.Context, cfg *Config, firstResponse bsoncore.Document) error
 }
 
 // SpeculativeAuthenticator represents an authenticator that supports speculative authentication.