GODRIVER-1592 Run CSE flow for change streams

Author: Divjot Arora

data/client-side-encryption-prose/change-streams-test.json

@@ -0,0 +1,70 @@
+{
+    "json_schema": {
+        "properties": {
+            "encrypted_string": {
+                "encrypt": {
+                    "keyId": [
+                        {
+                            "$binary": {
+                                "base64": "AAAAAAAAAAAAAAAAAAAAAA==",
+                                "subType": "04"
+                            }
+                        }
+                    ],
+                    "bsonType": "string",
+                    "algorithm": "AEAD_AES_256_CBC_HMAC_SHA_512-Deterministic"
+                }
+            }
+        },
+        "bsonType": "object"
+    },
+    "key_vault_data": [
+        {
+            "status": 1,
+            "_id": {
+                "$binary": {
+                    "base64": "AAAAAAAAAAAAAAAAAAAAAA==",
+                    "subType": "04"
+                }
+            },
+            "masterKey": {
+                "provider": "aws",
+                "key": "arn:aws:kms:us-east-1:579766882180:key/89fcc2c4-08b0-4bd9-9f25-e30687b580d0",
+                "region": "us-east-1"
+            },
+            "updateDate": {
+                "$date": {
+                    "$numberLong": "1552949630483"
+                }
+            },
+            "keyMaterial": {
+                "$binary": {
+                    "base64": "AQICAHhQNmWG2CzOm1dq3kWLM+iDUZhEqnhJwH9wZVpuZ94A8gEqnsxXlR51T5EbEVezUqqKAAAAwjCBvwYJKoZIhvcNAQcGoIGxMIGuAgEAMIGoBgkqhkiG9w0BBwEwHgYJYIZIAWUDBAEuMBEEDHa4jo6yp0Z18KgbUgIBEIB74sKxWtV8/YHje5lv5THTl0HIbhSwM6EqRlmBiFFatmEWaeMk4tO4xBX65eq670I5TWPSLMzpp8ncGHMmvHqRajNBnmFtbYxN3E3/WjxmdbOOe+OXpnGJPcGsftc7cB2shRfA4lICPnE26+oVNXT6p0Lo20nY5XC7jyCO",
+                    "subType": "00"
+                }
+            },
+            "creationDate": {
+                "$date": {
+                    "$numberLong": "1552949630483"
+                }
+            },
+            "keyAltNames": [
+                "altname",
+                "another_altname"
+            ]
+        }
+    ],
+    "encrypted_document": {
+        "_id": 1,
+        "encrypted_string": {
+            "$binary": {
+                "base64": "AQAAAAAAAAAAAAAAAAAAAAACwj+3zkv2VM+aTfk60RqhXq6a/77WlLwu/BxXFkL7EppGsju/m8f0x5kBDD3EZTtGALGXlym5jnpZAoSIkswHoA==",
+                "subType": "06"
+            }
+        }
+    },
+    "decrytped_document": {
+        "_id": 1,
+        "encrypted_string": "string0"
+    }
+}

mongo/change_stream.go

@@ -69,6 +69,7 @@ type changeStreamConfig struct {
 	streamType     StreamType
 	collectionName string
 	databaseName   string
+	crypt          *driver.Crypt
 }
 
 func newChangeStream(ctx context.Context, config changeStreamConfig, pipeline interface{},
@@ -100,8 +101,12 @@ func newChangeStream(ctx context.Context, config changeStreamConfig, pipeline in
 	cs.aggregate = operation.NewAggregate(nil).
 		ReadPreference(config.readPreference).ReadConcern(config.readConcern).
 		Deployment(cs.client.deployment).ClusterClock(cs.client.clock).
-		CommandMonitor(cs.client.monitor).Session(cs.sess).ServerSelector(cs.selector).Retry(driver.RetryNone)
+		CommandMonitor(cs.client.monitor).Session(cs.sess).ServerSelector(cs.selector).Retry(driver.RetryNone).
+		Crypt(config.crypt)
 
+	if config.crypt != nil {
+		cs.cursorOptions.Crypt = config.crypt
+	}
 	if cs.options.Collation != nil {
 		cs.aggregate.Collation(bsoncore.Document(cs.options.Collation.ToDocument()))
 	}

mongo/client.go

@@ -812,6 +812,7 @@ func (c *Client) Watch(ctx context.Context, pipeline interface{},
 		client:         c,
 		registry:       c.registry,
 		streamType:     ClientStream,
+		crypt:          c.crypt,
 	}
 
 	return newChangeStream(ctx, csConfig, pipeline, opts...)

mongo/collection.go

@@ -1553,6 +1553,7 @@ func (coll *Collection) Watch(ctx context.Context, pipeline interface{},
 		streamType:     CollectionStream,
 		collectionName: coll.Name(),
 		databaseName:   coll.db.Name(),
+		crypt:          coll.client.crypt,
 	}
 	return newChangeStream(ctx, csConfig, pipeline, opts...)
 }

mongo/database.go

@@ -429,6 +429,7 @@ func (db *Database) Watch(ctx context.Context, pipeline interface{},
 		registry:       db.registry,
 		streamType:     DatabaseStream,
 		databaseName:   db.Name(),
+		crypt:          db.client.crypt,
 	}
 	return newChangeStream(ctx, csConfig, pipeline, opts...)
 }

mongo/integration/client_side_encryption_prose_test.go

@@ -708,6 +708,105 @@ func TestClientSideEncryptionProse(t *testing.T) {
 			})
 		}
 	})
+	changeStreamOpts := mtest.NewOptions().
+		CreateClient(false).
+		Topologies(mtest.ReplicaSet)
+	mt.RunOpts("change streams", changeStreamOpts, func(mt *mtest.T) {
+		// Change streams can't easily fit into the spec test format because of their tailable nature, so there are two
+		// prose tests for them instead:
+		//
+		// 1. Auto-encryption errors for Watch operations. Collection-level change streams error because the
+		// $changeStream aggregation stage is not valid for encryption. Client and database-level streams error because
+		// only collection-level operations are valid for encryption.
+		//
+		// 2. Events are automatically decrypted: If the Watch() is done with BypassAutoEncryption=true, the Watch
+		// should succeed and subsequent getMore calls should decrypt documents when necessary.
+
+		var testConfig struct {
+			JSONSchema        bson.Raw   `bson:"json_schema"`
+			KeyVaultData      []bson.Raw `bson:"key_vault_data"`
+			EncryptedDocument bson.Raw   `bson:"encrypted_document"`
+			DecryptedDocument bson.Raw   `bson:"decrytped_document"`
+		}
+		decodeJSONFile(mt, "change-streams-test.json", &testConfig)
+
+		schemaMap := map[string]interface{}{
+			"db.coll": testConfig.JSONSchema,
+		}
+		kmsProviders := map[string]map[string]interface{}{
+			"aws": {
+				"accessKeyId":     keyID,
+				"secretAccessKey": secretAccessKey,
+			},
+		}
+
+		testCases := []struct {
+			name       string
+			streamType mongo.StreamType
+		}{
+			{"client", mongo.ClientStream},
+			{"database", mongo.DatabaseStream},
+			{"collection", mongo.CollectionStream},
+		}
+		mt.RunOpts("auto encryption errors", noClientOpts, func(mt *mtest.T) {
+			for _, tc := range testCases {
+				mt.Run(tc.name, func(mt *mtest.T) {
+					autoEncryptionOpts := options.AutoEncryption().
+						SetKmsProviders(kmsProviders).
+						SetKeyVaultNamespace(kvNamespace).
+						SetSchemaMap(schemaMap)
+					cpt := setup(mt, autoEncryptionOpts, nil, nil)
+					defer cpt.teardown(mt)
+
+					_, err := getWatcher(mt, tc.streamType, cpt).Watch(mtest.Background, mongo.Pipeline{})
+					assert.NotNil(mt, err, "expected Watch error: %v", err)
+				})
+			}
+		})
+		mt.RunOpts("events are automatically decrypted", noClientOpts, func(mt *mtest.T) {
+			for _, tc := range testCases {
+				mt.Run(tc.name, func(mt *mtest.T) {
+					autoEncryptionOpts := options.AutoEncryption().
+						SetKmsProviders(kmsProviders).
+						SetKeyVaultNamespace(kvNamespace).
+						SetSchemaMap(schemaMap).
+						SetBypassAutoEncryption(true)
+					cpt := setup(mt, autoEncryptionOpts, nil, nil)
+					defer cpt.teardown(mt)
+
+					// Insert key vault data so the key can be accessed when starting the change stream.
+					insertDocuments(mt, cpt.keyVaultColl, testConfig.KeyVaultData)
+
+					stream, err := getWatcher(mt, tc.streamType, cpt).Watch(mtest.Background, mongo.Pipeline{})
+					assert.Nil(mt, err, "Watch error: %v", err)
+					defer stream.Close(mtest.Background)
+
+					// Insert already encrypted data and verify that it is automatically decrypted by Next().
+					insertDocuments(mt, cpt.coll, []bson.Raw{testConfig.EncryptedDocument})
+					assert.True(mt, stream.Next(mtest.Background), "expected Next to return true, got false")
+					gotDocument := stream.Current.Lookup("fullDocument").Document()
+					err = compareDocs(mt, testConfig.DecryptedDocument, gotDocument)
+					assert.Nil(mt, err, "compareDocs error: %v", err)
+				})
+			}
+		})
+	})
+}
+
+func getWatcher(mt *mtest.T, streamType mongo.StreamType, cpt *cseProseTest) watcher {
+	mt.Helper()
+
+	switch streamType {
+	case mongo.ClientStream:
+		return cpt.cseClient
+	case mongo.DatabaseStream:
+		return cpt.cseColl.Database()
+	case mongo.CollectionStream:
+		return cpt.cseColl
+	default:
+		mt.Fatalf("unknown stream type %v", streamType)
+	}
+	return nil
 }
 
 type cseProseTest struct {
@@ -729,10 +828,12 @@ func setup(mt *mtest.T, aeo *options.AutoEncryptionOptions, kvClientOpts *option
 	cpt.coll = mt.CreateCollection(mtest.Collection{
 		Name: "coll",
 		DB:   "db",
+		Opts: options.Collection().SetWriteConcern(mtest.MajorityWc),
 	}, false)
 	cpt.keyVaultColl = mt.CreateCollection(mtest.Collection{
 		Name: "datakeys",
-		DB:   "admin",
+		DB:   "keyvault",
+		Opts: options.Collection().SetWriteConcern(mtest.MajorityWc),
 	}, false)
 
 	if aeo != nil {
@@ -781,6 +882,18 @@ func readJSONFile(mt *mtest.T, file string) bson.Raw {
 	return doc
 }
 
+func decodeJSONFile(mt *mtest.T, file string, val interface{}) bson.Raw {
+	mt.Helper()
+
+	content, err := ioutil.ReadFile(filepath.Join(clientEncryptionProseDir, file))
+	assert.Nil(mt, err, "ReadFile error for %v: %v", file, err)
+
+	var doc bson.Raw
+	err = bson.UnmarshalExtJSON(content, true, val)
+	assert.Nil(mt, err, "UnmarshalExtJSON error for file %v: %v", file, err)
+	return doc
+}
+
 func rawValueToCoreValue(rv bson.RawValue) bsoncore.Value {
 	return bsoncore.Value{Type: rv.Type, Data: rv.Value}
 }