SERVER-20651 do not allow mongos connect to mongod if mongod does not support COMMAND protocol

15 files changed, 144 insertions, 36 deletions

diff --git a/src/mongo/client/dbclient.cpp b/src/mongo/client/dbclient.cpp
index 66a7abba021..11f650abccb 100644
--- a/src/mongo/client/dbclient.cpp
+++ b/src/mongo/client/dbclient.cpp
@@ -926,6 +926,15 @@ Status DBClientConnection::connect(const HostAndPort& serverAddress) {
 
     _setServerRPCProtocols(swProtocolSet.getValue());
 
+    auto negotiatedProtocol =
+        rpc::negotiate(getServerRPCProtocols(),
+                       rpc::computeProtocolSet(WireSpec::instance().minWireVersionOutgoing,
+                                               WireSpec::instance().maxWireVersionOutgoing));
+
+    if (!negotiatedProtocol.isOK()) {
+        return negotiatedProtocol.getStatus();
+    }
+
     if (_hook) {
         auto validationStatus = _hook(swIsMasterReply.getValue());
         if (!validationStatus.isOK()) {
diff --git a/src/mongo/db/db.cpp b/src/mongo/db/db.cpp
index 1d5e3ac9967..1f0d32385bb 100644
--- a/src/mongo/db/db.cpp
+++ b/src/mongo/db/db.cpp
@@ -95,6 +95,7 @@
 #include "mongo/db/storage/wiredtiger/wiredtiger_customization_hooks.h"
 #include "mongo/db/storage/storage_options.h"
 #include "mongo/db/ttl.h"
+#include "mongo/db/wire_version.h"
 #include "mongo/executor/network_interface_factory.h"
 #include "mongo/platform/process_id.h"
 #include "mongo/scripting/engine.h"
@@ -405,9 +406,21 @@ static void repairDatabasesAndCheckVersion(OperationContext* txn) {
     LOG(1) << "done repairDatabases" << endl;
 }
 
+static void _initWireSpec() {
+    WireSpec& spec = WireSpec::instance();
+    // accept from any version
+    spec.minWireVersionIncoming = RELEASE_2_4_AND_BEFORE;
+    spec.maxWireVersionIncoming = FIND_COMMAND;
+    // connect to any version
+    spec.minWireVersionOutgoing = RELEASE_2_4_AND_BEFORE;
+    spec.maxWireVersionOutgoing = FIND_COMMAND;
+}
+
+
 static void _initAndListen(int listenPort) {
     Client::initThread("initandlisten");
 
+    _initWireSpec();
     getGlobalServiceContext()->setOpObserver(stdx::make_unique<OpObserver>());
 
     const repl::ReplSettings& replSettings = repl::getGlobalReplicationCoordinator()->getSettings();
diff --git a/src/mongo/db/dbdirectclient.cpp b/src/mongo/db/dbdirectclient.cpp
index 8842c1c4a96..a36df61ffe5 100644
--- a/src/mongo/db/dbdirectclient.cpp
+++ b/src/mongo/db/dbdirectclient.cpp
@@ -90,11 +90,14 @@ std::string DBDirectClient::getServerAddress() const {
     return "localhost";  // TODO: should this have the port?
 }
 
+// Returned version should match the incoming connections restrictions.
 int DBDirectClient::getMinWireVersion() {
-    return kMinWireVersion;
+    return WireSpec::instance().minWireVersionIncoming;
 }
+
+// Returned version should match the incoming connections restrictions.
 int DBDirectClient::getMaxWireVersion() {
-    return kMaxWireVersion;
+    return WireSpec::instance().maxWireVersionIncoming;
 }
 
 bool DBDirectClient::callRead(Message& toSend, Message& response) {
diff --git a/src/mongo/db/repl/replication_info.cpp b/src/mongo/db/repl/replication_info.cpp
index 9f8ac2fad0b..273deaa3bd9 100644
--- a/src/mongo/db/repl/replication_info.cpp
+++ b/src/mongo/db/repl/replication_info.cpp
@@ -237,8 +237,8 @@ public:
         result.appendNumber("maxMessageSizeBytes", MaxMessageSizeBytes);
         result.appendNumber("maxWriteBatchSize", BatchedCommandRequest::kMaxWriteBatchSize);
         result.appendDate("localTime", jsTime());
-        result.append("maxWireVersion", kMaxWireVersion);
-        result.append("minWireVersion", kMinWireVersion);
+        result.append("maxWireVersion", WireSpec::instance().maxWireVersionIncoming);
+        result.append("minWireVersion", WireSpec::instance().minWireVersionIncoming);
         return true;
     }
 } cmdismaster;
diff --git a/src/mongo/db/s/set_shard_version_command.cpp b/src/mongo/db/s/set_shard_version_command.cpp
index b21c5214b0b..4606720cb33 100644
--- a/src/mongo/db/s/set_shard_version_command.cpp
+++ b/src/mongo/db/s/set_shard_version_command.cpp
@@ -144,8 +144,8 @@ public:
 
             // TODO: SERVER-21397 remove post v3.3.
             // Send back wire version to let mongos know what protocol we can speak
-            result.append("minWireVersion", kMinWireVersion);
-            result.append("maxWireVersion", kMaxWireVersion);
+            result.append("minWireVersion", WireSpec::instance().minWireVersionIncoming);
+            result.append("maxWireVersion", WireSpec::instance().maxWireVersionIncoming);
 
             return true;
         }
diff --git a/src/mongo/db/wire_version.h b/src/mongo/db/wire_version.h
index 0fc729915cf..13667ce808d 100644
--- a/src/mongo/db/wire_version.h
+++ b/src/mongo/db/wire_version.h
@@ -60,12 +60,30 @@ enum WireVersion {
     FIND_COMMAND = 4,
 };
 
-// Latest version that the server accepts. This should always be at the latest entry in
-// WireVersion.
-static const int kMaxWireVersion = FIND_COMMAND;
+struct WireSpec {
+    MONGO_DISALLOW_COPYING(WireSpec);
+
+    static WireSpec& instance() {
+        static WireSpec instance;
+        return instance;
+    }
+
+    // Minimum version that the server accepts on incoming requests. We should bump this whenever
+    // we don't want to allow incoming connections from clients that are too old.
+    int minWireVersionIncoming;
+    // Latest version that the server accepts on incoming requests. This should always be at the
+    // latest entry in WireVersion.
+    int maxWireVersionIncoming;
+
+    // Minimum version allowed on remote nodes when the server sends requests. We should bump this
+    // whenever we don't want to connect to clients that are too old.
+    int minWireVersionOutgoing;
+    // Latest version allowed on remote nodes when the server sends requests.
+    int maxWireVersionOutgoing;
+
+private:
+    WireSpec() = default;
+};
 
-// Minimum version that the server accepts. We should bump this whenever we don't want
-// to allow communication with too old agents.
-static const int kMinWireVersion = RELEASE_2_4_AND_BEFORE;
 
 }  // namespace mongo
diff --git a/src/mongo/dbtests/dbtests.cpp b/src/mongo/dbtests/dbtests.cpp
index 1d80f0bc77c..54779f3a6a6 100644
--- a/src/mongo/dbtests/dbtests.cpp
+++ b/src/mongo/dbtests/dbtests.cpp
@@ -41,6 +41,7 @@
 #include "mongo/db/db_raii.h"
 #include "mongo/db/service_context_d.h"
 #include "mongo/db/service_context.h"
+#include "mongo/db/wire_version.h"
 #include "mongo/db/repl/replication_coordinator_global.h"
 #include "mongo/db/repl/replication_coordinator_mock.h"
 #include "mongo/dbtests/framework.h"
@@ -56,6 +57,16 @@ namespace dbtests {
 // This specifies default dbpath for our testing framework
 const std::string default_test_dbpath = "/tmp/unittest";
 
+void initWireSpec() {
+    WireSpec& spec = WireSpec::instance();
+    // accept from any version
+    spec.minWireVersionIncoming = RELEASE_2_4_AND_BEFORE;
+    spec.maxWireVersionIncoming = FIND_COMMAND;
+    // connect to any version
+    spec.minWireVersionOutgoing = RELEASE_2_4_AND_BEFORE;
+    spec.maxWireVersionOutgoing = FIND_COMMAND;
+}
+
 Status createIndex(OperationContext* txn, StringData ns, const BSONObj& keys, bool unique) {
     BSONObjBuilder specBuilder;
     specBuilder << "name" << DBClientBase::genIndexName(keys) << "ns" << ns << "key" << keys;
@@ -100,6 +111,7 @@ int dbtestsMain(int argc, char** argv, char** envp) {
     static StaticObserver StaticObserver;
     Command::testCommandsEnabled = true;
     ::mongo::setupSynchronousSignalHandlers();
+    mongo::dbtests::initWireSpec();
     mongo::runGlobalInitializersOrDie(argc, argv, envp);
     repl::ReplSettings replSettings;
     replSettings.oplogSize = 10 * 1024 * 1024;
diff --git a/src/mongo/executor/network_interface_asio.h b/src/mongo/executor/network_interface_asio.h
index 049c14192da..7c01ad281e3 100644
--- a/src/mongo/executor/network_interface_asio.h
+++ b/src/mongo/executor/network_interface_asio.h
@@ -158,7 +158,10 @@ private:
         std::unique_ptr<AsyncStreamInterface> _stream;
 
         rpc::ProtocolSet _serverProtocols;
-        rpc::ProtocolSet _clientProtocols{rpc::supports::kAll};
+
+        // Dynamically initialized from [min max]WireVersionOutgoing.
+        // Its expected that isMaster response is checked only on the caller.
+        rpc::ProtocolSet _clientProtocols{rpc::supports::kNone};
     };
 
     /**
diff --git a/src/mongo/executor/network_interface_asio_auth.cpp b/src/mongo/executor/network_interface_asio_auth.cpp
index 8a7eddad2fa..b9a341d13d6 100644
--- a/src/mongo/executor/network_interface_asio_auth.cpp
+++ b/src/mongo/executor/network_interface_asio_auth.cpp
@@ -90,21 +90,13 @@ void NetworkInterfaceASIO::_runIsMaster(AsyncOp* op) {
 
         auto commandReply = std::move(swCommandReply.getValue());
 
-        if (_hook) {
-            // Run the validation hook.
-            auto validHost = callNoexcept(
-                *_hook, &NetworkConnectionHook::validateHost, op->request().target, commandReply);
-            if (!validHost.isOK()) {
-                return _completeOperation(op, validHost);
-            }
-        }
-
         auto protocolSet = rpc::parseProtocolSetFromIsMasterReply(commandReply.data);
         if (!protocolSet.isOK())
             return _completeOperation(op, protocolSet.getStatus());
 
         op->connection().setServerProtocols(protocolSet.getValue());
 
+        invariant(op->connection().clientProtocols() != rpc::supports::kNone);
         // Set the operation protocol
         auto negotiatedProtocol =
             rpc::negotiate(op->connection().serverProtocols(), op->connection().clientProtocols());
@@ -115,6 +107,15 @@ void NetworkInterfaceASIO::_runIsMaster(AsyncOp* op) {
 
         op->setOperationProtocol(negotiatedProtocol.getValue());
 
+        if (_hook) {
+            // Run the validation hook.
+            auto validHost = callNoexcept(
+                *_hook, &NetworkConnectionHook::validateHost, op->request().target, commandReply);
+            if (!validHost.isOK()) {
+                return _completeOperation(op, validHost);
+            }
+        }
+
         return _authenticate(op);
 
     };
diff --git a/src/mongo/executor/network_interface_asio_connect.cpp b/src/mongo/executor/network_interface_asio_connect.cpp
index 72d4cba8ac1..a90e046c8a2 100644
--- a/src/mongo/executor/network_interface_asio_connect.cpp
+++ b/src/mongo/executor/network_interface_asio_connect.cpp
@@ -36,6 +36,7 @@
 
 #include "mongo/base/system_error.h"
 #include "mongo/config.h"
+#include "mongo/db/wire_version.h"
 #include "mongo/executor/async_stream.h"
 #include "mongo/executor/async_stream_factory.h"
 #include "mongo/executor/async_stream_interface.h"
@@ -50,7 +51,10 @@ using asio::ip::tcp;
 
 NetworkInterfaceASIO::AsyncConnection::AsyncConnection(std::unique_ptr<AsyncStreamInterface> stream,
                                                        rpc::ProtocolSet protocols)
-    : _stream(std::move(stream)), _serverProtocols(protocols) {}
+    : _stream(std::move(stream)),
+      _serverProtocols(protocols),
+      _clientProtocols(rpc::computeProtocolSet(WireSpec::instance().minWireVersionOutgoing,
+                                               WireSpec::instance().maxWireVersionOutgoing)) {}
 
 #if defined(_MSC_VER) && _MSC_VER < 1900
 NetworkInterfaceASIO::AsyncConnection::AsyncConnection(AsyncConnection&& other)
diff --git a/src/mongo/executor/network_interface_asio_test.cpp b/src/mongo/executor/network_interface_asio_test.cpp
index 21f8be0e280..2fe5d5e5e17 100644
--- a/src/mongo/executor/network_interface_asio_test.cpp
+++ b/src/mongo/executor/network_interface_asio_test.cpp
@@ -54,20 +54,32 @@ namespace {
 
 HostAndPort testHost{"localhost", 20000};
 
+void initWireSpecMongoD() {
+    WireSpec& spec = WireSpec::instance();
+    // accept from any version
+    spec.minWireVersionIncoming = RELEASE_2_4_AND_BEFORE;
+    spec.maxWireVersionIncoming = FIND_COMMAND;
+    // connect to any version
+    spec.minWireVersionOutgoing = RELEASE_2_4_AND_BEFORE;
+    spec.maxWireVersionOutgoing = FIND_COMMAND;
+}
+
 // Utility function to use with mock streams
 RemoteCommandResponse simulateIsMaster(RemoteCommandRequest request) {
     ASSERT_EQ(std::string{request.cmdObj.firstElementFieldName()}, "isMaster");
     ASSERT_EQ(request.dbname, "admin");
 
     RemoteCommandResponse response;
-    response.data = BSON("minWireVersion" << mongo::kMinWireVersion << "maxWireVersion"
-                                          << mongo::kMaxWireVersion);
+    response.data = BSON("minWireVersion" << mongo::WireSpec::instance().minWireVersionIncoming
+                                          << "maxWireVersion"
+                                          << mongo::WireSpec::instance().maxWireVersionIncoming);
     return response;
 }
 
 class NetworkInterfaceASIOTest : public mongo::unittest::Test {
 public:
     void setUp() override {
+        initWireSpecMongoD();
         NetworkInterfaceASIO::Options options;
 
         // Use mock timer factory
@@ -557,15 +569,17 @@ TEST_F(NetworkInterfaceASIOConnectionHookTest, ValidateHostInvalid) {
     ConnectEvent{stream}.skip();
 
     // simulate isMaster reply.
-    stream->simulateServer(rpc::Protocol::kOpQuery,
-                           [](RemoteCommandRequest request) -> RemoteCommandResponse {
-                               RemoteCommandResponse response;
-                               response.data = BSON("minWireVersion"
-                                                    << mongo::kMinWireVersion << "maxWireVersion"
-                                                    << mongo::kMaxWireVersion << "TESTKEY"
-                                                    << "TESTVALUE");
-                               return response;
-                           });
+    stream->simulateServer(
+        rpc::Protocol::kOpQuery,
+        [](RemoteCommandRequest request) -> RemoteCommandResponse {
+            RemoteCommandResponse response;
+            response.data =
+                BSON("minWireVersion"
+                     << mongo::WireSpec::instance().minWireVersionIncoming << "maxWireVersion"
+                     << mongo::WireSpec::instance().maxWireVersionIncoming << "TESTKEY"
+                     << "TESTVALUE");
+            return response;
+        });
 
     // we should stop here.
     auto& res = deferred.get();
diff --git a/src/mongo/rpc/protocol.cpp b/src/mongo/rpc/protocol.cpp
index 5bff4b05e4f..3c3cafb6479 100644
--- a/src/mongo/rpc/protocol.cpp
+++ b/src/mongo/rpc/protocol.cpp
@@ -151,5 +151,18 @@ bool supportsWireVersionForOpCommandInMongod(int minWireVersion, int maxWireVers
         (maxWireVersion >= WireVersion::FIND_COMMAND);
 }
 
+ProtocolSet computeProtocolSet(int minWireVersion, int maxWireVersion) {
+    ProtocolSet result = supports::kNone;
+    if (minWireVersion <= maxWireVersion) {
+        if (maxWireVersion >= WireVersion::FIND_COMMAND) {
+            result |= supports::kOpCommandOnly;
+        }
+        if (minWireVersion <= WireVersion::RELEASE_2_4_AND_BEFORE) {
+            result |= supports::kOpQueryOnly;
+        }
+    }
+    return result;
+}
+
 }  // namespace rpc
 }  // namespace mongo
diff --git a/src/mongo/rpc/protocol.h b/src/mongo/rpc/protocol.h
index 4efb5e0b833..643351106e0 100644
--- a/src/mongo/rpc/protocol.h
+++ b/src/mongo/rpc/protocol.h
@@ -106,5 +106,10 @@ StatusWith<ProtocolSet> parseProtocolSetFromIsMasterReply(const BSONObj& isMaste
  */
 bool supportsWireVersionForOpCommandInMongod(int minWireVersion, int maxWireVersion);
 
+/**
+  * Computes supported protocols from wire versions.
+  */
+ProtocolSet computeProtocolSet(int minWireVersion, int maxWireVersion);
+
 }  // namespace rpc
 }  // namespace mongo
diff --git a/src/mongo/s/commands/cluster_is_master_cmd.cpp b/src/mongo/s/commands/cluster_is_master_cmd.cpp
index 32ea059e71e..14a697808be 100644
--- a/src/mongo/s/commands/cluster_is_master_cmd.cpp
+++ b/src/mongo/s/commands/cluster_is_master_cmd.cpp
@@ -74,8 +74,8 @@ public:
 
         // Mongos tries to keep exactly the same version range of the server for which
         // it is compiled.
-        result.append("maxWireVersion", kMaxWireVersion);
-        result.append("minWireVersion", kMinWireVersion);
+        result.append("maxWireVersion", WireSpec::instance().maxWireVersionIncoming);
+        result.append("minWireVersion", WireSpec::instance().minWireVersionIncoming);
 
         return true;
     }
diff --git a/src/mongo/s/server.cpp b/src/mongo/s/server.cpp
index 00614a5f1cc..396eadb1f97 100644
--- a/src/mongo/s/server.cpp
+++ b/src/mongo/s/server.cpp
@@ -56,6 +56,7 @@
 #include "mongo/db/service_context.h"
 #include "mongo/db/service_context_noop.h"
 #include "mongo/db/startup_warnings_common.h"
+#include "mongo/db/wire_version.h"
 #include "mongo/platform/process_id.h"
 #include "mongo/s/balance.h"
 #include "mongo/s/catalog/forwarding_catalog_manager.h"
@@ -198,10 +199,22 @@ static Status initializeSharding(OperationContext* txn) {
     return Status::OK();
 }
 
+static void _initWireSpec() {
+    WireSpec& spec = WireSpec::instance();
+    // accept from any version
+    spec.minWireVersionIncoming = RELEASE_2_4_AND_BEFORE;
+    spec.maxWireVersionIncoming = FIND_COMMAND;
+    // connect to version supporting Find Command only
+    spec.minWireVersionOutgoing = FIND_COMMAND;
+    spec.maxWireVersionOutgoing = FIND_COMMAND;
+}
+
 static ExitCode runMongosServer() {
     Client::initThread("mongosMain");
     printShardingVersionInfo(false);
 
+    _initWireSpec();
+
     // Add sharding hooks to both connection pools - ShardingConnectionHook includes auth hooks
     globalConnPool.addHook(new ShardingConnectionHook(false));
     shardConnectionPool.addHook(new ShardingConnectionHook(true));