Support HTTPS downgrade when not using the proxy. Add support for wildcard DNS support for no proxy hosts. (#749) (#751)

wchau · web-flow · commit 1b3c6f59838f · 2025-06-16T23:50:31.000-07:00
diff --git a/client/src/main/scala/io/delta/sharing/client/DeltaSharingFileSystem.scala b/client/src/main/scala/io/delta/sharing/client/DeltaSharingFileSystem.scala
@@ -50,66 +50,30 @@ private[sharing] class DeltaSharingFileSystem extends FileSystem with Logging {
   private[sharing] def createHttpClient() = {
     val proxyConfigOpt = ConfUtils.getProxyConfig(getConf)
     val maxConnections = ConfUtils.maxConnections(getConf)
+    val neverUseHttps = ConfUtils.getNeverUseHttps(getConf)
     val config = RequestConfig.custom()
       .setConnectTimeout(timeoutInSeconds * 1000)
       .setConnectionRequestTimeout(timeoutInSeconds * 1000)
       .setSocketTimeout(timeoutInSeconds * 1000).build()
 
     logDebug(s"Creating delta sharing httpClient with timeoutInSeconds: $timeoutInSeconds.")
-    val clientBuilder = HttpClientBuilder.create()
+    val clientBuilder = DeltaSharingFileSystemHttpClientBuilder.create()
       .setMaxConnTotal(maxConnections)
       .setMaxConnPerRoute(maxConnections)
       .setDefaultRequestConfig(config)
       // Disable the default retry behavior because we have our own retry logic.
       // See `RetryUtils.runWithExponentialBackoff`.
       .disableAutomaticRetries()
 
+    if (neverUseHttps) {
+      clientBuilder.setDisableHttps()
+    }
+
     // Set proxy if provided.
     proxyConfigOpt.foreach { proxyConfig =>
-
       val proxy = new HttpHost(proxyConfig.host, proxyConfig.port)
       clientBuilder.setProxy(proxy)
-
-      val neverUseHttps = ConfUtils.getNeverUseHttps(getConf)
-      if (neverUseHttps) {
-        val httpRequestDowngradeExecutor = new HttpRequestExecutor {
-          override def execute(
-                                request: HttpRequest,
-                                connection: HttpClientConnection,
-                                context: HttpContext): HttpResponse = {
-            try {
-              val modifiedUri: URI = {
-                new URIBuilder(request.getRequestLine.getUri).setScheme("http").build()
-              }
-              val wrappedRequest = new RequestWrapper(request)
-              wrappedRequest.setURI(modifiedUri)
-
-              return super.execute(wrappedRequest, connection, context)
-            } catch {
-              case e: Exception =>
-                logInfo("Failed to downgrade the request to http", e)
-            }
-            super.execute(request, connection, context)
-          }
-        }
-        clientBuilder.setRequestExecutor(httpRequestDowngradeExecutor)
-      }
-      if (proxyConfig.noProxyHosts.nonEmpty || neverUseHttps) {
-        val routePlanner = new DefaultRoutePlanner(DefaultSchemePortResolver.INSTANCE) {
-          override def determineRoute(target: HttpHost,
-                                      request: HttpRequest,
-                                      context: HttpContext): HttpRoute = {
-            if (proxyConfig.noProxyHosts.contains(target.getHostName)) {
-              // Direct route (no proxy)
-              new HttpRoute(target)
-            } else {
-              // Route via proxy
-              new HttpRoute(target, proxy)
-            }
-          }
-        }
-        clientBuilder.setRoutePlanner(routePlanner)
-      }
+      clientBuilder.setNoProxyHosts(proxyConfig.noProxyHosts)
     }
     clientBuilder.build()
   }
diff --git a/client/src/main/scala/io/delta/sharing/client/DeltaSharingFileSystemHttpClient.scala b/client/src/main/scala/io/delta/sharing/client/DeltaSharingFileSystemHttpClient.scala
@@ -0,0 +1,81 @@
+/*
+ * Copyright (2021) The Delta Lake Project Authors.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package io.delta.sharing.client
+
+import java.net.URI
+
+import org.apache.http.{HttpHost, HttpRequest}
+import org.apache.http.client.methods.{CloseableHttpResponse, HttpRequestWrapper}
+import org.apache.http.client.utils.URIBuilder
+import org.apache.http.conn.ClientConnectionManager
+import org.apache.http.impl.client.CloseableHttpClient
+import org.apache.http.params.HttpParams
+import org.apache.http.protocol.HttpContext
+
+private[sharing] case class DeltaSharingFileSystemHttpClient(
+  noProxyHttpClient: CloseableHttpClient,
+  proxyHttpClient: CloseableHttpClient,
+  useProxy: Boolean,
+  noProxyHosts: Seq[String],
+  disableHttps: Boolean) extends CloseableHttpClient {
+
+  private[sharing] def hasNoProxyHostsMatch(host: String): Boolean = {
+    noProxyHosts.exists(record => {
+      // Wildcard DNS records support
+      if (record.startsWith("*.")) {
+        host.endsWith(record.drop(1))
+      } else {
+        host == record
+      }
+    })
+  }
+
+  override protected def doExecute(
+    target: HttpHost,
+    request: HttpRequest,
+    context: HttpContext
+  ): CloseableHttpResponse = {
+    val (updatedTarget, updatedRequest) = if (disableHttps && target.getSchemeName == "https") {
+      val modifiedUri: URI = new URIBuilder(request.getRequestLine.getUri).setScheme("http").build()
+      val wrappedRequest = HttpRequestWrapper.wrap(request)
+      wrappedRequest.setURI(modifiedUri)
+      (new HttpHost(target.getHostName, target.getPort, "http"), wrappedRequest)
+    } else {
+      (target, request)
+    }
+    if (useProxy && !hasNoProxyHostsMatch(target.getHostName)) {
+      proxyHttpClient.execute(updatedTarget, updatedRequest, context)
+    } else {
+      noProxyHttpClient.execute(updatedTarget, updatedRequest, context)
+    }
+  }
+
+  override def close(): Unit = {
+    noProxyHttpClient.close()
+    proxyHttpClient.close()
+  }
+
+  // This is deprecated since 4.3, so overriding with a dummy implementation
+  override def getParams(): HttpParams = {
+    throw new UnsupportedOperationException("getParams")
+  }
+
+  // This is deprecated since 4.3, so overriding with a dummy implementation
+  override def getConnectionManager(): ClientConnectionManager = {
+    throw new UnsupportedOperationException("getConnectionManager")
+  }
+}
diff --git a/client/src/main/scala/io/delta/sharing/client/DeltaSharingFileSystemHttpClientBuilder.scala b/client/src/main/scala/io/delta/sharing/client/DeltaSharingFileSystemHttpClientBuilder.scala
@@ -0,0 +1,85 @@
+/*
+ * Copyright (2021) The Delta Lake Project Authors.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package io.delta.sharing.client
+
+import org.apache.http.HttpHost
+import org.apache.http.client.config.RequestConfig
+import org.apache.http.impl.client.HttpClientBuilder
+
+private[sharing] class DeltaSharingFileSystemHttpClientBuilder {
+  private var noProxyHttpClientClientBuilder: HttpClientBuilder = HttpClientBuilder.create()
+  private var proxyHttpClientClientBuilder: HttpClientBuilder = HttpClientBuilder.create()
+  private var useProxy: Boolean = false
+  private var noProxyHosts: Seq[String] = Seq.empty
+  private var disableHttps: Boolean = false
+
+  def setMaxConnTotal(maxConnTotal: Int): DeltaSharingFileSystemHttpClientBuilder = {
+    noProxyHttpClientClientBuilder.setMaxConnTotal(maxConnTotal)
+    proxyHttpClientClientBuilder.setMaxConnTotal(maxConnTotal)
+    this
+  }
+
+  def setMaxConnPerRoute(maxConnPerRoute: Int): DeltaSharingFileSystemHttpClientBuilder = {
+    noProxyHttpClientClientBuilder.setMaxConnPerRoute(maxConnPerRoute)
+    proxyHttpClientClientBuilder.setMaxConnPerRoute(maxConnPerRoute)
+    this
+  }
+
+  def setDefaultRequestConfig(config: RequestConfig): DeltaSharingFileSystemHttpClientBuilder = {
+    noProxyHttpClientClientBuilder.setDefaultRequestConfig(config)
+    proxyHttpClientClientBuilder.setDefaultRequestConfig(config)
+    this
+  }
+
+  def disableAutomaticRetries(): DeltaSharingFileSystemHttpClientBuilder = {
+    noProxyHttpClientClientBuilder.disableAutomaticRetries()
+    proxyHttpClientClientBuilder.disableAutomaticRetries()
+    this
+  }
+
+  def setProxy(proxy: HttpHost): DeltaSharingFileSystemHttpClientBuilder = {
+    proxyHttpClientClientBuilder.setProxy(proxy)
+    useProxy = true
+    this
+  }
+
+  def setNoProxyHosts(noProxyHosts: Seq[String]): DeltaSharingFileSystemHttpClientBuilder = {
+    this.noProxyHosts = noProxyHosts
+    this
+  }
+
+  def setDisableHttps(): DeltaSharingFileSystemHttpClientBuilder = {
+    disableHttps = true
+    this
+  }
+
+  def build(): DeltaSharingFileSystemHttpClient = {
+    DeltaSharingFileSystemHttpClient(
+      noProxyHttpClient = noProxyHttpClientClientBuilder.build(),
+      proxyHttpClient = proxyHttpClientClientBuilder.build(),
+      useProxy = useProxy,
+      noProxyHosts = noProxyHosts,
+      disableHttps = disableHttps
+    )
+  }
+}
+
+private[sharing] object DeltaSharingFileSystemHttpClientBuilder {
+  def create(): DeltaSharingFileSystemHttpClientBuilder = {
+    new DeltaSharingFileSystemHttpClientBuilder
+  }
+}
diff --git a/client/src/main/scala/io/delta/sharing/client/util/ConfUtils.scala b/client/src/main/scala/io/delta/sharing/client/util/ConfUtils.scala
@@ -48,6 +48,10 @@ object ConfUtils {
   val TIMEOUT_CONF = "spark.delta.sharing.network.timeout"
   val TIMEOUT_DEFAULT = "320s"
 
+  // Note: There is a separate pool for proxy and non-proxy connections.
+  //       Thus, if you use both spark.delta.sharing.network.proxyHost and
+  //       spark.delta.sharing.network.noProxyHosts,
+  //       the max number of connections will be 2 * spark.delta.sharing.network.maxConnections.
   val MAX_CONNECTION_CONF = "spark.delta.sharing.network.maxConnections"
   val MAX_CONNECTION_DEFAULT = 64
 
diff --git a/client/src/test/scala/io/delta/sharing/client/DeltaSharingFileSystemHttpClientSuite.scala b/client/src/test/scala/io/delta/sharing/client/DeltaSharingFileSystemHttpClientSuite.scala
@@ -0,0 +1,84 @@
+/*
+ * Copyright (2021) The Delta Lake Project Authors.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package io.delta.sharing.client
+
+import org.apache.http.impl.client.HttpClients
+import org.apache.spark.SparkFunSuite
+
+class DeltaSharingFileSystemHttpClientSuite extends SparkFunSuite {
+  /**
+   * Helper method to create a test DeltaSharingFileSystemHttpClient with mock dependencies
+   */
+  private def createTestHttpClient(
+    noProxyHosts: Seq[String] = Seq.empty): DeltaSharingFileSystemHttpClient = {
+    val mockHttpClient = HttpClients.createDefault()
+    DeltaSharingFileSystemHttpClient(
+      noProxyHttpClient = mockHttpClient,
+      proxyHttpClient = mockHttpClient,
+      useProxy = true,
+      noProxyHosts = noProxyHosts,
+      disableHttps = false
+    )
+  }
+
+  test("hasNoProxyHostsMatch - exact host matches") {
+    val client = createTestHttpClient(noProxyHosts = Seq("example.com", "localhost", "127.0.0.1"))
+
+    assert(client.hasNoProxyHostsMatch("example.com"))
+    assert(client.hasNoProxyHostsMatch("localhost"))
+    assert(client.hasNoProxyHostsMatch("127.0.0.1"))
+    assert(!client.hasNoProxyHostsMatch("different.com"))
+    assert(!client.hasNoProxyHostsMatch("sub.example.com"))
+  }
+
+  test("hasNoProxyHostsMatch - wildcard DNS matches") {
+    val client = createTestHttpClient(noProxyHosts = Seq("*.example.com", "*.internal"))
+
+    // Should match wildcard patterns
+    assert(client.hasNoProxyHostsMatch("api.example.com"))
+    assert(client.hasNoProxyHostsMatch("sub.example.com"))
+    assert(client.hasNoProxyHostsMatch("deep.nested.example.com"))
+    assert(client.hasNoProxyHostsMatch("service.internal"))
+
+    // Should NOT match the wildcard domain itself
+    assert(!client.hasNoProxyHostsMatch("example.com"))
+    assert(!client.hasNoProxyHostsMatch("internal"))
+
+    // Should NOT match different domains
+    assert(!client.hasNoProxyHostsMatch("example.org"))
+    assert(!client.hasNoProxyHostsMatch("notexample.com"))
+    assert(!client.hasNoProxyHostsMatch("external"))
+  }
+
+  test("hasNoProxyHostsMatch - empty noProxyHosts list") {
+    val client = createTestHttpClient(noProxyHosts = Seq.empty)
+
+    assert(!client.hasNoProxyHostsMatch("example.com"))
+    assert(!client.hasNoProxyHostsMatch("localhost"))
+    assert(!client.hasNoProxyHostsMatch("127.0.0.1"))
+  }
+
+  test("hasNoProxyHostsMatch - edge cases") {
+    // These edge cases may not be correct, but for simplicity,
+    // we will allow some of these behaviors.
+    val client = createTestHttpClient(noProxyHosts = Seq("*.", "*.168.1.1"))
+
+    assert(!client.hasNoProxyHostsMatch("example.com"))
+    assert(client.hasNoProxyHostsMatch("."))
+    assert(client.hasNoProxyHostsMatch("192.168.1.1"))
+  }
+}
diff --git a/client/src/test/scala/io/delta/sharing/client/DeltaSharingFileSystemSuite.scala b/client/src/test/scala/io/delta/sharing/client/DeltaSharingFileSystemSuite.scala
