Defaults to {@code null}, in which case the driver connects to KMS hosts directly.
+ *
+ * @param kmsConnectCallback the KMS connect callback, or null to connect to KMS hosts directly
+ * @return this
+ * @see #getKmsConnectCallback()
+ * @since 5.11
+ */
+ public Builder kmsConnectCallback(@Nullable final KmsConnectCallback kmsConnectCallback) {
+ this.kmsConnectCallback = kmsConnectCallback;
+ return this;
+ }
+
/**
* Sets the map from namespace to local schema document
*
@@ -406,6 +426,17 @@ public Map getKmsProviderSslContextMap() {
return unmodifiableMap(kmsProviderSslContextMap);
}
+ /**
+ * Gets the callback that establishes connections to Key Management Service (KMS) hosts.
+ *
+ * @return the KMS connect callback, or null if the driver connects to KMS hosts directly
+ * @since 5.11
+ */
+ @Nullable
+ public KmsConnectCallback getKmsConnectCallback() {
+ return kmsConnectCallback;
+ }
+
/**
* Gets the map of namespace to local JSON schema.
*
@@ -529,6 +560,7 @@ private AutoEncryptionSettings(final Builder builder) {
this.keyVaultNamespace = notNull("keyVaultNamespace", builder.keyVaultNamespace);
this.kmsProviders = notNull("kmsProviders", builder.kmsProviders);
this.kmsProviderSslContextMap = notNull("kmsProviderSslContextMap", builder.kmsProviderSslContextMap);
+ this.kmsConnectCallback = builder.kmsConnectCallback;
this.kmsProviderPropertySuppliers = notNull("kmsProviderPropertySuppliers", builder.kmsProviderPropertySuppliers);
this.schemaMap = notNull("schemaMap", builder.schemaMap);
this.extraOptions = notNull("extraOptions", builder.extraOptions);
diff --git a/driver-core/src/main/com/mongodb/ClientEncryptionSettings.java b/driver-core/src/main/com/mongodb/ClientEncryptionSettings.java
index 252d9d0ff9c..df4fa117ffd 100644
--- a/driver-core/src/main/com/mongodb/ClientEncryptionSettings.java
+++ b/driver-core/src/main/com/mongodb/ClientEncryptionSettings.java
@@ -50,6 +50,8 @@ public final class ClientEncryptionSettings {
private final Map>> kmsProviderPropertySuppliers;
private final Map kmsProviderSslContextMap;
@Nullable
+ private final KmsConnectCallback kmsConnectCallback;
+ @Nullable
private final Long timeoutMS;
@Nullable
private final Long keyExpirationMS;
@@ -66,6 +68,8 @@ public static final class Builder {
private Map>> kmsProviderPropertySuppliers = new HashMap<>();
private Map kmsProviderSslContextMap = new HashMap<>();
@Nullable
+ private KmsConnectCallback kmsConnectCallback;
+ @Nullable
private Long timeoutMS;
@Nullable
private Long keyExpirationMS;
@@ -136,6 +140,22 @@ public Builder kmsProviderSslContextMap(final Map kmsProvide
return this;
}
+ /**
+ * Sets the callback that establishes connections to Key Management Service (KMS) hosts, enabling KMS requests
+ * to be routed through an intermediary such as an HTTP proxy.
+ *
+ * Defaults to {@code null}, in which case the driver connects to KMS hosts directly.
+ *
+ * @param kmsConnectCallback the KMS connect callback, or null to connect to KMS hosts directly
+ * @return this
+ * @see #getKmsConnectCallback()
+ * @since 5.11
+ */
+ public Builder kmsConnectCallback(@Nullable final KmsConnectCallback kmsConnectCallback) {
+ this.kmsConnectCallback = kmsConnectCallback;
+ return this;
+ }
+
/**
* The cache expiration time for data encryption keys.
* Defaults to {@code null} which defers to libmongocrypt's default which is currently 60000 ms. Set to 0 to disable key expiration.
@@ -335,6 +355,17 @@ public Map getKmsProviderSslContextMap() {
return unmodifiableMap(kmsProviderSslContextMap);
}
+ /**
+ * Gets the callback that establishes connections to Key Management Service (KMS) hosts.
+ *
+ * @return the KMS connect callback, or null if the driver connects to KMS hosts directly
+ * @since 5.11
+ */
+ @Nullable
+ public KmsConnectCallback getKmsConnectCallback() {
+ return kmsConnectCallback;
+ }
+
/**
* Returns the cache expiration time for data encryption keys.
*
@@ -399,6 +430,7 @@ private ClientEncryptionSettings(final Builder builder) {
this.kmsProviders = notNull("kmsProviders", builder.kmsProviders);
this.kmsProviderPropertySuppliers = notNull("kmsProviderPropertySuppliers", builder.kmsProviderPropertySuppliers);
this.kmsProviderSslContextMap = notNull("kmsProviderSslContextMap", builder.kmsProviderSslContextMap);
+ this.kmsConnectCallback = builder.kmsConnectCallback;
this.timeoutMS = builder.timeoutMS;
this.keyExpirationMS = builder.keyExpirationMS;
}
diff --git a/driver-core/src/main/com/mongodb/KmsConnectCallback.java b/driver-core/src/main/com/mongodb/KmsConnectCallback.java
new file mode 100644
index 00000000000..3036af89157
--- /dev/null
+++ b/driver-core/src/main/com/mongodb/KmsConnectCallback.java
@@ -0,0 +1,92 @@
+/*
+ * Copyright 2008-present MongoDB, Inc.
+ *
+ * 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 com.mongodb;
+
+import com.mongodb.annotations.ThreadSafe;
+
+import java.io.IOException;
+import java.net.Socket;
+
+/**
+ * A callback that establishes the connection used for a Key Management Service (KMS) request made by in-use encryption
+ * (client-side field level encryption or queryable encryption).
+ *
+ * When a callback is configured, the driver invokes it instead of connecting to the KMS host itself, and uses the
+ * socket it returns for the KMS request. This enables routing KMS requests through an intermediary, most commonly an
+ * HTTP proxy via the {@code HTTP CONNECT} method.
+ *
+ * The driver always negotiates TLS with the KMS host itself over the returned socket, using the
+ * {@link javax.net.ssl.SSLContext} configured for the KMS provider. Server Name Indication and certificate hostname
+ * verification target the KMS host named by the context, not the address the callback actually connected to.
+ * Implementations therefore MUST NOT negotiate TLS with the KMS host themselves; they must return a socket over which
+ * a TLS handshake with the KMS host can be performed. An implementation may use TLS for its own connection to an
+ * intermediary, in which case it returns an {@link javax.net.ssl.SSLSocket} and the driver layers the KMS host's TLS
+ * session on top of it.
+ *
+ * An {@link IOException} thrown by an implementation is treated as a transient network error.
+ *
+ * This is applicable only when using the synchronous variant of {@code MongoClient}. The reactive streams driver,
+ * and the drivers built on it, reject a configured callback rather than connecting to KMS hosts directly.
+ *
+ * Authenticating to an intermediary is the responsibility of the implementation. For a proxy requiring HTTP Basic
+ * authentication, for example, the implementation adds a {@code Proxy-Authorization} header to the {@code CONNECT}
+ * request.
+ *
+ * Example of an implementation that tunnels through an HTTP proxy:
+ * {@code
+ * KmsConnectCallback callback = context -> {
+ * Socket socket = new Socket();
+ * try {
+ * socket.connect(new InetSocketAddress("proxy.example.com", 8080));
+ *
+ * String target = context.getHost() + ":" + context.getPort();
+ * socket.getOutputStream().write(
+ * ("CONNECT " + target + " HTTP/1.1\r\nHost: " + target + "\r\n\r\n").getBytes(StandardCharsets.US_ASCII));
+ *
+ * // Read the status line and confirm a 2xx status, throwing an IOException otherwise. Match the status code
+ * // rather than the whole status line, as proxies differ in the HTTP version they reply with. Read the
+ * // response one byte at a time, up to the end of the header block, so that no byte of the TLS handshake that
+ * // the driver performs over this socket is consumed.
+ * readAndCheckProxyResponse(socket.getInputStream());
+ * } catch (IOException | RuntimeException e) {
+ * // The driver cannot close a socket that was never returned to it.
+ * socket.close();
+ * throw e;
+ * }
+ *
+ * return socket;
+ * };
+ * }
+ *
+ * @see ClientEncryptionSettings.Builder#kmsConnectCallback(KmsConnectCallback)
+ * @see AutoEncryptionSettings.Builder#kmsConnectCallback(KmsConnectCallback)
+ * @since 5.11
+ */
+@ThreadSafe
+@FunctionalInterface
+public interface KmsConnectCallback {
+ /**
+ * Returns a socket connected such that a TLS handshake with the KMS host can be performed over it.
+ *
+ * Ownership of the returned socket passes to the driver, which closes it once the KMS request completes.
+ *
+ * @param context the details of the connection to establish
+ * @return a connected socket, which must not have an established TLS session with the KMS host
+ * @throws IOException if the connection cannot be established. This is treated as a transient network error.
+ */
+ Socket connect(KmsConnectContext context) throws IOException;
+}
diff --git a/driver-core/src/main/com/mongodb/KmsConnectContext.java b/driver-core/src/main/com/mongodb/KmsConnectContext.java
new file mode 100644
index 00000000000..280db9b4983
--- /dev/null
+++ b/driver-core/src/main/com/mongodb/KmsConnectContext.java
@@ -0,0 +1,77 @@
+/*
+ * Copyright 2008-present MongoDB, Inc.
+ *
+ * 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 com.mongodb;
+
+import com.mongodb.annotations.Immutable;
+
+import static com.mongodb.assertions.Assertions.isTrueArgument;
+import static com.mongodb.assertions.Assertions.notNull;
+
+/**
+ * The details of a Key Management Service (KMS) connection that a {@link KmsConnectCallback} is asked to establish.
+ *
+ * @see KmsConnectCallback
+ * @since 5.11
+ */
+@Immutable
+public final class KmsConnectContext {
+ private final String host;
+ private final int port;
+
+ /**
+ * Construct a new instance.
+ *
+ * @param host the host name of the KMS host, which may not be null
+ * @param port the port of the KMS host, which must be positive
+ */
+ public KmsConnectContext(final String host, final int port) {
+ this.host = notNull("host", host);
+ isTrueArgument("port > 0", port > 0);
+ this.port = port;
+ }
+
+ /**
+ * Gets the host name of the KMS host that the connection must ultimately reach.
+ *
+ * This is not the host name of any intermediary such as a proxy. It is the host that the driver negotiates TLS
+ * with once the callback returns.
+ *
+ * @return the host name of the KMS host, never null
+ */
+ public String getHost() {
+ return host;
+ }
+
+ /**
+ * Gets the port of the KMS host that the connection must ultimately reach.
+ *
+ * This is not the port of any intermediary such as a proxy.
+ *
+ * @return the port of the KMS host, always positive
+ */
+ public int getPort() {
+ return port;
+ }
+
+ @Override
+ public String toString() {
+ return "KmsConnectContext{"
+ + "host='" + host + '\''
+ + ", port=" + port
+ + '}';
+ }
+}
diff --git a/driver-core/src/test/unit/com/mongodb/KmsConnectCallbackSettingsTest.java b/driver-core/src/test/unit/com/mongodb/KmsConnectCallbackSettingsTest.java
new file mode 100644
index 00000000000..e381fd95948
--- /dev/null
+++ b/driver-core/src/test/unit/com/mongodb/KmsConnectCallbackSettingsTest.java
@@ -0,0 +1,74 @@
+/*
+ * Copyright 2008-present MongoDB, Inc.
+ *
+ * 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 com.mongodb;
+
+import org.junit.jupiter.api.Test;
+
+import java.net.Socket;
+import java.util.HashMap;
+
+import static org.junit.jupiter.api.Assertions.assertNull;
+import static org.junit.jupiter.api.Assertions.assertSame;
+
+final class KmsConnectCallbackSettingsTest {
+
+ private static final KmsConnectCallback CALLBACK = context -> new Socket();
+
+ @Test
+ void autoEncryptionSettingsShouldDefaultToNoCallback() {
+ AutoEncryptionSettings settings = AutoEncryptionSettings.builder()
+ .keyVaultNamespace("keyvault.datakeys")
+ .kmsProviders(new HashMap<>())
+ .build();
+
+ assertNull(settings.getKmsConnectCallback());
+ }
+
+ @Test
+ void autoEncryptionSettingsShouldRoundTripCallback() {
+ AutoEncryptionSettings settings = AutoEncryptionSettings.builder()
+ .keyVaultNamespace("keyvault.datakeys")
+ .kmsProviders(new HashMap<>())
+ .kmsConnectCallback(CALLBACK)
+ .build();
+
+ assertSame(CALLBACK, settings.getKmsConnectCallback());
+ }
+
+ @Test
+ void clientEncryptionSettingsShouldDefaultToNoCallback() {
+ ClientEncryptionSettings settings = clientEncryptionSettingsBuilder().build();
+
+ assertNull(settings.getKmsConnectCallback());
+ }
+
+ @Test
+ void clientEncryptionSettingsShouldRoundTripCallback() {
+ ClientEncryptionSettings settings = clientEncryptionSettingsBuilder()
+ .kmsConnectCallback(CALLBACK)
+ .build();
+
+ assertSame(CALLBACK, settings.getKmsConnectCallback());
+ }
+
+ private static ClientEncryptionSettings.Builder clientEncryptionSettingsBuilder() {
+ return ClientEncryptionSettings.builder()
+ .keyVaultMongoClientSettings(MongoClientSettings.builder().build())
+ .keyVaultNamespace("keyvault.datakeys")
+ .kmsProviders(new HashMap<>());
+ }
+}
diff --git a/driver-reactive-streams/src/main/com/mongodb/reactivestreams/client/internal/crypt/Crypts.java b/driver-reactive-streams/src/main/com/mongodb/reactivestreams/client/internal/crypt/Crypts.java
index b06af01d476..c33c4355ae8 100644
--- a/driver-reactive-streams/src/main/com/mongodb/reactivestreams/client/internal/crypt/Crypts.java
+++ b/driver-reactive-streams/src/main/com/mongodb/reactivestreams/client/internal/crypt/Crypts.java
@@ -18,11 +18,13 @@
import com.mongodb.AutoEncryptionSettings;
import com.mongodb.ClientEncryptionSettings;
+import com.mongodb.KmsConnectCallback;
import com.mongodb.MongoClientException;
import com.mongodb.MongoClientSettings;
import com.mongodb.MongoNamespace;
import com.mongodb.internal.crypt.capi.MongoCrypt;
import com.mongodb.internal.crypt.capi.MongoCrypts;
+import com.mongodb.lang.Nullable;
import com.mongodb.reactivestreams.client.MongoClient;
import com.mongodb.reactivestreams.client.MongoClients;
@@ -41,6 +43,7 @@ private Crypts() {
}
public static Crypt createCrypt(final MongoClientSettings mongoClientSettings, final AutoEncryptionSettings autoEncryptionSettings) {
+ assertKmsConnectCallbackNotConfigured(autoEncryptionSettings.getKmsConnectCallback());
MongoClient sharedInternalClient = null;
MongoClientSettings keyVaultMongoClientSettings = autoEncryptionSettings.getKeyVaultMongoClientSettings();
if (keyVaultMongoClientSettings == null || !autoEncryptionSettings.isBypassAutoEncryption()) {
@@ -67,6 +70,7 @@ public static Crypt createCrypt(final MongoClientSettings mongoClientSettings, f
}
public static Crypt create(final MongoClient keyVaultClient, final ClientEncryptionSettings settings) {
+ assertKmsConnectCallbackNotConfigured(settings.getKmsConnectCallback());
return new Crypt(MongoCrypts.create(createMongoCryptOptions(settings)),
createKeyRetriever(keyVaultClient, settings.getKeyVaultNamespace()),
createKeyManagementService(settings.getKmsProviderSslContextMap()),
@@ -75,6 +79,19 @@ public static Crypt create(final MongoClient keyVaultClient, final ClientEncrypt
);
}
+ /**
+ * A {@link KmsConnectCallback} returns a {@link java.net.Socket}, which offers only blocking I/O and cannot be
+ * adapted to the non-blocking machinery this driver uses: a socket connected to an intermediary over TLS is never
+ * backed by a {@link java.nio.channels.SocketChannel}. Fail rather than silently connecting to KMS hosts directly,
+ * which mirrors how a proxy configured for connections to a MongoDB server is rejected in
+ * {@link MongoClients#create(MongoClientSettings, com.mongodb.MongoDriverInformation)}.
+ */
+ private static void assertKmsConnectCallbackNotConfigured(@Nullable final KmsConnectCallback kmsConnectCallback) {
+ if (kmsConnectCallback != null) {
+ throw new MongoClientException("kmsConnectCallback is not supported for reactive clients");
+ }
+ }
+
private static KeyRetriever createKeyRetriever(final MongoClient keyVaultClient,
final String keyVaultNamespaceString) {
return new KeyRetriever(keyVaultClient, new MongoNamespace(keyVaultNamespaceString));
diff --git a/driver-reactive-streams/src/test/unit/com/mongodb/reactivestreams/client/internal/crypt/CryptsKmsConnectCallbackNotSupportedTest.java b/driver-reactive-streams/src/test/unit/com/mongodb/reactivestreams/client/internal/crypt/CryptsKmsConnectCallbackNotSupportedTest.java
new file mode 100644
index 00000000000..022c6f5c79c
--- /dev/null
+++ b/driver-reactive-streams/src/test/unit/com/mongodb/reactivestreams/client/internal/crypt/CryptsKmsConnectCallbackNotSupportedTest.java
@@ -0,0 +1,82 @@
+/*
+ * Copyright 2008-present MongoDB, Inc.
+ *
+ * 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 com.mongodb.reactivestreams.client.internal.crypt;
+
+import com.mongodb.AutoEncryptionSettings;
+import com.mongodb.ClientEncryptionSettings;
+import com.mongodb.KmsConnectCallback;
+import com.mongodb.MongoClientException;
+import com.mongodb.MongoClientSettings;
+import org.junit.jupiter.api.Test;
+
+import java.net.Socket;
+import java.util.HashMap;
+
+import static org.junit.jupiter.api.Assertions.assertThrows;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+/**
+ * {@link KmsConnectCallback} is implemented only for the synchronous driver, so configuring one for a reactive client
+ * must fail rather than silently connecting to KMS hosts directly.
+ */
+final class CryptsKmsConnectCallbackNotSupportedTest {
+
+ private static final KmsConnectCallback CALLBACK = context -> new Socket();
+
+ @Test
+ void shouldRejectACallbackOnAutoEncryptionSettings() {
+ AutoEncryptionSettings settings = AutoEncryptionSettings.builder()
+ .keyVaultNamespace("keyvault.datakeys")
+ .kmsProviders(new HashMap<>())
+ .kmsConnectCallback(CALLBACK)
+ .build();
+
+ MongoClientException e = assertThrows(MongoClientException.class,
+ () -> Crypts.createCrypt(MongoClientSettings.builder().build(), settings));
+ assertTrue(e.getMessage().contains("not supported for reactive clients"),
+ () -> "unexpected message: " + e.getMessage());
+ }
+
+ @Test
+ void shouldRejectACallbackOnClientEncryptionSettings() {
+ ClientEncryptionSettings settings = ClientEncryptionSettings.builder()
+ .keyVaultMongoClientSettings(MongoClientSettings.builder().build())
+ .keyVaultNamespace("keyvault.datakeys")
+ .kmsProviders(new HashMap<>())
+ .kmsConnectCallback(CALLBACK)
+ .build();
+
+ MongoClientException e = assertThrows(MongoClientException.class, () -> Crypts.create(null, settings));
+ assertTrue(e.getMessage().contains("not supported for reactive clients"),
+ () -> "unexpected message: " + e.getMessage());
+ }
+
+ @Test
+ void shouldNotRejectSettingsWithoutACallback() {
+ ClientEncryptionSettings settings = ClientEncryptionSettings.builder()
+ .keyVaultMongoClientSettings(MongoClientSettings.builder().build())
+ .keyVaultNamespace("keyvault.datakeys")
+ .kmsProviders(new HashMap<>())
+ .build();
+
+ // Without a callback the guard must not fire; construction then fails for an unrelated reason, which is not a
+ // MongoClientException about callback support.
+ Throwable thrown = assertThrows(Throwable.class, () -> Crypts.create(null, settings));
+ assertTrue(thrown.getMessage() == null || !thrown.getMessage().contains("not supported for reactive clients"),
+ () -> "the callback guard fired unexpectedly: " + thrown.getMessage());
+ }
+}
diff --git a/driver-scala/src/main/scala/org/mongodb/scala/package.scala b/driver-scala/src/main/scala/org/mongodb/scala/package.scala
index 1cdc2d0a564..0cae5874bfb 100644
--- a/driver-scala/src/main/scala/org/mongodb/scala/package.scala
+++ b/driver-scala/src/main/scala/org/mongodb/scala/package.scala
@@ -442,6 +442,21 @@ package object scala extends ClientSessionImplicits with ObservableImplicits wit
*/
type AutoEncryptionSettings = com.mongodb.AutoEncryptionSettings
+ /**
+ * A callback that establishes the connection used for a Key Management Service (KMS) request made by in-use
+ * encryption.
+ *
+ * @since 5.11
+ */
+ type KmsConnectCallback = com.mongodb.KmsConnectCallback
+
+ /**
+ * The details of a Key Management Service (KMS) connection that a `KmsConnectCallback` is asked to establish.
+ *
+ * @since 5.11
+ */
+ type KmsConnectContext = com.mongodb.KmsConnectContext
+
/**
* The client-side settings for data key creation and explicit encryption.
*
diff --git a/driver-sync/src/main/com/mongodb/client/internal/Crypts.java b/driver-sync/src/main/com/mongodb/client/internal/Crypts.java
index 30319bbf4f8..4bc6703dd82 100644
--- a/driver-sync/src/main/com/mongodb/client/internal/Crypts.java
+++ b/driver-sync/src/main/com/mongodb/client/internal/Crypts.java
@@ -18,12 +18,14 @@
import com.mongodb.AutoEncryptionSettings;
import com.mongodb.ClientEncryptionSettings;
+import com.mongodb.KmsConnectCallback;
import com.mongodb.MongoClientSettings;
import com.mongodb.MongoNamespace;
import com.mongodb.client.MongoClient;
import com.mongodb.client.MongoClients;
import com.mongodb.internal.crypt.capi.MongoCrypt;
import com.mongodb.internal.crypt.capi.MongoCrypts;
+import com.mongodb.lang.Nullable;
import javax.net.ssl.SSLContext;
import java.util.Map;
@@ -51,7 +53,7 @@ public static Crypt createCrypt(final MongoClientSettings mongoClientSettings, f
return new Crypt(
mongoCrypt,
createKeyRetriever(keyVaultClient, settings.getKeyVaultNamespace()),
- createKeyManagementService(settings.getKmsProviderSslContextMap()),
+ createKeyManagementService(settings.getKmsProviderSslContextMap(), settings.getKmsConnectCallback()),
settings.getKmsProviders(),
settings.getKmsProviderPropertySuppliers(),
settings.isBypassAutoEncryption(),
@@ -63,7 +65,7 @@ public static Crypt createCrypt(final MongoClientSettings mongoClientSettings, f
static Crypt create(final MongoClient keyVaultClient, final ClientEncryptionSettings settings) {
return new Crypt(MongoCrypts.create(createMongoCryptOptions(settings)),
createKeyRetriever(keyVaultClient, settings.getKeyVaultNamespace()),
- createKeyManagementService(settings.getKmsProviderSslContextMap()),
+ createKeyManagementService(settings.getKmsProviderSslContextMap(), settings.getKmsConnectCallback()),
settings.getKmsProviders(),
settings.getKmsProviderPropertySuppliers()
);
@@ -73,8 +75,9 @@ private static KeyRetriever createKeyRetriever(final MongoClient keyVaultClient,
return new KeyRetriever(keyVaultClient, new MongoNamespace(keyVaultNamespaceString));
}
- private static KeyManagementService createKeyManagementService(final Map kmsProviderSslContextMap) {
- return new KeyManagementService(kmsProviderSslContextMap, 10000);
+ private static KeyManagementService createKeyManagementService(final Map kmsProviderSslContextMap,
+ @Nullable final KmsConnectCallback kmsConnectCallback) {
+ return new KeyManagementService(kmsProviderSslContextMap, kmsConnectCallback, 10000);
}
private Crypts() {
diff --git a/driver-sync/src/main/com/mongodb/client/internal/KeyManagementService.java b/driver-sync/src/main/com/mongodb/client/internal/KeyManagementService.java
index 806f768a923..0e423cae9fb 100644
--- a/driver-sync/src/main/com/mongodb/client/internal/KeyManagementService.java
+++ b/driver-sync/src/main/com/mongodb/client/internal/KeyManagementService.java
@@ -16,25 +16,20 @@
package com.mongodb.client.internal;
+import com.mongodb.KmsConnectCallback;
import com.mongodb.ServerAddress;
import com.mongodb.internal.TimeoutContext;
-import com.mongodb.internal.connection.SslHelper;
import com.mongodb.internal.diagnostics.logging.Logger;
import com.mongodb.internal.diagnostics.logging.Loggers;
import com.mongodb.internal.time.Timeout;
import com.mongodb.lang.Nullable;
import com.mongodb.lang.NonNull;
-import javax.net.SocketFactory;
import javax.net.ssl.SSLContext;
-import javax.net.ssl.SSLParameters;
import javax.net.ssl.SSLSocket;
-import javax.net.ssl.SSLSocketFactory;
import java.io.IOException;
import java.io.InputStream;
import java.io.OutputStream;
-import java.net.InetAddress;
-import java.net.InetSocketAddress;
import java.net.Socket;
import java.net.SocketException;
import java.nio.ByteBuffer;
@@ -48,10 +43,14 @@
class KeyManagementService {
private static final Logger LOGGER = Loggers.getLogger("client");
private final Map kmsProviderSslContextMap;
+ @Nullable
+ private final KmsConnectCallback kmsConnectCallback;
private final int timeoutMillis;
- KeyManagementService(final Map kmsProviderSslContextMap, final int timeoutMillis) {
+ KeyManagementService(final Map kmsProviderSslContextMap,
+ @Nullable final KmsConnectCallback kmsConnectCallback, final int timeoutMillis) {
this.kmsProviderSslContextMap = notNull("kmsProviderSslContextMap", kmsProviderSslContextMap);
+ this.kmsConnectCallback = kmsConnectCallback;
this.timeoutMillis = timeoutMillis;
}
@@ -61,18 +60,15 @@ public InputStream stream(final String kmsProvider, final String host, final Byt
LOGGER.info("Connecting to KMS server at " + serverAddress);
SSLContext sslContext = kmsProviderSslContextMap.get(kmsProvider);
- SocketFactory sslSocketFactory = sslContext == null
- ? SSLSocketFactory.getDefault() : sslContext.getSocketFactory();
- SSLSocket socket = (SSLSocket) sslSocketFactory.createSocket();
- enableHostNameVerification(socket);
+ SSLSocket socket = KmsSocketConnector.connect(sslContext, kmsConnectCallback, serverAddress,
+ timeoutMillis, remainingMillis(operationTimeout));
- try {
- socket.setSoTimeout(timeoutMillis);
- socket.connect(new InetSocketAddress(InetAddress.getByName(serverAddress.getHost()), serverAddress.getPort()), timeoutMillis);
- } catch (IOException e) {
+ // A KmsConnectCallback performs blocking I/O and may overrun the deadline it was given, which cannot be
+ // preempted. Re-check before issuing a request whose time budget is already spent.
+ Timeout.nullAsInfinite(operationTimeout).onExpired(() -> {
closeSocket(socket);
- throw e;
- }
+ TimeoutContext.throwMongoTimeoutException("Connecting to KMS server exceeded the timeout limit.");
+ });
try {
OutputStream outputStream = socket.getOutputStream();
@@ -94,13 +90,20 @@ public InputStream stream(final String kmsProvider, final String host, final Byt
}
}
- private void enableHostNameVerification(final SSLSocket socket) {
- SSLParameters sslParameters = socket.getSSLParameters();
- if (sslParameters == null) {
- sslParameters = new SSLParameters();
- }
- SslHelper.enableHostNameVerification(sslParameters);
- socket.setSSLParameters(sslParameters);
+ /**
+ * Determines the time available for reaching the KMS server, which the specification requires to be the time
+ * remaining in the operation when a client-side operation timeout is in use.
+ *
+ * Visible for testing.
+ *
+ * @return the remaining time available for connecting to the KMS server, in milliseconds, never larger than the
+ * configured connect timeout.
+ */
+ long remainingMillis(@Nullable final Timeout operationTimeout) {
+ return Timeout.nullAsInfinite(operationTimeout).call(MILLISECONDS,
+ () -> (long) timeoutMillis,
+ (ms) -> Math.min(ms, timeoutMillis),
+ () -> TimeoutContext.throwMongoTimeoutException("Connecting to KMS server exceeded the timeout limit."));
}
private void closeSocket(final Socket socket) {
diff --git a/driver-sync/src/main/com/mongodb/client/internal/KmsSocketConnector.java b/driver-sync/src/main/com/mongodb/client/internal/KmsSocketConnector.java
new file mode 100644
index 00000000000..7b464b76946
--- /dev/null
+++ b/driver-sync/src/main/com/mongodb/client/internal/KmsSocketConnector.java
@@ -0,0 +1,156 @@
+/*
+ * Copyright 2008-present MongoDB, Inc.
+ *
+ * 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 com.mongodb.client.internal;
+
+import com.mongodb.KmsConnectCallback;
+import com.mongodb.KmsConnectContext;
+import com.mongodb.ServerAddress;
+import com.mongodb.internal.connection.SslHelper;
+import com.mongodb.lang.Nullable;
+
+import javax.net.ssl.SSLContext;
+import javax.net.ssl.SSLParameters;
+import javax.net.ssl.SSLSocket;
+import javax.net.ssl.SSLSocketFactory;
+import java.io.IOException;
+import java.net.InetAddress;
+import java.net.InetSocketAddress;
+import java.net.Socket;
+
+import static com.mongodb.assertions.Assertions.notNull;
+
+/**
+ * Establishes the TLS connection used for a Key Management Service (KMS) request.
+ *
+ * When a {@link KmsConnectCallback} is configured, the callback establishes the underlying connection, which may be
+ * to an intermediary such as an HTTP proxy, and TLS for the KMS host is layered on top of the socket it returns. TLS is
+ * always negotiated end-to-end with the KMS host: Server Name Indication and certificate hostname verification are
+ * configured from the KMS address rather than from wherever the socket is actually connected, so an intermediary
+ * relays the session without being able to read it.
+ *
+ * This class is not part of the public API and may be removed or changed at any time
+ */
+public final class KmsSocketConnector {
+
+ /**
+ * Connects to the KMS host and returns a socket with an established TLS session.
+ *
+ * @param sslContext the SSL context configured for the KMS provider, or null to use the default
+ * @param kmsConnectCallback the callback that establishes the connection, or null to connect directly
+ * @param kmsAddress the address of the KMS host
+ * @param soTimeoutMillis the socket read timeout to apply
+ * @param connectTimeoutMillis the time available to establish the connection, which is never larger than
+ * {@code soTimeoutMillis}
+ * @return a connected socket with an established TLS session with the KMS host
+ * @throws IOException if the connection or the TLS handshake fails
+ */
+ public static SSLSocket connect(@Nullable final SSLContext sslContext,
+ @Nullable final KmsConnectCallback kmsConnectCallback, final ServerAddress kmsAddress,
+ final int soTimeoutMillis, final long connectTimeoutMillis) throws IOException {
+ SSLSocketFactory sslSocketFactory = sslContext == null
+ ? (SSLSocketFactory) SSLSocketFactory.getDefault() : sslContext.getSocketFactory();
+
+ if (kmsConnectCallback == null) {
+ return connectDirectly(sslSocketFactory, kmsAddress, soTimeoutMillis, connectTimeoutMillis);
+ }
+ return connectViaCallback(sslSocketFactory, kmsConnectCallback, kmsAddress, connectTimeoutMillis);
+ }
+
+ private static SSLSocket connectDirectly(final SSLSocketFactory sslSocketFactory, final ServerAddress kmsAddress,
+ final int soTimeoutMillis, final long connectTimeoutMillis) throws IOException {
+ SSLSocket socket = (SSLSocket) sslSocketFactory.createSocket();
+ enableHostNameVerification(socket, null);
+ try {
+ socket.setSoTimeout(soTimeoutMillis);
+ socket.connect(new InetSocketAddress(InetAddress.getByName(kmsAddress.getHost()), kmsAddress.getPort()),
+ Math.toIntExact(connectTimeoutMillis));
+ } catch (IOException | RuntimeException e) {
+ closeSocket(socket);
+ throw e;
+ }
+ return socket;
+ }
+
+ /**
+ * Obtains a socket from the callback, then layers TLS for the KMS host on top of it. The callback may have
+ * connected to an intermediary, so Server Name Indication and hostname verification are configured from
+ * {@code kmsAddress} rather than from the address the socket is actually connected to.
+ */
+ private static SSLSocket connectViaCallback(final SSLSocketFactory sslSocketFactory,
+ final KmsConnectCallback kmsConnectCallback, final ServerAddress kmsAddress,
+ final long connectTimeoutMillis) throws IOException {
+ Socket connectedSocket = notNull("socket returned by KmsConnectCallback",
+ kmsConnectCallback.connect(new KmsConnectContext(kmsAddress.getHost(), kmsAddress.getPort())));
+
+ SSLSocket socket;
+ try {
+ // Layers TLS over the already-connected socket. autoClose ensures that closing the returned socket also
+ // closes the socket the callback established.
+ socket = (SSLSocket) sslSocketFactory.createSocket(connectedSocket, kmsAddress.getHost(), kmsAddress.getPort(), true);
+ } catch (IOException | RuntimeException e) {
+ closeSocket(connectedSocket);
+ throw e;
+ }
+
+ try {
+ // Even though the callback's connection is already established, the TLS handshake has not been performed
+ // yet, so SSL parameters can still be set. They target the KMS host, not any intermediary.
+ enableHostNameVerification(socket, kmsAddress.getHost());
+ // The callback is not given a deadline, so bound the driver's own handshake by the time the operation has
+ // left. KeyManagementService re-checks expiry once this returns.
+ socket.setSoTimeout(Math.toIntExact(connectTimeoutMillis));
+ // Handshake explicitly so that a TLS failure is reported here rather than on the first write.
+ socket.startHandshake();
+ } catch (IOException | RuntimeException e) {
+ closeSocket(socket);
+ throw e;
+ }
+ return socket;
+ }
+
+ /**
+ * Moved from {@code KeyManagementService}, with {@code sniHost} added for the callback path: the socket returned by
+ * a {@link KmsConnectCallback} may be connected to an intermediary, so the KMS host has to be named explicitly.
+ *
+ * The null check is necessary despite the {@link SSLSocket#getSSLParameters()} contract, as some
+ * implementations return null. See JAVA-2876.
+ *
+ * @param sniHost the host to announce via Server Name Indication, or null to leave it unset
+ */
+ private static void enableHostNameVerification(final SSLSocket socket, @Nullable final String sniHost) {
+ SSLParameters sslParameters = socket.getSSLParameters();
+ if (sslParameters == null) {
+ sslParameters = new SSLParameters();
+ }
+ if (sniHost != null) {
+ SslHelper.enableSni(sniHost, sslParameters);
+ }
+ SslHelper.enableHostNameVerification(sslParameters);
+ socket.setSSLParameters(sslParameters);
+ }
+
+ private static void closeSocket(final Socket socket) {
+ try {
+ socket.close();
+ } catch (IOException | RuntimeException e) {
+ // ignore
+ }
+ }
+
+ private KmsSocketConnector() {
+ }
+}
diff --git a/driver-sync/src/test/functional/com/mongodb/client/AbstractClientSideEncryptionKmsConnectCallbackProseTest.java b/driver-sync/src/test/functional/com/mongodb/client/AbstractClientSideEncryptionKmsConnectCallbackProseTest.java
new file mode 100644
index 00000000000..91973e09d48
--- /dev/null
+++ b/driver-sync/src/test/functional/com/mongodb/client/AbstractClientSideEncryptionKmsConnectCallbackProseTest.java
@@ -0,0 +1,416 @@
+/*
+ * Copyright 2008-present MongoDB, Inc.
+ *
+ * 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 com.mongodb.client;
+
+import com.mongodb.AutoEncryptionSettings;
+import com.mongodb.ClientEncryptionSettings;
+import com.mongodb.MongoClientSettings;
+import com.mongodb.KmsConnectCallback;
+import com.mongodb.client.model.Filters;
+import com.mongodb.client.model.vault.DataKeyOptions;
+import com.mongodb.client.vault.ClientEncryption;
+import com.mongodb.lang.Nullable;
+import org.bson.BsonBinary;
+import org.bson.BsonDocument;
+import org.bson.Document;
+import org.junit.jupiter.api.DisplayName;
+import org.junit.jupiter.api.Test;
+
+import javax.net.ssl.HttpsURLConnection;
+import javax.net.ssl.SSLContext;
+import javax.net.ssl.TrustManagerFactory;
+import java.io.ByteArrayOutputStream;
+import java.io.File;
+import java.io.IOException;
+import java.io.InputStream;
+import java.io.UncheckedIOException;
+import java.net.InetSocketAddress;
+import java.net.Socket;
+import java.net.HttpURLConnection;
+import java.net.URL;
+import java.nio.charset.StandardCharsets;
+import java.nio.file.Files;
+import java.nio.file.Paths;
+import java.security.GeneralSecurityException;
+import java.security.KeyStore;
+import java.security.cert.CertificateFactory;
+import java.security.cert.X509Certificate;
+import java.util.Base64;
+import java.util.HashMap;
+import java.util.Map;
+import java.util.regex.Matcher;
+import java.util.regex.Pattern;
+
+import static com.mongodb.ClusterFixture.isClientSideEncryptionTest;
+import static com.mongodb.client.Fixture.getMongoClientSettingsBuilder;
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertInstanceOf;
+import static org.junit.jupiter.api.Assertions.assertNotNull;
+import static org.junit.jupiter.api.Assertions.assertThrows;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+import static org.junit.jupiter.api.Assumptions.assumeTrue;
+
+/**
+ * Prose test 28, "KMS Connect Callback".
+ *
+ * All cases require real AWS KMS credentials and are skipped when they are not available. The KMS HTTP proxy is
+ * started by {@code drivers-evergreen-tools}: {@code .evergreen/csfle/start-servers.sh} runs {@code kms_http_proxy.py}
+ * on port 9004 in plain HTTP mode and on port 9005 in HTTPS mode. Cases are skipped when the proxy is unreachable, so
+ * that this test does not fail when run outside that environment.
+ *
+ * Case 6, "Retry", is not implemented: it is to be skipped by drivers that do not implement DRIVERS-1541, and this
+ * driver does not retry KMS requests. TODO-JAVA-5391
+ *
+ * @see
+ * Prose test 28
+ */
+public abstract class AbstractClientSideEncryptionKmsConnectCallbackProseTest {
+
+ private static final String PROXY_HOST = "127.0.0.1";
+ private static final int HTTP_PROXY_PORT = 9004;
+ private static final int HTTPS_PROXY_PORT = 9005;
+ private static final int PROXY_TIMEOUT_MILLIS = 10000;
+
+ private static final String KEY_VAULT_NAMESPACE = "keyvault.datakeys";
+ private static final String MASTER_KEY = "{"
+ + "region: \"us-east-1\", "
+ + "key: \"arn:aws:kms:us-east-1:579766882180:key/89fcc2c4-08b0-4bd9-9f25-e30687b580d0\"}";
+
+ private static final Pattern CONNECT_COUNT_PATTERN = Pattern.compile("connect_count (\\d+)");
+
+ protected abstract ClientEncryption createClientEncryption(ClientEncryptionSettings settings);
+
+ protected abstract MongoClient createMongoClient(MongoClientSettings settings);
+
+
+ @Test
+ @DisplayName("Case 1: plain HTTP proxy")
+ void testPlainHttpProxy() throws IOException {
+ assumeTrue(isClientSideEncryptionTest(), "Requires AWS KMS credentials");
+ assumeProxyIsRunning(false);
+ resetMetrics(false);
+
+ try (ClientEncryption clientEncryption = createClientEncryption(clientEncryptionSettingsBuilder()
+ .kmsConnectCallback(proxyConnectCallback(false))
+ .build())) {
+ assertNotNull(createDataKey(clientEncryption));
+ }
+
+ assertTrue(getConnectCount(false) >= 1, "expected the KMS request to be routed through the proxy");
+ }
+
+ @Test
+ @DisplayName("Case 2: HTTPS proxy")
+ void testHttpsProxy() throws IOException {
+ assumeTrue(isClientSideEncryptionTest(), "Requires AWS KMS credentials");
+ assumeTrue(caFile() != null, "Requires the proxy's CA file, e.g. $DRIVERS_TOOLS/.evergreen/x509gen/ca.pem");
+ assumeProxyIsRunning(true);
+ resetMetrics(true);
+
+ // Two independent TLS layers are in play here: the driver's connection to the proxy, verified against the
+ // proxy's CA, and its connection to the KMS host, carried end-to-end through the CONNECT tunnel and verified
+ // against the real KMS host's certificate. Creating the data key confirms the driver verified the KMS host's
+ // identity rather than the proxy's.
+ try (ClientEncryption clientEncryption = createClientEncryption(clientEncryptionSettingsBuilder()
+ .kmsConnectCallback(proxyConnectCallback(true))
+ .build())) {
+ assertNotNull(createDataKey(clientEncryption));
+ }
+
+ assertTrue(getConnectCount(true) >= 1, "expected the KMS request to be routed through the proxy");
+ }
+
+ @Test
+ @DisplayName("Case 3: full auto encryption pipeline via proxy")
+ void testAutoEncryptionPipelineViaProxy() throws IOException {
+ assumeTrue(isClientSideEncryptionTest(), "Requires AWS KMS credentials");
+ assumeProxyIsRunning(false);
+
+ try (MongoClient client = createMongoClient(getMongoClientSettingsBuilder().build())) {
+ client.getDatabase("keyvault").getCollection("datakeys").drop();
+ client.getDatabase("db").getCollection("coll").drop();
+
+ BsonBinary dataKeyId;
+ try (ClientEncryption clientEncryption = createClientEncryption(clientEncryptionSettingsBuilder()
+ .kmsConnectCallback(proxyConnectCallback(false))
+ .build())) {
+ dataKeyId = createDataKey(clientEncryption);
+ assertNotNull(dataKeyId);
+ }
+
+ Map schemaMap = new HashMap<>();
+ schemaMap.put("db.coll", schemaForDataKey(dataKeyId));
+
+ resetMetrics(false);
+
+ AutoEncryptionSettings autoEncryptionSettings = AutoEncryptionSettings.builder()
+ .keyVaultNamespace(KEY_VAULT_NAMESPACE)
+ .kmsProviders(awsKmsProviders())
+ .schemaMap(schemaMap)
+ .kmsConnectCallback(proxyConnectCallback(false))
+ .build();
+
+ try (MongoClient encryptedClient = createMongoClient(getMongoClientSettingsBuilder()
+ .autoEncryptionSettings(autoEncryptionSettings)
+ .build())) {
+ MongoCollection encryptedColl = encryptedClient.getDatabase("db").getCollection("coll");
+ encryptedColl.insertOne(new Document("_id", 1).append("encrypted_string", "hello"));
+
+ Document decrypted = encryptedColl.find(Filters.eq("_id", 1)).first();
+ assertNotNull(decrypted);
+ assertEquals("hello", decrypted.get("encrypted_string"));
+ }
+
+ // read with the unencrypted client to confirm the value is stored encrypted
+ Document stored = client.getDatabase("db").getCollection("coll").find(Filters.eq("_id", 1)).first();
+ assertNotNull(stored);
+ assertInstanceOf(org.bson.types.Binary.class, stored.get("encrypted_string"));
+ }
+
+ // Exactly one KMS request is expected since the decrypted key is cached; more than one would indicate a
+ // data encryption key caching regression.
+ assertEquals(1, getConnectCount(false), "expected exactly one KMS request through the proxy");
+ }
+
+ @Test
+ @DisplayName("Case 4: Error")
+ void testCallbackError() {
+ // The callback fails before any KMS request is made, so no real credentials are needed and this case runs
+ // wherever a MongoDB deployment is available.
+ String failureMessage = "Proxy refused the CONNECT request";
+ KmsConnectCallback failingCallback = context -> {
+ throw new IOException(failureMessage);
+ };
+
+ try (ClientEncryption clientEncryption = createClientEncryption(clientEncryptionSettingsBuilder()
+ .kmsProviders(placeholderAwsKmsProviders())
+ .kmsConnectCallback(failingCallback)
+ .build())) {
+ RuntimeException e = assertThrows(RuntimeException.class, () -> createDataKey(clientEncryption));
+ assertTrue(causeChainContains(e, failureMessage),
+ () -> "expected the callback's failure to be reported, but got: " + e);
+ }
+ }
+
+ // Case 5 (callback receives timeout) is omitted because the callback is not given a deadline: the driver applies
+ // the operation's remaining time to its own connect and TLS handshake instead of exposing it to the callback.
+
+ /**
+ * Returns a callback that tunnels to the KMS host through the proxy using HTTP CONNECT, as described in the Setup
+ * section of prose test 28:
+ *
+ * - Accept {@code (host, port)} from the driver.
+ * - Open a connection to the proxy — plain TCP on port 9004, or TLS verified with {@code ca.pem} on 9005.
+ * - Send {@code CONNECT : HTTP/1.1\r\nHost: :\r\n\r\n}.
+ * - Read the response and verify it begins with {@code HTTP/1.1 200}.
+ * - Return the socket, over which the driver negotiates TLS with the KMS host.
+ *
+ */
+ private KmsConnectCallback proxyConnectCallback(final boolean useTls) {
+ return context -> {
+ // The context carries no deadline, so the callback chooses its own timeout for reaching the proxy.
+ int timeout = PROXY_TIMEOUT_MILLIS;
+ Socket socket = useTls
+ ? proxySslContext().getSocketFactory().createSocket()
+ : new Socket();
+ try {
+ socket.connect(new InetSocketAddress(PROXY_HOST, useTls ? HTTPS_PROXY_PORT : HTTP_PROXY_PORT), timeout);
+ socket.setSoTimeout(timeout);
+ String target = context.getHost() + ":" + context.getPort();
+ socket.getOutputStream().write(("CONNECT " + target + " HTTP/1.1\r\nHost: " + target + "\r\n\r\n")
+ .getBytes(StandardCharsets.US_ASCII));
+ socket.getOutputStream().flush();
+ // One byte at a time, so that no byte of the driver's TLS handshake is consumed.
+ StringBuilder headers = new StringBuilder();
+ while (!headers.toString().endsWith("\r\n\r\n")) {
+ int b = socket.getInputStream().read();
+ if (b == -1) {
+ throw new IOException("Proxy closed the connection before completing CONNECT: " + headers);
+ }
+ headers.append((char) b);
+ }
+ String statusLine = headers.substring(0, headers.indexOf("\r\n"));
+ if (!statusLine.startsWith("HTTP/1.1 200")) {
+ throw new IOException("Proxy refused the CONNECT request: " + statusLine);
+ }
+ return socket;
+ } catch (IOException | RuntimeException e) {
+ socket.close();
+ throw e;
+ }
+ };
+ }
+
+ private static boolean causeChainContains(final Throwable throwable, final String message) {
+ for (Throwable t = throwable; t != null; t = t.getCause()) {
+ if (t.getMessage() != null && t.getMessage().contains(message)) {
+ return true;
+ }
+ }
+ return false;
+ }
+
+ private ClientEncryptionSettings.Builder clientEncryptionSettingsBuilder() {
+ return ClientEncryptionSettings.builder()
+ .keyVaultMongoClientSettings(getMongoClientSettingsBuilder().build())
+ .keyVaultNamespace(KEY_VAULT_NAMESPACE)
+ .kmsProviders(awsKmsProviders());
+ }
+
+ /**
+ * Credentials that are never used, for the case whose callback fails before any KMS request is made.
+ */
+ private static Map> placeholderAwsKmsProviders() {
+ Map aws = new HashMap<>();
+ aws.put("accessKeyId", "fakeAccessKeyId");
+ aws.put("secretAccessKey", "fakeSecretAccessKey");
+ Map> kmsProviders = new HashMap<>();
+ kmsProviders.put("aws", aws);
+ return kmsProviders;
+ }
+
+ private static Map> awsKmsProviders() {
+ Map aws = new HashMap<>();
+ aws.put("accessKeyId", System.getenv("AWS_ACCESS_KEY_ID"));
+ aws.put("secretAccessKey", System.getenv("AWS_SECRET_ACCESS_KEY"));
+ Map> kmsProviders = new HashMap<>();
+ kmsProviders.put("aws", aws);
+ return kmsProviders;
+ }
+
+ private static BsonBinary createDataKey(final ClientEncryption clientEncryption) {
+ return clientEncryption.createDataKey("aws", new DataKeyOptions().masterKey(BsonDocument.parse(MASTER_KEY)));
+ }
+
+ private static BsonDocument schemaForDataKey(final BsonBinary dataKeyId) {
+ String base64DataKeyId = Base64.getEncoder().encodeToString(dataKeyId.getData());
+ return BsonDocument.parse("{"
+ + " bsonType: \"object\","
+ + " properties: {"
+ + " encrypted_string: {"
+ + " encrypt: {"
+ + " keyId: [{\"$binary\": {\"base64\": \"" + base64DataKeyId + "\", \"subType\": \"04\"}}],"
+ + " bsonType: \"string\","
+ + " algorithm: \"AEAD_AES_256_CBC_HMAC_SHA_512-Deterministic\""
+ + " }"
+ + " }"
+ + " }"
+ + "}");
+ }
+
+ // --- the proxy's control endpoints -------------------------------------------------------------------------
+
+ private void assumeProxyIsRunning(final boolean useTls) {
+ try {
+ getConnectCount(useTls);
+ } catch (IOException e) {
+ assumeTrue(false, "KMS HTTP proxy is not running on port "
+ + (useTls ? HTTPS_PROXY_PORT : HTTP_PROXY_PORT) + ": " + e.getMessage());
+ }
+ }
+
+ private void resetMetrics(final boolean useTls) throws IOException {
+ readControlResponse("/reset", "POST", useTls);
+ }
+
+ private int getConnectCount(final boolean useTls) throws IOException {
+ String body = readControlResponse("/metrics", "GET", useTls);
+ Matcher matcher = CONNECT_COUNT_PATTERN.matcher(body);
+ if (!matcher.find()) {
+ throw new IOException("Could not find connect_count in the proxy's metrics response: " + body);
+ }
+ return Integer.parseInt(matcher.group(1));
+ }
+
+ private String readControlResponse(final String path, final String method, final boolean useTls) throws IOException {
+ int port = useTls ? HTTPS_PROXY_PORT : HTTP_PROXY_PORT;
+ URL url = new URL((useTls ? "https" : "http") + "://" + PROXY_HOST + ":" + port + path);
+ HttpURLConnection connection = (HttpURLConnection) url.openConnection();
+ if (useTls) {
+ HttpsURLConnection httpsConnection = (HttpsURLConnection) connection;
+ httpsConnection.setSSLSocketFactory(proxySslContext().getSocketFactory());
+ // The proxy's certificate is verified against its CA above. Its subject does not necessarily match the
+ // loopback address that the control endpoints are reached on, which is immaterial for these tests.
+ httpsConnection.setHostnameVerifier((hostname, session) -> PROXY_HOST.equals(hostname));
+ }
+ connection.setRequestMethod(method);
+ connection.setConnectTimeout(5000);
+ connection.setReadTimeout(5000);
+ try (InputStream inputStream = connection.getInputStream()) {
+ ByteArrayOutputStream body = new ByteArrayOutputStream();
+ byte[] buffer = new byte[512];
+ int read;
+ while ((read = inputStream.read(buffer)) != -1) {
+ body.write(buffer, 0, read);
+ }
+ return new String(body.toByteArray(), StandardCharsets.UTF_8);
+ } finally {
+ connection.disconnect();
+ }
+ }
+
+ // --- the proxy's CA ----------------------------------------------------------------------------------------
+
+ private static SSLContext proxySslContext;
+
+ private static SSLContext proxySslContext() {
+ if (proxySslContext == null) {
+ proxySslContext = buildProxySslContext();
+ }
+ return proxySslContext;
+ }
+
+ private static SSLContext buildProxySslContext() {
+ String caFile = caFile();
+ assertNotNull(caFile, "the proxy's CA file could not be located");
+ try (InputStream caStream = Files.newInputStream(Paths.get(caFile))) {
+ X509Certificate ca = (X509Certificate) CertificateFactory.getInstance("X.509").generateCertificate(caStream);
+ KeyStore trustStore = KeyStore.getInstance(KeyStore.getDefaultType());
+ trustStore.load(null, null);
+ trustStore.setCertificateEntry("csfle-proxy-ca", ca);
+ TrustManagerFactory trustManagerFactory =
+ TrustManagerFactory.getInstance(TrustManagerFactory.getDefaultAlgorithm());
+ trustManagerFactory.init(trustStore);
+ SSLContext sslContext = SSLContext.getInstance("TLS");
+ sslContext.init(null, trustManagerFactory.getTrustManagers(), null);
+ return sslContext;
+ } catch (IOException e) {
+ throw new UncheckedIOException(e);
+ } catch (GeneralSecurityException e) {
+ throw new IllegalStateException("Could not build an SSLContext trusting the proxy's CA", e);
+ }
+ }
+
+ /**
+ * @return the path to the CA certificate that signed the HTTPS proxy's certificate, or null if it cannot be found,
+ * in which case the HTTPS proxy case is skipped.
+ */
+ @Nullable
+ private static String caFile() {
+ String caFile = System.getProperty("org.mongodb.test.csfle.tls.ca.file");
+ if (caFile == null) {
+ caFile = System.getenv("CSFLE_TLS_CA_FILE");
+ }
+ if (caFile == null) {
+ String driversTools = System.getenv("DRIVERS_TOOLS");
+ if (driversTools != null) {
+ caFile = driversTools + "/.evergreen/x509gen/ca.pem";
+ }
+ }
+ return caFile != null && new File(caFile).isFile() ? caFile : null;
+ }
+}
diff --git a/driver-sync/src/test/functional/com/mongodb/client/ClientSideEncryptionKmsConnectCallbackProseTest.java b/driver-sync/src/test/functional/com/mongodb/client/ClientSideEncryptionKmsConnectCallbackProseTest.java
new file mode 100644
index 00000000000..cd6f79494d8
--- /dev/null
+++ b/driver-sync/src/test/functional/com/mongodb/client/ClientSideEncryptionKmsConnectCallbackProseTest.java
@@ -0,0 +1,34 @@
+/*
+ * Copyright 2008-present MongoDB, Inc.
+ *
+ * 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 com.mongodb.client;
+
+import com.mongodb.ClientEncryptionSettings;
+import com.mongodb.MongoClientSettings;
+import com.mongodb.client.vault.ClientEncryption;
+import com.mongodb.client.vault.ClientEncryptions;
+
+public class ClientSideEncryptionKmsConnectCallbackProseTest extends AbstractClientSideEncryptionKmsConnectCallbackProseTest {
+ @Override
+ protected ClientEncryption createClientEncryption(final ClientEncryptionSettings settings) {
+ return ClientEncryptions.create(settings);
+ }
+
+ @Override
+ protected MongoClient createMongoClient(final MongoClientSettings settings) {
+ return MongoClients.create(settings);
+ }
+}
diff --git a/driver-sync/src/test/unit/com/mongodb/client/internal/KeyManagementServiceTest.java b/driver-sync/src/test/unit/com/mongodb/client/internal/KeyManagementServiceTest.java
new file mode 100644
index 00000000000..38b74035f40
--- /dev/null
+++ b/driver-sync/src/test/unit/com/mongodb/client/internal/KeyManagementServiceTest.java
@@ -0,0 +1,70 @@
+/*
+ * Copyright 2008-present MongoDB, Inc.
+ *
+ * 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 com.mongodb.client.internal;
+
+import com.mongodb.MongoOperationTimeoutException;
+import com.mongodb.internal.time.Timeout;
+import org.junit.jupiter.api.Test;
+
+import static com.mongodb.internal.time.Timeout.ZeroSemantics.ZERO_DURATION_MEANS_EXPIRED;
+import static java.util.Collections.emptyMap;
+import static java.util.concurrent.TimeUnit.MILLISECONDS;
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertThrows;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+/**
+ * The specification requires that a driver supporting CSOT pass the remaining {@code timeoutMS} when establishing a
+ * connection to a KMS host. Prose test 28 asserts this by observing a {@code kmsConnectCallback}; this covers the same
+ * requirement directly, including the branches that prose test does not reach.
+ */
+final class KeyManagementServiceTest {
+
+ private static final int CONNECT_TIMEOUT_MILLIS = 10_000;
+
+ private final KeyManagementService keyManagementService =
+ new KeyManagementService(emptyMap(), null, CONNECT_TIMEOUT_MILLIS);
+
+ @Test
+ void shouldUseConfiguredConnectTimeoutWhenNoOperationTimeoutApplies() {
+ assertEquals(CONNECT_TIMEOUT_MILLIS, keyManagementService.remainingMillis(null));
+ }
+
+ @Test
+ void shouldPassRemainingOperationTimeoutWhenItIsShorter() {
+ Timeout operationTimeout = Timeout.expiresIn(500, MILLISECONDS, ZERO_DURATION_MEANS_EXPIRED);
+
+ long remaining = keyManagementService.remainingMillis(operationTimeout);
+
+ assertTrue(remaining > 0 && remaining <= 500,
+ () -> "expected the remaining operation timeout to be passed, but got " + remaining);
+ }
+
+ @Test
+ void shouldNotExceedConfiguredConnectTimeoutWhenOperationTimeoutIsLonger() {
+ Timeout operationTimeout = Timeout.expiresIn(60_000, MILLISECONDS, ZERO_DURATION_MEANS_EXPIRED);
+
+ assertEquals(CONNECT_TIMEOUT_MILLIS, keyManagementService.remainingMillis(operationTimeout));
+ }
+
+ @Test
+ void shouldThrowWhenOperationTimeoutHasExpired() {
+ Timeout expired = Timeout.expiresIn(0, MILLISECONDS, ZERO_DURATION_MEANS_EXPIRED);
+
+ assertThrows(MongoOperationTimeoutException.class, () -> keyManagementService.remainingMillis(expired));
+ }
+}