DGS-19519 Enhance error handling for CSFLE configure method (#236)

* Enhance error handling for CSFLE configure method

* Minor cleanup

* Add test

* Add test

* Minor fix

Author: rayokota

package-lock.json

@@ -47,6 +47,7 @@ interface Dek {
 }
 
 interface DekClient {
+  config(): ClientConfig;
   registerKek(name: string, kmsType: string, kmsKeyId: string, shared: boolean,
               kmsProps?: { [key: string]: string }, doc?: string): Promise<Kek>;
   getKek(name: string, deleted: boolean): Promise<Kek>;
@@ -57,13 +58,15 @@ interface DekClient {
 }
 
 class DekRegistryClient implements DekClient {
+  private clientConfig: ClientConfig;
   private restService: RestService;
   private kekCache: LRUCache<string, Kek>;
   private dekCache: LRUCache<string, Dek>;
   private kekMutex: Mutex;
   private dekMutex: Mutex;
 
   constructor(config: ClientConfig) {
+    this.clientConfig = config;
     const cacheOptions = {
       max: config.cacheCapacity !== undefined ? config.cacheCapacity : 1000,
       ...(config.cacheLatestTtlSecs !== undefined && { ttl: config.cacheLatestTtlSecs * 1000 }),
@@ -82,7 +85,7 @@ class DekRegistryClient implements DekClient {
   static newClient(config: ClientConfig): DekClient {
     const url = config.baseURLs[0];
     if (url.startsWith("mock://")) {
-      return new MockDekRegistryClient()
+      return new MockDekRegistryClient(config)
     }
     return new DekRegistryClient(config)
   }
@@ -134,6 +137,10 @@ class DekRegistryClient implements DekClient {
     }
   }
 
+  config(): ClientConfig {
+    return this.clientConfig;
+  }
+
   async registerKek(name: string, kmsType: string, kmsKeyId: string, shared: boolean,
     kmsProps?: { [key: string]: string }, doc?: string): Promise<Kek> {
     const cacheKey = stringify({ name, deleted: false });

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

@@ -2,16 +2,23 @@ import { DekClient, Dek, Kek } from "./dekregistry-client";
 import { MOCK_TS } from "./constants";
 import stringify from "json-stringify-deterministic";
 import {RestError} from "../../../rest-error";
+import {ClientConfig} from "../../../rest-service";
 
 class MockDekRegistryClient implements DekClient {
+  private clientConfig?: ClientConfig;
   private kekCache: Map<string, Kek>;
   private dekCache: Map<string, Dek>;
 
-  constructor() {
+  constructor(config?: ClientConfig) {
+    this.clientConfig = config
     this.kekCache = new Map<string, Kek>();
     this.dekCache = new Map<string, Dek>();
   }
 
+  config(): ClientConfig {
+    return this.clientConfig!;
+  }
+
   async registerKek(name: string, kmsType: string, kmsKeyId: string, shared: boolean,
     kmsProps?: { [key: string]: string }, doc?: string): Promise<Kek> {
     const cacheKey = stringify({ name, deleted: false });

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

@@ -20,6 +20,7 @@ import {AesSivKey, AesSivKeySchema} from "./tink/proto/aes_siv_pb";
 import {create, fromBinary, toBinary} from "@bufbuild/protobuf";
 import {fromRawKey as aesGcmFromRawKey} from "./tink/aes_gcm";
 import {fromRawKey as aesSivFromRawKey} from "./tink/aes_siv";
+import {deepEqual} from "../../serde/json-util";
 
 // EncryptKekName represents a kek name
 const ENCRYPT_KEK_NAME = 'encrypt.kek.name'
@@ -83,8 +84,28 @@ export class FieldEncryptionExecutor extends FieldRuleExecutor {
   }
 
   override configure(clientConfig: ClientConfig, config: Map<string, string>) {
-    this.client = DekRegistryClient.newClient(clientConfig)
-    this.config = config
+    if (this.client != null) {
+      if (!deepEqual(this.client.config(), clientConfig)) {
+        throw new RuleError('executor already configured')
+      }
+    } else {
+      this.client = DekRegistryClient.newClient(clientConfig)
+    }
+
+    if (this.config != null) {
+      for (let [key, value] of config) {
+        let v = this.config.get(key)
+        if (v != null) {
+          if (v !== value) {
+            throw new RuleError('rule config key already set: {key}')
+          }
+        } else {
+          this.config.set(key, value)
+        }
+      }
+    } else {
+      this.config = config
+    }
   }
 
   override type(): string {

schemaregistry/rules/encryption/encrypt-executor.ts

@@ -570,6 +570,7 @@ function getType(fd: DescField): FieldType {
         case ScalarType.SFIXED64:
           return FieldType.LONG
         case ScalarType.FLOAT:
+          return FieldType.FLOAT
         case ScalarType.DOUBLE:
           return FieldType.DOUBLE
         case ScalarType.BOOL:

schemaregistry/serde/protobuf.ts

@@ -0,0 +1,32 @@
+import { describe, expect, it } from '@jest/globals';
+import {FieldEncryptionExecutor} from "../../../rules/encryption/encrypt-executor";
+import {ClientConfig} from "../../../rest-service";
+
+describe('FieldEncryptionExecutor', () => {
+  it('configure error', () => {
+    const executor = new FieldEncryptionExecutor()
+    const clientConfig: ClientConfig = {
+      baseURLs: ['mock://'],
+      cacheCapacity: 1000
+    }
+    const config = new Map<string, string>();
+    config.set('key', 'value');
+    executor.configure(clientConfig, config);
+    // configure with same args is fine
+    executor.configure(clientConfig, config);
+    const config2 = new Map<string, string>();
+    config2.set('key2', 'value2');
+    // configure with additional config keys is fine
+    executor.configure(clientConfig, config);
+
+    const clientConfig2: ClientConfig = {
+      baseURLs: ['blah://'],
+      cacheCapacity: 1000
+    }
+    expect(() => executor.configure(clientConfig2, config)).toThrowError()
+
+    const config3  = new Map<string, string>();
+    config3.set('key', 'value3');
+    expect(() => executor.configure(clientConfig, config3)).toThrowError()
+  })
+})

schemaregistry/test/rules/encryption/encrypt-executor.spec.ts

@@ -865,6 +865,7 @@ describe('AvroSerializer', () => {
     expect(obj2.bytesField).toEqual(obj.bytesField);
   })
   it('basic encryption with dek rotation', async () => {
+    const fieldEncryptionExecutor = FieldEncryptionExecutor.registerWithClock(new FakeClock());
     (fieldEncryptionExecutor.clock as FakeClock).fixedNow = Date.now()
     let conf: ClientConfig = {
       baseURLs: [baseURL],
@@ -972,6 +973,7 @@ describe('AvroSerializer', () => {
     expect(3).toEqual(dek.version);
   })
   it('basic encryption with preserialized data', async () => {
+    const fieldEncryptionExecutor = FieldEncryptionExecutor.registerWithClock(new FakeClock());
     let conf: ClientConfig = {
       baseURLs: [baseURL],
       cacheCapacity: 1000
@@ -1024,6 +1026,7 @@ describe('AvroSerializer', () => {
     expect(obj2.f1).toEqual(obj.f1);
   })
   it('deterministic encryption with preserialized data', async () => {
+    const fieldEncryptionExecutor = FieldEncryptionExecutor.registerWithClock(new FakeClock());
     let conf: ClientConfig = {
       baseURLs: [baseURL],
       cacheCapacity: 1000
@@ -1077,6 +1080,7 @@ describe('AvroSerializer', () => {
     expect(obj2.f1).toEqual(obj.f1);
   })
   it('dek rotation encryption with preserialized data', async () => {
+    const fieldEncryptionExecutor = FieldEncryptionExecutor.registerWithClock(new FakeClock());
     let conf: ClientConfig = {
       baseURLs: [baseURL],
       cacheCapacity: 1000