secret-persistence: Hashicorp Vault Secret Store

airbyte-config/config-models/src/main/java/io/airbyte/config/Configs.java

@@ -103,11 +103,28 @@ public interface Configs {
 
   /**
    * Defines the Secret Persistence type. None by default. Set to GOOGLE_SECRET_MANAGER to use Google
-   * Secret Manager. Set to TESTING_CONFIG_DB_TABLE to use the database as a test. Alpha support.
-   * Undefined behavior will result if this is turned on and then off.
+   * Secret Manager. Set to TESTING_CONFIG_DB_TABLE to use the database as a test. Set to VAULT to use
+   * Hashicorp Vault. Alpha support. Undefined behavior will result if this is turned on and then off.
    */
   SecretPersistenceType getSecretPersistenceType();
 
+  /**
+   * Define the vault address to read/write Airbyte Configuration to Hashicorp Vault. Alpha Support.
+   */
+  String getVaultAddress();
+
+  /**
+   * Define the vault path prefix to read/write Airbyte Configuration to Hashicorp Vault. Empty by
+   * default. Alpha Support.
+   */
+  String getVaultPrefix();
+
+  /**
+   * Define the vault token to read/write Airbyte Configuration to Hashicorp Vault. Empty by default.
+   * Alpha Support.
+   */
+  String getVaultToken();
+
   // Database
   /**
    * Define the Jobs Database user.
@@ -574,7 +591,8 @@ enum DeploymentMode {
   enum SecretPersistenceType {
     NONE,
     TESTING_CONFIG_DB_TABLE,
-    GOOGLE_SECRET_MANAGER
+    GOOGLE_SECRET_MANAGER,
+    VAULT
   }
 
 }

airbyte-config/config-models/src/main/java/io/airbyte/config/EnvConfigs.java

@@ -165,6 +165,10 @@ public class EnvConfigs implements Configs {
   private static final String DEFAULT_JOB_KUBE_CURL_IMAGE = "curlimages/curl:7.83.1";
   private static final int DEFAULT_DATABASE_INITIALIZATION_TIMEOUT_MS = 60 * 1000;
 
+  private static final String VAULT_ADDRESS = "VAULT_ADDRESS";
+  private static final String VAULT_PREFIX = "VAULT_PREFIX";
+  private static final String VAULT_AUTH_TOKEN = "VAULT_AUTH_TOKEN";
+
   public static final long DEFAULT_MAX_SPEC_WORKERS = 5;
   public static final long DEFAULT_MAX_CHECK_WORKERS = 5;
   public static final long DEFAULT_MAX_DISCOVER_WORKERS = 5;
@@ -337,6 +341,21 @@ public SecretPersistenceType getSecretPersistenceType() {
     return SecretPersistenceType.valueOf(secretPersistenceStr);
   }
 
+  @Override
+  public String getVaultAddress() {
+    return getEnv(VAULT_ADDRESS);
+  }
+
+  @Override
+  public String getVaultPrefix() {
+    return getEnvOrDefault(VAULT_PREFIX, "");
+  }
+
+  @Override
+  public String getVaultToken() {
+    return getEnv(VAULT_AUTH_TOKEN);
+  }
+
   // Database
   @Override
   public String getDatabaseUser() {

airbyte-config/config-persistence/build.gradle

@@ -14,11 +14,13 @@ dependencies {
 
     implementation 'commons-io:commons-io:2.7'
     implementation 'com.google.cloud:google-cloud-secretmanager:2.0.5'
+    implementation 'com.bettercloud:vault-java-driver:5.1.0'
 
     testImplementation 'org.hamcrest:hamcrest-all:1.3'
     testImplementation libs.platform.testcontainers.postgresql
     testImplementation libs.flyway.core
     testImplementation project(':airbyte-test-utils')
+    testImplementation "org.testcontainers:vault:1.17.2"
     integrationTestJavaImplementation project(':airbyte-config:config-persistence')
 }
 

airbyte-config/config-persistence/src/main/java/io/airbyte/config/persistence/split_secrets/SecretPersistence.java

@@ -31,6 +31,9 @@ static Optional<SecretPersistence> getLongLived(final DSLContext dslContext, fin
       case GOOGLE_SECRET_MANAGER -> {
         return Optional.of(GoogleSecretManagerPersistence.getLongLived(configs.getSecretStoreGcpProjectId(), configs.getSecretStoreGcpCredentials()));
       }
+      case VAULT -> {
+        return Optional.of(new VaultSecretPersistence(configs.getVaultAddress(), configs.getVaultPrefix(), configs.getVaultToken()));
+      }
       default -> {
         return Optional.empty();
       }
@@ -56,6 +59,9 @@ static Optional<SecretPersistence> getEphemeral(final DSLContext dslContext, fin
       case GOOGLE_SECRET_MANAGER -> {
         return Optional.of(GoogleSecretManagerPersistence.getEphemeral(configs.getSecretStoreGcpProjectId(), configs.getSecretStoreGcpCredentials()));
       }
+      case VAULT -> {
+        return Optional.of(new VaultSecretPersistence(configs.getVaultAddress(), configs.getVaultPrefix(), configs.getVaultToken()));
+      }
       default -> {
         return Optional.empty();
       }

airbyte-config/config-persistence/src/main/java/io/airbyte/config/persistence/split_secrets/VaultSecretPersistence.java

@@ -0,0 +1,65 @@
+/*
+ * Copyright (c) 2022 Airbyte, Inc., all rights reserved.
+ */
+
+package io.airbyte.config.persistence.split_secrets;
+
+import com.bettercloud.vault.Vault;
+import com.bettercloud.vault.VaultConfig;
+import com.bettercloud.vault.VaultException;
+import io.airbyte.commons.lang.Exceptions;
+import java.util.HashMap;
+import java.util.Optional;
+import lombok.extern.slf4j.Slf4j;
+import lombok.val;
+
+@Slf4j
+final public class VaultSecretPersistence implements SecretPersistence {
+
+  private final String SECRET_KEY = "value";
+  private final Vault vault;
+  private final String pathPrefix;
+
+  public VaultSecretPersistence(final String address, final String prefix, final String token) {
+    this.vault = Exceptions.toRuntime(() -> getVaultClient(address, token));
+    this.pathPrefix = prefix;
+  }
+
+  @Override
+  public Optional<String> read(final SecretCoordinate coordinate) {
+    try {
+      val response = vault.logical().read(pathPrefix + coordinate.getFullCoordinate());
+      val restResponse = response.getRestResponse();
+      val responseCode = restResponse.getStatus();
+      if (responseCode != 200) {
+        log.error("Vault failed on read. Response code: " + responseCode);
+        return Optional.empty();
+      }
+      val data = response.getData();
+      return Optional.of(data.get(SECRET_KEY));
+    } catch (final VaultException e) {
+      return Optional.empty();
+    }
+  }
+
+  @Override
+  public void write(final SecretCoordinate coordinate, final String payload) {
+    try {
+      val newSecret = new HashMap<String, Object>();
+      newSecret.put(SECRET_KEY, payload);
+      vault.logical().write(pathPrefix + coordinate.getFullCoordinate(), newSecret);
+    } catch (final VaultException e) {
+      log.error("Vault failed on write", e);
+    }
+  }
+
+  private static Vault getVaultClient(final String address, final String token) throws VaultException {
+    val config = new VaultConfig()
+        .address(address)
+        .token(token)
+        .engineVersion(2)
+        .build();
+    return new Vault(config);
+  }
+
+}

airbyte-config/config-persistence/src/test/java/io/airbyte/config/persistence/split_secrets/VaultSecretPersistenceTest.java

@@ -0,0 +1,64 @@
+/*
+ * Copyright (c) 2022 Airbyte, Inc., all rights reserved.
+ */
+
+package io.airbyte.config.persistence.split_secrets;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.assertj.core.api.Assertions.assertThat;
+
+import lombok.val;
+import org.apache.commons.lang3.RandomUtils;
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.testcontainers.vault.VaultContainer;
+
+public class VaultSecretPersistenceTest {
+
+  private VaultSecretPersistence persistence;
+  private String baseCoordinate;
+
+  private VaultContainer vaultContainer;
+
+  @BeforeEach
+  void setUp() {
+    vaultContainer = new VaultContainer("vault").withVaultToken("vault-dev-token-id");
+    vaultContainer.start();
+
+    val vaultAddress = "http://" + vaultContainer.getHost() + ":" + vaultContainer.getFirstMappedPort();
+
+    persistence = new VaultSecretPersistence(vaultAddress, "secret/testing", "vault-dev-token-id");
+    baseCoordinate = "VaultSecretPersistenceIntegrationTest_coordinate_" + RandomUtils.nextInt() % 20000;
+  }
+
+  @AfterEach
+  void tearDown() {
+    vaultContainer.stop();
+  }
+
+  @Test
+  void testReadWriteUpdate() {
+    val coordinate1 = new SecretCoordinate(baseCoordinate, 1);
+
+    // try reading non-existent value
+    val firstRead = persistence.read(coordinate1);
+    assertThat(firstRead.isEmpty()).isTrue();
+
+    // write
+    val firstPayload = "abc";
+    persistence.write(coordinate1, firstPayload);
+    val secondRead = persistence.read(coordinate1);
+    assertThat(secondRead.isPresent()).isTrue();
+    assertEquals(firstPayload, secondRead.get());
+
+    // update
+    val secondPayload = "def";
+    val coordinate2 = new SecretCoordinate(baseCoordinate, 2);
+    persistence.write(coordinate2, secondPayload);
+    val thirdRead = persistence.read(coordinate2);
+    assertThat(thirdRead.isPresent()).isTrue();
+    assertEquals(secondPayload, thirdRead.get());
+  }
+
+}

docs/operator-guides/configuring-airbyte.md

@@ -46,7 +46,11 @@ The following variables are relevant to both Docker and Kubernetes.
 #### Secrets
 1. `SECRET_STORE_GCP_PROJECT_ID` - Defines the GCP Project to store secrets in. Alpha support.
 2. `SECRET_STORE_GCP_CREDENTIALS` - Define the JSON credentials used to read/write Airbyte Configuration to Google Secret Manager. These credentials must have Secret Manager Read/Write access. Alpha support.
-3. `SECRET_PERSISTENCE` - Defines the Secret Persistence type. Defaults to NONE. Set to GOOGLE_SECRET_MANAGER to use Google Secret Manager. Set to TESTING_CONFIG_DB_TABLE to use the database as a test. Alpha support. Undefined behavior will result if this is turned on and then off.
+3. `VAULT_ADDRESS` - Define the vault address to read/write Airbyte Configuration to Hashicorp Vault. Alpha Support.
+4. `VAULT_PREFIX` - Define the vault path prefix. Empty by default. Alpha Support.
+5. `VAULT_AUTH_TOKEN` - The token used for vault authentication. Alpha Support.
+6. `VAULT_AUTH_METHOD` - How vault will preform authentication. Currently, only supports Token auth. Defaults to token. Alpha Support.
+7. `SECRET_PERSISTENCE` - Defines the Secret Persistence type. Defaults to NONE. Set to GOOGLE_SECRET_MANAGER to use Google Secret Manager. Set to TESTING_CONFIG_DB_TABLE to use the database as a test. Set to VAULT to use Hashicorp Vault, currently only the token based authentication is supported. Alpha support. Undefined behavior will result if this is turned on and then off.
 
 #### Database
 1. `DATABASE_USER` - Define the Jobs Database user.