diff --git a/bindings/c/test/fdb_c_client_config_tests.py b/bindings/c/test/fdb_c_client_config_tests.py
index 60355e44f2..f78a2b1420 100644
--- a/bindings/c/test/fdb_c_client_config_tests.py
+++ b/bindings/c/test/fdb_c_client_config_tests.py
@@ -14,7 +14,7 @@ from threading import Thread
 import time
 from fdb_version import CURRENT_VERSION, PREV_RELEASE_VERSION, PREV2_RELEASE_VERSION
 from binary_download import FdbBinaryDownloader
-from local_cluster import LocalCluster, PortProvider
+from local_cluster import LocalCluster, PortProvider, TLSConfig
 from test_util import random_alphanum_string
 
 args = None
@@ -36,6 +36,9 @@ class TestCluster(LocalCluster):
     def __init__(
         self,
         version: str,
+        tls_config: TLSConfig = None,
+        mkcert_binary: str = None,
+        disable_server_side_tls: bool = False,
     ):
         self.client_config_tester_bin = Path(args.client_config_tester_bin).resolve()
         assert self.client_config_tester_bin.exists(), "{} does not exist".format(
@@ -46,7 +49,16 @@ class TestCluster(LocalCluster):
         assert self.build_dir.is_dir(), "{} is not a directory".format(args.build_dir)
         self.tmp_dir = self.build_dir.joinpath("tmp", random_alphanum_string(16))
         print("Creating temp dir {}".format(self.tmp_dir), file=sys.stderr)
+
         self.tmp_dir.mkdir(parents=True)
+        if mkcert_binary:
+            self.mkcert_binary = Path(mkcert_binary).resolve()
+        else:
+            self.mkcert_binary = os.path.join(self.build_dir, "bin", "mkcert")
+        assert Path(self.mkcert_binary).exists(), "{} does not exist".format(
+            self.mkcert_binary
+        )
+
         self.version = version
         super().__init__(
             self.tmp_dir,
@@ -54,6 +66,9 @@ class TestCluster(LocalCluster):
             downloader.binary_path(version, "fdbmonitor"),
             downloader.binary_path(version, "fdbcli"),
             1,
+            tls_config=tls_config,
+            mkcert_binary=self.mkcert_binary,
+            disable_server_side_tls=disable_server_side_tls,
         )
         self.set_env_var("LD_LIBRARY_PATH", downloader.lib_dir(version))
 
@@ -97,6 +112,10 @@ class ClientConfigTest:
         self.status_json = None
 
         # Configuration parameters to be set directly as needed
+        self.tls_client_cert_file = None
+        self.tls_client_key_file = None
+        self.tls_client_ca_file = None
+        self.tls_client_disable_plaintext_connection = None
         self.disable_local_client = False
         self.disable_client_bypass = False
         self.ignore_external_client_failures = False
@@ -266,6 +285,18 @@ class ClientConfigTest:
         if self.disable_client_bypass:
             cmd_args += ["--network-option-disable_client_bypass", ""]
 
+        if self.tls_client_cert_file:
+            cmd_args += ["--network-option-tls_cert_path", self.tls_client_cert_file]
+
+        if self.tls_client_key_file:
+            cmd_args += ["--network-option-tls_key_path", self.tls_client_key_file]
+
+        if self.tls_client_ca_file:
+            cmd_args += ["--network-option-tls_ca_path", self.tls_client_ca_file]
+
+        if self.tls_client_disable_plaintext_connection:
+            cmd_args += ["--network-option-tls_disable_plaintext_connection", ""]
+
         if self.external_lib_path is not None:
             cmd_args += ["--external-client-library", self.external_lib_path]
 
@@ -337,6 +368,16 @@ class ClientConfigTests(unittest.TestCase):
     def tearDownClass(cls):
         cls.cluster.tear_down()
 
+    def test_disable_plaintext_connection(self):
+        # Local client only; Plaintext connections are disabled in a plaintext cluster; Timeout Expected
+        test = ClientConfigTest(self)
+        test.print_status = True
+        test.tls_client_disable_plaintext_connection = True
+        test.transaction_timeout = 100
+        test.expected_error = 1031  # Timeout
+        test.exec()
+        test.check_healthy_status(False)
+
     def test_local_client_only(self):
         # Local client only
         test = ClientConfigTest(self)
@@ -581,6 +622,78 @@ class ClientConfigSeparateCluster(unittest.TestCase):
         finally:
             self.cluster.tear_down()
 
+    def test_tls_cluster_tls_client(self):
+        # Test connecting successfully to a TLS-enabled cluster
+        self.cluster = TestCluster(CURRENT_VERSION, tls_config=TLSConfig())
+        self.cluster.setup()
+        try:
+            test = ClientConfigTest(self)
+            test.print_status = True
+            test.tls_client_cert_file = self.cluster.client_cert_file
+            test.tls_client_key_file = self.cluster.client_key_file
+            test.tls_client_ca_file = self.cluster.client_ca_file
+            test.tls_client_disable_plaintext_connection = True
+            test.exec()
+            test.check_healthy_status(True)
+        finally:
+            self.cluster.tear_down()
+
+    def test_plaintext_cluster_tls_client(self):
+        # Test connecting succesfully to a plaintext cluster with a TLS client
+        self.cluster = TestCluster(
+            CURRENT_VERSION, tls_config=TLSConfig(), disable_server_side_tls=True
+        )
+        self.cluster.setup()
+        try:
+            test = ClientConfigTest(self)
+            test.print_status = True
+            test.tls_client_cert_file = self.cluster.client_cert_file
+            test.tls_client_key_file = self.cluster.client_key_file
+            test.tls_client_ca_file = self.cluster.client_ca_file
+            test.exec()
+            test.check_healthy_status(True)
+        finally:
+            self.cluster.tear_down()
+
+    def test_tls_cluster_tls_client_plaintext_disabled(self):
+        # Test connecting successfully to a TLS-enabled cluster with plain-text connections
+        # disabled in a TLS-configured client
+        disable_plaintext_connection = True
+        tls_config = TLSConfig(
+            client_disable_plaintext_connection=disable_plaintext_connection
+        )
+        self.cluster = TestCluster(CURRENT_VERSION, tls_config=tls_config)
+        self.cluster.setup()
+        try:
+            test = ClientConfigTest(self)
+            test.print_status = True
+            test.tls_client_cert_file = self.cluster.client_cert_file
+            test.tls_client_key_file = self.cluster.client_key_file
+            test.tls_client_ca_file = self.cluster.client_ca_file
+            test.tls_client_disable_plaintext_connection = disable_plaintext_connection
+            test.exec()
+            test.check_healthy_status(True)
+        finally:
+            self.cluster.tear_down()
+
+    def test_plaintext_cluster_tls_client_plaintext_connection_disabled(self):
+        # Test connecting succesfully to a plaintext cluster with a TLS-configured client with plaintext connections disabled
+        self.cluster = TestCluster(
+            CURRENT_VERSION, tls_config=TLSConfig(), disable_server_side_tls=True
+        )
+        self.cluster.setup()
+        try:
+            test = ClientConfigTest(self)
+            test.tls_client_cert_file = self.cluster.client_cert_file
+            test.tls_client_key_file = self.cluster.client_key_file
+            test.tls_client_ca_file = self.cluster.client_ca_file
+            test.tls_client_disable_plaintext_connection = True
+            test.transaction_timeout = 100
+            test.expected_error = 1031  # Timeout
+            test.exec()
+        finally:
+            self.cluster.tear_down()
+
 
 # Test client-side tracing
 class ClientTracingTests(unittest.TestCase):
diff --git a/bindings/go/src/fdb/generated.go b/bindings/go/src/fdb/generated.go
index 114e8896b3..fd2203c2fd 100644
--- a/bindings/go/src/fdb/generated.go
+++ b/bindings/go/src/fdb/generated.go
@@ -213,6 +213,11 @@ func (o NetworkOptions) SetTLSPassword(param string) error {
 	return o.setOpt(54, []byte(param))
 }
 
+// Prevent client from connecting to a non-TLS endpoint by throwing network connection failed error.
+func (o NetworkOptions) SetTLSDisablePlaintextConnection() error {
+	return o.setOpt(55, nil)
+}
+
 // Disables the multi-version client API and instead uses the local client directly. Must be set before setting up the network.
 func (o NetworkOptions) SetDisableMultiVersionClientApi() error {
 	return o.setOpt(60, nil)
diff --git a/documentation/sphinx/source/api-common.rst.inc b/documentation/sphinx/source/api-common.rst.inc
index d3e48b7271..4209d0ad4f 100644
--- a/documentation/sphinx/source/api-common.rst.inc
+++ b/documentation/sphinx/source/api-common.rst.inc
@@ -612,6 +612,10 @@
 
     Sets the passphrase for encrypted private key. Password should be set before setting the key for the password to be used.
 
+.. |option-tls-disable-plaintext-connection| replace::
+
+    Disable non-TLS connections from the client, allowing only TLS connections. Plaintext connections will timeout.
+
 .. |option-set-disable-local-client| replace::
 
     Prevents connections through the local client, allowing only connections through externally loaded client libraries.
diff --git a/documentation/sphinx/source/api-python.rst b/documentation/sphinx/source/api-python.rst
index 09bb07327a..b0446ef0e6 100644
--- a/documentation/sphinx/source/api-python.rst
+++ b/documentation/sphinx/source/api-python.rst
@@ -206,6 +206,10 @@ After importing the ``fdb`` module and selecting an API version, you probably wa
 
        |option-tls-password|
 
+    .. method :: fdb.options.set_tls_disable_plaintext_connection()
+
+       |option-tls-disable-plaintext-connection|
+
     .. method :: fdb.options.set_disable_local_client()
 
       |option-set-disable-local-client|
diff --git a/documentation/sphinx/source/tls.rst b/documentation/sphinx/source/tls.rst
index dcb0c2c930..53c22fb49d 100644
--- a/documentation/sphinx/source/tls.rst
+++ b/documentation/sphinx/source/tls.rst
@@ -116,6 +116,8 @@ The value for each setting can be specified in more than one way.  The actual va
 
 For the password, rather than using the command-line option, it is recommended to use the environment variable ``FDB_TLS_PASSWORD``, as command-line options are more visible to other processes running on the same host.
 
+Clients can disable non-TLS or plaintext connections by setting ``--tls-disable-plaintext-connection``.
+
 As with all other command-line options to ``fdbserver``, the TLS settings can be specified in the :ref:`[fdbserver] section of the configuration file <foundationdb-conf-fdbserver>`.
 
 The settings for certificate file, key file, peer verification, password and CA file are interpreted by the software.
diff --git a/fdbcli/fdbcli.actor.cpp b/fdbcli/fdbcli.actor.cpp
index 88d1475017..0c742eece7 100644
--- a/fdbcli/fdbcli.actor.cpp
+++ b/fdbcli/fdbcli.actor.cpp
@@ -922,6 +922,7 @@ struct CLIOptions {
 	std::string tlsVerifyPeers;
 	std::string tlsCAPath;
 	std::string tlsPassword;
+	bool tlsDisablePlainTextConnection = false;
 	uint64_t memLimit = 8uLL << 30;
 
 	std::vector<std::pair<std::string, std::string>> knobs;
@@ -1041,6 +1042,9 @@ struct CLIOptions {
 		case TLSConfig::OPT_TLS_VERIFY_PEERS:
 			tlsVerifyPeers = args.OptionArg();
 			break;
+		case TLSConfig::OPT_TLS_DISABLE_PLAINTEXT_CONNECTION:
+			tlsDisablePlainTextConnection = true;
+			break;
 
 		case OPT_HELP:
 			printProgramUsage(program_name.c_str());
@@ -2409,6 +2413,15 @@ int main(int argc, char** argv) {
 		}
 	}
 
+	if (opt.tlsDisablePlainTextConnection) {
+		try {
+			setNetworkOption(FDBNetworkOptions::TLS_DISABLE_PLAINTEXT_CONNECTION);
+		} catch (Error& e) {
+			fprintf(stderr, "ERROR: cannot disable non-TLS connections (%s)\n", e.what());
+			return 1;
+		}
+	}
+
 	try {
 		setNetworkOption(FDBNetworkOptions::DISABLE_CLIENT_STATISTICS_LOGGING);
 	} catch (Error& e) {
@@ -2423,6 +2436,7 @@ int main(int argc, char** argv) {
 		printf("\tCertificate Path: %s\n", tlsConfig.getCertificatePathSync().c_str());
 		printf("\tKey Path: %s\n", tlsConfig.getKeyPathSync().c_str());
 		printf("\tCA Path: %s\n", tlsConfig.getCAPathSync().c_str());
+		printf("\tPlaintext Connection Disable: %s\n", tlsConfig.getDisablePlainTextConnection() ? "true" : "false");
 		try {
 			LoadedTLSConfig loaded = tlsConfig.loadSync();
 			printf("\tPassword: %s\n", loaded.getPassword().empty() ? "Not configured" : "Exists, but redacted");
diff --git a/fdbclient/NativeAPI.actor.cpp b/fdbclient/NativeAPI.actor.cpp
index 5a9ca300de..740510901c 100644
--- a/fdbclient/NativeAPI.actor.cpp
+++ b/fdbclient/NativeAPI.actor.cpp
@@ -2543,6 +2543,9 @@ void setNetworkOption(FDBNetworkOptions::Option option, Optional<StringRef> valu
 		tlsConfig.clearVerifyPeers();
 		tlsConfig.addVerifyPeers(value.get().toString());
 		break;
+	case FDBNetworkOptions::TLS_DISABLE_PLAINTEXT_CONNECTION:
+		tlsConfig.setDisablePlainTextConnection(true);
+		break;
 	case FDBNetworkOptions::CLIENT_BUGGIFY_ENABLE:
 		enableBuggify(true, BuggifyType::Client);
 		break;
diff --git a/fdbclient/vexillographer/fdb.options b/fdbclient/vexillographer/fdb.options
index 787b1d27cf..5d362cb814 100644
--- a/fdbclient/vexillographer/fdb.options
+++ b/fdbclient/vexillographer/fdb.options
@@ -104,6 +104,8 @@ description is not currently required but encouraged.
     <Option name="TLS_password" code="54"
             paramType="String" paramDescription="key passphrase"
             description="Set the passphrase for encrypted private key. Password should be set before setting the key for the password to be used." />
+    <Option name="TLS_disable_plaintext_connection" code="55"
+            description="Prevent client from connecting to a non-TLS endpoint by throwing network connection failed error." />
     <Option name="disable_multi_version_client_api" code="60"
             description="Disables the multi-version client API and instead uses the local client directly. Must be set before setting up the network." />
     <Option name="callbacks_on_external_threads" code="61"
diff --git a/fdbserver/fdbserver.actor.cpp b/fdbserver/fdbserver.actor.cpp
index 65473f8d84..ba3d478856 100644
--- a/fdbserver/fdbserver.actor.cpp
+++ b/fdbserver/fdbserver.actor.cpp
@@ -1694,6 +1694,9 @@ private:
 			case TLSConfig::OPT_TLS_VERIFY_PEERS:
 				tlsConfig.addVerifyPeers(args.OptionArg());
 				break;
+			case TLSConfig::OPT_TLS_DISABLE_PLAINTEXT_CONNECTION:
+				tlsConfig.setDisablePlainTextConnection(true);
+				break;
 			case OPT_KMS_CONN_DISCOVERY_URL_FILE: {
 				knobs.emplace_back("rest_kms_connector_discover_kms_url_file", args.OptionArg());
 				break;
diff --git a/flow/Net2.actor.cpp b/flow/Net2.actor.cpp
index 4987b80bbe..5a8c451e32 100644
--- a/flow/Net2.actor.cpp
+++ b/flow/Net2.actor.cpp
@@ -1321,7 +1321,8 @@ void Net2::initTLS(ETLSInitState targetState) {
 			    .detail("CertificatePath", tlsConfig.getCertificatePathSync())
 			    .detail("KeyPath", tlsConfig.getKeyPathSync())
 			    .detail("HasPassword", !loaded.getPassword().empty())
-			    .detail("VerifyPeers", boost::algorithm::join(loaded.getVerifyPeers(), "|"));
+			    .detail("VerifyPeers", boost::algorithm::join(loaded.getVerifyPeers(), "|"))
+			    .detail("DisablePlainTextConnection", tlsConfig.getDisablePlainTextConnection());
 			auto loadedTlsConfig = tlsConfig.loadSync();
 			ConfigureSSLContext(loadedTlsConfig, newContext);
 			activeTlsPolicy = makeReference<TLSPolicy>(loadedTlsConfig, onPolicyFailure);
@@ -1787,6 +1788,11 @@ Future<Reference<IConnection>> Net2::connect(NetworkAddress toAddr, tcp::socket*
 		return SSLConnection::connect(&this->reactor.ios, this->sslContextVar.get(), toAddr, existingSocket);
 	}
 
+	if (tlsConfig.getDisablePlainTextConnection()) {
+		TraceEvent(SevError, "PlainTextConnectionDisabled").detail("toAddr", toAddr);
+		throw connection_failed();
+	}
+
 	return Connection::connect(&this->reactor.ios, toAddr);
 }
 
diff --git a/flow/TLSConfig.actor.cpp b/flow/TLSConfig.actor.cpp
index 591960b724..52eafe267c 100644
--- a/flow/TLSConfig.actor.cpp
+++ b/flow/TLSConfig.actor.cpp
@@ -42,6 +42,7 @@ TLSPolicy::~TLSPolicy() {}
 #include <sstream>
 #include <utility>
 #include <boost/asio/ssl/context.hpp>
+#include <boost/lexical_cast.hpp>
 
 #include "flow/Platform.h"
 #include "flow/IAsyncFile.h"
@@ -64,6 +65,10 @@ std::vector<std::string> LoadedTLSConfig::getVerifyPeers() const {
 	return { "Check.Valid=1" };
 }
 
+bool LoadedTLSConfig::getDisablePlainTextConnection() const {
+	return tlsDisablePlainTextConnection;
+}
+
 std::string LoadedTLSConfig::getPassword() const {
 	if (tlsPassword.size()) {
 		return tlsPassword;
@@ -211,6 +216,22 @@ std::string TLSConfig::getCAPathSync() const {
 	return envCAPath;
 }
 
+bool TLSConfig::getDisablePlainTextConnection() const {
+	std::string envDisablePlainTextConnection;
+	if (platform::getEnvironmentVar("FDB_TLS_DISABLE_PLAINTEXT_CONNECTION", envDisablePlainTextConnection)) {
+		try {
+			return boost::lexical_cast<bool>(envDisablePlainTextConnection);
+		} catch (boost::bad_lexical_cast& e) {
+			fprintf(stderr,
+			        "Warning: Ignoring invalid FDB_TLS_DISABLE_PLAINTEXT_CONNECTION [%s]: %s\n",
+			        envDisablePlainTextConnection.c_str(),
+			        e.what());
+		}
+	}
+
+	return tlsDisablePlainTextConnection;
+}
+
 LoadedTLSConfig TLSConfig::loadSync() const {
 	LoadedTLSConfig loaded;
 
@@ -253,6 +274,7 @@ LoadedTLSConfig TLSConfig::loadSync() const {
 	loaded.tlsPassword = tlsPassword;
 	loaded.tlsVerifyPeers = tlsVerifyPeers;
 	loaded.endpointType = endpointType;
+	loaded.tlsDisablePlainTextConnection = tlsDisablePlainTextConnection;
 
 	return loaded;
 }
@@ -327,6 +349,7 @@ ACTOR Future<LoadedTLSConfig> TLSConfig::loadAsync(const TLSConfig* self) {
 	loaded.tlsPassword = self->tlsPassword;
 	loaded.tlsVerifyPeers = self->tlsVerifyPeers;
 	loaded.endpointType = self->endpointType;
+	loaded.tlsDisablePlainTextConnection = self->tlsDisablePlainTextConnection;
 
 	return loaded;
 }
diff --git a/flow/include/flow/TLSConfig.actor.h b/flow/include/flow/TLSConfig.actor.h
index b80ac7a7ee..14f2e4db99 100644
--- a/flow/include/flow/TLSConfig.actor.h
+++ b/flow/include/flow/TLSConfig.actor.h
@@ -96,6 +96,8 @@ public:
 	// If no environment setting exists, return an empty string
 	std::string getPassword() const;
 
+	bool getDisablePlainTextConnection() const;
+
 	TLSEndpointType getEndpointType() const { return endpointType; }
 
 	bool isTLSEnabled() const { return endpointType != TLSEndpointType::UNSET; }
@@ -109,6 +111,7 @@ private:
 	std::string tlsCertBytes, tlsKeyBytes, tlsCABytes;
 	std::string tlsPassword;
 	std::vector<std::string> tlsVerifyPeers;
+	bool tlsDisablePlainTextConnection;
 	TLSEndpointType endpointType = TLSEndpointType::UNSET;
 
 	friend class TLSConfig;
@@ -125,7 +128,8 @@ public:
 		OPT_TLS_KEY,
 		OPT_TLS_VERIFY_PEERS,
 		OPT_TLS_CA_FILE,
-		OPT_TLS_PASSWORD
+		OPT_TLS_PASSWORD,
+		OPT_TLS_DISABLE_PLAINTEXT_CONNECTION
 	};
 
 	TLSConfig() = default;
@@ -161,6 +165,8 @@ public:
 		tlsCAPath = "";
 	}
 
+	void setDisablePlainTextConnection(const bool val) { tlsDisablePlainTextConnection = val; }
+
 	void setPassword(const std::string& password) { tlsPassword = password; }
 
 	void clearVerifyPeers() { tlsVerifyPeers.clear(); }
@@ -187,6 +193,8 @@ public:
 	std::string getKeyPathSync() const;
 	std::string getCAPathSync() const;
 
+	bool getDisablePlainTextConnection() const;
+
 private:
 	ACTOR static Future<LoadedTLSConfig> loadAsync(const TLSConfig* self);
 	template <typename T>
@@ -195,6 +203,7 @@ private:
 	std::string tlsCertPath, tlsKeyPath, tlsCAPath;
 	std::string tlsCertBytes, tlsKeyBytes, tlsCABytes;
 	std::string tlsPassword;
+	bool tlsDisablePlainTextConnection = false;
 	std::vector<std::string> tlsVerifyPeers;
 	TLSEndpointType endpointType = TLSEndpointType::UNSET;
 };
@@ -251,14 +260,16 @@ public:
 #define TLS_VERIFY_PEERS_FLAG "--tls-verify-peers"
 #define TLS_CA_FILE_FLAG "--tls-ca-file"
 #define TLS_PASSWORD_FLAG "--tls-password"
+#define TLS_DISABLE_PLAINTEXT_CONNECTION_FLAG "--tls-disable-plaintext-connection"
 
 #define TLS_OPTION_FLAGS                                                                                               \
 	{ TLSConfig::OPT_TLS_PLUGIN, TLS_PLUGIN_FLAG, SO_REQ_SEP },                                                        \
 	    { TLSConfig::OPT_TLS_CERTIFICATES, TLS_CERTIFICATE_FILE_FLAG, SO_REQ_SEP },                                    \
 	    { TLSConfig::OPT_TLS_KEY, TLS_KEY_FILE_FLAG, SO_REQ_SEP },                                                     \
 	    { TLSConfig::OPT_TLS_VERIFY_PEERS, TLS_VERIFY_PEERS_FLAG, SO_REQ_SEP },                                        \
-	    { TLSConfig::OPT_TLS_PASSWORD, TLS_PASSWORD_FLAG, SO_REQ_SEP }, {                                              \
-		TLSConfig::OPT_TLS_CA_FILE, TLS_CA_FILE_FLAG, SO_REQ_SEP                                                       \
+	    { TLSConfig::OPT_TLS_PASSWORD, TLS_PASSWORD_FLAG, SO_REQ_SEP },                                                \
+	    { TLSConfig::OPT_TLS_CA_FILE, TLS_CA_FILE_FLAG, SO_REQ_SEP }, {                                                \
+		TLSConfig::OPT_TLS_DISABLE_PLAINTEXT_CONNECTION, TLS_DISABLE_PLAINTEXT_CONNECTION_FLAG, SO_NONE                \
 	}
 
 #define TLS_HELP                                                                                                       \
@@ -274,7 +285,9 @@ public:
 	"                 The passphrase of encrypted private key\n"                                                       \
 	"  " TLS_VERIFY_PEERS_FLAG " CONSTRAINTS\n"                                                                        \
 	"                 The constraints by which to validate TLS peers. The contents\n"                                  \
-	"                 and format of CONSTRAINTS are plugin-specific.\n"
+	"                 and format of CONSTRAINTS are plugin-specific.\n"                                                \
+	"  " TLS_DISABLE_PLAINTEXT_CONNECTION_FLAG "\n"                                                                    \
+	"                 Disable non-TLS connections. All plaintext connection attempts will timeout.\n"
 
 #include "flow/unactorcompiler.h"
 #endif
diff --git a/tests/TestRunner/local_cluster.py b/tests/TestRunner/local_cluster.py
index f891834df5..14ed7d6656 100644
--- a/tests/TestRunner/local_cluster.py
+++ b/tests/TestRunner/local_cluster.py
@@ -79,11 +79,13 @@ class TLSConfig:
         self,
         server_chain_len: int = 3,
         client_chain_len: int = 2,
-        verify_peers="Check.Valid=1",
+        verify_peers: str = "Check.Valid=1",
+        client_disable_plaintext_connection: bool = False,
     ):
         self.server_chain_len = server_chain_len
         self.client_chain_len = client_chain_len
         self.verify_peers = verify_peers
+        self.client_disable_plaintext_connection = client_disable_plaintext_connection
 
 
 class LocalCluster:
@@ -154,6 +156,7 @@ knob_min_trace_severity=5
         custom_config: dict = {},
         authorization_kty: str = "",
         authorization_keypair_id: str = "",
+        disable_server_side_tls: bool = False,
     ):
         self.port_provider = PortProvider()
         self.basedir = Path(basedir)
@@ -202,6 +205,7 @@ knob_min_trace_severity=5
         self.use_legacy_conf_syntax = False
         self.coordinators = set()
         self.active_servers = set(self.server_ports.keys())
+        self.disable_server_side_tls = disable_server_side_tls
         self.tls_config = tls_config
         self.public_key_jwks_str = None
         self.public_key_json_file = None
@@ -287,7 +291,7 @@ knob_min_trace_severity=5
                     encrypt_config=encrypt_config,
                     tls_config=self.tls_conf_string(),
                     authz_public_key_config=self.authz_public_key_conf_string(),
-                    optional_tls=":tls" if self.tls_config is not None else "",
+                    optional_tls=self.tls_optional_string(),
                     custom_config="\n".join(
                         [
                             "{} = {}".format(key, value)
@@ -333,7 +337,7 @@ knob_min_trace_severity=5
             secret=self.cluster_secret,
             ip_addr=self.ip_address,
             server_port=self.server_ports[0],
-            optional_tls=":tls" if self.tls_config else "",
+            optional_tls=self.tls_optional_string(),
         )
         return conn_str
 
@@ -405,6 +409,8 @@ knob_min_trace_severity=5
                 "--tls-ca-file",
                 self.server_ca_file,
             ]
+            if self.tls_config.client_disable_plaintext_connection:
+                args += ["--tls-disable-plaintext-connection"]
         if self.use_future_protocol_version:
             args += ["--use-future-protocol-version"]
         res = subprocess.run(
@@ -476,7 +482,7 @@ knob_min_trace_severity=5
 
     # Materialize server's TLS configuration section
     def tls_conf_string(self):
-        if self.tls_config is None:
+        if self.tls_config is None or self.disable_server_side_tls:
             return ""
         else:
             conf_map = {
@@ -487,6 +493,12 @@ knob_min_trace_severity=5
             }
             return "\n".join("{} = {}".format(k, v) for k, v in conf_map.items())
 
+    def tls_optional_string(self):
+        if self.tls_config is None or self.disable_server_side_tls:
+            return ""
+        else:
+            return ":tls"
+
     def authz_public_key_conf_string(self):
         if self.public_key_json_file is not None:
             return "authorization-public-key-file = {}".format(