Add delete method to Secret Persistence interface. (#12180)

Add a delete method that allows us to delete secrets.

This is to prep us for actually deleting secrets when we delete sources/destinations, which will happen in a follow up PR.

airbyte-config/config-secrets/src/main/kotlin/secrets/persistence/AwsSecretManagerPersistence.kt

@@ -118,7 +118,7 @@ class AwsSecretManagerPersistence(private val awsClient: AwsClient, private val
    *
    * @param coordinate SecretCoordinate to delete.
    */
-  private fun deleteSecret(coordinate: SecretCoordinate) {
+  override fun delete(coordinate: SecretCoordinate) {
     awsClient.client.deleteSecret(
       DeleteSecretRequest()
         .withSecretId(coordinate.coordinateBase)

airbyte-config/config-secrets/src/main/kotlin/secrets/persistence/GoogleSecretManagerPersistence.kt

@@ -105,6 +105,13 @@ class GoogleSecretManagerPersistence(
       client.addSecretVersion(name, secretPayload)
     }
   }
+
+  override fun delete(coordinate: SecretCoordinate) {
+    googleSecretManagerServiceClient.createClient().use { client ->
+      val secretName = SecretName.of(gcpProjectId, coordinate.fullCoordinate)
+      client.deleteSecret(secretName)
+    }
+  }
 }
 
 @Singleton

airbyte-config/config-secrets/src/main/kotlin/secrets/persistence/LocalTestingSecretPersistence.kt

@@ -60,4 +60,8 @@ open class LocalTestingSecretPersistence(
       coordinate.fullCoordinate,
     ).execute()
   }
+
+  override fun delete(coordinate: SecretCoordinate) {
+    return
+  }
 }

airbyte-config/config-secrets/src/main/kotlin/secrets/persistence/NoOpSecretPersistence.kt

@@ -19,4 +19,8 @@ class NoOpSecretPersistence : SecretPersistence {
   ) {
     return
   }
+
+  override fun delete(coordinate: SecretCoordinate) {
+    return
+  }
 }

airbyte-config/config-secrets/src/main/kotlin/secrets/persistence/RuntimeSecretPersistence.kt

@@ -77,6 +77,10 @@ class RuntimeSecretPersistence(private val secretPersistenceConfig: SecretPersis
     secretPersistence.write(coordinate, payload)
   }
 
+  override fun delete(coordinate: SecretCoordinate) {
+    return
+  }
+
   private fun buildAwsSecretManager(configuration: Map<String, String>): AwsSecretManagerPersistence {
     // We default to ACCESS_KEY auth
     val authType = configuration["auth_type"]?.uppercase() ?: AwsAuthType.ACCESS_KEY.value

airbyte-config/config-secrets/src/main/kotlin/secrets/persistence/SecretPersistence.kt

@@ -46,4 +46,6 @@ interface SecretPersistence : ReadOnlySecretPersistence {
     // Default implementation does not support expiry.
     write(coordinate, payload)
   }
+
+  fun delete(coordinate: SecretCoordinate)
 }

airbyte-config/config-secrets/src/main/kotlin/secrets/persistence/VaultSecretPersistence.kt

@@ -57,6 +57,10 @@ class VaultSecretPersistence(
     }
   }
 
+  override fun delete(coordinate: SecretCoordinate) {
+    return
+  }
+
   companion object {
     private const val SECRET_KEY = "value"
   }

airbyte-config/config-secrets/src/test/kotlin/secrets/persistence/AwsSecretManagerPersistenceTest.kt

@@ -7,6 +7,8 @@ package io.airbyte.config.secrets.persistence
 import com.amazonaws.secretsmanager.caching.SecretCache
 import com.amazonaws.services.secretsmanager.AWSSecretsManager
 import com.amazonaws.services.secretsmanager.model.CreateSecretResult
+import com.amazonaws.services.secretsmanager.model.DeleteSecretRequest
+import com.amazonaws.services.secretsmanager.model.DeleteSecretResult
 import com.amazonaws.services.secretsmanager.model.DescribeSecretResult
 import com.amazonaws.services.secretsmanager.model.GetSecretValueResult
 import com.amazonaws.services.secretsmanager.model.ResourceNotFoundException
@@ -174,4 +176,26 @@ class AwsSecretManagerPersistenceTest {
 
     verify { mockAwsClient.updateSecret(any()) }
   }
+
+  @Test
+  fun `test deleting a secret via the client deletes the secret`() {
+    val secret = "secret value"
+    val coordinate = SecretCoordinate.fromFullCoordinate("secret_coordinate_v1")
+    val mockClient: AwsClient = mockk()
+    val mockCache: AwsCache = mockk()
+    val mockAwsCache: SecretCache = mockk()
+    val mockAwsClient: AWSSecretsManager = mockk()
+    val persistence = AwsSecretManagerPersistence(mockClient, mockCache)
+    every { mockAwsCache.getSecretString(any()) } returns secret
+    every { mockAwsClient.deleteSecret(any()) } returns mockk<DeleteSecretResult>()
+    every { mockCache.cache } returns mockAwsCache
+    every { mockClient.client } returns mockAwsClient
+    every { mockClient.serializedConfig } returns null
+    every { mockClient.kmsKeyArn } returns null
+    every { mockClient.tags } returns emptyMap()
+
+    persistence.delete(coordinate)
+
+    verify { mockAwsClient.deleteSecret(any<DeleteSecretRequest>()) }
+  }
 }

airbyte-config/config-secrets/src/test/kotlin/secrets/persistence/GoogleSecretManagerPersistenceTest.kt

@@ -18,7 +18,9 @@ import com.google.protobuf.ByteString
 import io.airbyte.config.secrets.SecretCoordinate
 import io.airbyte.config.secrets.persistence.GoogleSecretManagerPersistence.Companion.replicationPolicy
 import io.grpc.Status
+import io.mockk.Runs
 import io.mockk.every
+import io.mockk.just
 import io.mockk.mockk
 import io.mockk.verify
 import org.junit.jupiter.api.Assertions
@@ -164,4 +166,26 @@ class GoogleSecretManagerPersistenceTest {
 
     verify { mockGoogleClient.addSecretVersion(any<SecretName>(), any<SecretPayload>()) }
   }
+
+  @Test
+  fun `test deleting a secret via the client deletes the secret`() {
+    val secret = "secret value"
+    val projectId = "test"
+    val coordinate = SecretCoordinate.fromFullCoordinate("secret_coordinate_v1")
+    val mockClient: GoogleSecretManagerServiceClient = mockk()
+    val mockGoogleClient: SecretManagerServiceClient = mockk()
+    val mockResponse: AccessSecretVersionResponse = mockk()
+    val mockPayload: SecretPayload = mockk()
+    val persistence = GoogleSecretManagerPersistence(projectId, mockClient)
+
+    every { mockPayload.data } returns ByteString.copyFromUtf8(secret)
+    every { mockResponse.payload } returns mockPayload
+    every { mockClient.createClient() } returns mockGoogleClient
+    every { mockGoogleClient.deleteSecret(ofType(SecretName::class)) } just Runs
+    every { mockGoogleClient.close() } returns Unit
+
+    persistence.delete(coordinate)
+
+    verify { mockGoogleClient.deleteSecret(any<SecretName>()) }
+  }
 }

airbyte-config/config-secrets/src/testFixtures/kotlin/secrets/MemorySecretPersistence.kt

@@ -23,6 +23,10 @@ class MemorySecretPersistence : SecretPersistence {
     secretMap[coordinate] = payload
   }
 
+  override fun delete(coordinate: SecretCoordinate) {
+    secretMap.remove(coordinate)
+  }
+
   val map: Map<SecretCoordinate, String>
     get() = secretMap.toMutableMap()
 }