Add the ability to customize socket creation.

Bug: https://github.com/dart-lang/sdk/issues/42716
Change-Id: I9e854007f15ed54cc2d85a372bc145e1b90f5967
Reviewed-on: https://dart-review.googlesource.com/c/sdk/+/231530
Reviewed-by: Alexander Aprelev <aam@google.com>
Commit-Queue: Brian Quinlan <bquinlan@google.com>

CHANGELOG.md

@@ -25,6 +25,21 @@
 - `IdbFactory.supportsDatabaseNames` has been deprecated. It will always return
   `false`.
 
+#### `dart:io`
+
+- **Breaking Change** [#47887](https://github.com/dart-lang/sdk/issues/47887):
+  `HttpClient` has a new `connectionFactory` property, which allows socket
+  creation to be customized. Classes that `implement HttpClient` may be broken
+  by this change. Add the following method to your classes to fix them:
+
+  ```dart
+  void set connectionFactory(
+      Future<ConnectionTask<Socket>> Function(
+              Uri url, String? proxyHost, int? proxyPort)?
+          f) =>
+      throw UnsupportedError("connectionFactory not implemented");
+  ```
+
 ### Tools
 
 #### Dart command line

sdk/lib/_http/http.dart

@@ -1516,6 +1516,46 @@ abstract class HttpClient {
   /// Add credentials to be used for authorizing HTTP requests.
   void addCredentials(Uri url, String realm, HttpClientCredentials credentials);
 
+  /// Sets the function used to create socket connections.
+  ///
+  /// The URL requested (e.g. through [getUrl]) and proxy configuration
+  /// ([f.proxyHost] and [f.proxyPort]) are passed as arguments. [f.proxyHost]
+  /// and [f.proxyPort] will be `null` if the connection is not made through
+  /// a proxy.
+  ///
+  /// Since connections may be reused based on host and port, it is important
+  /// that the function not ignore [f.proxyHost] and [f.proxyPort] if they are
+  /// not `null`. If proxies are not meaningful for the returned [Socket], you
+  /// can set [findProxy] to use a direct connection.
+  ///
+  /// For example:
+  ///
+  /// ```dart
+  /// import "dart:io";
+  ///
+  /// void main() async {
+  ///   HttpClient client = HttpClient()
+  ///     ..connectionFactory = (Uri uri, String? proxyHost, int? proxyPort) {
+  ///         assert(proxyHost == null);
+  ///         assert(proxyPort == null);
+  ///         var address = InternetAddress("/var/run/docker.sock",
+  ///             type: InternetAddressType.unix);
+  ///         return Socket.startConnect(address, 0);
+  ///     }
+  ///     ..findProxy = (Uri uri) => 'DIRECT';
+  ///
+  ///   final request = await client.getUrl(Uri.parse("http://ignored/v1.41/info"));
+  ///   final response = await request.close();
+  ///   print(response.statusCode);
+  ///   await response.drain();
+  ///   client.close();
+  /// }
+  /// ```
+  void set connectionFactory(
+      Future<ConnectionTask<Socket>> Function(
+              Uri url, String? proxyHost, int? proxyPort)?
+          f);
+
   /// Sets the function used to resolve the proxy server to be used for
   /// opening a HTTP connection to the specified [url]. If this
   /// function is not set, direct connections will always be used.

sdk/lib/_http/http_impl.dart

@@ -2338,14 +2338,16 @@ class _ConnectionTarget {
   final int port;
   final bool isSecure;
   final SecurityContext? context;
+  final Future<ConnectionTask<Socket>> Function(Uri, String?, int?)?
+      connectionFactory;
   final Set<_HttpClientConnection> _idle = HashSet();
   final Set<_HttpClientConnection> _active = HashSet();
   final Set<ConnectionTask> _socketTasks = HashSet();
   final _pending = ListQueue<void Function()>();
   int _connecting = 0;
 
-  _ConnectionTarget(
-      this.key, this.host, this.port, this.isSecure, this.context);
+  _ConnectionTarget(this.key, this.host, this.port, this.isSecure, this.context,
+      this.connectionFactory);
 
   bool get isEmpty => _idle.isEmpty && _active.isEmpty && _connecting == 0;
 
@@ -2410,8 +2412,8 @@ class _ConnectionTarget {
     }
   }
 
-  Future<_ConnectionInfo> connect(String uriHost, int uriPort, _Proxy proxy,
-      _HttpClient client, _HttpProfileData? profileData) {
+  Future<_ConnectionInfo> connect(Uri uri, String uriHost, int uriPort,
+      _Proxy proxy, _HttpClient client, _HttpProfileData? profileData) {
     if (hasIdle) {
       var connection = takeIdle();
       client._connectionsChanged();
@@ -2422,8 +2424,8 @@ class _ConnectionTarget {
         _active.length + _connecting >= maxConnectionsPerHost) {
       var completer = Completer<_ConnectionInfo>();
       _pending.add(() {
-        completer
-            .complete(connect(uriHost, uriPort, proxy, client, profileData));
+        completer.complete(
+            connect(uri, uriHost, uriPort, proxy, client, profileData));
       });
       return completer.future;
     }
@@ -2434,10 +2436,20 @@ class _ConnectionTarget {
       return currentBadCertificateCallback(certificate, uriHost, uriPort);
     }
 
-    Future<ConnectionTask> connectionTask = (isSecure && proxy.isDirect
-        ? SecureSocket.startConnect(host, port,
-            context: context, onBadCertificate: callback)
-        : Socket.startConnect(host, port));
+    Future<ConnectionTask> connectionTask;
+    final cf = connectionFactory;
+    if (cf != null) {
+      if (proxy.isDirect) {
+        connectionTask = cf(uri, null, null);
+      } else {
+        connectionTask = cf(uri, host, port);
+      }
+    } else {
+      connectionTask = (isSecure && proxy.isDirect
+          ? SecureSocket.startConnect(host, port,
+              context: context, onBadCertificate: callback)
+          : Socket.startConnect(host, port));
+    }
     _connecting++;
     return connectionTask.then((ConnectionTask task) {
       _socketTasks.add(task);
@@ -2506,6 +2518,8 @@ class _HttpClient implements HttpClient {
   final List<_Credentials> _credentials = [];
   final List<_ProxyCredentials> _proxyCredentials = [];
   final SecurityContext? _context;
+  Future<ConnectionTask<Socket>> Function(Uri, String?, int?)?
+      _connectionFactory;
   Future<bool> Function(Uri, String scheme, String? realm)? _authenticate;
   Future<bool> Function(String host, int port, String scheme, String? realm)?
       _authenticateProxy;
@@ -2631,6 +2645,12 @@ class _HttpClient implements HttpClient {
         _ProxyCredentials(host, port, realm, cr as _HttpClientCredentials));
   }
 
+  void set connectionFactory(
+          Future<ConnectionTask<Socket>> Function(
+                  Uri url, String? proxyHost, int? proxyPort)?
+              f) =>
+      _connectionFactory = f;
+
   set findProxy(String Function(Uri uri)? f) => _findProxy = f;
 
   static void _startRequestTimelineEvent(
@@ -2665,7 +2685,9 @@ class _HttpClient implements HttpClient {
     if (method != "CONNECT") {
       if (uri.host.isEmpty) {
         throw ArgumentError("No host specified in URI $uri");
-      } else if (!uri.isScheme("http") && !uri.isScheme("https")) {
+      } else if (this._connectionFactory == null &&
+          !uri.isScheme("http") &&
+          !uri.isScheme("https")) {
         throw ArgumentError("Unsupported scheme '${uri.scheme}' in URI $uri");
       }
     }
@@ -2695,7 +2717,7 @@ class _HttpClient implements HttpClient {
     if (HttpClient.enableTimelineLogging) {
       profileData = HttpProfiler.startRequest(method, uri);
     }
-    return _getConnection(uri.host, port, proxyConf, isSecure, profileData)
+    return _getConnection(uri, uri.host, port, proxyConf, isSecure, profileData)
         .then((_ConnectionInfo info) {
       _HttpClientRequest send(_ConnectionInfo info) {
         profileData?.requestEvent('Connection established');
@@ -2706,7 +2728,8 @@ class _HttpClient implements HttpClient {
       // If the connection was closed before the request was sent, create
       // and use another connection.
       if (info.connection.closed) {
-        return _getConnection(uri.host, port, proxyConf, isSecure, profileData)
+        return _getConnection(
+                uri, uri.host, port, proxyConf, isSecure, profileData)
             .then(send);
       }
       return send(info);
@@ -2812,12 +2835,14 @@ class _HttpClient implements HttpClient {
   _ConnectionTarget _getConnectionTarget(String host, int port, bool isSecure) {
     String key = _HttpClientConnection.makeKey(isSecure, host, port);
     return _connectionTargets.putIfAbsent(key, () {
-      return _ConnectionTarget(key, host, port, isSecure, _context);
+      return _ConnectionTarget(
+          key, host, port, isSecure, _context, _connectionFactory);
     });
   }
 
   // Get a new _HttpClientConnection, from the matching _ConnectionTarget.
   Future<_ConnectionInfo> _getConnection(
+      Uri uri,
       String uriHost,
       int uriPort,
       _ProxyConfiguration proxyConf,
@@ -2831,7 +2856,7 @@ class _HttpClient implements HttpClient {
       String host = proxy.isDirect ? uriHost : proxy.host!;
       int port = proxy.isDirect ? uriPort : proxy.port!;
       return _getConnectionTarget(host, port, isSecure)
-          .connect(uriHost, uriPort, proxy, this, profileData)
+          .connect(uri, uriHost, uriPort, proxy, this, profileData)
           // On error, continue with next proxy.
           .catchError(connect);
     }

tests/standalone/io/http_connection_factory_test.dart

@@ -0,0 +1,146 @@
+// Copyright (c) 2021, the Dart project authors.  Please see the AUTHORS file
+// for details. All rights reserved. Use of this source code is governed by a
+// BSD-style license that can be found in the LICENSE file.
+import "dart:io";
+import 'dart:convert';
+import "package:expect/expect.dart";
+import 'http_proxy_test.dart' show setupProxyServer;
+import 'test_utils.dart' show withTempDir;
+
+testDirectConnection() async {
+  var server = await HttpServer.bind(InternetAddress.anyIPv6, 0);
+  server.forEach((HttpRequest request) {
+    request.response.write('Hello, world!');
+    request.response.close();
+  });
+  final serverUri = Uri.http("127.0.0.1:${server.port}", "/");
+  var client = HttpClient()
+    ..connectionFactory = (uri, proxyHost, proxyPort) {
+      Expect.isNull(proxyHost);
+      Expect.isNull(proxyPort);
+      Expect.equals(serverUri, uri);
+      return Socket.startConnect(uri.host, uri.port);
+    }
+    ..findProxy = (uri) => 'DIRECT';
+  final response = await client.getUrl(serverUri).then((request) {
+    return request.close();
+  });
+  Expect.equals(200, response.statusCode);
+  final responseText = await response
+      .transform(utf8.decoder)
+      .fold('', (String x, String y) => x + y);
+  Expect.equals("Hello, world!", responseText);
+  client.close();
+  server.close();
+}
+
+testConnectionViaProxy() async {
+  var proxyServer = await setupProxyServer();
+  var server = await HttpServer.bind(InternetAddress.anyIPv6, 0);
+  server.forEach((HttpRequest request) {
+    request.response.write('Hello via Proxy');
+    request.response.close();
+  });
+  final serverUri = Uri.http("127.0.0.1:${server.port}", "/");
+  final client = HttpClient()
+    ..connectionFactory = (uri, proxyHost, proxyPort) {
+      Expect.equals("localhost", proxyHost);
+      Expect.equals(proxyServer.port, proxyPort);
+      Expect.equals(serverUri, uri);
+      return Socket.startConnect(proxyHost, proxyPort as int);
+    }
+    ..findProxy = (uri) => "PROXY localhost:${proxyServer.port}";
+  final response = await client.getUrl(serverUri).then((request) {
+    return request.close();
+  });
+  Expect.equals(200, response.statusCode);
+  final responseText = await response
+      .transform(utf8.decoder)
+      .fold('', (String x, String y) => x + y);
+  Expect.equals("Hello via Proxy", responseText);
+  client.close();
+  server.close();
+  proxyServer.shutdown();
+}
+
+testDifferentAddressFamiliesAndProxySettings(String dir) async {
+  // Test a custom connection factory for Unix domain sockets that also allows
+  // regular INET/INET6 access with and without a proxy.
+  var proxyServer = await setupProxyServer();
+  var inet6Server = await HttpServer.bind(InternetAddress.anyIPv6, 0);
+  inet6Server.forEach((HttpRequest request) {
+    request.response.write('Hello via Proxy');
+    request.response.close();
+  });
+  final inet6ServerUri = Uri.http("127.0.0.1:${inet6Server.port}", "/");
+  final unixPath = '$dir/sock';
+  final unixAddress = InternetAddress(unixPath, type: InternetAddressType.unix);
+  final unixServer = await HttpServer.bind(unixAddress, 0);
+  unixServer.forEach((HttpRequest request) {
+    request.response.write('Hello via Unix');
+    request.response.close();
+  });
+  final client = HttpClient()
+    ..connectionFactory = (uri, proxyHost, proxyPort) {
+      if (uri.scheme == 'unix') {
+        assert(proxyHost == null);
+        assert(proxyPort == null);
+        var address = InternetAddress(unixPath, type: InternetAddressType.unix);
+        return Socket.startConnect(address, 0);
+      } else {
+        if (proxyHost != null && proxyPort != null) {
+          return Socket.startConnect(proxyHost, proxyPort);
+        }
+        return Socket.startConnect(uri.host, uri.port);
+      }
+    }
+    ..findProxy = (uri) {
+      if (uri.scheme == 'unix') {
+        // Proxy settings are not meaningful for Unix domain sockets.
+        return 'DIRECT';
+      } else {
+        return "PROXY localhost:${proxyServer.port}";
+      }
+    };
+  // Fetch a URL from the INET6 server and verify the results.
+  final inet6Response = await client.getUrl(inet6ServerUri).then((request) {
+    return request.close();
+  });
+  Expect.equals(200, inet6Response.statusCode);
+  final inet6ResponseText = await inet6Response
+      .transform(utf8.decoder)
+      .fold('', (String x, String y) => x + y);
+  Expect.equals("Hello via Proxy", inet6ResponseText);
+  // Fetch a URL from the Unix server and verify the results.
+  final unixResponse = await client
+      .getUrl(Uri(
+          scheme: "unix",
+          // Connection pooling is based on the host/port combination
+          // so ensure that the host is unique for unique logical
+          // endpoints. Also, the `host` property is converted to
+          // lowercase so you cannot use it directly for file paths.
+          host: 'dummy',
+          path: "/"))
+      .then((request) {
+    return request.close();
+  });
+  Expect.equals(200, unixResponse.statusCode);
+  final unixResponseText = await unixResponse
+      .transform(utf8.decoder)
+      .fold('', (String x, String y) => x + y);
+  Expect.equals("Hello via Unix", unixResponseText);
+  client.close();
+  inet6Server.close();
+  unixServer.close();
+  proxyServer.shutdown();
+}
+
+main() async {
+  await testDirectConnection();
+  await testConnectionViaProxy();
+  if (Platform.isMacOS || Platform.isLinux || Platform.isAndroid) {
+    await withTempDir('unix_socket_test', (Directory dir) async {
+      await testDifferentAddressFamiliesAndProxySettings('${dir.path}');
+    });
+  }
+}

tests/standalone/io/http_override_test.dart

@@ -41,6 +41,10 @@ class MyHttpClient1 implements HttpClient {
   set authenticate(Future<bool> f(Uri url, String scheme, String realm)?) {}
   void addCredentials(
       Uri url, String realm, HttpClientCredentials credentials) {}
+  set connectionFactory(
+      Future<ConnectionTask<Socket>> Function(
+              Uri url, String? proxyHost, int? proxyPort)?
+          f) {}
   set findProxy(String f(Uri url)?) {}
   set authenticateProxy(
       Future<bool> f(String host, int port, String scheme, String realm)?) {}
@@ -85,6 +89,10 @@ class MyHttpClient2 implements HttpClient {
   set authenticate(Future<bool> f(Uri url, String scheme, String realm)?) {}
   void addCredentials(
       Uri url, String realm, HttpClientCredentials credentials) {}
+  set connectionFactory(
+      Future<ConnectionTask<Socket>> Function(
+              Uri url, String? proxyHost, int? proxyPort)?
+          f) {}
   set findProxy(String f(Uri url)?) {}
   set authenticateProxy(
       Future<bool> f(String host, int port, String scheme, String realm)?) {}

tests/standalone_2/io/http_connection_factory_test.dart

@@ -0,0 +1,147 @@
+// Copyright (c) 2021, the Dart project authors.  Please see the AUTHORS file
+// for details. All rights reserved. Use of this source code is governed by a
+// BSD-style license that can be found in the LICENSE file.
+// @dart = 2.9
+import "dart:io";
+import 'dart:convert';
+import "package:expect/expect.dart";
+import 'http_proxy_test.dart' show setupProxyServer;
+import 'test_utils.dart' show withTempDir;
+
+testDirectConnection() async {
+  var server = await HttpServer.bind(InternetAddress.anyIPv6, 0);
+  server.forEach((HttpRequest request) {
+    request.response.write('Hello, world!');
+    request.response.close();
+  });
+  final serverUri = Uri.http("127.0.0.1:${server.port}", "/");
+  var client = HttpClient()
+    ..connectionFactory = (uri, proxyHost, proxyPort) {
+      Expect.isNull(proxyHost);
+      Expect.isNull(proxyPort);
+      Expect.equals(serverUri, uri);
+      return Socket.startConnect(uri.host, uri.port);
+    }
+    ..findProxy = (uri) => 'DIRECT';
+  final response = await client.getUrl(serverUri).then((request) {
+    return request.close();
+  });
+  Expect.equals(200, response.statusCode);
+  final responseText = await response
+      .transform(utf8.decoder)
+      .fold('', (String x, String y) => x + y);
+  Expect.equals("Hello, world!", responseText);
+  client.close();
+  server.close();
+}
+
+testConnectionViaProxy() async {
+  var proxyServer = await setupProxyServer();
+  var server = await HttpServer.bind(InternetAddress.anyIPv6, 0);
+  server.forEach((HttpRequest request) {
+    request.response.write('Hello via Proxy');
+    request.response.close();
+  });
+  final serverUri = Uri.http("127.0.0.1:${server.port}", "/");
+  final client = HttpClient()
+    ..connectionFactory = (uri, proxyHost, proxyPort) {
+      Expect.equals("localhost", proxyHost);
+      Expect.equals(proxyServer.port, proxyPort);
+      Expect.equals(serverUri, uri);
+      return Socket.startConnect(proxyHost, proxyPort as int);
+    }
+    ..findProxy = (uri) => "PROXY localhost:${proxyServer.port}";
+  final response = await client.getUrl(serverUri).then((request) {
+    return request.close();
+  });
+  Expect.equals(200, response.statusCode);
+  final responseText = await response
+      .transform(utf8.decoder)
+      .fold('', (String x, String y) => x + y);
+  Expect.equals("Hello via Proxy", responseText);
+  client.close();
+  server.close();
+  proxyServer.shutdown();
+}
+
+testDifferentAddressFamiliesAndProxySettings(String dir) async {
+  // Test a custom connection factory for Unix domain sockets that also allows
+  // regular INET/INET6 access with and without a proxy.
+  var proxyServer = await setupProxyServer();
+  var inet6Server = await HttpServer.bind(InternetAddress.anyIPv6, 0);
+  inet6Server.forEach((HttpRequest request) {
+    request.response.write('Hello via Proxy');
+    request.response.close();
+  });
+  final inet6ServerUri = Uri.http("127.0.0.1:${inet6Server.port}", "/");
+  final unixPath = '$dir/sock';
+  final unixAddress = InternetAddress(unixPath, type: InternetAddressType.unix);
+  final unixServer = await HttpServer.bind(unixAddress, 0);
+  unixServer.forEach((HttpRequest request) {
+    request.response.write('Hello via Unix');
+    request.response.close();
+  });
+  final client = HttpClient()
+    ..connectionFactory = (uri, proxyHost, proxyPort) {
+      if (uri.scheme == 'unix') {
+        assert(proxyHost == null);
+        assert(proxyPort == null);
+        var address = InternetAddress(unixPath, type: InternetAddressType.unix);
+        return Socket.startConnect(address, 0);
+      } else {
+        if (proxyHost != null && proxyPort != null) {
+          return Socket.startConnect(proxyHost, proxyPort);
+        }
+        return Socket.startConnect(uri.host, uri.port);
+      }
+    }
+    ..findProxy = (uri) {
+      if (uri.scheme == 'unix') {
+        // Proxy settings are not meaningful for Unix domain sockets.
+        return 'DIRECT';
+      } else {
+        return "PROXY localhost:${proxyServer.port}";
+      }
+    };
+// Fetch a URL from the INET6 server and verify the results.
+  final inet6Response = await client.getUrl(inet6ServerUri).then((request) {
+    return request.close();
+  });
+  Expect.equals(200, inet6Response.statusCode);
+  final inet6ResponseText = await inet6Response
+      .transform(utf8.decoder)
+      .fold('', (String x, String y) => x + y);
+  Expect.equals("Hello via Proxy", inet6ResponseText);
+// Fetch a URL from the Unix server and verify the results.
+  final unixResponse = await client
+      .getUrl(Uri(
+          scheme: "unix",
+          // Connection pooling is based on the host/port combination
+          // so ensure that the host is unique for unique logical
+          // endpoints. Also, the `host` property is converted to
+          // lowercase so you cannot use it directly for file paths.
+          host: 'dummy',
+          path: "/"))
+      .then((request) {
+    return request.close();
+  });
+  Expect.equals(200, unixResponse.statusCode);
+  final unixResponseText = await unixResponse
+      .transform(utf8.decoder)
+      .fold('', (String x, String y) => x + y);
+  Expect.equals("Hello via Unix", unixResponseText);
+  client.close();
+  inet6Server.close();
+  unixServer.close();
+  proxyServer.shutdown();
+}
+
+main() async {
+  await testDirectConnection();
+  await testConnectionViaProxy();
+  if (Platform.isMacOS || Platform.isLinux || Platform.isAndroid) {
+    await withTempDir('unix_socket_test', (Directory dir) async {
+      await testDifferentAddressFamiliesAndProxySettings('${dir.path}');
+    });
+  }
+}

tests/standalone_2/io/http_override_test.dart

@@ -39,6 +39,10 @@ class MyHttpClient1 implements HttpClient {
   set authenticate(Future<bool> f(Uri url, String scheme, String realm)) {}
   void addCredentials(
       Uri url, String realm, HttpClientCredentials credentials) {}
+  set connectionFactory(
+      Future<ConnectionTask<Socket>> Function(
+              Uri url, String proxyHost, int proxyPort)
+          f) {}
   set findProxy(String f(Uri url)) {}
   set authenticateProxy(
       Future<bool> f(String host, int port, String scheme, String realm)) {}
@@ -79,6 +83,10 @@ class MyHttpClient2 implements HttpClient {
   set authenticate(Future<bool> f(Uri url, String scheme, String realm)) {}
   void addCredentials(
       Uri url, String realm, HttpClientCredentials credentials) {}
+  set connectionFactory(
+      Future<ConnectionTask<Socket>> Function(
+              Uri url, String proxyHost, int proxyPort)
+          f) {}
   set findProxy(String f(Uri url)) {}
   set authenticateProxy(
       Future<bool> f(String host, int port, String scheme, String realm)) {}