change tlsVersion to minTlsVersion but keep the tls-version config name

discovery-kubernetes-api/src/main/resources/reference.conf

@@ -17,7 +17,7 @@ pekko.discovery {
     api-service-host-env-name = "KUBERNETES_SERVICE_HOST"
     api-service-port-env-name = "KUBERNETES_SERVICE_PORT"
 
-    # the TLS version to use when connecting to the API server
+    # the minimum TLS version to use when connecting to the API server
     tls-version = "TLSv1.2"
 
     # Namespace discovery path

discovery-kubernetes-api/src/main/scala/org/apache/pekko/discovery/kubernetes/KubernetesApiServiceDiscovery.scala

@@ -231,8 +231,9 @@ class KubernetesApiServiceDiscovery(settings: Settings)(
    * This uses blocking IO, and so should only be used at startup from blocking dispatcher.
    */
   private def clientHttpsConnectionContext(): HttpsConnectionContext = {
-    val sslContext = PemManagersProvider.createSslContext(settings.apiCaPath, settings.tlsVersion)
-    ConnectionContext.httpsClient(sslContext)
+    val sslContext = PemManagersProvider.createSslContext(settings.apiCaPath)
+    ConnectionContext.httpsClient((host, port) =>
+      PemManagersProvider.configureClientEngine(sslContext.createSSLEngine(host, port), settings.minTlsVersion))
   }
 
   /**

discovery-kubernetes-api/src/main/scala/org/apache/pekko/discovery/kubernetes/Settings.scala

@@ -49,7 +49,7 @@ final class Settings(kubernetesApi: Config) extends Extension {
   val apiServicePortEnvName: String =
     kubernetesApi.getString("api-service-port-env-name")
 
-  val tlsVersion: String =
+  val minTlsVersion: String =
     kubernetesApi.getString("tls-version")
 
   val podNamespacePath: String =
@@ -76,7 +76,8 @@ final class Settings(kubernetesApi: Config) extends Extension {
 
   override def toString =
     s"Settings($apiCaPath, $apiTokenPath, $apiServiceHostEnvName, $apiServicePortEnvName, " +
-    s"$podNamespacePath, $podNamespace, $podDomain, httpRequestAcceptEncoding=$httpRequestAcceptEncoding)"
+    s"$podNamespacePath, $podNamespace, $podDomain, minTlsVersion=$minTlsVersion, " +
+    s"httpRequestAcceptEncoding=$httpRequestAcceptEncoding)"
 }
 
 object Settings extends ExtensionId[Settings] with ExtensionIdProvider {

discovery-kubernetes-api/src/test/scala/org/apache/pekko/discovery/kubernetes/SettingsSpec.scala

@@ -26,11 +26,11 @@ import org.scalatest.wordspec.AnyWordSpec
 class SettingsSpec extends AnyWordSpec with Matchers {
 
   "Settings" should {
-    "default tls-version to v1.2" in {
+    "default min-tls-version to v1.2" in {
       val system = ActorSystem("test")
       try {
         val settings = Settings(system)
-        settings.tlsVersion shouldBe "TLSv1.2"
+        settings.minTlsVersion shouldBe "TLSv1.2"
       } finally {
         system.terminate()
       }
@@ -44,7 +44,7 @@ class SettingsSpec extends AnyWordSpec with Matchers {
       val system = ActorSystem("test", config)
       try {
         val settings = Settings(system)
-        settings.tlsVersion shouldBe "TLSv1.3"
+        settings.minTlsVersion shouldBe "TLSv1.3"
       } finally {
         system.terminate()
       }

lease-kubernetes/src/main/resources/reference.conf

@@ -45,7 +45,7 @@ pekko.coordination.lease.kubernetes {
     # set to false for plain text with no auth
     secure-api-server = true
 
-    # the TLS version to use when connecting to the API server
+    # the minimum TLS version to use when connecting to the API server
     tls-version = "TLSv1.2"
 
     # The amount of time to wait for a lease to be acquired or released. This includes all requests to the API

lease-kubernetes/src/main/scala/org/apache/pekko/coordination/lease/kubernetes/KubernetesSettings.scala

@@ -71,7 +71,7 @@ private[pekko] object KubernetesSettings {
       config.getString("namespace-path"),
       apiServerRequestTimeout,
       secure = config.getBoolean("secure-api-server"),
-      tlsVersion = config.getString("tls-version"),
+      minTlsVersion = config.getString("tls-version"),
       bodyReadTimeout = apiServerRequestTimeout / 2,
       tokenRetrySettings = tokenRetrySettings,
       leaseLabelMaxLength = config.getInt("lease-name-max-length"),
@@ -102,7 +102,7 @@ private[pekko] class KubernetesSettings(
     val namespacePath: String,
     val apiServerRequestTimeout: FiniteDuration,
     val secure: Boolean = true,
-    val tlsVersion: String = "TLSv1.2",
+    val minTlsVersion: String = "TLSv1.2",
     val bodyReadTimeout: FiniteDuration = 1.second,
     val tokenRetrySettings: TokenRetrySettings = new TokenRetrySettings(
       5,

lease-kubernetes/src/main/scala/org/apache/pekko/coordination/lease/kubernetes/internal/AbstractKubernetesApiImpl.scala

@@ -51,9 +51,11 @@ import scala.util.control.NonFatal
   private val http: HttpExt = Http()(system)
 
   private lazy val sslContext: SSLContext =
-    PemManagersProvider.createSslContext(settings.apiCaPath, settings.tlsVersion)
+    PemManagersProvider.createSslContext(settings.apiCaPath)
 
-  private lazy val clientSslContext: HttpsConnectionContext = ConnectionContext.httpsClient(sslContext)
+  private lazy val clientSslContext: HttpsConnectionContext =
+    ConnectionContext.httpsClient((host, port) =>
+      PemManagersProvider.configureClientEngine(sslContext.createSSLEngine(host, port), settings.minTlsVersion))
 
   protected val namespace: Future[String] = {
     settings.namespace match {

lease-kubernetes/src/test/scala/org/apache/pekko/coordination/lease/kubernetes/KubernetesSettingsSpec.scala

@@ -41,11 +41,11 @@ class KubernetesSettingsSpec extends AnyWordSpec with Matchers {
            api-server-request-timeout=4s
         """.stripMargin).apiServerRequestTimeout shouldEqual 4.seconds
     }
-    "default tls-version to v1.2" in {
-      conf("").tlsVersion shouldEqual "TLSv1.2"
+    "default min-tls-version to v1.2" in {
+      conf("").minTlsVersion shouldEqual "TLSv1.2"
     }
     "support tls-version override" in {
-      conf("tls-version=TLSv1.3").tlsVersion shouldEqual "TLSv1.3"
+      conf("tls-version=TLSv1.3").minTlsVersion shouldEqual "TLSv1.3"
     }
     "default lease-name-max-length to 63" in {
       conf("").leaseLabelMaxLength shouldEqual 63

management-cluster-bootstrap/src/main/resources/reference.conf

@@ -140,7 +140,7 @@ pekko.management {
         # if this is left empty, the default Java Runtime trust store will be used
         # pekko-management-cluster-bootstrap 1.2.0 did not default to the Java Runtime trust store
         ca-path = ""
-        # the TLS version to use when connecting to contact points
+        # the minimum TLS version to use when connecting to contact points
         tls-version = "TLSv1.2"
       }
     }

management-cluster-bootstrap/src/main/scala/org/apache/pekko/management/cluster/bootstrap/ClusterBootstrapSettings.scala

@@ -139,7 +139,7 @@ final class ClusterBootstrapSettings(config: Config, log: LoggingAdapter) {
       private val httpClientConfig = contactPointConfig.getConfig("http-client")
 
       val caPath: String = httpClientConfig.getString("ca-path")
-      val tlsVersion: String = httpClientConfig.getString("tls-version")
+      val minTlsVersion: String = httpClientConfig.getString("tls-version")
     }
 
     val fallbackPort: Int =

management-cluster-bootstrap/src/main/scala/org/apache/pekko/management/cluster/bootstrap/internal/HttpContactPointBootstrap.scala

@@ -14,10 +14,9 @@
 package org.apache.pekko.management.cluster.bootstrap.internal
 
 import java.time.LocalDateTime
-import java.security.{ KeyStore, SecureRandom }
 import java.util.concurrent.ThreadLocalRandom
 import java.util.concurrent.TimeoutException
-import javax.net.ssl.{ KeyManager, KeyManagerFactory, SSLContext, TrustManager }
+import javax.net.ssl.SSLContext
 import scala.concurrent.{ Future, Promise }
 import scala.concurrent.duration._
 
@@ -67,24 +66,16 @@ private[bootstrap] object HttpContactPointBootstrap {
   private val DefaultTlsVersion = "TLSv1.2" // keep in sync with default in reference.conf
 
   def generateSSLContext(settings: ClusterBootstrapSettings): SSLContext = {
-    val factory = KeyManagerFactory.getInstance(KeyManagerFactory.getDefaultAlgorithm)
-    val keyStore = KeyStore.getInstance("PKCS12")
-    keyStore.load(null)
-    factory.init(keyStore, Array.empty)
-    val km: Array[KeyManager] = factory.getKeyManagers
     val caPath = settings.contactPoint.httpClient.caPath.trim
-    val tm: Array[TrustManager] = if (caPath.isEmpty) {
-      // null means use the default JVM trust store, which is what we want if no CA path is configured
-      None.orNull
-    } else {
-      val certificates = PemManagersProvider.loadCertificates(caPath)
-      PemManagersProvider.buildTrustManagers(certificates)
-    }
-    val tlsVersion = settings.contactPoint.httpClient.tlsVersion.trim
-    val random: SecureRandom = new SecureRandom
-    val sslContext = SSLContext.getInstance(tlsVersion)
-    sslContext.init(km, tm, random)
-    sslContext
+    // an empty ca-path means use the default JVM trust store
+    PemManagersProvider.createSslContext(if (caPath.nonEmpty) Some(caPath) else None)
+  }
+
+  def generateClientConnectionContext(settings: ClusterBootstrapSettings): HttpsConnectionContext = {
+    val sslContext = generateSSLContext(settings)
+    val minTlsVersion = settings.contactPoint.httpClient.minTlsVersion.trim
+    ConnectionContext.httpsClient((host, port) =>
+      PemManagersProvider.configureClientEngine(sslContext.createSSLEngine(host, port), minTlsVersion))
   }
 }
 
@@ -120,10 +111,10 @@ private[bootstrap] class HttpContactPointBootstrap(
 
   private val useCustomSslContext: Boolean =
     settings.contactPoint.httpClient.caPath.trim.nonEmpty ||
-    settings.contactPoint.httpClient.tlsVersion != DefaultTlsVersion
+    settings.contactPoint.httpClient.minTlsVersion != DefaultTlsVersion
 
   private lazy val clientSslContext: HttpsConnectionContext =
-    ConnectionContext.httpsClient(generateSSLContext(settings))
+    generateClientConnectionContext(settings)
 
   private val http = Http()
 

management-cluster-bootstrap/src/test/scala/org/apache/pekko/management/cluster/bootstrap/internal/HttpContactPointBootstrapSpec.scala

@@ -19,6 +19,7 @@ import org.apache.pekko
 import pekko.actor.{ ActorPath, ActorSystem }
 import pekko.event.Logging
 import pekko.management.cluster.bootstrap.ClusterBootstrapSettings
+import pekko.pki.kubernetes.PemManagersProvider
 import pekko.http.scaladsl.model.Uri.Host
 import com.typesafe.config.ConfigFactory
 import org.scalatest.matchers.should.Matchers
@@ -73,7 +74,7 @@ class HttpContactPointBootstrapSpec extends AnyWordSpec with Matchers {
         sys.terminate()
       }
     }
-    "fail to generate SSLContext with bad tls-version" in {
+    "fail with bad tls-version when a connection engine is created" in {
       val sys = ActorSystem("HttpContactPointBootstrapSpec")
       val log = Logging(sys, classOf[HttpContactPointBootstrapSpec])
       try {
@@ -83,10 +84,37 @@ class HttpContactPointBootstrapSpec extends AnyWordSpec with Matchers {
             tls-version = "BAD_VERSION"
           }""").withFallback(sys.settings.config)
         val settings = new ClusterBootstrapSettings(cfg, log)
-        val noSuchAlgorithmException = intercept[java.security.NoSuchAlgorithmException] {
-          HttpContactPointBootstrap.generateSSLContext(settings)
+        // the context builds fine; the version is validated per connection engine
+        HttpContactPointBootstrap.generateClientConnectionContext(settings) should not be null
+        val sslContext = HttpContactPointBootstrap.generateSSLContext(settings)
+        val illegalArgumentException = intercept[IllegalArgumentException] {
+          PemManagersProvider.configureClientEngine(
+            sslContext.createSSLEngine("example.com", 443),
+            settings.contactPoint.httpClient.minTlsVersion.trim)
         }
-        noSuchAlgorithmException.getMessage.contains("BAD_VERSION") should be(true)
+        illegalArgumentException.getMessage.contains("BAD_VERSION") should be(true)
+      } finally {
+        sys.terminate()
+      }
+    }
+
+    "restrict the enabled protocols to the configured minimum" in {
+      val sys = ActorSystem("HttpContactPointBootstrapSpec")
+      val log = Logging(sys, classOf[HttpContactPointBootstrapSpec])
+      try {
+        val cfg = ConfigFactory.parseString(s"""
+          pekko.management.cluster.bootstrap.contact-point.http-client {
+            ca-path = "${userDir}/src/test/files/ca.crt"
+            tls-version = "TLSv1.3"
+          }""").withFallback(sys.settings.config)
+        val settings = new ClusterBootstrapSettings(cfg, log)
+        HttpContactPointBootstrap.generateClientConnectionContext(settings) should not be null
+        val sslContext = HttpContactPointBootstrap.generateSSLContext(settings)
+        val engine = PemManagersProvider.configureClientEngine(
+          sslContext.createSSLEngine("example.com", 443),
+          settings.contactPoint.httpClient.minTlsVersion.trim)
+        engine.getEnabledProtocols.toSeq shouldEqual Seq("TLSv1.3")
+        engine.getSSLParameters.getEndpointIdentificationAlgorithm shouldEqual "https"
       } finally {
         sys.terminate()
       }

management-pki/src/main/scala/org/apache/pekko/pki/kubernetes/PemManagersProvider.scala

@@ -27,7 +27,7 @@ import org.apache.pekko
 import pekko.annotation.InternalApi
 import pekko.pki.pem.{ DERPrivateKeyLoader, PEMDecoder }
 
-import javax.net.ssl.{ KeyManagerFactory, SSLContext, TrustManager, TrustManagerFactory }
+import javax.net.ssl.{ KeyManagerFactory, SSLContext, SSLEngine, TrustManager, TrustManagerFactory }
 
 /**
  * INTERNAL API
@@ -68,20 +68,87 @@ private[pekko] object PemManagersProvider {
     certFactory.generateCertificates(Files.newInputStream(new File(filename).toPath)).asScala
   }
 
+  /**
+   * INTERNAL API
+   *
+   * TLS protocol versions in ascending order of preference. Anything not listed here
+   * (for example SSLv3 or the "SSLv2Hello" pseudo-protocol) is never selected by
+   * [[protocolsAtOrAbove]].
+   */
+  private val TlsVersionOrder = Map("TLSv1" -> 1, "TLSv1.1" -> 2, "TLSv1.2" -> 3, "TLSv1.3" -> 4)
+
+  /**
+   * INTERNAL API
+   *
+   * The subset of `supportedProtocols` that is at or above `minTlsVersion`.
+   *
+   * A minimum version can only be enforced on an `SSLSocket` or `SSLEngine`; an `SSLContext`
+   * has no mutable protocol list, so callers must apply the result via
+   * [[configureClientEngine]] or `SSLSocket.setEnabledProtocols`.
+   *
+   * @throws IllegalArgumentException if `minTlsVersion` is not a known TLS version, or if no
+   *                                  supported protocol satisfies it.
+   */
+  @InternalApi def protocolsAtOrAbove(minTlsVersion: String, supportedProtocols: Array[String]): Array[String] = {
+    val minOrder = TlsVersionOrder.getOrElse(
+      minTlsVersion,
+      throw new IllegalArgumentException(
+        s"Unknown TLS version [$minTlsVersion]. Supported values: " +
+        TlsVersionOrder.keys.toSeq.sorted.mkString(", ")))
+    val filtered = supportedProtocols.filter(protocol => TlsVersionOrder.get(protocol).exists(_ >= minOrder))
+    if (filtered.isEmpty)
+      throw new IllegalArgumentException(
+        s"No TLS protocol at or above [$minTlsVersion] is supported. Supported protocols: " +
+        supportedProtocols.mkString(", "))
+    filtered
+  }
+
+  /**
+   * INTERNAL API
+   *
+   * Configures `engine` as a client engine that only negotiates TLS versions at or above
+   * `minTlsVersion`, keeping the "https" endpoint identification algorithm that
+   * `ConnectionContext.httpsClient(sslContext)` would otherwise apply, so hostname
+   * verification is not lost.
+   */
+  @InternalApi def configureClientEngine(engine: SSLEngine, minTlsVersion: String): SSLEngine = {
+    engine.setUseClientMode(true)
+    engine.setEnabledProtocols(protocolsAtOrAbove(minTlsVersion, engine.getSupportedProtocols))
+    val params = engine.getSSLParameters
+    params.setEndpointIdentificationAlgorithm("https")
+    engine.setSSLParameters(params)
+    engine
+  }
+
   /**
    * INTERNAL API
    *
    * Creates an SSLContext that trusts the given CA certificate file, with no client key material.
+   *
+   * The returned context supports every TLS version the JVM enables by default; a minimum
+   * version cannot be pinned on an `SSLContext`, so restrict the connection itself with
+   * [[configureClientEngine]].
+   */
+  @InternalApi def createSslContext(caCertPath: String): SSLContext =
+    createSslContext(Some(caCertPath))
+
+  /**
+   * INTERNAL API
+   *
+   * Creates an SSLContext with no client key material. If `caCertPath` is `Some(path)` the CA
+   * certificates in that file are trusted, otherwise the default JVM trust store is used.
    */
-  @InternalApi def createSslContext(caCertPath: String, tlsVersion: String): SSLContext = {
-    val certificates = loadCertificates(caCertPath)
+  @InternalApi def createSslContext(caCertPath: Option[String]): SSLContext = {
+    val tm = caCertPath match {
+      case Some(path) => buildTrustManagers(loadCertificates(path))
+      case None       => null // null means use the default JVM trust store
+    }
     val factory = KeyManagerFactory.getInstance(KeyManagerFactory.getDefaultAlgorithm)
     val ks = KeyStore.getInstance("PKCS12")
     ks.load(null)
     factory.init(ks, Array.empty)
     val km = factory.getKeyManagers
-    val tm = buildTrustManagers(certificates)
-    val sslContext = SSLContext.getInstance(tlsVersion)
+    val sslContext = SSLContext.getInstance("TLS")
     sslContext.init(km, tm, new SecureRandom)
     sslContext
   }