Don't return secrets in the API & Only update credentials when reques…

…ted (#1022)

airbyte-commons/src/main/java/io/airbyte/commons/json/JsonSecretsProcessor.java

@@ -0,0 +1,113 @@
+/*
+ * MIT License
+ *
+ * Copyright (c) 2020 Airbyte
+ *
+ * Permission is hereby granted, free of charge, to any person obtaining a copy
+ * of this software and associated documentation files (the "Software"), to deal
+ * in the Software without restriction, including without limitation the rights
+ * to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
+ * copies of the Software, and to permit persons to whom the Software is
+ * furnished to do so, subject to the following conditions:
+ *
+ * The above copyright notice and this permission notice shall be included in all
+ * copies or substantial portions of the Software.
+ *
+ * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
+ * IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
+ * FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
+ * AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
+ * LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
+ * OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
+ * SOFTWARE.
+ */
+
+package io.airbyte.commons.json;
+
+import com.fasterxml.jackson.databind.JsonNode;
+import com.fasterxml.jackson.databind.node.ObjectNode;
+import com.google.common.annotations.VisibleForTesting;
+import com.google.common.base.Preconditions;
+
+public class JsonSecretsProcessor {
+
+  public static String AIRBYTE_SECRET_FIELD = "airbyte_secret";
+
+  @VisibleForTesting
+  static String SECRETS_MASK = "**********";
+
+  private static String PROPERTIES_FIELD = "properties";
+
+  /**
+   * Returns a copy of the input object wherein any fields annotated with "airbyte_secret" in the
+   * input schema are masked.
+   * <p>
+   * TODO this method only masks secrets at the top level of the configuration object. It does not
+   * support the keywords anyOf, allOf, oneOf, not, and dependencies. This will be fixed in the
+   * future.
+   *
+   * @param schema Schema containing secret annotations
+   * @param obj Object containing potentially secret fields
+   * @return
+   */
+  public JsonNode maskSecrets(JsonNode obj, JsonNode schema) {
+    if (!canBeProcessed(schema)) {
+      return obj;
+    }
+    Preconditions.checkArgument(schema.isObject());
+
+    ObjectNode properties = (ObjectNode) schema.get(PROPERTIES_FIELD);
+    JsonNode copy = obj.deepCopy();
+    for (String key : Jsons.keys(properties)) {
+      if (isSecret(properties.get(key)) && copy.has(key)) {
+        ((ObjectNode) copy).put(key, SECRETS_MASK);
+      }
+    }
+
+    return copy;
+  }
+
+  /**
+   * Returns a copy of the destination object in which any secret fields (as denoted by the input
+   * schema) found in the source object are added.
+   * <p>
+   * TODO this method only absorbs secrets at the top level of the configuration object. It does not
+   * support the keywords anyOf, allOf, oneOf, not, and dependencies. This will be fixed in the
+   * future.
+   *
+   * @param src The object potentially containing secrets
+   * @param dst The object to absorb secrets into
+   * @param schema
+   * @return
+   */
+  public JsonNode copySecrets(JsonNode src, JsonNode dst, JsonNode schema) {
+    if (!canBeProcessed(schema)) {
+      return dst;
+    }
+    Preconditions.checkArgument(dst.isObject());
+    Preconditions.checkArgument(src.isObject());
+
+    ObjectNode dstCopy = dst.deepCopy();
+
+    ObjectNode properties = (ObjectNode) schema.get(PROPERTIES_FIELD);
+    for (String key : Jsons.keys(properties)) {
+      // We only copy the original secret if the destination object isn't attempting to overwrite it
+      // i.e: if the value of the secret isn't set to the mask
+      if (isSecret(properties.get(key)) && src.has(key)) {
+        if (dst.has(key) && dst.get(key).asText().equals(SECRETS_MASK))
+          dstCopy.set(key, src.get(key));
+      }
+    }
+
+    return dstCopy;
+  }
+
+  private static boolean isSecret(JsonNode obj) {
+    return obj.isObject() && obj.has(AIRBYTE_SECRET_FIELD) && obj.get(AIRBYTE_SECRET_FIELD).asBoolean();
+  }
+
+  private static boolean canBeProcessed(JsonNode schema) {
+    return schema.isObject() && schema.has(PROPERTIES_FIELD) && schema.get(PROPERTIES_FIELD).isObject();
+  }
+
+}

airbyte-commons/src/test/java/io/airbyte/commons/json/JsonSecretsProcessorTest.java

@@ -0,0 +1,123 @@
+/*
+ * MIT License
+ *
+ * Copyright (c) 2020 Airbyte
+ *
+ * Permission is hereby granted, free of charge, to any person obtaining a copy
+ * of this software and associated documentation files (the "Software"), to deal
+ * in the Software without restriction, including without limitation the rights
+ * to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
+ * copies of the Software, and to permit persons to whom the Software is
+ * furnished to do so, subject to the following conditions:
+ *
+ * The above copyright notice and this permission notice shall be included in all
+ * copies or substantial portions of the Software.
+ *
+ * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
+ * IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
+ * FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
+ * AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
+ * LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
+ * OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
+ * SOFTWARE.
+ */
+
+package io.airbyte.commons.json;
+
+import static io.airbyte.commons.json.JsonSecretsProcessor.SECRETS_MASK;
+import static org.junit.jupiter.api.Assertions.*;
+
+import com.fasterxml.jackson.databind.JsonNode;
+import com.google.common.collect.ImmutableMap;
+import io.airbyte.commons.resources.MoreResources;
+import java.io.IOException;
+import org.junit.jupiter.api.Test;
+
+public class JsonSecretsProcessorTest {
+
+  JsonSecretsProcessor processor = new JsonSecretsProcessor();
+
+  @Test
+  public void testMaskSecrets() throws IOException {
+    JsonNode obj = Jsons.jsonNode(ImmutableMap.builder()
+        .put("field1", "value1")
+        .put("field2", 2)
+        .put("secret1", "donttellanyone")
+        .put("secret2", "verysecret").build());
+    JsonNode schema = Jsons.deserialize(MoreResources.readResource("secrets_json_schema.json"));
+
+    JsonNode sanitized = processor.maskSecrets(obj, schema);
+
+    JsonNode expected = Jsons.jsonNode(ImmutableMap.builder()
+        .put("field1", "value1")
+        .put("field2", 2)
+        .put("secret1", SECRETS_MASK)
+        .put("secret2", SECRETS_MASK).build());
+    assertEquals(expected, sanitized);
+  }
+
+  @Test
+  public void testMaskSecretsNotInObj() throws IOException {
+    JsonNode schema = Jsons.deserialize(MoreResources.readResource("secrets_json_schema.json"));
+    JsonNode obj = Jsons.jsonNode(ImmutableMap.builder()
+        .put("field1", "value1")
+        .put("field2", 2).build());
+
+    JsonNode actual = processor.maskSecrets(obj, schema);
+
+    // Didn't have secrets, no fields should have been impacted.
+    assertEquals(obj, actual);
+  }
+
+  @Test
+  public void testCopySecrets() throws IOException {
+    JsonNode src = Jsons.jsonNode(ImmutableMap.builder()
+        .put("field1", "value1")
+        .put("field2", 2)
+        .put("additional_field", "dont_copy_me")
+        .put("secret1", "donttellanyone")
+        .put("secret2", "updateme")
+        .build());
+
+    JsonNode dst = Jsons.jsonNode(ImmutableMap.builder()
+        .put("field1", "value1")
+        .put("field2", 2)
+        .put("secret1", SECRETS_MASK)
+        .put("secret2", "newvalue")
+        .build());
+
+    JsonNode schema = Jsons.deserialize(MoreResources.readResource("secrets_json_schema.json"));
+
+    JsonNode actual = processor.copySecrets(src, dst, schema);
+
+    JsonNode expected = Jsons.jsonNode(ImmutableMap.builder()
+        .put("field1", "value1")
+        .put("field2", 2)
+        .put("secret1", "donttellanyone")
+        .put("secret2", "newvalue")
+        .build());
+
+    assertEquals(expected, actual);
+  }
+
+  @Test
+  public void testCopySecretsNotInSrc() throws IOException {
+    JsonNode schema = Jsons.deserialize(MoreResources.readResource("secrets_json_schema.json"));
+    JsonNode src = Jsons.jsonNode(ImmutableMap.builder()
+        .put("field1", "value1")
+        .put("field2", 2)
+        .put("additional_field", "dont_copy_me")
+        .build());
+
+    JsonNode dst = Jsons.jsonNode(ImmutableMap.builder()
+        .put("field1", "value1")
+        .put("field2", 2)
+        .build());
+
+    JsonNode expected = dst.deepCopy();
+    JsonNode actual = processor.copySecrets(src, dst, schema);
+
+    assertEquals(expected, actual);
+  }
+
+}

airbyte-commons/src/test/resources/secrets_json_schema.json

@@ -0,0 +1,18 @@
+{
+  "properties": {
+    "secret1": {
+      "type": "string",
+      "airbyte_secret": true
+    },
+    "secret2": {
+      "type": "string",
+      "airbyte_secret": "true"
+    },
+    "field1": {
+      "type": "string"
+    },
+    "field2": {
+      "type": "number"
+    }
+  }
+}

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

@@ -91,12 +91,6 @@ public StandardSourceDefinition getSourceDefinitionFromConnection(UUID connectio
     }
   }
 
-  public StandardSourceDefinition getStandardSource(final UUID sourceId)
-      throws JsonValidationException, IOException, ConfigNotFoundException {
-    return persistence.getConfig(
-        ConfigSchema.STANDARD_SOURCE_DEFINITION, sourceId.toString(), StandardSourceDefinition.class);
-  }
-
   public List<StandardSourceDefinition> listStandardSources()
       throws JsonValidationException, IOException, ConfigNotFoundException {
     return persistence.listConfigs(ConfigSchema.STANDARD_SOURCE_DEFINITION, StandardSourceDefinition.class);

airbyte-scheduler/src/main/java/io/airbyte/scheduler/job_factory/DefaultSyncJobFactory.java

@@ -57,7 +57,7 @@ public Long create(final UUID connectionId) {
       final DestinationConnection destinationConnection =
           configRepository.getDestinationConnection(standardSync.getDestinationId());
 
-      final StandardSourceDefinition sourceDefinition = configRepository.getStandardSource(sourceConnection.getSourceDefinitionId());
+      final StandardSourceDefinition sourceDefinition = configRepository.getStandardSourceDefinition(sourceConnection.getSourceDefinitionId());
       final StandardDestinationDefinition destinationDefinition =
           configRepository.getStandardDestinationDefinition(destinationConnection.getDestinationDefinitionId());
 

airbyte-scheduler/src/test/java/io/airbyte/scheduler/job_factory/DefaultSyncJobFactoryTest.java

@@ -77,7 +77,7 @@ void createSyncJobFromConnectionId() throws JsonValidationException, ConfigNotFo
     when(schedulerPersistence
         .createSyncJob(sourceConnection, destinationConnection, standardSync, srcDockerImage, dstDockerImage))
             .thenReturn(jobId);
-    when(configRepository.getStandardSource(sourceDefinitionId))
+    when(configRepository.getStandardSourceDefinition(sourceDefinitionId))
         .thenReturn(new StandardSourceDefinition().withSourceDefinitionId(sourceDefinitionId).withDockerRepository(srcDockerRepo)
             .withDockerImageTag(srcDockerTag));