CSHARP-2431: Support Client-side Field Level Encryption.

src/MongoDB.Bson/IO/ElementAppendingBsonWriter.cs

@@ -74,6 +74,30 @@ public override void WriteEndDocument()
             base.WriteEndDocument();
         }
 
+        public override void WriteRawBsonDocument(IByteBuffer slice)
+        {
+            WriteStartDocument();
+
+            if (Wrapped is BsonBinaryWriter binaryWriter)
+            {
+                // just copy the bytes (without the length and terminating null)
+                var lengthBytes = new byte[4];
+                slice.GetBytes(0, lengthBytes, 0, 4);
+                var length = BitConverter.ToInt32(lengthBytes, 0);
+                using (var elements = slice.GetSlice(4, length - 5))
+                {
+                    var stream = binaryWriter.BsonStream;
+                    stream.WriteSlice(elements);
+                }
+            }
+            else
+            {
+                throw new NotSupportedException("WriteRawBsonDocument supports only BsonBinaryWriter.");
+            }
+
+            WriteEndDocument();
+        }
+
         /// <inheritdoc />
         public override void WriteStartDocument()
         {

src/MongoDB.Bson/IO/JsonReader.cs

@@ -1195,7 +1195,7 @@ private BsonValue ParseDateTimeExtendedJson()
             }
             else if (valueToken.Type == JsonTokenType.BeginObject)
             {
-                VerifyToken("$numberLong");
+                VerifyString("$numberLong");
                 VerifyToken(":");
                 var millisecondsSinceEpochToken = PopToken();
                 if (millisecondsSinceEpochToken.Type == JsonTokenType.String)

src/MongoDB.Bson/ObjectModel/BsonBinarySubType.cs

@@ -51,6 +51,10 @@ public enum BsonBinarySubType
         /// </summary>
         MD5 = 0x05,
         /// <summary>
+        /// Encrypted binary data.
+        /// </summary>
+        Encrypted = 0x06,
+        /// <summary>
         /// User defined binary data.
         /// </summary>
         UserDefined = 0x80

src/MongoDB.Driver.Core/Core/Clusters/Cluster.cs

@@ -26,6 +26,7 @@
 using MongoDB.Driver.Core.Events;
 using MongoDB.Driver.Core.Misc;
 using MongoDB.Driver.Core.Servers;
+using MongoDB.Libmongocrypt;
 
 namespace MongoDB.Driver.Core.Clusters
 {
@@ -62,6 +63,7 @@ internal abstract class Cluster : ICluster
         // fields
         private readonly IClusterClock _clusterClock = new ClusterClock();
         private readonly ClusterId _clusterId;
+        private CryptClient _cryptClient = null;
         private ClusterDescription _description;
         private TaskCompletionSource<bool> _descriptionChangedTaskCompletionSource;
         private readonly object _descriptionLock = new object();
@@ -109,6 +111,11 @@ public ClusterId ClusterId
             get { return _clusterId; }
         }
 
+        public CryptClient CryptClient
+        {
+            get { return _cryptClient; }
+        }
+
         public ClusterDescription Description
         {
             get
@@ -188,7 +195,13 @@ private void ExitServerSelectionWaitQueue()
         public virtual void Initialize()
         {
             ThrowIfDisposed();
-            _state.TryChange(State.Initial, State.Open);
+            if (_state.TryChange(State.Initial, State.Open))
+            {
+                if (_settings.KmsProviders != null || _settings.SchemaMap != null)
+                {
+                    _cryptClient = CryptClientCreator.CreateCryptClient(_settings.KmsProviders, _settings.SchemaMap);
+                }
+            }
         }
 
         private void RapidHeartbeatTimerCallback(object args)
@@ -350,7 +363,7 @@ private async Task WaitForDescriptionChangedAsync(IServerSelector selector, Clus
         {
             using (var helper = new WaitForDescriptionChangedHelper(this, selector, description, descriptionChangedTask, timeout, cancellationToken))
             {
-                var completedTask  = await Task.WhenAny(helper.Tasks).ConfigureAwait(false);
+                var completedTask = await Task.WhenAny(helper.Tasks).ConfigureAwait(false);
                 helper.HandleCompletedTask(completedTask);
             }
         }
@@ -528,7 +541,7 @@ private sealed class WaitForDescriptionChangedHelper : IDisposable
             private readonly CancellationTokenSource _timeoutCancellationTokenSource;
             private readonly Task _timeoutTask;
 
-            public  WaitForDescriptionChangedHelper(Cluster cluster, IServerSelector selector, ClusterDescription description, Task descriptionChangedTask , TimeSpan timeout, CancellationToken cancellationToken)
+            public WaitForDescriptionChangedHelper(Cluster cluster, IServerSelector selector, ClusterDescription description, Task descriptionChangedTask, TimeSpan timeout, CancellationToken cancellationToken)
             {
                 _cluster = cluster;
                 _description = description;

src/MongoDB.Driver.Core/Core/Clusters/CryptClientCreator.cs

@@ -0,0 +1,105 @@
+/* Copyright 2019-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.
+*/
+
+using System;
+using System.Collections.Generic;
+using System.Linq;
+using MongoDB.Bson;
+using MongoDB.Bson.IO;
+using MongoDB.Driver.Core.Misc;
+using MongoDB.Libmongocrypt;
+
+namespace MongoDB.Driver.Core.Clusters
+{
+    /// <summary>
+    /// Represents a creator for CryptClient.
+    /// </summary>
+    public sealed class CryptClientCreator
+    {
+        #region static
+#pragma warning disable 3002
+        /// <summary>
+        /// Create a CryptClient instance.
+        /// </summary>
+        /// <param name="kmsProviders">The kms providers.</param>
+        /// <param name="schemaMap">The schema map.</param>
+        /// <returns>The CryptClient instance.</returns>
+        public static CryptClient CreateCryptClient(
+            IReadOnlyDictionary<string, IReadOnlyDictionary<string, object>> kmsProviders,
+            IReadOnlyDictionary<string, BsonDocument> schemaMap)
+        {
+            var helper = new CryptClientCreator(kmsProviders, schemaMap);
+            var cryptOptions = helper.CreateCryptOptions();
+            return helper.CreateCryptClient(cryptOptions);
+        }
+#pragma warning restore
+        #endregion
+
+        private readonly IReadOnlyDictionary<string, IReadOnlyDictionary<string, object>> _kmsProviders;
+        private readonly IReadOnlyDictionary<string, BsonDocument> _schemaMap;
+
+        private CryptClientCreator(
+            IReadOnlyDictionary<string, IReadOnlyDictionary<string, object>> kmsProviders,
+            IReadOnlyDictionary<string, BsonDocument> schemaMap)
+        {
+            _kmsProviders = Ensure.IsNotNull(kmsProviders, nameof(kmsProviders));
+            _schemaMap = schemaMap;
+        }
+
+        private CryptClient CreateCryptClient(CryptOptions options)
+        {
+            return CryptClientFactory.Create(options);
+        }
+
+        private CryptOptions CreateCryptOptions()
+        {
+            Dictionary<KmsType, IKmsCredentials> kmsProvidersMap = null;
+            if (_kmsProviders != null && _kmsProviders.Count > 0)
+            {
+                kmsProvidersMap = new Dictionary<KmsType, IKmsCredentials>();
+                if (_kmsProviders.TryGetValue("aws", out var awsProvider))
+                {
+                    if (awsProvider.TryGetValue("accessKeyId", out var accessKeyId) &&
+                        awsProvider.TryGetValue("secretAccessKey", out var secretAccessKey))
+                    {
+                        kmsProvidersMap.Add(KmsType.Aws, new AwsKmsCredentials((string)secretAccessKey, (string)accessKeyId));
+                    }
+                }
+                if (_kmsProviders.TryGetValue("local", out var localProvider))
+                {
+                    if (localProvider.TryGetValue("key", out var keyObject) && keyObject is byte[] key)
+                    {
+                        kmsProvidersMap.Add(KmsType.Local, new LocalKmsCredentials(key));
+                    }
+                }
+            }
+            else
+            {
+                throw new ArgumentException("At least one kms provider must be specified");
+            }
+
+            byte[] schemaBytes = null;
+            if (_schemaMap != null)
+            {
+                var schemaMapElements = _schemaMap.Select(c => new BsonElement(c.Key, c.Value));
+                var schemaDocument = new BsonDocument(schemaMapElements);
+                var writerSettings = new BsonBinaryWriterSettings { GuidRepresentation = GuidRepresentation.Unspecified };
+                schemaBytes = schemaDocument.ToBson(writerSettings: writerSettings);
+            }
+
+            return new CryptOptions(kmsProvidersMap, schemaBytes);
+        }
+    }
+}

src/MongoDB.Driver.Core/Core/Clusters/ICluster.cs

@@ -20,6 +20,7 @@
 using MongoDB.Driver.Core.Clusters.ServerSelectors;
 using MongoDB.Driver.Core.Configuration;
 using MongoDB.Driver.Core.Servers;
+using MongoDB.Libmongocrypt;
 
 namespace MongoDB.Driver.Core.Clusters
 {
@@ -66,6 +67,14 @@ public interface ICluster : IDisposable
         /// <returns>A core server session.</returns>
         ICoreServerSession AcquireServerSession();
 
+        /// <summary>
+        /// Gets the crypt client.
+        /// </summary>
+        /// <returns>A crypt client.</returns>
+#pragma warning disable CS3003
+        CryptClient CryptClient { get; }
+#pragma warning restore
+
         /// <summary>
         /// Initializes the cluster.
         /// </summary>

src/MongoDB.Driver.Core/Core/Configuration/ClusterSettings.cs

@@ -17,6 +17,7 @@
 using System.Collections.Generic;
 using System.Linq;
 using System.Net;
+using MongoDB.Bson;
 using MongoDB.Driver.Core.Clusters;
 using MongoDB.Driver.Core.Clusters.ServerSelectors;
 using MongoDB.Driver.Core.Misc;
@@ -36,8 +37,10 @@ public class ClusterSettings
         // fields
         private readonly ClusterConnectionMode _connectionMode;
         private readonly IReadOnlyList<EndPoint> _endPoints;
+        private readonly IReadOnlyDictionary<string, IReadOnlyDictionary<string, object>> _kmsProviders;
         private readonly int _maxServerSelectionWaitQueueSize;
         private readonly string _replicaSetName;
+        private readonly IReadOnlyDictionary<string, BsonDocument> _schemaMap;
         private readonly ConnectionStringScheme _scheme;
         private readonly TimeSpan _serverSelectionTimeout;
         private readonly IServerSelector _preServerSelector;
@@ -49,30 +52,36 @@ public class ClusterSettings
         /// </summary>
         /// <param name="connectionMode">The connection mode.</param>
         /// <param name="endPoints">The end points.</param>
+        /// <param name="kmsProviders">The kms providers.</param>
         /// <param name="maxServerSelectionWaitQueueSize">Maximum size of the server selection wait queue.</param>
         /// <param name="replicaSetName">Name of the replica set.</param>
         /// <param name="serverSelectionTimeout">The server selection timeout.</param>
         /// <param name="preServerSelector">The pre server selector.</param>
         /// <param name="postServerSelector">The post server selector.</param>
+        /// <param name="schemaMap">The schema map.</param>
         /// <param name="scheme">The connection string scheme.</param>
         public ClusterSettings(
             Optional<ClusterConnectionMode> connectionMode = default(Optional<ClusterConnectionMode>),
             Optional<IEnumerable<EndPoint>> endPoints = default(Optional<IEnumerable<EndPoint>>),
+            Optional<IReadOnlyDictionary<string, IReadOnlyDictionary<string, object>>> kmsProviders = default(Optional<IReadOnlyDictionary<string, IReadOnlyDictionary<string, object>>>),
             Optional<int> maxServerSelectionWaitQueueSize = default(Optional<int>),
             Optional<string> replicaSetName = default(Optional<string>),
             Optional<TimeSpan> serverSelectionTimeout = default(Optional<TimeSpan>),
             Optional<IServerSelector> preServerSelector = default(Optional<IServerSelector>),
             Optional<IServerSelector> postServerSelector = default(Optional<IServerSelector>),
+            Optional<IReadOnlyDictionary<string, BsonDocument>> schemaMap = default(Optional<IReadOnlyDictionary<string, BsonDocument>>),
             Optional<ConnectionStringScheme> scheme = default(Optional<ConnectionStringScheme>))
         {
             _connectionMode = connectionMode.WithDefault(ClusterConnectionMode.Automatic);
             _endPoints = Ensure.IsNotNull(endPoints.WithDefault(__defaultEndPoints), "endPoints").ToList();
+            _kmsProviders = kmsProviders.WithDefault(null);
             _maxServerSelectionWaitQueueSize = Ensure.IsGreaterThanOrEqualToZero(maxServerSelectionWaitQueueSize.WithDefault(500), "maxServerSelectionWaitQueueSize");
             _replicaSetName = replicaSetName.WithDefault(null);
             _serverSelectionTimeout = Ensure.IsGreaterThanOrEqualToZero(serverSelectionTimeout.WithDefault(TimeSpan.FromSeconds(30)), "serverSelectionTimeout");
             _preServerSelector = preServerSelector.WithDefault(null);
             _postServerSelector = postServerSelector.WithDefault(null);
             _scheme = scheme.WithDefault(ConnectionStringScheme.MongoDB);
+            _schemaMap = schemaMap.WithDefault(null);
         }
 
         // properties
@@ -98,6 +107,17 @@ public IReadOnlyList<EndPoint> EndPoints
             get { return _endPoints; }
         }
 
+        /// <summary>
+        /// Gets the kms providers.
+        /// </summary>
+        /// <value>
+        /// The kms providers.
+        /// </value>
+        public IReadOnlyDictionary<string, IReadOnlyDictionary<string, object>> KmsProviders
+        {
+            get { return _kmsProviders; }
+        }
+
         /// <summary>
         /// Gets the maximum size of the server selection wait queue.
         /// </summary>
@@ -120,6 +140,17 @@ public string ReplicaSetName
             get { return _replicaSetName; }
         }
 
+        /// <summary>
+        /// Gets the schema map.
+        /// </summary>
+        /// <value>
+        /// The schema map.
+        /// </value>
+        public IReadOnlyDictionary<string, BsonDocument> SchemaMap
+        {
+            get { return _schemaMap; }
+        }
+
         /// <summary>
         /// Gets the connection string scheme.
         /// </summary>
@@ -170,31 +201,37 @@ public IServerSelector PostServerSelector
         /// </summary>
         /// <param name="connectionMode">The connection mode.</param>
         /// <param name="endPoints">The end points.</param>
+        /// <param name="kmsProviders">The kms providers.</param>
         /// <param name="maxServerSelectionWaitQueueSize">Maximum size of the server selection wait queue.</param>
         /// <param name="replicaSetName">Name of the replica set.</param>
         /// <param name="serverSelectionTimeout">The server selection timeout.</param>
         /// <param name="preServerSelector">The pre server selector.</param>
         /// <param name="postServerSelector">The post server selector.</param>
+        /// <param name="schemaMap">The schema map.</param>
         /// <param name="scheme">The connection string scheme.</param>
         /// <returns>A new ClusterSettings instance.</returns>
         public ClusterSettings With(
             Optional<ClusterConnectionMode> connectionMode = default(Optional<ClusterConnectionMode>),
             Optional<IEnumerable<EndPoint>> endPoints = default(Optional<IEnumerable<EndPoint>>),
+            Optional<IReadOnlyDictionary<string, IReadOnlyDictionary<string, object>>> kmsProviders = default(Optional<IReadOnlyDictionary<string, IReadOnlyDictionary<string, object>>>),
             Optional<int> maxServerSelectionWaitQueueSize = default(Optional<int>),
             Optional<string> replicaSetName = default(Optional<string>),
             Optional<TimeSpan> serverSelectionTimeout = default(Optional<TimeSpan>),
             Optional<IServerSelector> preServerSelector = default(Optional<IServerSelector>),
             Optional<IServerSelector> postServerSelector = default(Optional<IServerSelector>),
+            Optional<IReadOnlyDictionary<string, BsonDocument>> schemaMap = default(Optional<IReadOnlyDictionary<string, BsonDocument>>),
             Optional<ConnectionStringScheme> scheme = default(Optional<ConnectionStringScheme>))
         {
             return new ClusterSettings(
                 connectionMode: connectionMode.WithDefault(_connectionMode),
                 endPoints: Optional.Enumerable(endPoints.WithDefault(_endPoints)),
+                kmsProviders: Optional.Create(kmsProviders.WithDefault(_kmsProviders)),
                 maxServerSelectionWaitQueueSize: maxServerSelectionWaitQueueSize.WithDefault(_maxServerSelectionWaitQueueSize),
                 replicaSetName: replicaSetName.WithDefault(_replicaSetName),
                 serverSelectionTimeout: serverSelectionTimeout.WithDefault(_serverSelectionTimeout),
                 preServerSelector: Optional.Create(preServerSelector.WithDefault(_preServerSelector)),
                 postServerSelector: Optional.Create(postServerSelector.WithDefault(_postServerSelector)),
+                schemaMap: Optional.Create(schemaMap.WithDefault(_schemaMap)),
                 scheme: scheme.WithDefault(_scheme));
         }
     }

src/MongoDB.Driver.Core/Core/Misc/Feature.cs

@@ -41,6 +41,7 @@ public class Feature
         private static readonly Feature __bypassDocumentValidation = new Feature("BypassDocumentValidation", new SemanticVersion(3, 2, 0));
         private static readonly Feature __changeStreamStage = new Feature("ChangeStreamStage", new SemanticVersion(3, 5, 11));
         private static readonly Feature __changeStreamPostBatchResumeToken = new Feature("ChangeStreamPostBatchResumeToken", new SemanticVersion(4, 0 ,7));
+        private static readonly Feature __clientSideEncryption = new Feature("ClientSideEncryption", new SemanticVersion(4, 1, 9));
         private static readonly CollationFeature __collation = new CollationFeature("Collation", new SemanticVersion(3, 3, 11));
         private static readonly Feature __commandMessage = new Feature("CommandMessage", new SemanticVersion(3, 6, 0));
         private static readonly CommandsThatWriteAcceptWriteConcernFeature __commandsThatWriteAcceptWriteConcern = new CommandsThatWriteAcceptWriteConcernFeature("CommandsThatWriteAcceptWriteConcern", new SemanticVersion(3, 3, 11));
@@ -170,6 +171,11 @@ public class Feature
         /// </summary>
         public static Feature ChangeStreamPostBatchResumeToken => __changeStreamPostBatchResumeToken;
 
+        /// <summary>
+        /// Gets the client side encryption feature.
+        /// </summary>
+        public static Feature ClientSideEncryption => __clientSideEncryption;
+
         /// <summary>
         /// Gets the collation feature.
         /// </summary>