Ensure algorithm query param is passed for CSFLE (#230)

* Add missing algorithm query param

* Add test

Author: rayokota

schemaregistry/rules/encryption/dekregistry/dekregistry-client.ts

@@ -224,7 +224,7 @@ class DekRegistryClient implements DekClient {
       subject = encodeURIComponent(subject);
 
       const response = await this.restService.handleRequest<Dek>(
-        `/dek-registry/v1/keks/${kekName}/deks/${subject}/versions/${version}?deleted=${deleted}`,
+        `/dek-registry/v1/keks/${kekName}/deks/${subject}/versions/${version}?algorithm=${algorithm}&deleted=${deleted}`,
         'GET');
       this.dekCache.set(cacheKey, response.data);
       return response.data;

schemaregistry/test/serde/avro.spec.ts

@@ -675,6 +675,90 @@ describe('AvroSerializer', () => {
     expect(obj2.stringField).not.toEqual("hi");
     expect(obj2.bytesField).not.toEqual(Buffer.from([1, 2]));
   })
+  it('deterministic encryption', async () => {
+    let conf: ClientConfig = {
+      baseURLs: [baseURL],
+      cacheCapacity: 1000
+    }
+    let client = SchemaRegistryClient.newClient(conf)
+    let serConfig: AvroSerializerConfig = {
+      useLatestVersion: true,
+      ruleConfig: {
+        secret: 'mysecret'
+      }
+    }
+    let ser = new AvroSerializer(client, SerdeType.VALUE, serConfig)
+    let dekClient = fieldEncryptionExecutor.client!
+
+    let encRule: Rule = {
+      name: 'test-encrypt',
+      kind: 'TRANSFORM',
+      mode: RuleMode.WRITEREAD,
+      type: 'ENCRYPT',
+      tags: ['PII'],
+      params: {
+        'encrypt.kek.name': 'kek1',
+        'encrypt.kms.type': 'local-kms',
+        'encrypt.kms.key.id': 'mykey',
+        'encrypt.dek.algorithm': 'AES256_SIV',
+      },
+      onFailure: 'ERROR,NONE'
+    }
+    let ruleSet: RuleSet = {
+      domainRules: [encRule]
+    }
+
+    let info: SchemaInfo = {
+      schemaType: 'AVRO',
+      schema: demoSchema,
+      ruleSet
+    }
+
+    await client.register(subject, info, false)
+
+    let obj = {
+      intField: 123,
+      doubleField: 45.67,
+      stringField: 'hi',
+      boolField: true,
+      bytesField: Buffer.from([1, 2]),
+    }
+    let bytes = await ser.serialize(topic, obj)
+
+    // reset encrypted field
+    obj.stringField = 'hi'
+    obj.bytesField = Buffer.from([1, 2])
+
+    let deserConfig: AvroDeserializerConfig = {
+      ruleConfig: {
+        secret: 'mysecret'
+      }
+    }
+    let deser = new AvroDeserializer(client, SerdeType.VALUE, deserConfig)
+    fieldEncryptionExecutor.client = dekClient
+    let obj2 = await deser.deserialize(topic, bytes)
+    expect(obj2.intField).toEqual(obj.intField);
+    expect(obj2.doubleField).toBeCloseTo(obj.doubleField, 0.001);
+    expect(obj2.stringField).toEqual(obj.stringField);
+    expect(obj2.boolField).toEqual(obj.boolField);
+    expect(obj2.bytesField).toEqual(obj.bytesField);
+
+    let registry = new RuleRegistry()
+    registry.registerExecutor(new FieldEncryptionExecutor())
+    registry.registerOverride({type: 'ENCRYPT', disabled: true})
+    deser = new AvroDeserializer(client, SerdeType.VALUE, deserConfig, registry)
+    obj2 = await deser.deserialize(topic, bytes)
+    expect(obj2.stringField).not.toEqual("hi");
+    expect(obj2.bytesField).not.toEqual(Buffer.from([1, 2]));
+
+    clearKmsClients()
+    registry = new RuleRegistry()
+    registry.registerExecutor(new FieldEncryptionExecutor())
+    deser = new AvroDeserializer(client, SerdeType.VALUE, {}, registry)
+    obj2 = await deser.deserialize(topic, bytes)
+    expect(obj2.stringField).not.toEqual("hi");
+    expect(obj2.bytesField).not.toEqual(Buffer.from([1, 2]));
+  })
   it('basic encryption with logical type', async () => {
     let conf: ClientConfig = {
       baseURLs: [baseURL],