GH-1319: [JDBC] Add hostnameOverride connection property for TLS certificate verification and SNI

docs/source/flight_sql_jdbc_driver.rst

@@ -118,6 +118,17 @@ case-sensitive. The supported parameters are:
      - Path to PEM-encoded root certificates for TLS - use this as
        an alternative to ``trustStore``
 
+   * - hostnameOverride
+     - null
+     - When TLS is enabled, the hostname to verify the server certificate
+       against and to send for SNI and as the gRPC ``:authority`` header,
+       instead of **HOSTNAME**. Use this when
+       connecting through a tunnel, a private endpoint, or an IP address
+       that does not match the server certificate. Endpoints returned by
+       the server are only verified against this hostname if they are on
+       the same host as the connection. Ignored when ``useEncryption`` is
+       false.
+
    * - clientCertificate
      - null
      - Path to PEM-encoded client mTLS certificate when the Flight

flight/flight-sql-jdbc-core/src/main/java/org/apache/arrow/driver/jdbc/ArrowFlightConnection.java

@@ -119,6 +119,7 @@ private static ArrowFlightSqlClientHandler createNewClientHandler(
           .withBufferAllocator(allocator)
           .withEncryption(config.useEncryption())
           .withDisableCertificateVerification(config.getDisableCertificateVerification())
+          .withHostnameOverride(config.getHostnameOverride())
           .withToken(config.getToken())
           .withCallOptions(config.toCallOption())
           .withRetainCookies(config.retainCookies())

flight/flight-sql-jdbc-core/src/main/java/org/apache/arrow/driver/jdbc/client/ArrowFlightSqlClientHandler.java

@@ -188,6 +188,7 @@ public List<CloseableEndpointStreamPair> getStreams(final FlightInfo flightInfo)
                     .withHost(endpointUri.getHost())
                     .withPort(endpointUri.getPort())
                     .withEncryption(endpointUri.getScheme().equals(LocationSchemes.GRPC_TLS))
+                    .withHostnameOverride(builder.getHostnameOverrideForEndpoint(endpointUri))
                     .withClientCache(flightClientCache)
                     .withConnectTimeout(builder.connectTimeout);
 
@@ -677,6 +678,8 @@ public static final class Builder {
 
     @VisibleForTesting boolean disableCertificateVerification;
 
+    @VisibleForTesting @Nullable String hostnameOverride;
+
     @VisibleForTesting boolean useSystemTrustStore = true;
 
     @VisibleForTesting String tlsRootCertificatesPath;
@@ -732,6 +735,7 @@ public Builder() {}
       this.token = original.token;
       this.useEncryption = original.useEncryption;
       this.disableCertificateVerification = original.disableCertificateVerification;
+      this.hostnameOverride = original.hostnameOverride;
       this.useSystemTrustStore = original.useSystemTrustStore;
       this.tlsRootCertificatesPath = original.tlsRootCertificatesPath;
       this.clientCertificatePath = original.clientCertificatePath;
@@ -840,6 +844,42 @@ public Builder withDisableCertificateVerification(
       return this;
     }
 
+    /**
+     * Sets the hostname to use for TLS certificate verification and SNI instead of the host this
+     * handler connects to. It is also sent as the gRPC authority ({@code :authority} header).
+     * Ignored when encryption is disabled.
+     *
+     * @param hostnameOverride the hostname to verify the server certificate against, or null or
+     *     blank to use the host.
+     * @return this instance.
+     */
+    public Builder withHostnameOverride(@Nullable final String hostnameOverride) {
+      final String trimmed = hostnameOverride == null ? null : hostnameOverride.trim();
+      this.hostnameOverride = trimmed == null || trimmed.isEmpty() ? null : trimmed;
+      return this;
+    }
+
+    /**
+     * Gets the hostname override to use when connecting to an endpoint location. The override is
+     * only valid for the host it was configured for, so endpoints on any other host are verified
+     * against their own hostname. Like the override itself, it is ignored when the original
+     * connection does not use encryption.
+     *
+     * @param endpointUri the endpoint location.
+     * @return the hostname override if the endpoint is on the original host, otherwise null.
+     */
+    @VisibleForTesting
+    @Nullable String getHostnameOverrideForEndpoint(final URI endpointUri) {
+      final String endpointHost = endpointUri.getHost();
+      if (!useEncryption
+          || hostnameOverride == null
+          || endpointHost == null
+          || !endpointHost.equalsIgnoreCase(host)) {
+        return null;
+      }
+      return hostnameOverride;
+    }
+
     /**
      * Sets whether to use the certificates from the operating system.
      *
@@ -1070,6 +1110,10 @@ public ArrowFlightSqlClientHandler build() throws SQLException {
         clientBuilder.location(location);
 
         if (useEncryption) {
+          if (hostnameOverride != null) {
+            clientBuilder.overrideHostname(hostnameOverride);
+          }
+
           if (disableCertificateVerification) {
             clientBuilder.verifyServer(false);
           } else {

flight/flight-sql-jdbc-core/src/main/java/org/apache/arrow/driver/jdbc/utils/ArrowFlightConnectionConfigImpl.java

@@ -112,6 +112,15 @@ public String getTlsRootCertificatesPath() {
     return ArrowFlightConnectionProperty.TLS_ROOT_CERTS.getString(properties);
   }
 
+  /**
+   * Gets the hostname to use for TLS certificate verification and SNI instead of the host.
+   *
+   * @return the hostname override, or null if not set.
+   */
+  public String getHostnameOverride() {
+    return ArrowFlightConnectionProperty.HOSTNAME_OVERRIDE.getString(properties);
+  }
+
   public String getClientCertificatePath() {
     return ArrowFlightConnectionProperty.CLIENT_CERTIFICATE.getString(properties);
   }
@@ -258,6 +267,7 @@ public enum ArrowFlightConnectionProperty implements ConnectionProperty {
     TRUST_STORE_PASSWORD("trustStorePassword", null, Type.STRING, false),
     USE_SYSTEM_TRUST_STORE("useSystemTrustStore", true, Type.BOOLEAN, false),
     TLS_ROOT_CERTS("tlsRootCerts", null, Type.STRING, false),
+    HOSTNAME_OVERRIDE("hostnameOverride", null, Type.STRING, false),
     CLIENT_CERTIFICATE("clientCertificate", null, Type.STRING, false),
     CLIENT_KEY("clientKey", null, Type.STRING, false),
     THREAD_POOL_SIZE("threadPoolSize", 1, Type.NUMBER, false),

flight/flight-sql-jdbc-core/src/test/java/org/apache/arrow/driver/jdbc/ConnectionTlsHostnameOverrideTest.java

@@ -0,0 +1,238 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You 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 org.apache.arrow.driver.jdbc;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertThrows;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+import static org.junit.jupiter.api.Assertions.fail;
+
+import java.net.URLEncoder;
+import java.nio.charset.StandardCharsets;
+import java.security.cert.CertificateException;
+import java.sql.Connection;
+import java.sql.DriverManager;
+import java.sql.ResultSet;
+import java.sql.SQLException;
+import java.sql.Statement;
+import java.util.ArrayDeque;
+import java.util.Arrays;
+import java.util.Deque;
+import java.util.Properties;
+import org.apache.arrow.driver.jdbc.authentication.UserPasswordAuthentication;
+import org.apache.arrow.driver.jdbc.utils.ArrowFlightConnectionConfigImpl.ArrowFlightConnectionProperty;
+import org.apache.arrow.driver.jdbc.utils.FlightSqlTestCertificates;
+import org.apache.arrow.driver.jdbc.utils.MockFlightSqlProducer;
+import org.apache.arrow.driver.jdbc.utils.PartitionedFlightSqlProducer;
+import org.apache.arrow.flight.FlightEndpoint;
+import org.apache.arrow.flight.FlightServer;
+import org.apache.arrow.flight.Location;
+import org.apache.arrow.flight.Ticket;
+import org.apache.arrow.memory.BufferAllocator;
+import org.apache.arrow.memory.RootAllocator;
+import org.apache.arrow.vector.IntVector;
+import org.apache.arrow.vector.VectorSchemaRoot;
+import org.apache.arrow.vector.types.pojo.ArrowType;
+import org.apache.arrow.vector.types.pojo.Field;
+import org.apache.arrow.vector.types.pojo.Schema;
+import org.junit.jupiter.api.AfterAll;
+import org.junit.jupiter.api.BeforeAll;
+import org.junit.jupiter.api.Test;
+
+/**
+ * Tests the hostnameOverride connection property. The test certificate is issued for {@code
+ * localhost} and has no IP address in its subject alternative names, so connecting by IP address
+ * only passes verification when the hostname override is used. Servers bind to the IP address
+ * explicitly so the tests don't depend on how {@code localhost} resolves.
+ */
+public class ConnectionTlsHostnameOverrideTest {
+
+  private static final String IP_ADDRESS = "127.0.0.1";
+  private static final String CERTIFICATE_HOSTNAME = "localhost";
+  // Connecting with credentials performs a handshake, so the TLS connection is made eagerly.
+  private static final String USER = "user1";
+  private static final String PASSWORD = "pass1";
+
+  private static final FlightSqlTestCertificates.CertKeyPair certKey =
+      FlightSqlTestCertificates.exampleTlsCerts().get(0);
+  private static final String tlsRootCertsPath = certKey.cert.getPath();
+
+  private static BufferAllocator allocator;
+  private static FlightServer tlsServer;
+
+  @BeforeAll
+  public static void setUp() throws Exception {
+    DriverManager.registerDriver(new ArrowFlightJdbcDriver());
+    allocator = new RootAllocator(Long.MAX_VALUE);
+    tlsServer =
+        FlightServer.builder(
+                allocator, Location.forGrpcTls(IP_ADDRESS, 0), new MockFlightSqlProducer())
+            .headerAuthenticator(
+                new UserPasswordAuthentication.Builder()
+                    .user(USER, PASSWORD)
+                    .build()
+                    .authenticate())
+            .useTls(certKey.cert, certKey.key)
+            .build()
+            .start();
+  }
+
+  @AfterAll
+  public static void tearDown() throws Exception {
+    tlsServer.close();
+    allocator.close();
+  }
+
+  private static String tlsUrl(final String host, final int port) throws Exception {
+    return String.format(
+        "jdbc:arrow-flight-sql://%s:%d?useEncryption=true&useSystemTrustStore=false&%s=%s",
+        host,
+        port,
+        ArrowFlightConnectionProperty.TLS_ROOT_CERTS.camelName(),
+        URLEncoder.encode(tlsRootCertsPath, "UTF-8"));
+  }
+
+  private static String serverUrl(final String host) throws Exception {
+    return tlsUrl(host, tlsServer.getPort())
+        + String.format("&user=%s&password=%s", USER, PASSWORD);
+  }
+
+  /** Asserts that connecting fails because the server certificate was rejected. */
+  private static void assertCertificateRejected(final String url) {
+    final SQLException exception =
+        assertThrows(
+            SQLException.class,
+            () -> {
+              try (Connection connection = DriverManager.getConnection(url)) {
+                connection.isValid(0);
+              }
+            });
+    final Deque<Throwable> pending = new ArrayDeque<>();
+    pending.add(exception);
+    while (!pending.isEmpty()) {
+      final Throwable throwable = pending.pop();
+      if (throwable instanceof CertificateException) {
+        return;
+      }
+      if (throwable.getCause() != null) {
+        pending.add(throwable.getCause());
+      }
+      pending.addAll(Arrays.asList(throwable.getSuppressed()));
+    }
+    fail("Expected the server certificate to be rejected", exception);
+  }
+
+  @Test
+  public void testConnectionByIpAddressFailsWithoutHostnameOverride() throws Exception {
+    assertCertificateRejected(serverUrl(IP_ADDRESS));
+  }
+
+  @Test
+  public void testConnectionByIpAddressSucceedsWithHostnameOverrideInUrl() throws Exception {
+    try (Connection connection =
+        DriverManager.getConnection(
+            serverUrl(IP_ADDRESS) + "&hostnameOverride=" + CERTIFICATE_HOSTNAME)) {
+      assertTrue(connection.isValid(0));
+    }
+  }
+
+  @Test
+  public void testConnectionByIpAddressSucceedsWithHostnameOverrideInProperties() throws Exception {
+    final Properties properties = new Properties();
+    properties.setProperty(
+        ArrowFlightConnectionProperty.HOSTNAME_OVERRIDE.camelName(), CERTIFICATE_HOSTNAME);
+
+    try (Connection connection = DriverManager.getConnection(serverUrl(IP_ADDRESS), properties)) {
+      assertTrue(connection.isValid(0));
+    }
+  }
+
+  @Test
+  public void testServerCertificateIsVerifiedAgainstHostnameOverride() throws Exception {
+    // Same connection as the passing override tests, but the override doesn't match the
+    // certificate, so verification must use the override's value.
+    assertCertificateRejected(serverUrl(IP_ADDRESS) + "&hostnameOverride=other.example.com");
+  }
+
+  @Test
+  public void testBlankHostnameOverrideIsIgnored() throws Exception {
+    // A blank value behaves as if the property were unset rather than being sent to gRPC.
+    assertCertificateRejected(serverUrl(IP_ADDRESS) + "&hostnameOverride=");
+  }
+
+  @Test
+  public void testHostnameOverrideIsIgnoredWithoutEncryption() throws Exception {
+    try (FlightServer server =
+            FlightServer.builder(
+                    allocator, Location.forGrpcInsecure(IP_ADDRESS, 0), new MockFlightSqlProducer())
+                .build()
+                .start();
+        Connection connection =
+            DriverManager.getConnection(
+                String.format(
+                    "jdbc:arrow-flight-sql://%s:%d?useEncryption=false&hostnameOverride=%s",
+                    IP_ADDRESS, server.getPort(), "other.example.com"))) {
+      assertTrue(connection.isValid(0));
+    }
+  }
+
+  @Test
+  public void testHostnameOverrideIsAppliedToEndpointOnConnectionHost() throws Exception {
+    final Schema schema =
+        new Schema(
+            Arrays.asList(Field.nullablePrimitive("int_column", new ArrowType.Int(32, true))));
+    final Ticket ticket = new Ticket("partition".getBytes(StandardCharsets.UTF_8));
+
+    try (VectorSchemaRoot partition = VectorSchemaRoot.create(schema, allocator);
+        FlightServer dataServer =
+            FlightServer.builder(
+                    allocator,
+                    Location.forGrpcTls(IP_ADDRESS, 0),
+                    new PartitionedFlightSqlProducer.DataOnlyFlightSqlProducer(ticket, partition))
+                .useTls(certKey.cert, certKey.key)
+                .build()
+                .start()) {
+      partition.setRowCount(1);
+      ((IntVector) partition.getVector(0)).set(0, 42);
+
+      // The endpoint is on the same IP address as the connection, so it must be verified against
+      // the hostname override as well.
+      final FlightEndpoint endpoint =
+          new FlightEndpoint(ticket, Location.forGrpcTls(IP_ADDRESS, dataServer.getPort()));
+
+      try (PartitionedFlightSqlProducer rootProducer =
+              new PartitionedFlightSqlProducer(schema, endpoint);
+          FlightServer rootServer =
+              FlightServer.builder(allocator, Location.forGrpcTls(IP_ADDRESS, 0), rootProducer)
+                  .useTls(certKey.cert, certKey.key)
+                  .build()
+                  .start();
+          Connection connection =
+              DriverManager.getConnection(
+                  tlsUrl(IP_ADDRESS, rootServer.getPort())
+                      + "&hostnameOverride="
+                      + CERTIFICATE_HOSTNAME);
+          Statement statement = connection.createStatement();
+          ResultSet resultSet = statement.executeQuery("SELECT partitioned_data")) {
+        assertTrue(resultSet.next());
+        assertEquals(42, resultSet.getInt(1));
+        assertFalse(resultSet.next());
+      }
+    }
+  }
+}