implement secret handling for workspace_service_account table

airbyte-config/persistence/src/main/java/io/airbyte/config/persistence/ConfigRepository.java

@@ -37,6 +37,7 @@
 import io.airbyte.config.StandardSyncState;
 import io.airbyte.config.StandardWorkspace;
 import io.airbyte.config.State;
+import io.airbyte.config.WorkspaceServiceAccount;
 import io.airbyte.db.Database;
 import io.airbyte.db.ExceptionWrappingDatabase;
 import io.airbyte.db.instance.configs.jooq.enums.ActorType;
@@ -970,4 +971,15 @@ private Condition includeTombstones(final Field<Boolean> tombstoneField, final b
     }
   }
 
+  public WorkspaceServiceAccount getWorkspaceServiceAccountNoSecrets(final UUID workspaceId)
+      throws JsonValidationException, IOException, ConfigNotFoundException {
+    return persistence.getConfig(ConfigSchema.WORKSPACE_SERVICE_ACCOUNT, workspaceId.toString(), WorkspaceServiceAccount.class);
+  }
+
+  public void writeWorkspaceServiceAccountNoSecrets(final WorkspaceServiceAccount workspaceServiceAccount)
+      throws JsonValidationException, IOException {
+    persistence.writeConfig(ConfigSchema.WORKSPACE_SERVICE_ACCOUNT, workspaceServiceAccount.getWorkspaceId().toString(),
+        workspaceServiceAccount);
+  }
+
 }

airbyte-config/persistence/src/main/java/io/airbyte/config/persistence/SecretsRepositoryReader.java

@@ -10,6 +10,7 @@
 import io.airbyte.config.ConfigSchema;
 import io.airbyte.config.DestinationConnection;
 import io.airbyte.config.SourceConnection;
+import io.airbyte.config.WorkspaceServiceAccount;
 import io.airbyte.config.persistence.split_secrets.SecretsHydrator;
 import io.airbyte.validation.json.JsonValidationException;
 import java.io.IOException;
@@ -96,4 +97,18 @@ private void hydrateValuesIfKeyPresent(final String key, final Map<String, Strea
     }
   }
 
+  public WorkspaceServiceAccount getWorkspaceServiceAccountWithSecrets(final UUID workspaceId)
+      throws JsonValidationException, ConfigNotFoundException, IOException {
+    final WorkspaceServiceAccount workspaceServiceAccount = configRepository.getWorkspaceServiceAccountNoSecrets(workspaceId);
+
+    final JsonNode jsonCredential =
+        workspaceServiceAccount.getJsonCredential() != null ? secretsHydrator.hydrateSecretCoordinate(workspaceServiceAccount.getJsonCredential())
+            : null;
+
+    final JsonNode hmacKey =
+        workspaceServiceAccount.getHmacKey() != null ? secretsHydrator.hydrateSecretCoordinate(workspaceServiceAccount.getHmacKey()) : null;
+
+    return Jsons.clone(workspaceServiceAccount).withJsonCredential(jsonCredential).withHmacKey(hmacKey);
+  }
+
 }

airbyte-config/persistence/src/main/java/io/airbyte/config/persistence/SecretsRepositoryWriter.java

@@ -13,6 +13,8 @@
 import io.airbyte.config.SourceConnection;
 import io.airbyte.config.StandardDestinationDefinition;
 import io.airbyte.config.StandardSourceDefinition;
+import io.airbyte.config.WorkspaceServiceAccount;
+import io.airbyte.config.persistence.split_secrets.SecretCoordinateToPayload;
 import io.airbyte.config.persistence.split_secrets.SecretPersistence;
 import io.airbyte.config.persistence.split_secrets.SecretsHelpers;
 import io.airbyte.config.persistence.split_secrets.SplitSecretConfig;
@@ -264,4 +266,61 @@ public void replaceAllConfigs(final Map<AirbyteConfig, Stream<?>> configs, final
     }
   }
 
+  public void writeServiceAccountJsonCredentials(final WorkspaceServiceAccount workspaceServiceAccount)
+      throws JsonValidationException, IOException {
+    final WorkspaceServiceAccount workspaceServiceAccountForDB = getWorkspaceServiceAccountWithSecretCoordinate(workspaceServiceAccount);
+    configRepository.writeWorkspaceServiceAccountNoSecrets(workspaceServiceAccountForDB);
+  }
+
+  /**
+   * This method is to encrypt the secret JSON key and HMAC key of a GCP service account a associated
+   * with a workspace. If in future we build a similar feature i.e. an AWS account associated with a
+   * workspace, we will have to build new implementation for it
+   */
+  private WorkspaceServiceAccount getWorkspaceServiceAccountWithSecretCoordinate(final WorkspaceServiceAccount workspaceServiceAccount)
+      throws JsonValidationException, IOException {
+    if (longLivedSecretPersistence.isPresent()) {
+      final WorkspaceServiceAccount clonedWorkspaceServiceAccount = Jsons.clone(workspaceServiceAccount);
+      final Optional<WorkspaceServiceAccount> optionalWorkspaceServiceAccount = getOptionalWorkspaceServiceAccount(
+          workspaceServiceAccount.getWorkspaceId());
+      // Convert the JSON key of Service Account into secret co-oridnate. Ref :
+      // https://cloud.google.com/iam/docs/service-accounts#key-types
+      if (workspaceServiceAccount.getJsonCredential() != null) {
+        final SecretCoordinateToPayload jsonCredSecretCoordinateToPayload =
+            SecretsHelpers.convertServiceAccountCredsToSecret(workspaceServiceAccount.getJsonCredential().toPrettyString(),
+                longLivedSecretPersistence.get(),
+                workspaceServiceAccount.getWorkspaceId(),
+                UUID::randomUUID,
+                optionalWorkspaceServiceAccount.map(WorkspaceServiceAccount::getJsonCredential).orElse(null),
+                "json");
+        longLivedSecretPersistence.get().write(jsonCredSecretCoordinateToPayload.secretCoordinate(), jsonCredSecretCoordinateToPayload.payload());
+        clonedWorkspaceServiceAccount.setJsonCredential(jsonCredSecretCoordinateToPayload.secretCoordinateForDB());
+      }
+      // Convert the HMAC key of Service Account into secret co-oridnate. Ref :
+      // https://cloud.google.com/storage/docs/authentication/hmackeys
+      if (workspaceServiceAccount.getHmacKey() != null) {
+        final SecretCoordinateToPayload hmackKeySecretCoordinateToPayload =
+            SecretsHelpers.convertServiceAccountCredsToSecret(workspaceServiceAccount.getHmacKey().toString(),
+                longLivedSecretPersistence.get(),
+                workspaceServiceAccount.getWorkspaceId(),
+                UUID::randomUUID,
+                optionalWorkspaceServiceAccount.map(WorkspaceServiceAccount::getHmacKey).orElse(null),
+                "hmac");
+        longLivedSecretPersistence.get().write(hmackKeySecretCoordinateToPayload.secretCoordinate(), hmackKeySecretCoordinateToPayload.payload());
+        clonedWorkspaceServiceAccount.setHmacKey(hmackKeySecretCoordinateToPayload.secretCoordinateForDB());
+      }
+      return clonedWorkspaceServiceAccount;
+    }
+    return workspaceServiceAccount;
+  }
+
+  public Optional<WorkspaceServiceAccount> getOptionalWorkspaceServiceAccount(final UUID workspaceId)
+      throws JsonValidationException, IOException {
+    try {
+      return Optional.of(configRepository.getWorkspaceServiceAccountNoSecrets(workspaceId));
+    } catch (ConfigNotFoundException e) {
+      return Optional.empty();
+    }
+  }
+
 }

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

@@ -16,4 +16,9 @@ public JsonNode hydrate(final JsonNode partialConfig) {
     return partialConfig;
   }
 
+  @Override
+  public JsonNode hydrateSecretCoordinate(JsonNode secretCoordinate) {
+    return secretCoordinate;
+  }
+
 }

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

@@ -22,4 +22,9 @@ public JsonNode hydrate(final JsonNode partialConfig) {
     return SecretsHelpers.combineConfig(partialConfig, readOnlySecretPersistence);
   }
 
+  @Override
+  public JsonNode hydrateSecretCoordinate(final JsonNode secretCoordinate) {
+    return SecretsHelpers.hydrateSecretCoordinate(secretCoordinate, readOnlySecretPersistence);
+  }
+
 }

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

@@ -0,0 +1,13 @@
+/*
+ * Copyright (c) 2021 Airbyte, Inc., all rights reserved.
+ */
+
+package io.airbyte.config.persistence.split_secrets;
+
+import com.fasterxml.jackson.databind.JsonNode;
+
+public record SecretCoordinateToPayload(SecretCoordinate secretCoordinate,
+                                        String payload,
+                                        JsonNode secretCoordinateForDB) {
+
+}

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

@@ -116,7 +116,7 @@ public static JsonNode combineConfig(final JsonNode partialConfig, final ReadOnl
     if (config.has(COORDINATE_FIELD)) {
       final var coordinateNode = config.get(COORDINATE_FIELD);
       final var coordinate = getCoordinateFromTextNode(coordinateNode);
-      return getOrThrowSecretValueNode(secretPersistence, coordinate);
+      return new TextNode(getOrThrowSecretValue(secretPersistence, coordinate));
     }
 
     // otherwise iterate through all object fields
@@ -336,16 +336,16 @@ private static JsonNode getFieldOrEmptyNode(final JsonNode node, final int field
    * @param secretPersistence storage layer for secrets
    * @param coordinate reference to a secret in the persistence
    * @throws RuntimeException when a secret at that coordinate is not available in the persistence
-   * @return a json text node containing the secret value
+   * @return a json string containing the secret value or a JSON
    */
-  private static TextNode getOrThrowSecretValueNode(final ReadOnlySecretPersistence secretPersistence, final SecretCoordinate coordinate) {
+  private static String getOrThrowSecretValue(final ReadOnlySecretPersistence secretPersistence,
+                                              final SecretCoordinate coordinate) {
     final var secretValue = secretPersistence.read(coordinate);
 
     if (secretValue.isEmpty()) {
       throw new RuntimeException(String.format("That secret was not found in the store! Coordinate: %s", coordinate.getFullCoordinate()));
     }
-
-    return new TextNode(secretValue.get());
+    return secretValue.get();
   }
 
   private static SecretCoordinate getCoordinateFromTextNode(final JsonNode node) {
@@ -379,6 +379,15 @@ protected static SecretCoordinate getCoordinate(
                                                   final UUID workspaceId,
                                                   final Supplier<UUID> uuidSupplier,
                                                   final @Nullable String oldSecretFullCoordinate) {
+    return getSecretCoordinate("airbyte_workspace_", newSecret, secretReader, workspaceId, uuidSupplier, oldSecretFullCoordinate);
+  }
+
+  private static SecretCoordinate getSecretCoordinate(final String secretBasePrefix,
+                                                      final String newSecret,
+                                                      final ReadOnlySecretPersistence secretReader,
+                                                      final UUID secretBaseId,
+                                                      final Supplier<UUID> uuidSupplier,
+                                                      final @Nullable String oldSecretFullCoordinate) {
     String coordinateBase = null;
     Long version = null;
 
@@ -398,7 +407,7 @@ protected static SecretCoordinate getCoordinate(
     if (coordinateBase == null) {
       // IMPORTANT: format of this cannot be changed without introducing migrations for secrets
       // persistences
-      coordinateBase = "airbyte_workspace_" + workspaceId + "_secret_" + uuidSupplier.get();
+      coordinateBase = secretBasePrefix + secretBaseId + "_secret_" + uuidSupplier.get();
     }
 
     if (version == null) {
@@ -408,4 +417,54 @@ protected static SecretCoordinate getCoordinate(
     return new SecretCoordinate(coordinateBase, version);
   }
 
+  /**
+   * This method takes in the key (JSON key or HMAC key) of a workspace service account as a secret
+   * and generates a co-ordinate for the secret so that the secret can be written in secret
+   * persistence at the generated co-ordinate
+   *
+   * @param newSecret The JSON key or HMAC key value
+   * @param secretReader To read the value from secret persistence for comparison with the new value
+   * @param workspaceId of the service account
+   * @param uuidSupplier provided to allow a test case to produce known UUIDs in order for easy *
+   *        fixture creation.
+   * @param oldSecretCoordinate a nullable full coordinate (base+version) retrieved from the *
+   *        previous config
+   * @param keyType HMAC ot JSON key
+   * @return a coordinate (versioned reference to where the secret is stored in the persistence)
+   */
+  public static SecretCoordinateToPayload convertServiceAccountCredsToSecret(final String newSecret,
+                                                                             final ReadOnlySecretPersistence secretReader,
+                                                                             final UUID workspaceId,
+                                                                             final Supplier<UUID> uuidSupplier,
+                                                                             final @Nullable JsonNode oldSecretCoordinate,
+                                                                             final String keyType) {
+    final String oldSecretFullCoordinate =
+        (oldSecretCoordinate != null && oldSecretCoordinate.has(COORDINATE_FIELD)) ? oldSecretCoordinate.get(COORDINATE_FIELD).asText()
+            : null;
+    final SecretCoordinate coordinateForStagingConfig = getSecretCoordinate("service_account_" + keyType + "_",
+        newSecret,
+        secretReader,
+        workspaceId,
+        uuidSupplier,
+        oldSecretFullCoordinate);
+    return new SecretCoordinateToPayload(coordinateForStagingConfig,
+        newSecret,
+        Jsons.jsonNode(Map.of(COORDINATE_FIELD,
+            coordinateForStagingConfig.getFullCoordinate())));
+  }
+
+  /**
+   * Takes in the secret coordinate in form of a JSON and fetches the secret from the store
+   *
+   * @param secretCoordinateAsJson The co-ordinate at which we expect the secret value to be present
+   *        in the secret persistence
+   * @param readOnlySecretPersistence The secret persistence
+   * @return Original secret value as JsonNode
+   */
+  public static JsonNode hydrateSecretCoordinate(final JsonNode secretCoordinateAsJson,
+                                                 final ReadOnlySecretPersistence readOnlySecretPersistence) {
+    final var secretCoordinate = getCoordinateFromTextNode(secretCoordinateAsJson.get(COORDINATE_FIELD));
+    return Jsons.deserialize(getOrThrowSecretValue(readOnlySecretPersistence, secretCoordinate));
+  }
+
 }

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

@@ -19,4 +19,12 @@ public interface SecretsHydrator {
    */
   JsonNode hydrate(JsonNode partialConfig);
 
+  /**
+   * Takes in the secret coordinate in form of a JSON and fetches the secret from the store
+   *
+   * @param secretCoordinate The co-ordinate of the secret in the store in JSON format
+   * @return original secret value
+   */
+  JsonNode hydrateSecretCoordinate(final JsonNode secretCoordinate);
+
 }