From d41c3f7d6f9e95b5c53087d2aa9a09956b0940ec Mon Sep 17 00:00:00 2001
From: Charles <charles@airbyte.io>
Date: Thu, 14 Apr 2022 17:37:06 -0700
Subject: [PATCH] json schema traversal + secrets (#11847)

---
 .../io/airbyte/commons/json/JsonPaths.java    |  22 +-
 .../io/airbyte/commons/json/JsonSchemas.java  | 176 +++++++++-
 .../airbyte/commons/json/JsonSchemasTest.java | 121 +++++++
 .../io/airbyte/commons/json/JsonsTest.java    |  18 -
 .../json_schemas/composite_json_schema.json   |  39 +++
 .../json_schemas/json_with_all_types.json     |  34 ++
 .../json_with_array_type_fields.json          |  18 +
 ...ith_array_type_fields_with_composites.json |  32 ++
 .../split_secrets/JsonSecretsProcessor.java   | 128 ++-----
 .../split_secrets/SecretsHelpers.java         | 202 ++++-------
 .../JsonSecretsProcessorTest.java             | 326 ++++++++++--------
 .../split_secrets/SecretsHelpersTest.java     |  18 +-
 .../split_secrets/SecretsTestCase.java        |  15 +
 .../test_cases/ArrayTestCase.java             |  12 +-
 .../test_cases/NestedObjectTestCase.java      |  20 +-
 .../src/test/resources/array/expectedPaths    |   1 +
 .../test/resources/array/partial_config.json  |  12 +-
 .../array/updated_partial_config.json         |  12 +-
 .../src/test/resources/array2/expected.json   |   3 +
 .../test/resources/array2/full_config.json    |   3 +
 .../test/resources/array2/partial_config.json |   7 +
 .../src/test/resources/array2/spec.json       |  20 ++
 .../test/resources/array2/update_config.json  |   3 +
 .../array2/updated_partial_config.json        |   7 +
 .../resources/array_of_oneof/expectedPaths    |   1 +
 .../resources/nested_object/expectedPaths     |   1 +
 .../nested_object/partial_config.json         |   4 +-
 .../nested_object/updated_partial_config.json |   4 +-
 .../updated_partial_config_update1.json       |   4 +-
 .../updated_partial_config_update2.json       |   4 +-
 .../test/resources/nested_oneof/expectedPaths |   1 +
 .../src/test/resources/oneof/expectedPaths    |   1 +
 .../resources/optional_password/expectedPaths |   1 +
 .../resources/postgres_ssh_key/expectedPaths  |   1 +
 .../src/test/resources/simple/expectedPaths   |   1 +
 35 files changed, 840 insertions(+), 432 deletions(-)
 create mode 100644 airbyte-commons/src/test/java/io/airbyte/commons/json/JsonSchemasTest.java
 create mode 100644 airbyte-commons/src/test/resources/json_schemas/composite_json_schema.json
 create mode 100644 airbyte-commons/src/test/resources/json_schemas/json_with_all_types.json
 create mode 100644 airbyte-commons/src/test/resources/json_schemas/json_with_array_type_fields.json
 create mode 100644 airbyte-commons/src/test/resources/json_schemas/json_with_array_type_fields_with_composites.json
 create mode 100644 airbyte-config/persistence/src/test/resources/array/expectedPaths
 create mode 100644 airbyte-config/persistence/src/test/resources/array2/expected.json
 create mode 100644 airbyte-config/persistence/src/test/resources/array2/full_config.json
 create mode 100644 airbyte-config/persistence/src/test/resources/array2/partial_config.json
 create mode 100644 airbyte-config/persistence/src/test/resources/array2/spec.json
 create mode 100644 airbyte-config/persistence/src/test/resources/array2/update_config.json
 create mode 100644 airbyte-config/persistence/src/test/resources/array2/updated_partial_config.json
 create mode 100644 airbyte-config/persistence/src/test/resources/array_of_oneof/expectedPaths
 create mode 100644 airbyte-config/persistence/src/test/resources/nested_object/expectedPaths
 create mode 100644 airbyte-config/persistence/src/test/resources/nested_oneof/expectedPaths
 create mode 100644 airbyte-config/persistence/src/test/resources/oneof/expectedPaths
 create mode 100644 airbyte-config/persistence/src/test/resources/optional_password/expectedPaths
 create mode 100644 airbyte-config/persistence/src/test/resources/postgres_ssh_key/expectedPaths
 create mode 100644 airbyte-config/persistence/src/test/resources/simple/expectedPaths

diff --git a/airbyte-commons/src/main/java/io/airbyte/commons/json/JsonPaths.java b/airbyte-commons/src/main/java/io/airbyte/commons/json/JsonPaths.java
index 428299aa958b..0b75b981500d 100644
--- a/airbyte-commons/src/main/java/io/airbyte/commons/json/JsonPaths.java
+++ b/airbyte-commons/src/main/java/io/airbyte/commons/json/JsonPaths.java
@@ -42,15 +42,15 @@
  * returning a list for query results. In addition, we provide helper functions that will just
  * return a single value (see: {@link JsonPaths#getSingleValue(JsonNode, String)}). These should
  * only be used if it is not possible for a query to return more than one value.
- *
- * Note: Package private as most uses of JsonPaths seems like they can be hidden inside other
- * commons libraries (i.e. Jsons and JsonsSchemas). If this assumption proves incorrect, we can open
- * it up.
  */
-class JsonPaths {
+public class JsonPaths {
 
   private static final Logger LOGGER = LoggerFactory.getLogger(JsonPaths.class);
 
+  static final String JSON_PATH_START_CHARACTER = "$";
+  static final String JSON_PATH_LIST_SPLAT = "[*]";
+  static final String JSON_PATH_FIELD_SEPARATOR = ".";
+
   // set default configurations at start up to match our JSON setup.
   static {
     Configuration.setDefaults(new Configuration.Defaults() {
@@ -82,6 +82,18 @@ public Set<Option> options() {
     });
   }
 
+  public static String empty() {
+    return JSON_PATH_START_CHARACTER;
+  }
+
+  public static String appendField(final String jsonPath, final String field) {
+    return jsonPath + JSON_PATH_FIELD_SEPARATOR + field;
+  }
+
+  public static String appendAppendListSplat(final String jsonPath) {
+    return jsonPath + JSON_PATH_LIST_SPLAT;
+  }
+
   /*
    * This version of the JsonPath Configuration object allows queries to return to the path of values
    * instead of the values that were found.
diff --git a/airbyte-commons/src/main/java/io/airbyte/commons/json/JsonSchemas.java b/airbyte-commons/src/main/java/io/airbyte/commons/json/JsonSchemas.java
index 9ff7ed735e14..47e18010ff1d 100644
--- a/airbyte-commons/src/main/java/io/airbyte/commons/json/JsonSchemas.java
+++ b/airbyte-commons/src/main/java/io/airbyte/commons/json/JsonSchemas.java
@@ -8,15 +8,48 @@
 import com.fasterxml.jackson.databind.node.ObjectNode;
 import io.airbyte.commons.io.IOs;
 import io.airbyte.commons.resources.MoreResources;
+import io.airbyte.commons.util.MoreIterators;
 import java.io.IOException;
 import java.nio.file.Files;
 import java.nio.file.Path;
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.Collections;
+import java.util.HashSet;
+import java.util.Iterator;
 import java.util.List;
+import java.util.Map.Entry;
+import java.util.Optional;
+import java.util.Set;
+import java.util.function.BiConsumer;
+import java.util.function.BiFunction;
+import java.util.function.Predicate;
 import java.util.stream.Collectors;
 import java.util.stream.Stream;
 
+// todo (cgardens) - we need the ability to identify jsonschemas that Airbyte considers invalid for
+// a connector (e.g. "not" keyword).
 public class JsonSchemas {
 
+  private static final String JSON_SCHEMA_TYPE_KEY = "type";
+  private static final String JSON_SCHEMA_PROPERTIES_KEY = "properties";
+  private static final String JSON_SCHEMA_ITEMS_KEY = "items";
+
+  // all JSONSchema types.
+  private static final String ARRAY_TYPE = "array";
+  private static final String OBJECT_TYPE = "object";
+  private static final String STRING_TYPE = "string";
+  private static final String NUMBER_TYPE = "number";
+  private static final String BOOLEAN_TYPE = "boolean";
+  private static final String NULL_TYPE = "null";
+  private static final String ONE_OF_TYPE = "oneOf";
+  private static final String ALL_OF_TYPE = "allOf";
+  private static final String ANY_OF_TYPE = "anyOf";
+
+  private static final String ARRAY_JSON_PATH = "[]";
+
+  private static final Set<String> COMPOSITE_KEYWORDS = Set.of(ONE_OF_TYPE, ALL_OF_TYPE, ANY_OF_TYPE);
+
   /**
    * JsonSchema supports to ways of declaring type. `type: "string"` and `type: ["null", "string"]`.
    * This method will mutate a JsonNode with a type field so that the output type is the array
@@ -25,16 +58,16 @@ public class JsonSchemas {
    * @param jsonNode - a json object with children that contain types.
    */
   public static void mutateTypeToArrayStandard(final JsonNode jsonNode) {
-    if (jsonNode.get("type") != null && !jsonNode.get("type").isArray()) {
-      final JsonNode type = jsonNode.get("type");
-      ((ObjectNode) jsonNode).putArray("type").add(type);
+    if (jsonNode.get(JSON_SCHEMA_TYPE_KEY) != null && !jsonNode.get(JSON_SCHEMA_TYPE_KEY).isArray()) {
+      final JsonNode type = jsonNode.get(JSON_SCHEMA_TYPE_KEY);
+      ((ObjectNode) jsonNode).putArray(JSON_SCHEMA_TYPE_KEY).add(type);
     }
   }
 
   /*
-   * JsonReferenceProcessor relies on all of the json in consumes being in a file system (not in a
-   * jar). This method copies all of the json configs out of the jar into a temporary directory so
-   * that JsonReferenceProcessor can find them.
+   * JsonReferenceProcessor relies on all the json in consumes being in a file system (not in a jar).
+   * This method copies all the json configs out of the jar into a temporary directory so that
+   * JsonReferenceProcessor can find them.
    */
   public static <T> Path prepareSchemas(final String resourceDir, final Class<T> klass) {
     try {
@@ -59,4 +92,135 @@ public static <T> Path prepareSchemas(final String resourceDir, final Class<T> k
     }
   }
 
+  public static void traverseJsonSchema(final JsonNode jsonSchemaNode, final BiConsumer<JsonNode, String> consumer) {
+    traverseJsonSchemaInternal(jsonSchemaNode, JsonPaths.empty(), consumer);
+  }
+
+  /**
+   * Traverse a JsonSchema object. At each node, optionally map a value.
+   *
+   * @param jsonSchema - JsonSchema object to traverse
+   * @param mapper - accepts the current node and the path to that node. if it returns an empty
+   *        optional, nothing will be collected, otherwise, whatever is returned will be collected and
+   *        returned by the final collection.
+   * @param <T> - type of objects being collected
+   * @return - collection of all items that were collected during the traversal. Returns a { @link
+   *         Collection } because there is no order or uniqueness guarantee so neither List nor Set
+   *         make sense.
+   */
+  public static <T> Collection<T> traverseJsonSchemaWithCollector(final JsonNode jsonSchema, final BiFunction<JsonNode, String, Optional<T>> mapper) {
+    final List<T> collectors = new ArrayList<>();
+    traverseJsonSchema(jsonSchema, (node, path) -> mapper.apply(node, path).ifPresent(collectors::add));
+    return collectors;
+  }
+
+  /**
+   * Traverses a JsonSchema object. It returns the path to each node that meet the provided condition.
+   * The paths are return in JsonPath format
+   *
+   * @param obj - JsonSchema object to traverse
+   * @param predicate - predicate to determine if the path for a node should be collected.
+   * @return - collection of all paths that were collected during the traversal.
+   */
+  public static Set<String> collectJsonPathsThatMeetCondition(final JsonNode obj, final Predicate<JsonNode> predicate) {
+    return new HashSet<>(traverseJsonSchemaWithCollector(obj, (node, path) -> {
+      if (predicate.test(node)) {
+        return Optional.of(path);
+      } else {
+        return Optional.empty();
+      }
+    }));
+  }
+
+  /**
+   * Recursive, depth-first implementation of { @link JsonSchemas#traverseJsonSchema(final JsonNode
+   * jsonNode, final BiConsumer<JsonNode, List<String>> consumer) }. Takes path as argument so that
+   * the path can be passsed to the consumer.
+   *
+   * @param jsonSchemaNode - jsonschema object to traverse.
+   * @param path - path from the first call of traverseJsonSchema to the current node.
+   * @param consumer - consumer to be called at each node. it accepts the current node and the path to
+   *        the node from the root of the object passed at the root level invocation
+   */
+  // todo (cgardens) - replace with easier to understand traversal logic from SecretsHelper.
+  private static void traverseJsonSchemaInternal(final JsonNode jsonSchemaNode,
+                                                 final String path,
+                                                 final BiConsumer<JsonNode, String> consumer) {
+    if (!jsonSchemaNode.isObject()) {
+      throw new IllegalArgumentException(String.format("json schema nodes should always be object nodes. path: %s actual: %s", path, jsonSchemaNode));
+    }
+
+    consumer.accept(jsonSchemaNode, path);
+    // if type is missing assume object. not official JsonSchema, but it seems to be a common
+    // compromise.
+    final List<String> nodeTypes = getTypeOrObject(jsonSchemaNode);
+
+    for (final String nodeType : nodeTypes) {
+      switch (nodeType) {
+        // case BOOLEAN_TYPE, NUMBER_TYPE, STRING_TYPE, NULL_TYPE -> do nothing after consumer.accept above.
+        case ARRAY_TYPE -> {
+          final String newPath = JsonPaths.appendAppendListSplat(path);
+          // hit every node.
+          traverseJsonSchemaInternal(jsonSchemaNode.get(JSON_SCHEMA_ITEMS_KEY), newPath, consumer);
+        }
+        case OBJECT_TYPE -> {
+          final Optional<String> comboKeyWordOptional = getKeywordIfComposite(jsonSchemaNode);
+          if (jsonSchemaNode.has(JSON_SCHEMA_PROPERTIES_KEY)) {
+            for (final Iterator<Entry<String, JsonNode>> it = jsonSchemaNode.get(JSON_SCHEMA_PROPERTIES_KEY).fields(); it.hasNext();) {
+              final Entry<String, JsonNode> child = it.next();
+              final String newPath = JsonPaths.appendField(path, child.getKey());
+              traverseJsonSchemaInternal(child.getValue(), newPath, consumer);
+            }
+          } else if (comboKeyWordOptional.isPresent()) {
+            for (final JsonNode arrayItem : jsonSchemaNode.get(comboKeyWordOptional.get())) {
+              traverseJsonSchemaInternal(arrayItem, path, consumer);
+            }
+          } else {
+            throw new IllegalArgumentException(
+                "malformed JsonSchema object type, must have one of the following fields: properties, oneOf, allOf, anyOf");
+          }
+        }
+      }
+    }
+  }
+
+  /**
+   * If the object uses JSONSchema composite functionality (e.g. oneOf, anyOf, allOf), detect it and
+   * return which one it is using.
+   *
+   * @param node - object to detect use of composite functionality.
+   * @return the composite functionality being used, if not using composite functionality, empty.
+   */
+  private static Optional<String> getKeywordIfComposite(final JsonNode node) {
+    for (final String keyWord : COMPOSITE_KEYWORDS) {
+      if (node.has(keyWord)) {
+        return Optional.ofNullable(keyWord);
+      }
+    }
+    return Optional.empty();
+  }
+
+  public static List<String> getTypeOrObject(final JsonNode jsonNode) {
+    final List<String> types = getType(jsonNode);
+    if (types.isEmpty()) {
+      return List.of(OBJECT_TYPE);
+    } else {
+      return types;
+    }
+  }
+
+  public static List<String> getType(final JsonNode jsonNode) {
+    if (jsonNode.has(JSON_SCHEMA_TYPE_KEY)) {
+      if (jsonNode.get(JSON_SCHEMA_TYPE_KEY).isArray()) {
+        return MoreIterators.toList(jsonNode.get(JSON_SCHEMA_TYPE_KEY).iterator())
+            .stream()
+            .map(JsonNode::asText)
+            .collect(Collectors.toList());
+      } else {
+        return List.of(jsonNode.get(JSON_SCHEMA_TYPE_KEY).asText());
+      }
+    }
+    return Collections.emptyList();
+  }
+
 }
diff --git a/airbyte-commons/src/test/java/io/airbyte/commons/json/JsonSchemasTest.java b/airbyte-commons/src/test/java/io/airbyte/commons/json/JsonSchemasTest.java
new file mode 100644
index 000000000000..e974612bb33c
--- /dev/null
+++ b/airbyte-commons/src/test/java/io/airbyte/commons/json/JsonSchemasTest.java
@@ -0,0 +1,121 @@
+/*
+ * Copyright (c) 2021 Airbyte, Inc., all rights reserved.
+ */
+
+package io.airbyte.commons.json;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.mockito.Mockito.mock;
+
+import com.fasterxml.jackson.databind.JsonNode;
+import io.airbyte.commons.resources.MoreResources;
+import java.io.IOException;
+import java.util.function.BiConsumer;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.ValueSource;
+import org.mockito.InOrder;
+import org.mockito.Mockito;
+
+class JsonSchemasTest {
+
+  @Test
+  void testMutateTypeToArrayStandard() {
+    final JsonNode expectedWithoutType = Jsons.deserialize("{\"test\":\"abc\"}");
+    final JsonNode actualWithoutType = Jsons.clone(expectedWithoutType);
+    JsonSchemas.mutateTypeToArrayStandard(expectedWithoutType);
+    assertEquals(expectedWithoutType, actualWithoutType);
+
+    final JsonNode expectedWithArrayType = Jsons.deserialize("{\"test\":\"abc\", \"type\":[\"object\"]}");
+    final JsonNode actualWithArrayType = Jsons.clone(expectedWithArrayType);
+    JsonSchemas.mutateTypeToArrayStandard(actualWithArrayType);
+    assertEquals(expectedWithoutType, actualWithoutType);
+
+    final JsonNode expectedWithoutArrayType = Jsons.deserialize("{\"test\":\"abc\", \"type\":[\"object\"]}");
+    final JsonNode actualWithStringType = Jsons.deserialize("{\"test\":\"abc\", \"type\":\"object\"}");
+    JsonSchemas.mutateTypeToArrayStandard(actualWithStringType);
+    assertEquals(expectedWithoutArrayType, actualWithStringType);
+  }
+
+  @SuppressWarnings("unchecked")
+  @Test
+  void testTraverse() throws IOException {
+    final JsonNode jsonWithAllTypes = Jsons.deserialize(MoreResources.readResource("json_schemas/json_with_all_types.json"));
+    final BiConsumer<JsonNode, String> mock = mock(BiConsumer.class);
+
+    JsonSchemas.traverseJsonSchema(jsonWithAllTypes, mock);
+    final InOrder inOrder = Mockito.inOrder(mock);
+    inOrder.verify(mock).accept(jsonWithAllTypes, JsonPaths.empty());
+    inOrder.verify(mock).accept(jsonWithAllTypes.get("properties").get("name"), "$.name");
+    inOrder.verify(mock).accept(jsonWithAllTypes.get("properties").get("name").get("properties").get("first"), "$.name.first");
+    inOrder.verify(mock).accept(jsonWithAllTypes.get("properties").get("name").get("properties").get("last"), "$.name.last");
+    inOrder.verify(mock).accept(jsonWithAllTypes.get("properties").get("company"), "$.company");
+    inOrder.verify(mock).accept(jsonWithAllTypes.get("properties").get("pets"), "$.pets");
+    inOrder.verify(mock).accept(jsonWithAllTypes.get("properties").get("pets").get("items"), "$.pets[*]");
+    inOrder.verify(mock).accept(jsonWithAllTypes.get("properties").get("pets").get("items").get("properties").get("type"), "$.pets[*].type");
+    inOrder.verify(mock).accept(jsonWithAllTypes.get("properties").get("pets").get("items").get("properties").get("number"), "$.pets[*].number");
+    inOrder.verifyNoMoreInteractions();
+  }
+
+  @SuppressWarnings("unchecked")
+  @ValueSource(strings = {
+    "anyOf",
+    "oneOf",
+    "allOf"
+  })
+  @ParameterizedTest
+  void testTraverseComposite(final String compositeKeyword) throws IOException {
+    final String jsonSchemaString = MoreResources.readResource("json_schemas/composite_json_schema.json")
+        .replaceAll("<composite-placeholder>", compositeKeyword);
+    final JsonNode jsonWithAllTypes = Jsons.deserialize(jsonSchemaString);
+    final BiConsumer<JsonNode, String> mock = mock(BiConsumer.class);
+
+    JsonSchemas.traverseJsonSchema(jsonWithAllTypes, mock);
+
+    final InOrder inOrder = Mockito.inOrder(mock);
+    inOrder.verify(mock).accept(jsonWithAllTypes, JsonPaths.empty());
+    inOrder.verify(mock).accept(jsonWithAllTypes.get(compositeKeyword).get(0), JsonPaths.empty());
+    inOrder.verify(mock).accept(jsonWithAllTypes.get(compositeKeyword).get(1), JsonPaths.empty());
+    inOrder.verify(mock).accept(jsonWithAllTypes.get(compositeKeyword).get(1).get("properties").get("prop1"), "$.prop1");
+    inOrder.verify(mock).accept(jsonWithAllTypes.get(compositeKeyword).get(2), JsonPaths.empty());
+    inOrder.verify(mock).accept(jsonWithAllTypes.get(compositeKeyword).get(2).get("items"), "$[*]");
+    inOrder.verify(mock).accept(jsonWithAllTypes.get(compositeKeyword).get(3).get(compositeKeyword).get(0), JsonPaths.empty());
+    inOrder.verify(mock).accept(jsonWithAllTypes.get(compositeKeyword).get(3).get(compositeKeyword).get(1), JsonPaths.empty());
+    inOrder.verify(mock).accept(jsonWithAllTypes.get(compositeKeyword).get(3).get(compositeKeyword).get(1).get("items"), "$[*]");
+    inOrder.verifyNoMoreInteractions();
+  }
+
+  @SuppressWarnings("unchecked")
+  @Test
+  void testTraverseMultiType() throws IOException {
+    final JsonNode jsonWithAllTypes = Jsons.deserialize(MoreResources.readResource("json_schemas/json_with_array_type_fields.json"));
+    final BiConsumer<JsonNode, String> mock = mock(BiConsumer.class);
+
+    JsonSchemas.traverseJsonSchema(jsonWithAllTypes, mock);
+    final InOrder inOrder = Mockito.inOrder(mock);
+    inOrder.verify(mock).accept(jsonWithAllTypes, JsonPaths.empty());
+    inOrder.verify(mock).accept(jsonWithAllTypes.get("properties").get("company"), "$.company");
+    inOrder.verify(mock).accept(jsonWithAllTypes.get("items"), "$[*]");
+    inOrder.verify(mock).accept(jsonWithAllTypes.get("items").get("properties").get("user"), "$[*].user");
+    inOrder.verifyNoMoreInteractions();
+  }
+
+  @SuppressWarnings("unchecked")
+  @Test
+  void testTraverseMultiTypeComposite() throws IOException {
+    final String compositeKeyword = "anyOf";
+    final JsonNode jsonWithAllTypes = Jsons.deserialize(MoreResources.readResource("json_schemas/json_with_array_type_fields_with_composites.json"));
+    final BiConsumer<JsonNode, String> mock = mock(BiConsumer.class);
+
+    JsonSchemas.traverseJsonSchema(jsonWithAllTypes, mock);
+
+    final InOrder inOrder = Mockito.inOrder(mock);
+    inOrder.verify(mock).accept(jsonWithAllTypes, JsonPaths.empty());
+    inOrder.verify(mock).accept(jsonWithAllTypes.get(compositeKeyword).get(0).get("properties").get("company"), "$.company");
+    inOrder.verify(mock).accept(jsonWithAllTypes.get(compositeKeyword).get(1).get("properties").get("organization"), "$.organization");
+    inOrder.verify(mock).accept(jsonWithAllTypes.get("items"), "$[*]");
+    inOrder.verify(mock).accept(jsonWithAllTypes.get("items").get("properties").get("user"), "$[*].user");
+    inOrder.verifyNoMoreInteractions();
+  }
+
+}
diff --git a/airbyte-commons/src/test/java/io/airbyte/commons/json/JsonsTest.java b/airbyte-commons/src/test/java/io/airbyte/commons/json/JsonsTest.java
index 8ae3b104f58a..4dbc291f9e7b 100644
--- a/airbyte-commons/src/test/java/io/airbyte/commons/json/JsonsTest.java
+++ b/airbyte-commons/src/test/java/io/airbyte/commons/json/JsonsTest.java
@@ -186,24 +186,6 @@ void testClone() {
     assertEquals(expected, actual);
   }
 
-  @Test
-  void testMutateTypeToArrayStandard() {
-    final JsonNode expectedWithoutType = Jsons.deserialize("{\"test\":\"abc\"}");
-    final JsonNode actualWithoutType = Jsons.clone(expectedWithoutType);
-    JsonSchemas.mutateTypeToArrayStandard(expectedWithoutType);
-    assertEquals(expectedWithoutType, actualWithoutType);
-
-    final JsonNode expectedWithArrayType = Jsons.deserialize("{\"test\":\"abc\", \"type\":[\"object\"]}");
-    final JsonNode actualWithArrayType = Jsons.clone(expectedWithArrayType);
-    JsonSchemas.mutateTypeToArrayStandard(actualWithArrayType);
-    assertEquals(expectedWithoutType, actualWithoutType);
-
-    final JsonNode expectedWithoutArrayType = Jsons.deserialize("{\"test\":\"abc\", \"type\":[\"object\"]}");
-    final JsonNode actualWithStringType = Jsons.deserialize("{\"test\":\"abc\", \"type\":\"object\"}");
-    JsonSchemas.mutateTypeToArrayStandard(actualWithStringType);
-    assertEquals(expectedWithoutArrayType, actualWithStringType);
-  }
-
   @Test
   void testToBytes() {
     final String jsonString = "{\"test\":\"abc\",\"type\":[\"object\"]}";
diff --git a/airbyte-commons/src/test/resources/json_schemas/composite_json_schema.json b/airbyte-commons/src/test/resources/json_schemas/composite_json_schema.json
new file mode 100644
index 000000000000..ba2ec2918e36
--- /dev/null
+++ b/airbyte-commons/src/test/resources/json_schemas/composite_json_schema.json
@@ -0,0 +1,39 @@
+{
+  "<composite-placeholder>": [
+    {
+      "type": "string",
+      "description": "top level"
+    },
+    {
+      "type": "object",
+      "properties": {
+        "prop1": {
+          "type": "string"
+        }
+      }
+    },
+    {
+      "type": "array",
+      "items": {
+        "type": "string",
+        "description": "in array"
+      }
+    },
+    {
+      "type": "object",
+      "<composite-placeholder>": [
+        {
+          "type": "string",
+          "description": "once nested"
+        },
+        {
+          "type": "array",
+          "items": {
+            "type": "string",
+            "description": "nested in array"
+          }
+        }
+      ]
+    }
+  ]
+}
diff --git a/airbyte-commons/src/test/resources/json_schemas/json_with_all_types.json b/airbyte-commons/src/test/resources/json_schemas/json_with_all_types.json
new file mode 100644
index 000000000000..53c385817eaa
--- /dev/null
+++ b/airbyte-commons/src/test/resources/json_schemas/json_with_all_types.json
@@ -0,0 +1,34 @@
+{
+  "title": "Jackson founder",
+  "type": "object",
+  "properties": {
+    "name": {
+      "type": "object",
+      "properties": {
+        "first": {
+          "type": "string"
+        },
+        "last": {
+          "type": "string"
+        }
+      }
+    },
+    "company": {
+      "type": "string"
+    },
+    "pets": {
+      "type": "array",
+      "items": {
+        "type": "object",
+        "properties": {
+          "type": {
+            "type": "string"
+          },
+          "number": {
+            "type": "string"
+          }
+        }
+      }
+    }
+  }
+}
diff --git a/airbyte-commons/src/test/resources/json_schemas/json_with_array_type_fields.json b/airbyte-commons/src/test/resources/json_schemas/json_with_array_type_fields.json
new file mode 100644
index 000000000000..56748ecc1375
--- /dev/null
+++ b/airbyte-commons/src/test/resources/json_schemas/json_with_array_type_fields.json
@@ -0,0 +1,18 @@
+{
+  "type": ["object", "array", "string"],
+  "properties": {
+    "company": {
+      "type": "string",
+      "description": "company name"
+    }
+  },
+  "items": {
+    "type": "object",
+    "properties": {
+      "user": {
+        "type": "string",
+        "description": "username"
+      }
+    }
+  }
+}
diff --git a/airbyte-commons/src/test/resources/json_schemas/json_with_array_type_fields_with_composites.json b/airbyte-commons/src/test/resources/json_schemas/json_with_array_type_fields_with_composites.json
new file mode 100644
index 000000000000..9fcab7b1006b
--- /dev/null
+++ b/airbyte-commons/src/test/resources/json_schemas/json_with_array_type_fields_with_composites.json
@@ -0,0 +1,32 @@
+{
+  "type": ["object", "array", "string"],
+  "anyOf": [
+    {
+      "type": "object",
+      "properties": {
+        "company": {
+          "type": "string",
+          "description": "company name"
+        }
+      }
+    },
+    {
+      "type": "object",
+      "properties": {
+        "organization": {
+          "type": "string",
+          "description": "organization name"
+        }
+      }
+    }
+  ],
+  "items": {
+    "type": "object",
+    "properties": {
+      "user": {
+        "type": "string",
+        "description": "username"
+      }
+    }
+  }
+}
diff --git a/airbyte-config/persistence/src/main/java/io/airbyte/config/persistence/split_secrets/JsonSecretsProcessor.java b/airbyte-config/persistence/src/main/java/io/airbyte/config/persistence/split_secrets/JsonSecretsProcessor.java
index 677a3051e8f7..e132fbb022c3 100644
--- a/airbyte-config/persistence/src/main/java/io/airbyte/config/persistence/split_secrets/JsonSecretsProcessor.java
+++ b/airbyte-config/persistence/src/main/java/io/airbyte/config/persistence/split_secrets/JsonSecretsProcessor.java
@@ -6,20 +6,18 @@
 
 import com.fasterxml.jackson.databind.JsonNode;
 import com.fasterxml.jackson.databind.node.ArrayNode;
-import com.fasterxml.jackson.databind.node.JsonNodeFactory;
 import com.fasterxml.jackson.databind.node.ObjectNode;
 import com.google.common.annotations.VisibleForTesting;
 import com.google.common.base.Preconditions;
+import io.airbyte.commons.json.JsonPaths;
+import io.airbyte.commons.json.JsonSchemas;
 import io.airbyte.commons.json.Jsons;
+import io.airbyte.commons.util.MoreIterators;
 import io.airbyte.validation.json.JsonSchemaValidator;
-import java.util.HashSet;
-import java.util.LinkedList;
 import java.util.List;
 import java.util.Optional;
-import java.util.Queue;
 import java.util.Set;
 import lombok.Builder;
-import lombok.Value;
 
 @Builder
 public class JsonSecretsProcessor {
@@ -53,18 +51,41 @@ public class JsonSecretsProcessor {
    */
   public JsonNode prepareSecretsForOutput(final JsonNode obj, final JsonNode schema) {
     if (maskSecrets) {
+      // todo (cgardens) this is not safe. should throw.
       // if schema is an object and has a properties field
-      if (!canBeProcessed(schema)) {
+      if (!isValidJsonSchema(schema)) {
         return obj;
       }
 
-      final SecretKeys secretKeys = getAllSecretKeys(schema);
-      return maskAllSecrets(obj, secretKeys);
+      return maskAllSecrets(obj, schema);
     }
 
     return obj;
   }
 
+  /**
+   * Given a JSONSchema object and an object that conforms to that schema, obfuscate all fields in the
+   * object that are a secret.
+   *
+   * @param json - json object that conforms to the schema
+   * @param schema - jsonschema object
+   * @return json object with all secrets masked.
+   */
+  public static JsonNode maskAllSecrets(final JsonNode json, final JsonNode schema) {
+    final Set<String> pathsWithSecrets = JsonSchemas.collectJsonPathsThatMeetCondition(
+        schema,
+        node -> MoreIterators.toList(node.fields())
+            .stream()
+            .anyMatch(field -> field.getKey().equals(AIRBYTE_SECRET_FIELD)));
+
+    JsonNode copy = Jsons.clone(json);
+    for (final String path : pathsWithSecrets) {
+      copy = JsonPaths.replaceAtString(copy, path, 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.
@@ -74,12 +95,14 @@ public JsonNode prepareSecretsForOutput(final JsonNode obj, final JsonNode schem
    *
    * @param src The object potentially containing secrets
    * @param dst The object to absorb secrets into
-   * @param schema
-   * @return
+   * @param schema Schema of objects
+   * @return dst object with secrets absorbed from src object
    */
+  // todo (cgardens) - figure out how to reused JsonSchemas and JsonPaths for this traversal as well.
   public JsonNode copySecrets(final JsonNode src, final JsonNode dst, final JsonNode schema) {
     if (copySecrets) {
-      if (!canBeProcessed(schema)) {
+      // todo (cgardens) this is not safe. should throw.
+      if (!isValidJsonSchema(schema)) {
         return dst;
       }
       Preconditions.checkArgument(dst.isObject());
@@ -141,78 +164,7 @@ static boolean isSecret(final JsonNode obj) {
     return obj.isObject() && obj.has(AIRBYTE_SECRET_FIELD) && obj.get(AIRBYTE_SECRET_FIELD).asBoolean();
   }
 
-  protected JsonNode maskAllSecrets(final JsonNode obj, final SecretKeys secretKeys) {
-    final JsonNode copiedObj = obj.deepCopy();
-    final Queue<JsonNode> toProcess = new LinkedList<>();
-    toProcess.add(copiedObj);
-
-    while (!toProcess.isEmpty()) {
-      final JsonNode currentNode = toProcess.remove();
-      for (final String key : Jsons.keys(currentNode)) {
-        if (secretKeys.fieldSecretKey.contains(key)) {
-          ((ObjectNode) currentNode).put(key, SECRETS_MASK);
-        } else if (currentNode.get(key).isObject()) {
-          toProcess.add(currentNode.get(key));
-        } else if (currentNode.get(key).isArray()) {
-          if (secretKeys.arraySecretKey.contains(key)) {
-            final ArrayNode sanitizedArrayNode = new ArrayNode(JsonNodeFactory.instance);
-            currentNode.get(key).forEach((secret) -> sanitizedArrayNode.add(SECRETS_MASK));
-            ((ObjectNode) currentNode).put(key, sanitizedArrayNode);
-          } else {
-            final ArrayNode arrayNode = (ArrayNode) currentNode.get(key);
-            arrayNode.forEach((node) -> {
-              toProcess.add(node);
-            });
-          }
-        }
-      }
-    }
-
-    return copiedObj;
-  }
-
-  @Value
-  protected class SecretKeys {
-
-    private final Set<String> fieldSecretKey;
-    private final Set<String> arraySecretKey;
-
-  }
-
-  protected SecretKeys getAllSecretKeys(final JsonNode schema) {
-    final Set<String> fieldSecretKeys = new HashSet<>();
-    final Set<String> arraySecretKeys = new HashSet<>();
-
-    final Queue<JsonNode> toProcess = new LinkedList<>();
-    toProcess.add(schema);
-
-    while (!toProcess.isEmpty()) {
-      final JsonNode currentNode = toProcess.remove();
-      for (final String key : Jsons.keys(currentNode)) {
-        if (isArrayDefinition(currentNode.get(key))) {
-          final JsonNode arrayItems = currentNode.get(key).get(ITEMS_FIELD);
-          if (arrayItems.has(AIRBYTE_SECRET_FIELD) && arrayItems.get(AIRBYTE_SECRET_FIELD).asBoolean()) {
-            arraySecretKeys.add(key);
-          } else {
-            toProcess.add(arrayItems);
-          }
-        } else if (JsonSecretsProcessor.isSecret(currentNode.get(key))) {
-          fieldSecretKeys.add(key);
-        } else if (currentNode.get(key).isObject()) {
-          toProcess.add(currentNode.get(key));
-        } else if (currentNode.get(key).isArray()) {
-          final ArrayNode arrayNode = (ArrayNode) currentNode.get(key);
-          arrayNode.forEach((node) -> {
-            toProcess.add(node);
-          });
-        }
-      }
-    }
-
-    return new SecretKeys(fieldSecretKeys, arraySecretKeys);
-  }
-
-  public static Optional<String> findJsonCombinationNode(final JsonNode node) {
+  private static Optional<String> findJsonCombinationNode(final JsonNode node) {
     for (final String combinationNode : List.of("allOf", "anyOf", "oneOf")) {
       if (node.has(combinationNode) && node.get(combinationNode).isArray()) {
         return Optional.of(combinationNode);
@@ -221,15 +173,9 @@ public static Optional<String> findJsonCombinationNode(final JsonNode node) {
     return Optional.empty();
   }
 
-  public static boolean canBeProcessed(final JsonNode schema) {
+  @SuppressWarnings("BooleanMethodIsAlwaysInverted")
+  private static boolean isValidJsonSchema(final JsonNode schema) {
     return schema.isObject() && schema.has(PROPERTIES_FIELD) && schema.get(PROPERTIES_FIELD).isObject();
   }
 
-  public static boolean isArrayDefinition(final JsonNode obj) {
-    return obj.isObject()
-        && obj.has(TYPE_FIELD)
-        && obj.get(TYPE_FIELD).asText().equals(ARRAY_TYPE_FIELD)
-        && obj.has(ITEMS_FIELD);
-  }
-
 }
diff --git a/airbyte-config/persistence/src/main/java/io/airbyte/config/persistence/split_secrets/SecretsHelpers.java b/airbyte-config/persistence/src/main/java/io/airbyte/config/persistence/split_secrets/SecretsHelpers.java
index 579ec254dd65..ea41388a6fef 100644
--- a/airbyte-config/persistence/src/main/java/io/airbyte/config/persistence/split_secrets/SecretsHelpers.java
+++ b/airbyte-config/persistence/src/main/java/io/airbyte/config/persistence/split_secrets/SecretsHelpers.java
@@ -9,15 +9,17 @@
 import com.fasterxml.jackson.databind.node.ObjectNode;
 import com.fasterxml.jackson.databind.node.TextNode;
 import com.google.common.annotations.VisibleForTesting;
+import io.airbyte.commons.json.JsonPaths;
+import io.airbyte.commons.json.JsonSchemas;
 import io.airbyte.commons.json.Jsons;
+import io.airbyte.commons.util.MoreIterators;
 import io.airbyte.protocol.models.ConnectorSpecification;
-import io.airbyte.validation.json.JsonSchemaValidator;
 import java.util.HashMap;
+import java.util.List;
 import java.util.Map;
 import java.util.Optional;
 import java.util.UUID;
 import java.util.function.Supplier;
-import java.util.stream.Collectors;
 import javax.annotation.Nullable;
 
 /**
@@ -163,6 +165,39 @@ public static SplitSecretConfig splitAndUpdateConfig(final Supplier<UUID> uuidSu
     return internalSplitAndUpdateConfig(uuidSupplier, workspaceId, secretReader, oldPartialConfig, newFullConfig, spec.getConnectionSpecification());
   }
 
+  /**
+   * This returns all the path to the airbyte secrets based on a schema spec. The path will be return
+   * in an ascending alphabetical order.
+   */
+  public static List<String> getSortedSecretPaths(final JsonNode spec) {
+    return JsonSchemas.collectJsonPathsThatMeetCondition(
+        spec,
+        node -> MoreIterators.toList(node.fields())
+            .stream()
+            .anyMatch(field -> field.getKey().equals(JsonSecretsProcessor.AIRBYTE_SECRET_FIELD)))
+        .stream()
+        .sorted()
+        .toList();
+  }
+
+  private static Optional<String> getExistingCoordinateIfExists(final JsonNode json) {
+    if (json != null && json.has(COORDINATE_FIELD)) {
+      return Optional.ofNullable(json.get(COORDINATE_FIELD).asText());
+    } else {
+      return Optional.empty();
+    }
+  }
+
+  private static SecretCoordinate getOrCreateCoordinate(final ReadOnlySecretPersistence secretReader,
+                                                        final UUID workspaceId,
+                                                        final Supplier<UUID> uuidSupplier,
+                                                        final JsonNode newJson,
+                                                        final JsonNode persistedJson) {
+
+    final Optional<String> existingCoordinateOptional = getExistingCoordinateIfExists(persistedJson);
+    return getCoordinate(newJson.asText(), secretReader, workspaceId, uuidSupplier, existingCoordinateOptional.orElse(null));
+  }
+
   /**
    * Internal method used to support both "split config" and "split and update config" operations.
    *
@@ -179,154 +214,41 @@ public static SplitSecretConfig splitAndUpdateConfig(final Supplier<UUID> uuidSu
    *        stored for
    * @param secretReader provides a way to determine if a secret is the same or updated at a specific
    *        location in a config
-   * @param oldPartialConfig previous partial config for this specific connector configuration
-   * @param originalFullConfig new config containing secrets that will be used to update the partial
-   *        config for this connector
+   * @param persistedPartialConfig previous partial config for this specific connector configuration
+   * @param newFullConfig new config containing secrets that will be used to update the partial config
+   *        for this connector
    * @param spec connector specification
    * @return a partial config + a map of coordinates to secret payloads
    */
-  private static SplitSecretConfig internalSplitAndUpdateConfig(final Supplier<UUID> uuidSupplier,
-                                                                final UUID workspaceId,
-                                                                final ReadOnlySecretPersistence secretReader,
-                                                                final JsonNode oldPartialConfig,
-                                                                final JsonNode originalFullConfig,
-                                                                final JsonNode spec) {
-    final var fullConfig = originalFullConfig.deepCopy();
+  @VisibleForTesting
+  static SplitSecretConfig internalSplitAndUpdateConfig(final Supplier<UUID> uuidSupplier,
+                                                        final UUID workspaceId,
+                                                        final ReadOnlySecretPersistence secretReader,
+                                                        final JsonNode persistedPartialConfig,
+                                                        final JsonNode newFullConfig,
+                                                        final JsonNode spec) {
+    var fullConfigCopy = newFullConfig.deepCopy();
     final var secretMap = new HashMap<SecretCoordinate, String>();
 
-    // provide a lambda for hiding repeated arguments to improve readability
-    final InternalSplitter splitter =
-        (final JsonNode partialConfig, final JsonNode newFullConfig, final JsonNode newFullConfigSpec) -> internalSplitAndUpdateConfig(uuidSupplier,
-            workspaceId,
-            secretReader, partialConfig, newFullConfig, newFullConfigSpec);
-
-    final var specTypeToHandle = getSpecTypeToHandle(spec);
-
-    switch (specTypeToHandle) {
-      case STRING -> {
-        if (JsonSecretsProcessor.isSecret(spec)) {
-          final var oldFullSecretCoordinate = oldPartialConfig.has(COORDINATE_FIELD) ? oldPartialConfig.get(COORDINATE_FIELD).asText() : null;
-          final var secretCoordinate = getCoordinate(fullConfig.asText(), secretReader, workspaceId, uuidSupplier, oldFullSecretCoordinate);
+    final List<String> paths = getSortedSecretPaths(spec);
 
-          final var newPartialConfig = Jsons.jsonNode(Map.of(
-              COORDINATE_FIELD, secretCoordinate.getFullCoordinate()));
-
-          final var coordinateToPayload = Map.of(
-              secretCoordinate,
-              fullConfig.asText());
-
-          return new SplitSecretConfig(newPartialConfig, coordinateToPayload);
-        }
-      }
-      case OBJECT -> {
-        final var specPropertiesObject = (ObjectNode) spec.get(JsonSecretsProcessor.PROPERTIES_FIELD);
-        final var specProperties = Jsons.keys(specPropertiesObject).stream()
-            .filter(fullConfig::has)
-            .collect(Collectors.toList());
-
-        // if the input config is specified as an object, we go through and handle each type of property
-        for (final String specProperty : specProperties) {
-          final var nextOldPartialConfig = getFieldOrEmptyNode(oldPartialConfig, specProperty);
-
-          final var nestedSplitConfig =
-              splitter.splitAndUpdateConfig(nextOldPartialConfig, fullConfig.get(specProperty), spec.get("properties").get(specProperty));
-          ((ObjectNode) fullConfig).replace(specProperty, nestedSplitConfig.getPartialConfig());
-          secretMap.putAll(nestedSplitConfig.getCoordinateToPayload());
-        }
-      }
-      case ARRAY -> {
-        for (int i = 0; i < fullConfig.size(); i++) {
-          final var partialConfigElement = getFieldOrEmptyNode(oldPartialConfig, i);
-          final var fullConfigElement = fullConfig.get(i);
-          final var splitConfig = splitter.splitAndUpdateConfig(partialConfigElement, fullConfigElement, spec.get("items"));
-          secretMap.putAll(splitConfig.getCoordinateToPayload());
-          ((ArrayNode) fullConfig).set(i, splitConfig.getPartialConfig());
-        }
-      }
-      case ONE_OF -> {
-        final var possibleSchemas = (ArrayNode) spec.get("oneOf");
-
-        for (int i = 0; i < possibleSchemas.size(); i++) {
-          final var possibleSchema = possibleSchemas.get(i);
-          final var set = new JsonSchemaValidator().validate(possibleSchema, fullConfig);
-          if (set.isEmpty()) {
-            final var splitConfig = splitter.splitAndUpdateConfig(oldPartialConfig, fullConfig, possibleSchema);
-            if (!splitConfig.getPartialConfig().equals(fullConfig)) {
-              return splitConfig;
-            }
-          }
-        }
-      }
-    }
-
-    return new SplitSecretConfig(fullConfig, secretMap);
-  }
+    for (final String path : paths) {
+      fullConfigCopy = JsonPaths.replaceAt(fullConfigCopy, path, (json, pathOfNode) -> {
+        final Optional<JsonNode> persistedNode = JsonPaths.getSingleValue(persistedPartialConfig, pathOfNode);
+        final SecretCoordinate coordinate = getOrCreateCoordinate(
+            secretReader,
+            workspaceId,
+            uuidSupplier,
+            json,
+            persistedNode.orElse(null));
 
-  /**
-   * Enum that allows us to switch on different types seen in a config / spec.
-   *
-   * Unrecognized types refers to values that cannot contain string airbyte_secret secrets such as
-   * numbers, binary fields, etc. They also may contain allOf or other mechanisms that aren't fully
-   * supported in Airbyte connector configurations.
-   */
-  private enum JsonSchemaSpecType {
-    OBJECT,
-    ARRAY,
-    STRING,
-    ONE_OF,
-    UNRECOGNIZED_TYPE
-  }
+        secretMap.put(coordinate, json.asText());
 
-  /**
-   * Determines what the spec is referring to.
-   *
-   * @param spec connector specification or a sub-node of the specification
-   * @return a type used to process a config or sub-node of a config
-   */
-  private static JsonSchemaSpecType getSpecTypeToHandle(final JsonNode spec) {
-    if (isObjectSchema(spec)) {
-      return JsonSchemaSpecType.OBJECT;
-    } else if (isArraySchema(spec)) {
-      return JsonSchemaSpecType.ARRAY;
-    } else if (isStringSchema(spec)) {
-      return JsonSchemaSpecType.STRING;
-    } else if (spec.has("oneOf") && spec.get("oneOf").isArray()) {
-      return JsonSchemaSpecType.ONE_OF;
-    } else {
-      return JsonSchemaSpecType.UNRECOGNIZED_TYPE;
+        return Jsons.jsonNode(Map.of(COORDINATE_FIELD, coordinate.getFullCoordinate()));
+      });
     }
-  }
-
-  /**
-   * Interface used to make calls to
-   * {@link SecretsHelpers#internalSplitAndUpdateConfig(Supplier, UUID, ReadOnlySecretPersistence, JsonNode, JsonNode, JsonNode)}
-   * more readable by arguments that are the same across the entire method.
-   */
-  @FunctionalInterface
-  public interface InternalSplitter {
-
-    SplitSecretConfig splitAndUpdateConfig(JsonNode oldPartialConfig, JsonNode fullConfig, JsonNode spec);
-
-  }
-
-  private static boolean isStringSchema(final JsonNode schema) {
-    return schema.has("type") && schema.get("type").asText().equals("string");
-  }
-
-  private static boolean isObjectSchema(final JsonNode schema) {
-    return schema.has("properties") && schema.get("properties").isObject();
-  }
-
-  private static boolean isArraySchema(final JsonNode schema) {
-    return schema.has("items") && schema.get("items").isObject();
-  }
-
-  private static JsonNode getFieldOrEmptyNode(final JsonNode node, final String field) {
-    return node.has(field) ? node.get(field) : Jsons.emptyObject();
-  }
 
-  private static JsonNode getFieldOrEmptyNode(final JsonNode node, final int field) {
-    return node.has(field) ? node.get(field) : Jsons.emptyObject();
+    return new SplitSecretConfig(fullConfigCopy, secretMap);
   }
 
   /**
diff --git a/airbyte-config/persistence/src/test/java/io/airbyte/config/persistence/split_secrets/JsonSecretsProcessorTest.java b/airbyte-config/persistence/src/test/java/io/airbyte/config/persistence/split_secrets/JsonSecretsProcessorTest.java
index 92b44d0ef8ee..8d2f340454e8 100644
--- a/airbyte-config/persistence/src/test/java/io/airbyte/config/persistence/split_secrets/JsonSecretsProcessorTest.java
+++ b/airbyte-config/persistence/src/test/java/io/airbyte/config/persistence/split_secrets/JsonSecretsProcessorTest.java
@@ -24,144 +24,172 @@
 public class JsonSecretsProcessorTest {
 
   private static final JsonNode SCHEMA_ONE_LAYER = Jsons.deserialize(
-      "{\n"
-          + "  \"properties\": {\n"
-          + "    \"secret1\": {\n"
-          + "      \"type\": \"string\",\n"
-          + "      \"airbyte_secret\": true\n"
-          + "    },\n"
-          + "    \"secret2\": {\n"
-          + "      \"type\": \"string\",\n"
-          + "      \"airbyte_secret\": \"true\"\n"
-          + "    },\n"
-          + "    \"field1\": {\n"
-          + "      \"type\": \"string\"\n"
-          + "    },\n"
-          + "    \"field2\": {\n"
-          + "      \"type\": \"number\"\n"
-          + "    }\n"
-          + "  }\n"
-          + "}\n");
+      """
+      {
+        "type": "object",  "properties": {
+          "secret1": {
+            "type": "string",
+            "airbyte_secret": true
+          },
+          "secret2": {
+            "type": "string",
+            "airbyte_secret": "true"
+          },
+          "field1": {
+            "type": "string"
+          },
+          "field2": {
+            "type": "number"
+          }
+        }
+      }
+      """);
+
+  private static final JsonNode SCHEMA_WITH_ARRAY = Jsons.deserialize(
+      """
+      {
+        "type": "object",  "properties": {
+          "secret1": {
+            "type": "array",      "items": {
+              "type": "string",
+              "airbyte_secret": true
+            }
+          },
+          "secret2": {
+            "type": "string",
+            "airbyte_secret": "true"
+          },
+          "field1": {
+            "type": "string"
+          },
+          "field2": {
+            "type": "number"
+          }
+        }
+      }
+      """);
 
   private static final JsonNode SCHEMA_INNER_OBJECT = Jsons.deserialize(
-      "{\n"
-          + "    \"type\": \"object\",\n"
-          + "    \"properties\": {\n"
-          + "      \"warehouse\": {\n"
-          + "        \"type\": \"string\"\n"
-          + "      },\n"
-          + "      \"loading_method\": {\n"
-          + "        \"type\": \"object\",\n"
-          + "        \"oneOf\": [\n"
-          + "          {\n"
-          + "            \"properties\": {}\n"
-          + "          },\n"
-          + "          {\n"
-          + "            \"properties\": {\n"
-          + "              \"s3_bucket_name\": {\n"
-          + "                \"type\": \"string\"\n"
-          + "              },\n"
-          + "              \"secret_access_key\": {\n"
-          + "                \"type\": \"string\",\n"
-          + "                \"airbyte_secret\": true\n"
-          + "              }\n"
-          + "            }\n"
-          + "          }\n"
-          + "        ]\n"
-          + "      }\n"
-          + "    }\n"
-          + "  }");
+      """
+      {
+          "type": "object",
+          "properties": {
+            "warehouse": {
+              "type": "string"
+            },
+            "loading_method": {
+              "type": "object",
+              "oneOf": [
+                {
+                  "properties": {}
+                },
+                {
+                  "properties": {
+                    "s3_bucket_name": {
+                      "type": "string"
+                    },
+                    "secret_access_key": {
+                      "type": "string",
+                      "airbyte_secret": true
+                    }
+                  }
+                }
+              ]
+            }
+          }
+        }""");
 
   private static final JsonNode ONE_OF_WITH_SAME_KEY_IN_SUB_SCHEMAS = Jsons.deserialize(
-      "{\n"
-          + "    \"$schema\": \"http://json-schema.org/draft-07/schema#\",\n"
-          + "    \"title\": \"S3 Destination Spec\",\n"
-          + "    \"type\": \"object\",\n"
-          + "    \"required\": [\n"
-          + "      \"client_id\",\n"
-          + "      \"format\"\n"
-          + "    ],\n"
-          + "    \"additionalProperties\": false,\n"
-          + "    \"properties\": {\n"
-          + "      \"client_id\": {\n"
-          + "        \"title\": \"client it\",\n"
-          + "        \"type\": \"string\",\n"
-          + "        \"default\": \"\"\n"
-          + "      },\n"
-          + "      \"format\": {\n"
-          + "        \"title\": \"Output Format\",\n"
-          + "        \"type\": \"object\",\n"
-          + "        \"description\": \"Output data format\",\n"
-          + "        \"oneOf\": [\n"
-          + "          {\n"
-          + "            \"title\": \"Avro: Apache Avro\",\n"
-          + "            \"required\": [\"format_type\", \"compression_codec\"],\n"
-          + "            \"properties\": {\n"
-          + "              \"format_type\": {\n"
-          + "                \"type\": \"string\",\n"
-          + "                \"enum\": [\"Avro\"],\n"
-          + "                \"default\": \"Avro\"\n"
-          + "              },\n"
-          + "              \"compression_codec\": {\n"
-          + "                \"title\": \"Compression Codec\",\n"
-          + "                \"description\": \"The compression algorithm used to compress data. Default to no compression.\",\n"
-          + "                \"type\": \"object\",\n"
-          + "                \"oneOf\": [\n"
-          + "                  {\n"
-          + "                    \"title\": \"no compression\",\n"
-          + "                    \"required\": [\"codec\"],\n"
-          + "                    \"properties\": {\n"
-          + "                      \"codec\": {\n"
-          + "                        \"type\": \"string\",\n"
-          + "                        \"enum\": [\"no compression\"],\n"
-          + "                        \"default\": \"no compression\"\n"
-          + "                      }\n"
-          + "                    }\n"
-          + "                  },\n"
-          + "                  {\n"
-          + "                    \"title\": \"Deflate\",\n"
-          + "                    \"required\": [\"codec\", \"compression_level\"],\n"
-          + "                    \"properties\": {\n"
-          + "                      \"codec\": {\n"
-          + "                        \"type\": \"string\",\n"
-          + "                        \"enum\": [\"Deflate\"],\n"
-          + "                        \"default\": \"Deflate\"\n"
-          + "                      },\n"
-          + "                      \"compression_level\": {\n"
-          + "                        \"type\": \"integer\",\n"
-          + "                        \"default\": 0,\n"
-          + "                        \"minimum\": 0,\n"
-          + "                        \"maximum\": 9\n"
-          + "                      }\n"
-          + "                    }\n"
-          + "                  }\n"
-          + "                ]\n"
-          + "              }\n"
-          + "            }\n"
-          + "          },\n"
-          + "          {\n"
-          + "            \"title\": \"Parquet: Columnar Storage\",\n"
-          + "            \"required\": [\"format_type\"],\n"
-          + "            \"properties\": {\n"
-          + "              \"format_type\": {\n"
-          + "                \"type\": \"string\",\n"
-          + "                \"enum\": [\"Parquet\"],\n"
-          + "                \"default\": \"Parquet\"\n"
-          + "              },\n"
-          + "              \"compression_codec\": {\n"
-          + "                \"type\": \"string\",\n"
-          + "                \"enum\": [\n"
-          + "                  \"UNCOMPRESSED\",\n"
-          + "                  \"GZIP\"\n"
-          + "                ],\n"
-          + "                \"default\": \"UNCOMPRESSED\"\n"
-          + "              }\n"
-          + "            }\n"
-          + "          }\n"
-          + "        ]\n"
-          + "      }\n"
-          + "    }\n"
-          + "  }");
+      """
+      {
+          "$schema": "http://json-schema.org/draft-07/schema#",
+          "title": "S3 Destination Spec",
+          "type": "object",
+          "required": [
+            "client_id",
+            "format"
+          ],
+          "additionalProperties": false,
+          "properties": {
+            "client_id": {
+              "title": "client it",
+              "type": "string",
+              "default": ""
+            },
+            "format": {
+              "title": "Output Format",
+              "type": "object",
+              "description": "Output data format",
+              "oneOf": [
+                {
+                  "title": "Avro: Apache Avro",
+                  "required": ["format_type", "compression_codec"],
+                  "properties": {
+                    "format_type": {
+                      "type": "string",
+                      "enum": ["Avro"],
+                      "default": "Avro"
+                    },
+                    "compression_codec": {
+                      "title": "Compression Codec",
+                      "description": "The compression algorithm used to compress data. Default to no compression.",
+                      "type": "object",
+                      "oneOf": [
+                        {
+                          "title": "no compression",
+                          "required": ["codec"],
+                          "properties": {
+                            "codec": {
+                              "type": "string",
+                              "enum": ["no compression"],
+                              "default": "no compression"
+                            }
+                          }
+                        },
+                        {
+                          "title": "Deflate",
+                          "required": ["codec", "compression_level"],
+                          "properties": {
+                            "codec": {
+                              "type": "string",
+                              "enum": ["Deflate"],
+                              "default": "Deflate"
+                            },
+                            "compression_level": {
+                              "type": "integer",
+                              "default": 0,
+                              "minimum": 0,
+                              "maximum": 9
+                            }
+                          }
+                        }
+                      ]
+                    }
+                  }
+                },
+                {
+                  "title": "Parquet: Columnar Storage",
+                  "required": ["format_type"],
+                  "properties": {
+                    "format_type": {
+                      "type": "string",
+                      "enum": ["Parquet"],
+                      "default": "Parquet"
+                    },
+                    "compression_codec": {
+                      "type": "string",
+                      "enum": [
+                        "UNCOMPRESSED",
+                        "GZIP"
+                      ],
+                      "default": "UNCOMPRESSED"
+                    }
+                  }
+                }
+              ]
+            }
+          }
+        }""");
 
   private JsonSecretsProcessor processor;
 
@@ -331,10 +359,38 @@ void testSecretScenario(final String folder, final boolean partial) throws IOExc
     final JsonNode expected = objectMapper.readTree(expectedIs);
 
     final JsonNode actual = processor.prepareSecretsForOutput(input, specs);
-
     assertEquals(expected, actual);
   }
 
+  // todo (cgardens) - example of a case that is not properly handled. we should explicitly call out
+  // that this type of jsonschema object is not allowed to do secrets.
+  // private static Stream<Arguments> scenarioProvider2() {
+  // return Stream.of(
+  // Arguments.of("array2", true),
+  // Arguments.of("array2", false));
+  // }
+  //
+  // @ParameterizedTest
+  // @MethodSource("scenarioProvider2")
+  // void testSecretScenario2(final String folder, final boolean partial) throws IOException {
+  // final ObjectMapper objectMapper = new ObjectMapper();
+  //
+  // final InputStream specIs = getClass().getClassLoader().getResourceAsStream(folder +
+  // "/spec.json");
+  // final JsonNode specs = objectMapper.readTree(specIs);
+  //
+  // final String inputFilePath = folder + (partial ? "/partial_config.json" : "/full_config.json");
+  // final InputStream inputIs = getClass().getClassLoader().getResourceAsStream(inputFilePath);
+  // final JsonNode input = objectMapper.readTree(inputIs);
+  //
+  // final String expectedFilePath = folder + "/expected.json";
+  // final InputStream expectedIs = getClass().getClassLoader().getResourceAsStream(expectedFilePath);
+  // final JsonNode expected = objectMapper.readTree(expectedIs);
+  //
+  // final JsonNode actual = Secrets.maskAllSecrets(input, specs);
+  // assertEquals(expected, actual);
+  // }
+
   @Test
   public void copiesSecrets_inNestedNonCombinationNode() throws JsonProcessingException {
     final ObjectMapper objectMapper = new ObjectMapper();
diff --git a/airbyte-config/persistence/src/test/java/io/airbyte/config/persistence/split_secrets/SecretsHelpersTest.java b/airbyte-config/persistence/src/test/java/io/airbyte/config/persistence/split_secrets/SecretsHelpersTest.java
index a1a8fe14735b..87b92efe4643 100644
--- a/airbyte-config/persistence/src/test/java/io/airbyte/config/persistence/split_secrets/SecretsHelpersTest.java
+++ b/airbyte-config/persistence/src/test/java/io/airbyte/config/persistence/split_secrets/SecretsHelpersTest.java
@@ -4,8 +4,12 @@
 
 package io.airbyte.config.persistence.split_secrets;
 
-import static org.junit.jupiter.api.Assertions.*;
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertThrows;
+import static org.junit.jupiter.api.Assertions.assertTrue;
 
+import com.fasterxml.jackson.databind.JsonNode;
 import io.airbyte.config.persistence.split_secrets.test_cases.ArrayOneOfTestCase;
 import io.airbyte.config.persistence.split_secrets.test_cases.ArrayTestCase;
 import io.airbyte.config.persistence.split_secrets.test_cases.NestedObjectTestCase;
@@ -16,11 +20,13 @@
 import io.airbyte.config.persistence.split_secrets.test_cases.SimpleTestCase;
 import io.airbyte.validation.json.JsonSchemaValidator;
 import io.airbyte.validation.json.JsonValidationException;
+import java.io.IOException;
 import java.util.List;
 import java.util.Map;
 import java.util.UUID;
 import java.util.regex.Pattern;
 import java.util.stream.Stream;
+import org.assertj.core.api.Assertions;
 import org.junit.jupiter.api.Test;
 import org.junit.jupiter.params.ParameterizedTest;
 import org.junit.jupiter.params.provider.Arguments;
@@ -204,4 +210,14 @@ void testUpdatingSecretsOneAtATime() {
     assertEquals(testCase.getSecretMapAfterUpdate2(), updatedSplit2.getCoordinateToPayload());
   }
 
+  @ParameterizedTest
+  @MethodSource("provideTestCases")
+  void testSecretPath(final SecretsTestCase testCase) throws IOException {
+    final JsonNode spec = testCase.getSpec().getConnectionSpecification();
+
+    final List<String> secretsPaths = SecretsHelpers.getSortedSecretPaths(spec);
+
+    Assertions.assertThat(secretsPaths).containsExactlyElementsOf(testCase.getExpectedSecretsPaths());
+  }
+
 }
diff --git a/airbyte-config/persistence/src/test/java/io/airbyte/config/persistence/split_secrets/SecretsTestCase.java b/airbyte-config/persistence/src/test/java/io/airbyte/config/persistence/split_secrets/SecretsTestCase.java
index 3244a3c2d032..33004273376a 100644
--- a/airbyte-config/persistence/src/test/java/io/airbyte/config/persistence/split_secrets/SecretsTestCase.java
+++ b/airbyte-config/persistence/src/test/java/io/airbyte/config/persistence/split_secrets/SecretsTestCase.java
@@ -10,6 +10,8 @@
 import io.airbyte.commons.resources.MoreResources;
 import io.airbyte.protocol.models.ConnectorSpecification;
 import java.io.IOException;
+import java.util.Arrays;
+import java.util.List;
 import java.util.Map;
 import java.util.function.Consumer;
 
@@ -39,6 +41,10 @@ default JsonNode getPartialConfig() {
     return Exceptions.toRuntime(() -> getNodeResource(getName(), "partial_config.json"));
   }
 
+  default JsonNode getSortedPartialConfig() {
+    return Exceptions.toRuntime(() -> getNodeResource(getName(), "partial_config.json"));
+  }
+
   default JsonNode getUpdateConfig() {
     return Exceptions.toRuntime(() -> getNodeResource(getName(), "update_config.json"));
   }
@@ -51,4 +57,13 @@ default JsonNode getNodeResource(final String testCase, final String fileName) t
     return Jsons.deserialize(MoreResources.readResource(testCase + "/" + fileName));
   }
 
+  default List<String> getExpectedSecretsPaths() throws IOException {
+    return Arrays.stream(
+        MoreResources.readResource(getName() + "/" + "expectedPaths")
+            .trim()
+            .split(";"))
+        .sorted()
+        .toList();
+  }
+
 }
diff --git a/airbyte-config/persistence/src/test/java/io/airbyte/config/persistence/split_secrets/test_cases/ArrayTestCase.java b/airbyte-config/persistence/src/test/java/io/airbyte/config/persistence/split_secrets/test_cases/ArrayTestCase.java
index 0c27191985fc..af4d87ac5813 100644
--- a/airbyte-config/persistence/src/test/java/io/airbyte/config/persistence/split_secrets/test_cases/ArrayTestCase.java
+++ b/airbyte-config/persistence/src/test/java/io/airbyte/config/persistence/split_secrets/test_cases/ArrayTestCase.java
@@ -32,12 +32,12 @@ public Map<SecretCoordinate, String> getFirstSecretMap() {
   @Override
   public Map<SecretCoordinate, String> getSecondSecretMap() {
     return Map.of(
-        new SecretCoordinate("airbyte_workspace_" + SecretsHelpersTest.WORKSPACE_ID + "_secret_" + SecretsHelpersTest.UUIDS.get(0), 2), "key5",
-        new SecretCoordinate("airbyte_workspace_" + SecretsHelpersTest.WORKSPACE_ID + "_secret_" + SecretsHelpersTest.UUIDS.get(1), 2), "key6",
-        new SecretCoordinate("airbyte_workspace_" + SecretsHelpersTest.WORKSPACE_ID + "_secret_" + SecretsHelpersTest.UUIDS.get(2), 2), "key7",
-        new SecretCoordinate("airbyte_workspace_" + SecretsHelpersTest.WORKSPACE_ID + "_secret_" + SecretsHelpersTest.UUIDS.get(3), 2), "key8",
-        new SecretCoordinate("airbyte_workspace_" + SecretsHelpersTest.WORKSPACE_ID + "_secret_" + SecretsHelpersTest.UUIDS.get(4), 2), "key9",
-        new SecretCoordinate("airbyte_workspace_" + SecretsHelpersTest.WORKSPACE_ID + "_secret_" + SecretsHelpersTest.UUIDS.get(5), 2), "key10");
+        new SecretCoordinate("airbyte_workspace_" + SecretsHelpersTest.WORKSPACE_ID + "_secret_" + SecretsHelpersTest.UUIDS.get(0), 2), "key8",
+        new SecretCoordinate("airbyte_workspace_" + SecretsHelpersTest.WORKSPACE_ID + "_secret_" + SecretsHelpersTest.UUIDS.get(1), 2), "key9",
+        new SecretCoordinate("airbyte_workspace_" + SecretsHelpersTest.WORKSPACE_ID + "_secret_" + SecretsHelpersTest.UUIDS.get(2), 2), "key10",
+        new SecretCoordinate("airbyte_workspace_" + SecretsHelpersTest.WORKSPACE_ID + "_secret_" + SecretsHelpersTest.UUIDS.get(3), 2), "key5",
+        new SecretCoordinate("airbyte_workspace_" + SecretsHelpersTest.WORKSPACE_ID + "_secret_" + SecretsHelpersTest.UUIDS.get(4), 2), "key6",
+        new SecretCoordinate("airbyte_workspace_" + SecretsHelpersTest.WORKSPACE_ID + "_secret_" + SecretsHelpersTest.UUIDS.get(5), 2), "key7");
   }
 
   @Override
diff --git a/airbyte-config/persistence/src/test/java/io/airbyte/config/persistence/split_secrets/test_cases/NestedObjectTestCase.java b/airbyte-config/persistence/src/test/java/io/airbyte/config/persistence/split_secrets/test_cases/NestedObjectTestCase.java
index 9cc4782790cd..f111ea4a271b 100644
--- a/airbyte-config/persistence/src/test/java/io/airbyte/config/persistence/split_secrets/test_cases/NestedObjectTestCase.java
+++ b/airbyte-config/persistence/src/test/java/io/airbyte/config/persistence/split_secrets/test_cases/NestedObjectTestCase.java
@@ -23,25 +23,25 @@ public String getName() {
   @Override
   public Map<SecretCoordinate, String> getFirstSecretMap() {
     return Map.of(
-        new SecretCoordinate("airbyte_workspace_" + SecretsHelpersTest.WORKSPACE_ID + "_secret_" + SecretsHelpersTest.UUIDS.get(0), 1), "hunter1",
-        new SecretCoordinate("airbyte_workspace_" + SecretsHelpersTest.WORKSPACE_ID + "_secret_" + SecretsHelpersTest.UUIDS.get(1), 1), "hunter2");
+        new SecretCoordinate("airbyte_workspace_" + SecretsHelpersTest.WORKSPACE_ID + "_secret_" + SecretsHelpersTest.UUIDS.get(1), 1), "hunter1",
+        new SecretCoordinate("airbyte_workspace_" + SecretsHelpersTest.WORKSPACE_ID + "_secret_" + SecretsHelpersTest.UUIDS.get(0), 1), "hunter2");
   }
 
   @Override
   public Map<SecretCoordinate, String> getSecondSecretMap() {
     return Map.of(
-        new SecretCoordinate("airbyte_workspace_" + SecretsHelpersTest.WORKSPACE_ID + "_secret_" + SecretsHelpersTest.UUIDS.get(0), 2), "hunter3",
-        new SecretCoordinate("airbyte_workspace_" + SecretsHelpersTest.WORKSPACE_ID + "_secret_" + SecretsHelpersTest.UUIDS.get(1), 2), "hunter4");
+        new SecretCoordinate("airbyte_workspace_" + SecretsHelpersTest.WORKSPACE_ID + "_secret_" + SecretsHelpersTest.UUIDS.get(1), 2), "hunter3",
+        new SecretCoordinate("airbyte_workspace_" + SecretsHelpersTest.WORKSPACE_ID + "_secret_" + SecretsHelpersTest.UUIDS.get(0), 2), "hunter4");
   }
 
   @Override
   public Consumer<SecretPersistence> getPersistenceUpdater() {
     return secretPersistence -> {
       secretPersistence.write(
-          new SecretCoordinate("airbyte_workspace_" + SecretsHelpersTest.WORKSPACE_ID + "_secret_" + SecretsHelpersTest.UUIDS.get(0), 1),
+          new SecretCoordinate("airbyte_workspace_" + SecretsHelpersTest.WORKSPACE_ID + "_secret_" + SecretsHelpersTest.UUIDS.get(1), 1),
           "hunter1");
       secretPersistence.write(
-          new SecretCoordinate("airbyte_workspace_" + SecretsHelpersTest.WORKSPACE_ID + "_secret_" + SecretsHelpersTest.UUIDS.get(1), 1),
+          new SecretCoordinate("airbyte_workspace_" + SecretsHelpersTest.WORKSPACE_ID + "_secret_" + SecretsHelpersTest.UUIDS.get(0), 1),
           "hunter2");
     };
   }
@@ -67,14 +67,14 @@ public JsonNode getFullConfigUpdate2() {
 
   public Map<SecretCoordinate, String> getSecretMapAfterUpdate1() {
     return Map.of(
-        new SecretCoordinate("airbyte_workspace_" + SecretsHelpersTest.WORKSPACE_ID + "_secret_" + SecretsHelpersTest.UUIDS.get(0), 2), "hunter3",
-        new SecretCoordinate("airbyte_workspace_" + SecretsHelpersTest.WORKSPACE_ID + "_secret_" + SecretsHelpersTest.UUIDS.get(1), 1), "hunter2");
+        new SecretCoordinate("airbyte_workspace_" + SecretsHelpersTest.WORKSPACE_ID + "_secret_" + SecretsHelpersTest.UUIDS.get(1), 2), "hunter3",
+        new SecretCoordinate("airbyte_workspace_" + SecretsHelpersTest.WORKSPACE_ID + "_secret_" + SecretsHelpersTest.UUIDS.get(0), 1), "hunter2");
   }
 
   public Map<SecretCoordinate, String> getSecretMapAfterUpdate2() {
     return Map.of(
-        new SecretCoordinate("airbyte_workspace_" + SecretsHelpersTest.WORKSPACE_ID + "_secret_" + SecretsHelpersTest.UUIDS.get(0), 2), "hunter3",
-        new SecretCoordinate("airbyte_workspace_" + SecretsHelpersTest.WORKSPACE_ID + "_secret_" + SecretsHelpersTest.UUIDS.get(1), 2), "hunter4");
+        new SecretCoordinate("airbyte_workspace_" + SecretsHelpersTest.WORKSPACE_ID + "_secret_" + SecretsHelpersTest.UUIDS.get(1), 2), "hunter3",
+        new SecretCoordinate("airbyte_workspace_" + SecretsHelpersTest.WORKSPACE_ID + "_secret_" + SecretsHelpersTest.UUIDS.get(0), 2), "hunter4");
   }
 
 }
diff --git a/airbyte-config/persistence/src/test/resources/array/expectedPaths b/airbyte-config/persistence/src/test/resources/array/expectedPaths
new file mode 100644
index 000000000000..61c3501c4f3f
--- /dev/null
+++ b/airbyte-config/persistence/src/test/resources/array/expectedPaths
@@ -0,0 +1 @@
+$.rotating_key_strings[*];$.rotating_key_objects[*].a
diff --git a/airbyte-config/persistence/src/test/resources/array/partial_config.json b/airbyte-config/persistence/src/test/resources/array/partial_config.json
index 516b030fa2a3..e7a6dd7909bb 100644
--- a/airbyte-config/persistence/src/test/resources/array/partial_config.json
+++ b/airbyte-config/persistence/src/test/resources/array/partial_config.json
@@ -2,29 +2,29 @@
   "username": "charles",
   "rotating_key_strings": [
     {
-      "_secret": "airbyte_workspace_e0eb0554-ffe0-4e9c-9dc0-ed7f52023eb2_secret_9eba44d8-51e7-48f1-bde2-619af0e42c22_v1"
+      "_secret": "airbyte_workspace_e0eb0554-ffe0-4e9c-9dc0-ed7f52023eb2_secret_c03ef566-79a7-4e77-b6f3-d23d2528f25a_v1"
     },
     {
-      "_secret": "airbyte_workspace_e0eb0554-ffe0-4e9c-9dc0-ed7f52023eb2_secret_2c2ef2b3-259a-4e73-96d1-f56dacee2e5e_v1"
+      "_secret": "airbyte_workspace_e0eb0554-ffe0-4e9c-9dc0-ed7f52023eb2_secret_35f08b15-bfd9-44fe-a8c7-5aa9e156c0f5_v1"
     },
     {
-      "_secret": "airbyte_workspace_e0eb0554-ffe0-4e9c-9dc0-ed7f52023eb2_secret_1206db5b-b968-4df1-9a76-f3fcdae7e307_v1"
+      "_secret": "airbyte_workspace_e0eb0554-ffe0-4e9c-9dc0-ed7f52023eb2_secret_159c0b6f-f9ae-48b4-b7f3-bcac4ba15743_v1"
     }
   ],
   "rotating_key_objects": [
     {
       "a": {
-        "_secret": "airbyte_workspace_e0eb0554-ffe0-4e9c-9dc0-ed7f52023eb2_secret_c03ef566-79a7-4e77-b6f3-d23d2528f25a_v1"
+        "_secret": "airbyte_workspace_e0eb0554-ffe0-4e9c-9dc0-ed7f52023eb2_secret_9eba44d8-51e7-48f1-bde2-619af0e42c22_v1"
       }
     },
     {
       "a": {
-        "_secret": "airbyte_workspace_e0eb0554-ffe0-4e9c-9dc0-ed7f52023eb2_secret_35f08b15-bfd9-44fe-a8c7-5aa9e156c0f5_v1"
+        "_secret": "airbyte_workspace_e0eb0554-ffe0-4e9c-9dc0-ed7f52023eb2_secret_2c2ef2b3-259a-4e73-96d1-f56dacee2e5e_v1"
       }
     },
     {
       "a": {
-        "_secret": "airbyte_workspace_e0eb0554-ffe0-4e9c-9dc0-ed7f52023eb2_secret_159c0b6f-f9ae-48b4-b7f3-bcac4ba15743_v1"
+        "_secret": "airbyte_workspace_e0eb0554-ffe0-4e9c-9dc0-ed7f52023eb2_secret_1206db5b-b968-4df1-9a76-f3fcdae7e307_v1"
       }
     }
   ]
diff --git a/airbyte-config/persistence/src/test/resources/array/updated_partial_config.json b/airbyte-config/persistence/src/test/resources/array/updated_partial_config.json
index b1ff8557d9ef..672a7de1bd2c 100644
--- a/airbyte-config/persistence/src/test/resources/array/updated_partial_config.json
+++ b/airbyte-config/persistence/src/test/resources/array/updated_partial_config.json
@@ -2,29 +2,29 @@
   "username": "charles",
   "rotating_key_strings": [
     {
-      "_secret": "airbyte_workspace_e0eb0554-ffe0-4e9c-9dc0-ed7f52023eb2_secret_9eba44d8-51e7-48f1-bde2-619af0e42c22_v2"
+      "_secret": "airbyte_workspace_e0eb0554-ffe0-4e9c-9dc0-ed7f52023eb2_secret_c03ef566-79a7-4e77-b6f3-d23d2528f25a_v2"
     },
     {
-      "_secret": "airbyte_workspace_e0eb0554-ffe0-4e9c-9dc0-ed7f52023eb2_secret_2c2ef2b3-259a-4e73-96d1-f56dacee2e5e_v2"
+      "_secret": "airbyte_workspace_e0eb0554-ffe0-4e9c-9dc0-ed7f52023eb2_secret_35f08b15-bfd9-44fe-a8c7-5aa9e156c0f5_v2"
     },
     {
-      "_secret": "airbyte_workspace_e0eb0554-ffe0-4e9c-9dc0-ed7f52023eb2_secret_1206db5b-b968-4df1-9a76-f3fcdae7e307_v2"
+      "_secret": "airbyte_workspace_e0eb0554-ffe0-4e9c-9dc0-ed7f52023eb2_secret_159c0b6f-f9ae-48b4-b7f3-bcac4ba15743_v2"
     }
   ],
   "rotating_key_objects": [
     {
       "a": {
-        "_secret": "airbyte_workspace_e0eb0554-ffe0-4e9c-9dc0-ed7f52023eb2_secret_c03ef566-79a7-4e77-b6f3-d23d2528f25a_v2"
+        "_secret": "airbyte_workspace_e0eb0554-ffe0-4e9c-9dc0-ed7f52023eb2_secret_9eba44d8-51e7-48f1-bde2-619af0e42c22_v2"
       }
     },
     {
       "a": {
-        "_secret": "airbyte_workspace_e0eb0554-ffe0-4e9c-9dc0-ed7f52023eb2_secret_35f08b15-bfd9-44fe-a8c7-5aa9e156c0f5_v2"
+        "_secret": "airbyte_workspace_e0eb0554-ffe0-4e9c-9dc0-ed7f52023eb2_secret_2c2ef2b3-259a-4e73-96d1-f56dacee2e5e_v2"
       }
     },
     {
       "a": {
-        "_secret": "airbyte_workspace_e0eb0554-ffe0-4e9c-9dc0-ed7f52023eb2_secret_159c0b6f-f9ae-48b4-b7f3-bcac4ba15743_v2"
+        "_secret": "airbyte_workspace_e0eb0554-ffe0-4e9c-9dc0-ed7f52023eb2_secret_1206db5b-b968-4df1-9a76-f3fcdae7e307_v2"
       }
     }
   ]
diff --git a/airbyte-config/persistence/src/test/resources/array2/expected.json b/airbyte-config/persistence/src/test/resources/array2/expected.json
new file mode 100644
index 000000000000..abaa3e233eef
--- /dev/null
+++ b/airbyte-config/persistence/src/test/resources/array2/expected.json
@@ -0,0 +1,3 @@
+{
+  "cred": ["**********"]
+}
diff --git a/airbyte-config/persistence/src/test/resources/array2/full_config.json b/airbyte-config/persistence/src/test/resources/array2/full_config.json
new file mode 100644
index 000000000000..7edbb656fb36
--- /dev/null
+++ b/airbyte-config/persistence/src/test/resources/array2/full_config.json
@@ -0,0 +1,3 @@
+{
+  "cred": ["hunter2"]
+}
diff --git a/airbyte-config/persistence/src/test/resources/array2/partial_config.json b/airbyte-config/persistence/src/test/resources/array2/partial_config.json
new file mode 100644
index 000000000000..7ba03d02f4e5
--- /dev/null
+++ b/airbyte-config/persistence/src/test/resources/array2/partial_config.json
@@ -0,0 +1,7 @@
+{
+  "cred": [
+    {
+      "_secret": "airbyte_workspace_e0eb0554-ffe0-4e9c-9dc0-ed7f52023eb2_secret_159c0b6f-f9ae-48b4-b7f3-bcac4ba15743_v1"
+    }
+  ]
+}
diff --git a/airbyte-config/persistence/src/test/resources/array2/spec.json b/airbyte-config/persistence/src/test/resources/array2/spec.json
new file mode 100644
index 000000000000..05c3bbea3900
--- /dev/null
+++ b/airbyte-config/persistence/src/test/resources/array2/spec.json
@@ -0,0 +1,20 @@
+{
+  "type": "object",
+  "properties": {
+    "cred": {
+      "oneOf": [
+        {
+          "type": "string",
+          "airbyte_secret": true
+        },
+        {
+          "type": "array",
+          "items": {
+            "type": "string",
+            "airbyte_secret": true
+          }
+        }
+      ]
+    }
+  }
+}
diff --git a/airbyte-config/persistence/src/test/resources/array2/update_config.json b/airbyte-config/persistence/src/test/resources/array2/update_config.json
new file mode 100644
index 000000000000..e8270b1653cf
--- /dev/null
+++ b/airbyte-config/persistence/src/test/resources/array2/update_config.json
@@ -0,0 +1,3 @@
+{
+  "cred": ["hunter20"]
+}
diff --git a/airbyte-config/persistence/src/test/resources/array2/updated_partial_config.json b/airbyte-config/persistence/src/test/resources/array2/updated_partial_config.json
new file mode 100644
index 000000000000..aab04a199c11
--- /dev/null
+++ b/airbyte-config/persistence/src/test/resources/array2/updated_partial_config.json
@@ -0,0 +1,7 @@
+{
+  "cred": [
+    {
+      "_secret": "airbyte_workspace_e0eb0554-ffe0-4e9c-9dc0-ed7f52023eb2_secret_159c0b6f-f9ae-48b4-b7f3-bcac4ba15743_v2"
+    }
+  ]
+}
diff --git a/airbyte-config/persistence/src/test/resources/array_of_oneof/expectedPaths b/airbyte-config/persistence/src/test/resources/array_of_oneof/expectedPaths
new file mode 100644
index 000000000000..29b1c466d2d6
--- /dev/null
+++ b/airbyte-config/persistence/src/test/resources/array_of_oneof/expectedPaths
@@ -0,0 +1 @@
+$.rotating_keys[*].key1;$.rotating_keys[*].key2
diff --git a/airbyte-config/persistence/src/test/resources/nested_object/expectedPaths b/airbyte-config/persistence/src/test/resources/nested_object/expectedPaths
new file mode 100644
index 000000000000..4160e1476229
--- /dev/null
+++ b/airbyte-config/persistence/src/test/resources/nested_object/expectedPaths
@@ -0,0 +1 @@
+$.password;$.nested.password
diff --git a/airbyte-config/persistence/src/test/resources/nested_object/partial_config.json b/airbyte-config/persistence/src/test/resources/nested_object/partial_config.json
index d056a87ba1c3..e9e44d7ce52c 100644
--- a/airbyte-config/persistence/src/test/resources/nested_object/partial_config.json
+++ b/airbyte-config/persistence/src/test/resources/nested_object/partial_config.json
@@ -1,11 +1,11 @@
 {
   "username": "charles",
   "password": {
-    "_secret": "airbyte_workspace_e0eb0554-ffe0-4e9c-9dc0-ed7f52023eb2_secret_9eba44d8-51e7-48f1-bde2-619af0e42c22_v1"
+    "_secret": "airbyte_workspace_e0eb0554-ffe0-4e9c-9dc0-ed7f52023eb2_secret_2c2ef2b3-259a-4e73-96d1-f56dacee2e5e_v1"
   },
   "nested": {
     "password": {
-      "_secret": "airbyte_workspace_e0eb0554-ffe0-4e9c-9dc0-ed7f52023eb2_secret_2c2ef2b3-259a-4e73-96d1-f56dacee2e5e_v1"
+      "_secret": "airbyte_workspace_e0eb0554-ffe0-4e9c-9dc0-ed7f52023eb2_secret_9eba44d8-51e7-48f1-bde2-619af0e42c22_v1"
     },
     "not_password": 13
   }
diff --git a/airbyte-config/persistence/src/test/resources/nested_object/updated_partial_config.json b/airbyte-config/persistence/src/test/resources/nested_object/updated_partial_config.json
index 0b9ba7d8eb41..24c73c6d182c 100644
--- a/airbyte-config/persistence/src/test/resources/nested_object/updated_partial_config.json
+++ b/airbyte-config/persistence/src/test/resources/nested_object/updated_partial_config.json
@@ -1,11 +1,11 @@
 {
   "username": "charles",
   "password": {
-    "_secret": "airbyte_workspace_e0eb0554-ffe0-4e9c-9dc0-ed7f52023eb2_secret_9eba44d8-51e7-48f1-bde2-619af0e42c22_v2"
+    "_secret": "airbyte_workspace_e0eb0554-ffe0-4e9c-9dc0-ed7f52023eb2_secret_2c2ef2b3-259a-4e73-96d1-f56dacee2e5e_v2"
   },
   "nested": {
     "password": {
-      "_secret": "airbyte_workspace_e0eb0554-ffe0-4e9c-9dc0-ed7f52023eb2_secret_2c2ef2b3-259a-4e73-96d1-f56dacee2e5e_v2"
+      "_secret": "airbyte_workspace_e0eb0554-ffe0-4e9c-9dc0-ed7f52023eb2_secret_9eba44d8-51e7-48f1-bde2-619af0e42c22_v2"
     },
     "not_password": 13
   }
diff --git a/airbyte-config/persistence/src/test/resources/nested_object/updated_partial_config_update1.json b/airbyte-config/persistence/src/test/resources/nested_object/updated_partial_config_update1.json
index 4d76d121174b..02accccc92ee 100644
--- a/airbyte-config/persistence/src/test/resources/nested_object/updated_partial_config_update1.json
+++ b/airbyte-config/persistence/src/test/resources/nested_object/updated_partial_config_update1.json
@@ -1,11 +1,11 @@
 {
   "username": "charles",
   "password": {
-    "_secret": "airbyte_workspace_e0eb0554-ffe0-4e9c-9dc0-ed7f52023eb2_secret_9eba44d8-51e7-48f1-bde2-619af0e42c22_v2"
+    "_secret": "airbyte_workspace_e0eb0554-ffe0-4e9c-9dc0-ed7f52023eb2_secret_2c2ef2b3-259a-4e73-96d1-f56dacee2e5e_v2"
   },
   "nested": {
     "password": {
-      "_secret": "airbyte_workspace_e0eb0554-ffe0-4e9c-9dc0-ed7f52023eb2_secret_2c2ef2b3-259a-4e73-96d1-f56dacee2e5e_v1"
+      "_secret": "airbyte_workspace_e0eb0554-ffe0-4e9c-9dc0-ed7f52023eb2_secret_9eba44d8-51e7-48f1-bde2-619af0e42c22_v1"
     },
     "not_password": 13
   }
diff --git a/airbyte-config/persistence/src/test/resources/nested_object/updated_partial_config_update2.json b/airbyte-config/persistence/src/test/resources/nested_object/updated_partial_config_update2.json
index 0b9ba7d8eb41..24c73c6d182c 100644
--- a/airbyte-config/persistence/src/test/resources/nested_object/updated_partial_config_update2.json
+++ b/airbyte-config/persistence/src/test/resources/nested_object/updated_partial_config_update2.json
@@ -1,11 +1,11 @@
 {
   "username": "charles",
   "password": {
-    "_secret": "airbyte_workspace_e0eb0554-ffe0-4e9c-9dc0-ed7f52023eb2_secret_9eba44d8-51e7-48f1-bde2-619af0e42c22_v2"
+    "_secret": "airbyte_workspace_e0eb0554-ffe0-4e9c-9dc0-ed7f52023eb2_secret_2c2ef2b3-259a-4e73-96d1-f56dacee2e5e_v2"
   },
   "nested": {
     "password": {
-      "_secret": "airbyte_workspace_e0eb0554-ffe0-4e9c-9dc0-ed7f52023eb2_secret_2c2ef2b3-259a-4e73-96d1-f56dacee2e5e_v2"
+      "_secret": "airbyte_workspace_e0eb0554-ffe0-4e9c-9dc0-ed7f52023eb2_secret_9eba44d8-51e7-48f1-bde2-619af0e42c22_v2"
     },
     "not_password": 13
   }
diff --git a/airbyte-config/persistence/src/test/resources/nested_oneof/expectedPaths b/airbyte-config/persistence/src/test/resources/nested_oneof/expectedPaths
new file mode 100644
index 000000000000..75e6804e2014
--- /dev/null
+++ b/airbyte-config/persistence/src/test/resources/nested_oneof/expectedPaths
@@ -0,0 +1 @@
+$.password1.key1;$.password1.key2.key3;$.password1.key2.key4;$.password2.key5;$.password2.key6.key7;$.password2.key6.key8
diff --git a/airbyte-config/persistence/src/test/resources/oneof/expectedPaths b/airbyte-config/persistence/src/test/resources/oneof/expectedPaths
new file mode 100644
index 000000000000..7fdcd68edaea
--- /dev/null
+++ b/airbyte-config/persistence/src/test/resources/oneof/expectedPaths
@@ -0,0 +1 @@
+$.password1.key1;$.password2.key1;$.password3.key1
diff --git a/airbyte-config/persistence/src/test/resources/optional_password/expectedPaths b/airbyte-config/persistence/src/test/resources/optional_password/expectedPaths
new file mode 100644
index 000000000000..deb2239ded28
--- /dev/null
+++ b/airbyte-config/persistence/src/test/resources/optional_password/expectedPaths
@@ -0,0 +1 @@
+$.password
diff --git a/airbyte-config/persistence/src/test/resources/postgres_ssh_key/expectedPaths b/airbyte-config/persistence/src/test/resources/postgres_ssh_key/expectedPaths
new file mode 100644
index 000000000000..d60e732b5110
--- /dev/null
+++ b/airbyte-config/persistence/src/test/resources/postgres_ssh_key/expectedPaths
@@ -0,0 +1 @@
+$.password;$.tunnel_method.ssh_key;$.tunnel_method.tunnel_user_password
diff --git a/airbyte-config/persistence/src/test/resources/simple/expectedPaths b/airbyte-config/persistence/src/test/resources/simple/expectedPaths
new file mode 100644
index 000000000000..deb2239ded28
--- /dev/null
+++ b/airbyte-config/persistence/src/test/resources/simple/expectedPaths
@@ -0,0 +1 @@
+$.password