feat(NODE-3689): require hello command for connection handshake to use OP_MSG disallowing OP_QUERY

src/sdam/topology.ts

@@ -382,6 +382,10 @@ export class Topology extends TypedEventEmitter<TopologyEvents> {
     return this.s.options.loadBalanced;
   }
 
+  get serverApi(): ServerApi | undefined {
+    return this.s.options.serverApi;
+  }
+
   get capabilities(): ServerCapabilities {
     return new ServerCapabilities(this.lastHello());
   }

src/utils.ts

@@ -371,9 +371,10 @@ export function uuidV4(): Buffer {
  */
 export function maxWireVersion(topologyOrServer?: Connection | Topology | Server): number {
   if (topologyOrServer) {
-    if (topologyOrServer.loadBalanced) {
-      // Since we do not have a monitor, we assume the load balanced server is always
-      // pointed at the latest mongodb version. There is a risk that for on-prem
+    if (topologyOrServer.loadBalanced || topologyOrServer.serverApi?.version) {
+      // Since we do not have a monitor, we assume
+      // when a server API version or load-balanced topology are requested
+      // the server is always pointed at the latest mongodb version. There is a risk that for on-prem
       // deployments that don't upgrade immediately that this could alert to the
       // application that a feature is available that is actually not.
       return MAX_SUPPORTED_WIRE_VERSION;

test/integration/change-streams/change_stream.test.ts

@@ -1262,7 +1262,9 @@ describe('Change Streams', function () {
           }
           req.reply({ ok: 1 });
         });
-        const client = this.configuration.newClient(`mongodb://${mockServer.uri()}/`);
+        const client = this.configuration.newClient(`mongodb://${mockServer.uri()}/`, {
+          serverApi: null // TODO(NODE-3807): remove resetting serverApi when the usage of mongodb mock server is removed
+        });
         client.connect(err => {
           expect(err).to.not.exist;
           const collection = client.db('cs').collection('test');

test/integration/change-streams/change_streams.prose.test.ts

@@ -332,7 +332,10 @@ describe('Change Stream prose tests', function () {
           }
           request.reply(this.applyOpTime(response));
         });
-        this.client = this.config.newClient(this.mongodbURI, { monitorCommands: true });
+        this.client = this.config.newClient(this.mongodbURI, {
+          monitorCommands: true,
+          serverApi: null // TODO(NODE-3807): remove resetting serverApi when the usage of mongodb mock server is removed
+        });
         this.apm = { started: [], succeeded: [], failed: [] };
 
         (

test/integration/collection-management/collection.test.ts

@@ -474,7 +474,9 @@ describe('Collection', function () {
     afterEach(() => mock.cleanup());
 
     function testCountDocMock(testConfiguration, config, done) {
-      const client = testConfiguration.newClient(`mongodb://${server.uri()}/test`);
+      const client = testConfiguration.newClient(`mongodb://${server.uri()}/test`, {
+        serverApi: null // TODO(NODE-3807): remove resetting serverApi when the usage of mongodb mock server is removed
+      });
       const close = e => client.close(() => done(e));
 
       server.setMessageHandler(request => {

test/integration/max-staleness/max_staleness.test.js

@@ -54,7 +54,8 @@ describe('Max Staleness', function () {
       var self = this;
       const configuration = this.configuration;
       const client = configuration.newClient(
-        `mongodb://${test.server.uri()}/test?readPreference=secondary&maxStalenessSeconds=250`
+        `mongodb://${test.server.uri()}/test?readPreference=secondary&maxStalenessSeconds=250`,
+        { serverApi: null } // TODO(NODE-3807): remove resetting serverApi when the usage of mongodb mock server is removed
       );
 
       client.connect(function (err, client) {
@@ -86,7 +87,9 @@ describe('Max Staleness', function () {
 
     test: function (done) {
       const configuration = this.configuration;
-      const client = configuration.newClient(`mongodb://${test.server.uri()}/test`);
+      const client = configuration.newClient(`mongodb://${test.server.uri()}/test`, {
+        serverApi: null // TODO(NODE-3807): remove resetting serverApi when the usage of mongodb mock server is removed
+      });
       client.connect(function (err, client) {
         expect(err).to.not.exist;
 
@@ -124,7 +127,9 @@ describe('Max Staleness', function () {
       test: function (done) {
         var self = this;
         const configuration = this.configuration;
-        const client = configuration.newClient(`mongodb://${test.server.uri()}/test`);
+        const client = configuration.newClient(`mongodb://${test.server.uri()}/test`, {
+          serverApi: null // TODO(NODE-3807): remove resetting serverApi when the usage of mongodb mock server is removed
+        });
         client.connect(function (err, client) {
           expect(err).to.not.exist;
           var db = client.db(self.configuration.db);
@@ -159,7 +164,9 @@ describe('Max Staleness', function () {
     test: function (done) {
       var self = this;
       const configuration = this.configuration;
-      const client = configuration.newClient(`mongodb://${test.server.uri()}/test`);
+      const client = configuration.newClient(`mongodb://${test.server.uri()}/test`, {
+        serverApi: null // TODO(NODE-3807): remove resetting serverApi when the usage of mongodb mock server is removed
+      });
       client.connect(function (err, client) {
         expect(err).to.not.exist;
         var db = client.db(self.configuration.db);

test/integration/mongodb-handshake/mongodb-handshake.test.ts

@@ -5,8 +5,12 @@ import * as sinon from 'sinon';
 import {
   Connection,
   LEGACY_HELLO_COMMAND,
+  MessageStream,
   MongoServerError,
-  MongoServerSelectionError
+  MongoServerSelectionError,
+  OpMsgRequest,
+  OpQueryRequest,
+  ServerApiVersion
 } from '../../mongodb';
 
 describe('MongoDB Handshake', () => {
@@ -48,6 +52,7 @@ describe('MongoDB Handshake', () => {
 
   context('when compressors are provided on the mongo client', () => {
     let spy: Sinon.SinonSpy;
+
     before(() => {
       spy = sinon.spy(Connection.prototype, 'command');
     });
@@ -57,9 +62,93 @@ describe('MongoDB Handshake', () => {
     it('constructs a handshake with the specified compressors', async function () {
       client = this.configuration.newClient({ compressors: ['snappy'] });
       await client.connect();
+      // The load-balanced mode doesn’t perform SDAM,
+      // so `connect` doesn’t do anything unless authentication is enabled.
+      // Force the driver to send a command to the server in the noauth mode.
+      await client.db('admin').command({ ping: 1 });
       expect(spy.called).to.be.true;
       const handshakeDoc = spy.getCall(0).args[1];
       expect(handshakeDoc).to.have.property('compression').to.deep.equal(['snappy']);
     });
   });
+
+  context('when load-balanced', function () {
+    let writeCommandSpy: Sinon.SinonSpy;
+
+    beforeEach(() => {
+      writeCommandSpy = sinon.spy(MessageStream.prototype, 'writeCommand');
+    });
+
+    afterEach(() => sinon.restore());
+
+    it('should send the hello command as OP_MSG', {
+      metadata: { requires: { topology: 'load-balanced' } },
+      test: async function () {
+        client = this.configuration.newClient({ loadBalanced: true });
+        await client.connect();
+        // The load-balanced mode doesn’t perform SDAM,
+        // so `connect` doesn’t do anything unless authentication is enabled.
+        // Force the driver to send a command to the server in the noauth mode.
+        await client.db('admin').command({ ping: 1 });
+        expect(writeCommandSpy).to.have.been.called;
+        expect(writeCommandSpy.firstCall.args[0] instanceof OpMsgRequest).to.equal(true);
+      }
+    });
+  });
+
+  context('when serverApi version is present', function () {
+    let writeCommandSpy: Sinon.SinonSpy;
+
+    beforeEach(() => {
+      writeCommandSpy = sinon.spy(MessageStream.prototype, 'writeCommand');
+    });
+
+    afterEach(() => sinon.restore());
+
+    it('should send the hello command as OP_MSG', {
+      metadata: { requires: { topology: '!load-balanced', mongodb: '>=5.0' } },
+      test: async function () {
+        client = this.configuration.newClient({}, { serverApi: { version: ServerApiVersion.v1 } });
+        await client.connect();
+        // The load-balanced mode doesn’t perform SDAM,
+        // so `connect` doesn’t do anything unless authentication is enabled.
+        // Force the driver to send a command to the server in the noauth mode.
+        await client.db('admin').command({ ping: 1 });
+        expect(writeCommandSpy).to.have.been.called;
+        expect(writeCommandSpy.firstCall.args[0] instanceof OpMsgRequest).to.equal(true);
+      }
+    });
+  });
+
+  context('when not load-balanced and serverApi version is not present', function () {
+    let writeCommandSpy: Sinon.SinonSpy;
+
+    beforeEach(() => {
+      writeCommandSpy = sinon.spy(MessageStream.prototype, 'writeCommand');
+    });
+
+    afterEach(() => sinon.restore());
+
+    it('should send the hello command as OP_MSG', {
+      metadata: { requires: { topology: '!load-balanced', mongodb: '>=5.0' } },
+      test: async function () {
+        if (this.configuration.serverApi) {
+          this.skipReason = 'Test requires serverApi to NOT be enabled';
+          return this.skip();
+        }
+        client = this.configuration.newClient();
+        await client.connect();
+        // The load-balanced mode doesn’t perform SDAM,
+        // so `connect` doesn’t do anything unless authentication is enabled.
+        // Force the driver to send a command to the server in the noauth mode.
+        await client.db('admin').command({ ping: 1 });
+        expect(writeCommandSpy).to.have.been.called;
+
+        const opRequests = writeCommandSpy.getCalls().map(items => items.args[0]);
+        expect(opRequests[0] instanceof OpQueryRequest).to.equal(true);
+        const isOpMsgRequestSent = !!opRequests.find(op => op instanceof OpMsgRequest);
+        expect(isOpMsgRequestSent).to.equal(true);
+      }
+    });
+  });
 });

test/readme.md

@@ -330,7 +330,7 @@ The following steps will walk you through how to start and test a load balancer.
    A new file name `lb.env` is automatically created.
 1. Source the environment variables using a command like `source lb.env`.
 1. Export **each** of the environment variables that were created in `lb.env`. For example: `export SINGLE_MONGOS_LB_URI`.
-1. Export the `LOAD_BALANCED` environment variable to 'true': `export LOAD_BALANCED='true'`
+1. Export the `LOAD_BALANCER` environment variable to 'true': `export LOAD_BALANCER='true'`
 1. Disable auth for tests: `export AUTH='noauth'`
 1. Run the test suite as you normally would:
    ```sh

test/unit/cmap/connection.test.ts

@@ -1135,4 +1135,109 @@ describe('new Connection()', function () {
       expect(writeCommandSpy.firstCall.args[0] instanceof OpQueryRequest).to.equal(true);
     });
   });
+
+  describe('when serverApi version is present', () => {
+    const CONNECT_DEFAULTS = {
+      id: 1,
+      tls: false,
+      generation: 1,
+      monitorCommands: false,
+      metadata: {} as ClientMetadata
+    };
+    let server;
+    let connectOptions;
+    let connection: Connection;
+    let writeCommandSpy;
+
+    beforeEach(async () => {
+      server = await mock.createServer();
+      server.setMessageHandler(request => {
+        request.reply(mock.HELLO);
+      });
+      writeCommandSpy = sinon.spy(MessageStream.prototype, 'writeCommand');
+    });
+
+    afterEach(async () => {
+      connection?.destroy({ force: true });
+      sinon.restore();
+      await mock.cleanup();
+    });
+
+    it('sends the first command as OP_MSG', async () => {
+      try {
+        connectOptions = {
+          ...CONNECT_DEFAULTS,
+          hostAddress: server.hostAddress() as HostAddress,
+          socketTimeoutMS: 100,
+          serverApi: { version: '1' }
+        };
+
+        connection = await promisify<Connection>(callback =>
+          //@ts-expect-error: Callbacks do not have mutual exclusion for error/result existence
+          connect(connectOptions, callback)
+        )();
+
+        await promisify(callback =>
+          connection.command(ns('admin.$cmd'), { hello: 1 }, {}, callback)
+        )();
+      } catch (error) {
+        /** Connection timeouts, but the handshake message is sent. */
+      }
+
+      expect(writeCommandSpy).to.have.been.called;
+      expect(writeCommandSpy.firstCall.args[0] instanceof OpMsgRequest).to.equal(true);
+    });
+  });
+
+  describe('when serverApi version is not present', () => {
+    const CONNECT_DEFAULTS = {
+      id: 1,
+      tls: false,
+      generation: 1,
+      monitorCommands: false,
+      metadata: {} as ClientMetadata
+    };
+    let server;
+    let connectOptions;
+    let connection: Connection;
+    let writeCommandSpy;
+
+    beforeEach(async () => {
+      server = await mock.createServer();
+      server.setMessageHandler(request => {
+        request.reply(mock.HELLO);
+      });
+      writeCommandSpy = sinon.spy(MessageStream.prototype, 'writeCommand');
+    });
+
+    afterEach(async () => {
+      connection?.destroy({ force: true });
+      sinon.restore();
+      await mock.cleanup();
+    });
+
+    it('sends the first command as OP_MSG', async () => {
+      try {
+        connectOptions = {
+          ...CONNECT_DEFAULTS,
+          hostAddress: server.hostAddress() as HostAddress,
+          socketTimeoutMS: 100
+        };
+
+        connection = await promisify<Connection>(callback =>
+          //@ts-expect-error: Callbacks do not have mutual exclusion for error/result existence
+          connect(connectOptions, callback)
+        )();
+
+        await promisify(callback =>
+          connection.command(ns('admin.$cmd'), { hello: 1 }, {}, callback)
+        )();
+      } catch (error) {
+        /** Connection timeouts, but the handshake message is sent. */
+      }
+
+      expect(writeCommandSpy).to.have.been.called;
+      expect(writeCommandSpy.firstCall.args[0] instanceof OpQueryRequest).to.equal(true);
+    });
+  });
 });