Enforce early socket read timeout setting. (#863)

Set socket read timeout (`fs.gs.http.read-timeout`) as early as possible
on new sockets returned from the custom `SSLSocketFactory`. This
guarantees the timeout is enforced during TLS handshakes when using
Conscrypt as the security provider.

See also google/conscrypt#864 .

gcs/CHANGES.md

@@ -74,6 +74,11 @@
 
 1.  Set default value for `fs.gs.list.max.items.per.call` property to `5000`.
 
+1.  Set socket read timeout (`fs.gs.http.read-timeout`) as early as possible on
+    new sockets returned from the custom `SSLSocketFactory`. This guarantees the
+    timeout is enforced during TLS handshakes when using Conscrypt as the
+    security provider.
+
 ### 2.2.2 - 2021-06-25
 
 1.  Support footer prefetch in gRPC read channel.

gcs/src/main/java/com/google/cloud/hadoop/fs/gcs/GoogleHadoopFileSystemConfiguration.java

@@ -23,7 +23,6 @@
 import static com.google.cloud.hadoop.gcsio.GoogleCloudStorageOptions.DEFAULT_GRPC_MESSAGE_TIMEOUT_CHECK_INTERVAL_MILLIS;
 import static com.google.cloud.hadoop.gcsio.GoogleCloudStorageOptions.DEFAULT_TRAFFIC_DIRECTOR_ENABLED;
 import static com.google.cloud.hadoop.gcsio.GoogleCloudStorageOptions.HTTP_REQUEST_CONNECT_TIMEOUT;
-import static com.google.cloud.hadoop.gcsio.GoogleCloudStorageOptions.HTTP_REQUEST_READ_TIMEOUT;
 import static com.google.cloud.hadoop.gcsio.GoogleCloudStorageOptions.MAX_BYTES_REWRITTEN_PER_CALL_DEFAULT;
 import static com.google.cloud.hadoop.gcsio.GoogleCloudStorageOptions.MAX_HTTP_REQUEST_RETRIES;
 import static com.google.cloud.hadoop.gcsio.GoogleCloudStorageOptions.MAX_LIST_ITEMS_PER_CALL_DEFAULT;
@@ -40,6 +39,7 @@
 import static com.google.cloud.hadoop.util.HadoopCredentialsConfiguration.PROXY_ADDRESS_SUFFIX;
 import static com.google.cloud.hadoop.util.HadoopCredentialsConfiguration.PROXY_PASSWORD_SUFFIX;
 import static com.google.cloud.hadoop.util.HadoopCredentialsConfiguration.PROXY_USERNAME_SUFFIX;
+import static com.google.cloud.hadoop.util.HadoopCredentialsConfiguration.READ_TIMEOUT_SUFFIX;
 import static com.google.cloud.hadoop.util.HadoopCredentialsConfiguration.getConfigKeyPrefixes;
 import static com.google.common.base.Strings.nullToEmpty;
 
@@ -255,10 +255,6 @@ public class GoogleHadoopFileSystemConfiguration {
   public static final HadoopConfigurationProperty<Integer> GCS_HTTP_CONNECT_TIMEOUT =
       new HadoopConfigurationProperty<>("fs.gs.http.connect-timeout", HTTP_REQUEST_CONNECT_TIMEOUT);
 
-  /** Configuration key for the connect timeout (in millisecond) for HTTP request to GCS. */
-  public static final HadoopConfigurationProperty<Integer> GCS_HTTP_READ_TIMEOUT =
-      new HadoopConfigurationProperty<>("fs.gs.http.read-timeout", HTTP_REQUEST_READ_TIMEOUT);
-
   /** Configuration key for adding a suffix to the GHFS application name sent to GCS. */
   public static final HadoopConfigurationProperty<String> GCS_APPLICATION_NAME_SUFFIX =
       new HadoopConfigurationProperty<>("fs.gs.application.name.suffix", "");
@@ -478,7 +474,8 @@ static GoogleCloudStorageOptions.Builder getGcsOptionsBuilder(Configuration conf
         .setBatchThreads(GCS_BATCH_THREADS.get(config, config::getInt))
         .setMaxHttpRequestRetries(GCS_HTTP_MAX_RETRY.get(config, config::getInt))
         .setHttpRequestConnectTimeout(GCS_HTTP_CONNECT_TIMEOUT.get(config, config::getInt))
-        .setHttpRequestReadTimeout(GCS_HTTP_READ_TIMEOUT.get(config, config::getInt))
+        .setHttpRequestReadTimeout(
+            READ_TIMEOUT_SUFFIX.withPrefixes(CONFIG_KEY_PREFIXES).get(config, config::getInt))
         .setAppName(getApplicationName(config))
         .setMaxWaitMillisForEmptyObjectCreation(
             GCS_MAX_WAIT_MILLIS_EMPTY_OBJECT_CREATE.get(config, config::getInt))

gcs/src/test/java/com/google/cloud/hadoop/fs/gcs/GoogleHadoopFileSystemConfigurationTest.java

@@ -93,7 +93,6 @@ public class GoogleHadoopFileSystemConfigurationTest {
           put("fs.gs.grpc.write.message.timeout.ms", 3_000L);
           put("fs.gs.http.connect-timeout", 20_000);
           put("fs.gs.http.max.retry", 10);
-          put("fs.gs.http.read-timeout", 20_000);
           put("fs.gs.implicit.dir.repair.enable", true);
           put("fs.gs.inputstream.fadvise", Fadvise.AUTO);
           put("fs.gs.inputstream.fast.fail.on.not.found.enable", true);

gcsio/src/main/java/com/google/cloud/hadoop/gcsio/GoogleCloudStorageImpl.java

@@ -400,7 +400,10 @@ private static Storage createStorage(
       throws IOException {
     HttpTransport httpTransport =
         HttpTransportFactory.createHttpTransport(
-            options.getProxyAddress(), options.getProxyUsername(), options.getProxyPassword());
+            options.getProxyAddress(),
+            options.getProxyUsername(),
+            options.getProxyPassword(),
+            Duration.ofMillis(options.getHttpRequestReadTimeout()));
     return new Storage.Builder(
             httpTransport, GsonFactory.getDefaultInstance(), httpRequestInitializer)
         .setRootUrl(options.getStorageRootUrl())

util-hadoop/src/main/java/com/google/cloud/hadoop/util/HadoopCredentialsConfiguration.java

@@ -33,6 +33,7 @@
 import java.io.IOException;
 import java.io.UncheckedIOException;
 import java.net.URI;
+import java.time.Duration;
 import java.time.Instant;
 import java.util.Date;
 import java.util.List;
@@ -137,6 +138,10 @@ public class HadoopCredentialsConfiguration {
   public static final HadoopConfigurationProperty<RedactedString> PROXY_PASSWORD_SUFFIX =
       new HadoopConfigurationProperty<>(".proxy.password");
 
+  /** Key suffix for setting the read timeout (in milliseconds) for HTTP request. */
+  public static final HadoopConfigurationProperty<Integer> READ_TIMEOUT_SUFFIX =
+      new HadoopConfigurationProperty<>(".http.read-timeout", 20 * 1000);
+
   /**
    * Configuration key for defining the OAUth2 client ID. Required when the authentication type is
    * USER_CREDENTIALS
@@ -313,7 +318,9 @@ private static Supplier<HttpTransport> getHttpTransport(
             return HttpTransportFactory.createHttpTransport(
                 PROXY_ADDRESS_SUFFIX.withPrefixes(keyPrefixes).get(config, config::get),
                 PROXY_USERNAME_SUFFIX.withPrefixes(keyPrefixes).getPassword(config),
-                PROXY_PASSWORD_SUFFIX.withPrefixes(keyPrefixes).getPassword(config));
+                PROXY_PASSWORD_SUFFIX.withPrefixes(keyPrefixes).getPassword(config),
+                Duration.ofMillis(
+                    READ_TIMEOUT_SUFFIX.withPrefixes(keyPrefixes).get(config, config::getInt)));
           } catch (IOException e) {
             throw new UncheckedIOException(e);
           }

util-hadoop/src/test/java/com/google/cloud/hadoop/util/HadoopCredentialsConfigurationTest.java

@@ -72,6 +72,7 @@ public class HadoopCredentialsConfigurationTest {
           put(".auth.refresh.token", null);
           put(".auth.service.account.json.keyfile", null);
           put(".auth.type", AuthenticationType.COMPUTE_ENGINE);
+          put(".http.read-timeout", 20 * 1000);
           put(".proxy.address", null);
           put(".proxy.password", null);
           put(".proxy.username", null);

util/src/main/java/com/google/cloud/hadoop/util/HttpTransportFactory.java

@@ -35,6 +35,7 @@
 import java.net.URI;
 import java.net.URISyntaxException;
 import java.security.GeneralSecurityException;
+import java.time.Duration;
 import javax.annotation.Nullable;
 import javax.net.ssl.HttpsURLConnection;
 import javax.net.ssl.SSLSocketFactory;
@@ -73,8 +74,34 @@ public static HttpTransport createHttpTransport(
       @Nullable RedactedString proxyUsername,
       @Nullable RedactedString proxyPassword)
       throws IOException {
+    return createHttpTransport(proxyAddress, proxyUsername, proxyPassword, /* readTimeout= */ null);
+  }
+
+  /**
+   * Create an {@link HttpTransport} based on a type class, optional HTTP proxy and optional socket
+   * read timeout.
+   *
+   * @param proxyAddress The HTTP proxy to use with the transport. Of the form hostname:port. If
+   *     empty no proxy will be used.
+   * @param proxyUsername The HTTP proxy username to use with the transport. If empty no proxy
+   *     username will be used.
+   * @param proxyPassword The HTTP proxy password to use with the transport. If empty no proxy
+   *     password will be used.
+   * @param readTimeout The socket read timeout to apply immediately on all HTTP requests. If empty,
+   *     no socket read timeout will be applied.
+   * @return The resulting HttpTransport.
+   * @throws IllegalArgumentException If the proxy address is invalid.
+   * @throws IOException If there is an issue connecting to Google's Certification server.
+   */
+  public static HttpTransport createHttpTransport(
+      @Nullable String proxyAddress,
+      @Nullable RedactedString proxyUsername,
+      @Nullable RedactedString proxyPassword,
+      @Nullable Duration readTimeout)
+      throws IOException {
     logger.atFiner().log(
-        "createHttpTransport(%s, %s, %s)", proxyAddress, proxyUsername, proxyPassword);
+        "createHttpTransport(%s, %s, %s, %s)",
+        proxyAddress, proxyUsername, proxyPassword, readTimeout);
     checkArgument(
         proxyAddress != null || (proxyUsername == null && proxyPassword == null),
         "if proxyAddress is null then proxyUsername and proxyPassword should be null too");
@@ -88,7 +115,7 @@ public static HttpTransport createHttpTransport(
               ? new PasswordAuthentication(
                   proxyUsername.value(), proxyPassword.value().toCharArray())
               : null;
-      return createNetHttpTransport(proxyUri, proxyAuth);
+      return createNetHttpTransport(proxyUri, proxyAuth, readTimeout);
     } catch (GeneralSecurityException e) {
       throw new IOException(e);
     }
@@ -99,12 +126,15 @@ public static HttpTransport createHttpTransport(
    *
    * @param proxyUri Optional HTTP proxy URI to use with the transport.
    * @param proxyAuth Optional HTTP proxy credentials to authenticate with the transport proxy.
+   * @param readTimeout Optional socket read timeout to apply immediately on all HTTP requests.
    * @return The resulting HttpTransport.
    * @throws IOException If there is an issue connecting to Google's certification server.
    * @throws GeneralSecurityException If there is a security issue with the keystore.
    */
   public static NetHttpTransport createNetHttpTransport(
-      @Nullable URI proxyUri, @Nullable PasswordAuthentication proxyAuth)
+      @Nullable URI proxyUri,
+      @Nullable PasswordAuthentication proxyAuth,
+      @Nullable Duration readTimeout)
       throws IOException, GeneralSecurityException {
     checkArgument(
         proxyUri != null || proxyAuth == null,
@@ -127,19 +157,20 @@ && getRequestingPort() == proxyUri.getPort()) {
             }
           });
     }
-    return createNetHttpTransportBuilder(proxyUri).build();
+    return createNetHttpTransportBuilder(proxyUri, readTimeout).build();
   }
 
   @VisibleForTesting
-  static NetHttpTransport.Builder createNetHttpTransportBuilder(@Nullable URI proxyUri)
+  static NetHttpTransport.Builder createNetHttpTransportBuilder(
+      @Nullable URI proxyUri, @Nullable Duration readTimeout)
       throws IOException, GeneralSecurityException {
     NetHttpTransport.Builder builder =
         new NetHttpTransport.Builder().trustCertificates(GoogleUtils.getCertificateTrustStore());
+    SSLSocketFactory wrappedSslSocketFactory =
+        requireNonNullElseGet(
+            builder.getSslSocketFactory(), HttpsURLConnection::getDefaultSSLSocketFactory);
     return builder
-        .setSslSocketFactory(
-            new SslKeepAliveSocketFactory(
-                requireNonNullElseGet(
-                    builder.getSslSocketFactory(), HttpsURLConnection::getDefaultSSLSocketFactory)))
+        .setSslSocketFactory(new CustomSslSocketFactory(wrappedSslSocketFactory, readTimeout))
         .setProxy(
             proxyUri == null
                 ? null
@@ -186,12 +217,14 @@ static URI parseProxyAddress(@Nullable String proxyAddress) {
 
   /** Wrapper class to have socketKeepAlive property while creating the socket */
   @VisibleForTesting
-  static class SslKeepAliveSocketFactory extends SSLSocketFactory {
+  static final class CustomSslSocketFactory extends SSLSocketFactory {
 
     private final SSLSocketFactory wrappedSockedFactory;
+    private final Integer readTimeoutMillis;
 
-    public SslKeepAliveSocketFactory(SSLSocketFactory wrappedSocketFactory) {
+    public CustomSslSocketFactory(SSLSocketFactory wrappedSocketFactory, Duration readTimeout) {
       this.wrappedSockedFactory = wrappedSocketFactory;
+      this.readTimeoutMillis = readTimeout != null ? Math.toIntExact(readTimeout.toMillis()) : null;
     }
 
     @Override
@@ -206,44 +239,55 @@ public String[] getSupportedCipherSuites() {
 
     @Override
     public Socket createSocket() throws IOException {
-      return setSocketKeepAlive(wrappedSockedFactory.createSocket());
+      return customizeSocket(wrappedSockedFactory.createSocket());
     }
 
     @Override
     public Socket createSocket(Socket s, InputStream consumed, boolean autoClose)
         throws IOException {
-      return setSocketKeepAlive(wrappedSockedFactory.createSocket(s, consumed, autoClose));
+      return customizeSocket(wrappedSockedFactory.createSocket(s, consumed, autoClose));
     }
 
     @Override
     public Socket createSocket(Socket s, String host, int port, boolean autoClose)
         throws IOException {
-      return setSocketKeepAlive(wrappedSockedFactory.createSocket(s, host, port, autoClose));
+      return customizeSocket(wrappedSockedFactory.createSocket(s, host, port, autoClose));
     }
 
     public Socket createSocket(String host, int port) throws IOException {
-      return setSocketKeepAlive(wrappedSockedFactory.createSocket(host, port));
+      return customizeSocket(wrappedSockedFactory.createSocket(host, port));
     }
 
     public Socket createSocket(InetAddress address, int port) throws IOException {
-      return setSocketKeepAlive(wrappedSockedFactory.createSocket(address, port));
+      return customizeSocket(wrappedSockedFactory.createSocket(address, port));
     }
 
     public Socket createSocket(String host, int port, InetAddress clientAddress, int clientPort)
         throws IOException {
-      return setSocketKeepAlive(
+      return customizeSocket(
           wrappedSockedFactory.createSocket(host, port, clientAddress, clientPort));
     }
 
     public Socket createSocket(
         InetAddress address, int port, InetAddress clientAddress, int clientPort)
         throws IOException {
-      return setSocketKeepAlive(
+      return customizeSocket(
           wrappedSockedFactory.createSocket(address, port, clientAddress, clientPort));
     }
 
-    private static Socket setSocketKeepAlive(Socket socket) throws SocketException {
+    private Socket customizeSocket(Socket socket) throws SocketException {
+      // Enable TCP keep-alive.
       socket.setKeepAlive(true);
+
+      // Set socket read timeout. This shouldn't be necessary, because we generally set the timeout
+      // through other layers, such as com.google.api.client.http.HttpRequest#setReadTimeout(int).
+      // However, setting it here guarantees that the timeout is enforced during TLS handshake when
+      // using Conscrypt as the security provider. (See discussion in
+      // https://github.com/google/conscrypt/issues/864 .)
+      if (readTimeoutMillis != null) {
+        socket.setSoTimeout(readTimeoutMillis);
+      }
+
       return socket;
     }
   }