Allow create kek/dek to act like undelete (#2807)

* Allow create kek/dek to act like undelete

If a kek/dek is created with the exact same args as the original
that was deleted, allow it to be undeleted.  This will be used
when copying kek/deks during schema linking.

* Fix checkstyle

checkstyle/suppressions.xml

@@ -40,7 +40,7 @@
               files="(KafkaSchemaRegistry|SchemaRegistryConfig|ProtobufData|JsonSchemaData).java"/>
 
     <suppress checks="BooleanExpressionComplexity"
-              files="(AvroData|JsonSchema|KafkaSchemaRegistry|ProtobufData|ProtobufSchema|CelExecutor|FieldRuleExecutor|MetadataEncoderService|MockDekRegistryClient).java"/>
+              files="(AvroData|JsonSchema|KafkaSchemaRegistry|ProtobufData|ProtobufSchema|CelExecutor|FieldRuleExecutor|MetadataEncoderService|MockDekRegistryClient|DataEncryptionKey|KeyEncryptionKey).java"/>
 
     <suppress checks="MemberName"
               files="(DynamicSchema|EnumDefinition|MessageDefinition|ServiceDefinition).java"/>

dek-registry/src/main/java/io/confluent/dekregistry/storage/DataEncryptionKey.java

@@ -89,6 +89,15 @@ public void setKeyMaterial(String keyMaterial) {
     this.keyMaterial = keyMaterial;
   }
 
+  @JsonIgnore
+  public boolean isEquivalent(DataEncryptionKey that) {
+    return Objects.equals(kekName, that.kekName)
+        && Objects.equals(subject, that.subject)
+        && algorithm == that.algorithm
+        && version == that.version
+        && Objects.equals(encryptedKeyMaterial, that.encryptedKeyMaterial);
+  }
+
   @Override
   public boolean equals(Object o) {
     if (this == o) {

dek-registry/src/main/java/io/confluent/dekregistry/storage/DekRegistry.java

@@ -450,10 +450,6 @@ private KeyEncryptionKey createKek(CreateKekRequest request)
     if (metricsManager.getKeyCount(tenant, KeyType.KEK) > config.maxKeys()) {
       throw new TooManyKeysException(KeyType.KEK.name());
     }
-    KeyEncryptionKeyId keyId = new KeyEncryptionKeyId(tenant, request.getName());
-    if (keys.containsKey(keyId)) {
-      throw new AlreadyExistsException(request.getName());
-    }
 
     String kmsType = normalizeKmsType(request.getKmsType());
 
@@ -462,6 +458,13 @@ private KeyEncryptionKey createKek(CreateKekRequest request)
         : Collections.emptySortedMap();
     KeyEncryptionKey key = new KeyEncryptionKey(request.getName(), kmsType,
         request.getKmsKeyId(), kmsProps, request.getDoc(), request.isShared(), false);
+
+    KeyEncryptionKeyId keyId = new KeyEncryptionKeyId(tenant, request.getName());
+    KeyEncryptionKey oldKey = (KeyEncryptionKey) keys.get(keyId);
+    // Allow create to act like undelete if the kek is deleted
+    if (oldKey != null && (!oldKey.isDeleted() || !oldKey.isEquivalent(key))) {
+      throw new AlreadyExistsException(request.getName());
+    }
     keys.put(keyId, key);
     // Retrieve key with ts set
     key = (KeyEncryptionKey) keys.get(keyId);
@@ -537,11 +540,6 @@ private DataEncryptionKey createDek(String kekName, CreateDekRequest request)
         ? request.getAlgorithm()
         : DekFormat.AES256_GCM;
     int version = request.getVersion() != null ? request.getVersion() : MIN_VERSION;
-    DataEncryptionKeyId keyId = new DataEncryptionKeyId(
-        tenant, kekName, request.getSubject(), algorithm, version);
-    if (keys.containsKey(keyId)) {
-      throw new AlreadyExistsException(request.getSubject());
-    }
     DataEncryptionKey key = new DataEncryptionKey(kekName, request.getSubject(),
         algorithm, version, request.getEncryptedKeyMaterial(), false);
     KeyEncryptionKey kek = getKek(key.getKekName(), true);
@@ -552,6 +550,13 @@ private DataEncryptionKey createDek(String kekName, CreateDekRequest request)
         throw new InvalidKeyException("encryptedKeyMaterial");
       }
     }
+    DataEncryptionKeyId keyId = new DataEncryptionKeyId(
+        tenant, kekName, request.getSubject(), algorithm, version);
+    DataEncryptionKey oldKey = (DataEncryptionKey) keys.get(keyId);
+    // Allow create to act like undelete if the dek is deleted
+    if (oldKey != null && (!oldKey.isDeleted() || !oldKey.isEquivalent(key))) {
+      throw new AlreadyExistsException(request.getSubject());
+    }
     keys.put(keyId, key);
     // Retrieve key with ts set
     key = (DataEncryptionKey) keys.get(keyId);

dek-registry/src/main/java/io/confluent/dekregistry/storage/KeyEncryptionKey.java

@@ -16,6 +16,7 @@
 package io.confluent.dekregistry.storage;
 
 import com.fasterxml.jackson.annotation.JsonCreator;
+import com.fasterxml.jackson.annotation.JsonIgnore;
 import com.fasterxml.jackson.annotation.JsonIgnoreProperties;
 import com.fasterxml.jackson.annotation.JsonInclude;
 import com.fasterxml.jackson.annotation.JsonInclude.Include;
@@ -90,6 +91,16 @@ public boolean isShared() {
     return this.shared;
   }
 
+  @JsonIgnore
+  public boolean isEquivalent(KeyEncryptionKey that) {
+    return shared == that.shared
+        && Objects.equals(name, that.name)
+        && Objects.equals(kmsType, that.kmsType)
+        && Objects.equals(kmsKeyId, that.kmsKeyId)
+        && Objects.equals(kmsProps, that.kmsProps)
+        && Objects.equals(doc, that.doc);
+  }
+
   @Override
   public boolean equals(Object o) {
     if (this == o) {

dek-registry/src/test/java/io/confluent/dekregistry/web/rest/RestApiTest.java

@@ -134,6 +134,13 @@ private void testBasic(Map<String, String> headers) throws Exception {
     Kek newKek = client.createKek(headers, kekName, kmsType, kmsKeyId, null, null, false);
     assertEquals(kek, newKek);
 
+    // Delete kek
+    client.deleteKek(headers, kekName, false);
+
+    // Allow create to act like undelete
+    newKek = client.createKek(headers, kekName, kmsType, kmsKeyId, null, null, false);
+    assertEquals(kek, newKek);
+
     newKek = client.getKek(kekName, false);
     assertEquals(kek, newKek);
 
@@ -217,6 +224,12 @@ private void testBasic(Map<String, String> headers) throws Exception {
     Dek newDek = client.createDek(headers, kekName, subject, null, algorithm, encryptedDekStr);
     assertEquals(dek, newDek);
 
+    client.deleteDek(headers, kekName, subject, algorithm, false);
+
+    // Allow create to act like undelete
+    newDek = client.createDek(headers, kekName, subject, null, algorithm, encryptedDekStr);
+    assertEquals(dek, newDek);
+
     newDek = client.getDek(kekName, subject, algorithm, false);
     assertEquals(dek, newDek);
     assertNotNull(newDek.getTimestamp());